diff --git a/author_issue_bootstrap.py b/author_issue_bootstrap.py index 4f8d771..66d2a48 100644 --- a/author_issue_bootstrap.py +++ b/author_issue_bootstrap.py @@ -196,6 +196,58 @@ def _verify_assignment_and_lease_ids( # Some lease rows may not yet have an assignment join; still require # the lease itself to exist and bind to the claimed session/issue. pass + # #943 review 622 B2: a lease that is no longer live confers no ownership. + # Existence alone previously satisfied this gate, so a released or expired + # lease could still authorize a bootstrap for a claim its session had given + # up. Checked before the session comparison so the reason names the real + # problem rather than reporting a mismatch. + from datetime import datetime, timezone + + lease_status = str(lease.get("status") or "").strip().lower() + if lease_status and lease_status != "active": + return { + "success": False, + "reason_code": "lease_not_live", + "message": ( + f"lease_id '{lid}' is '{lease_status}', not active; a lease that " + "is not live confers no ownership (fail closed)." + ), + "exact_next_action": ( + "Re-allocate the work item and pass the live assignment/lease pair." + ), + } + expires_raw = str(lease.get("expires_at") or "").strip() + if expires_raw: + try: + expires_at = datetime.fromisoformat(expires_raw.replace("Z", "+00:00")) + except ValueError: + return { + "success": False, + "reason_code": "lease_not_live", + "message": ( + f"lease_id '{lid}' records an unparseable expiry " + f"'{expires_raw}' (fail closed)." + ), + "exact_next_action": ( + "Re-allocate the work item and pass the live " + "assignment/lease pair." + ), + } + if expires_at.tzinfo is None: + expires_at = expires_at.replace(tzinfo=timezone.utc) + if expires_at <= datetime.now(timezone.utc): + return { + "success": False, + "reason_code": "lease_not_live", + "message": ( + f"lease_id '{lid}' expired at {expires_raw}; an expired lease " + "confers no ownership (fail closed)." + ), + "exact_next_action": ( + "Reclaim or re-allocate the lease, then retry with the live pair." + ), + } + lease_session = str(lease.get("session_id") or "").strip() if lease_session and lease_session != owner_session: return { diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 67dd53d..3721039 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -3665,6 +3665,293 @@ def _authenticated_username(host: str): return user +# #943 review 622 F3/F4: one coherent authority snapshot per mutation claim. +# +# The reviewed implementation read the identity from the pinned #714 session +# context and the profile from the live ``get_profile()``, so a sanctioned +# rebind could produce a claimant pair whose two halves came from different +# snapshots — and that pair is written durably into the issue lock. The +# canonical pairing is ``record_mutation_authority``: profile from +# ``get_profile()``, identity from ``_authenticated_username(host)``. This +# resolver reproduces that pairing, and uses the pinned session context only for +# drift detection, never as a value source. +_AUTHORITY_PROFILE_UNRESOLVED = "authority_profile_unresolved" +_AUTHORITY_IDENTITY_UNRESOLVED = "authority_identity_unresolved" +_AUTHORITY_IDENTITY_DRIFT = "authority_identity_drift" +_AUTHORITY_PROFILE_DRIFT = "authority_profile_drift" + + +def _active_mutation_authority(host: str | None) -> dict: + """Resolve one coherent (identity, profile) authority pair, or a refusal. + + Both halves come from the same live snapshot. The immutable #714 session + context is consulted only to detect drift: when it disagrees with the live + snapshot the result is a fail-closed refusal, never a silently blended pair. + + Returns a dict with ``ok`` True plus ``identity``/``profile_name``, or ``ok`` + False plus ``reason_code``, ``reasons`` and ``expected``/``actual`` when a + drift or resolution failure is detected. Never raises for an unresolved + profile: the caller converts the refusal into a structured author block so + reason code, retryability and transport survival are preserved (F4). + """ + try: + profile = get_profile() or {} + except (RuntimeError, ValueError, TypeError, KeyError, OSError) as exc: + # Narrow, and never a silent fallback to a previously cached name: an + # unresolvable profile is a fail-closed condition, matching + # record_mutation_authority's "active profile unresolved (fail closed)". + return { + "ok": False, + "reason_code": _AUTHORITY_PROFILE_UNRESOLVED, + "reasons": [ + "active profile could not be resolved: " + f"{_redact(str(exc))} (fail closed)" + ], + } + profile_name = (profile.get("profile_name") or "").strip() + if not profile_name: + return { + "ok": False, + "reason_code": _AUTHORITY_PROFILE_UNRESOLVED, + "reasons": [ + "active profile carries no profile_name (fail closed)" + ], + } + + identity = "" + if host: + identity = (_authenticated_username(host) or "").strip() + if not identity: + # A profile's expected_username is configuration, not proof of an + # authenticated actor, so it is never substituted here. + return { + "ok": False, + "reason_code": _AUTHORITY_IDENTITY_UNRESOLVED, + "reasons": [ + "authenticated identity could not be resolved for the active " + "host (fail closed); call gitea_whoami and retry" + ], + } + + ctx = session_ctx.get_session_context() or {} + pinned_identity = (ctx.get("identity") or "").strip() + pinned_profile = (ctx.get("profile_name") or "").strip() + if pinned_identity and pinned_identity != identity: + return { + "ok": False, + "reason_code": _AUTHORITY_IDENTITY_DRIFT, + "reasons": [ + "live authenticated identity disagrees with the bound session " + "context identity (fail closed)" + ], + "expected": pinned_identity, + "actual": identity, + } + if pinned_profile and pinned_profile != profile_name: + return { + "ok": False, + "reason_code": _AUTHORITY_PROFILE_DRIFT, + "reasons": [ + "live active profile disagrees with the bound session context " + "profile (fail closed)" + ], + "expected": pinned_profile, + "actual": profile_name, + } + return { + "ok": True, + "identity": identity, + "profile_name": profile_name, + "session_context_bound": bool(pinned_identity or pinned_profile), + } + + +def _active_username(host: str | None = None) -> str | None: + """Authenticated identity for the active mutation authority, else None. + + Refuses unbound, blank and whitespace-only identities, and never substitutes + a profile's ``expected_username`` for an authenticated one. + """ + authority = _active_mutation_authority(host) + return authority.get("identity") if authority.get("ok") else None + + +def _active_profile_name(host: str | None = None) -> str | None: + """Active profile name for the mutation authority, else None. + + Shares the single snapshot with :func:`_active_username`, so the two can + never describe different authority states. + """ + authority = _active_mutation_authority(host) + return authority.get("profile_name") if authority.get("ok") else None + + +# #943 review 622 B1: the workflow session that owns the work — never a +# process-lifetime or PID-derived value. +# +# The reviewed implementation minted "--" once per process. +# The MCP daemon outlives every task it serves, so that identifier conflates +# sequential author tasks and can never equal the control-plane session that +# owns an allocator-created lease; ``_verify_assignment_and_lease_ids`` therefore +# refused with lease_session_mismatch on the canonical allocated path. +# ``issue_lock_store.mint_task_session_id`` states the rule directly: an +# ownership key "deliberately contains no process identifier", because the PID +# "cannot identify *which* task holds a claim". +_SESSION_UNVERIFIED = "workflow_session_unverified" +_SESSION_REQUIRED = "workflow_session_required_for_allocated_work" +_SESSION_LOCK_OWNER_MISMATCH = "issue_lock_owner_mismatch" + + +def _resolve_owner_workflow_session( + *, + issue_number: int, + assignment_id: str | None, + lease_id: str | None, + session_id: str | None, + identity: str, + profile_name: str, + remote: str, + org: str | None, + repo: str | None, +) -> dict: + """Resolve the authoritative workflow session that owns this work. + + Precedence, each step fail-closed: + + 1. An explicit ``session_id`` is verified against the control-plane + ``sessions`` table: it must exist, be active, and match this role and + profile. An unverifiable identifier is refused, never trusted. + 2. Otherwise an existing issue lock for this issue supplies its per-task + ``task_session_id``, but only when the lock's recorded claimant matches + the resolved authority pair. + 3. Otherwise, when allocator identifiers are supplied, the request is + refused: ownership of an allocated assignment cannot be established + without its owning session, and the presence of the identifiers is not + itself evidence of ownership. + 4. Otherwise — no allocator identifiers and no existing lock — a fresh + per-task key is minted through the canonical + ``issue_lock_store.mint_task_session_id``. It carries no process + identifier and is never memoised, so sequential tasks on one daemon never + share an owner. + + Whether the resolved session actually owns a supplied lease stays the + decision of ``author_issue_bootstrap._verify_assignment_and_lease_ids``; + this function never pre-empts, duplicates or bypasses that gate. + """ + declared = (session_id or "").strip() + if declared: + db, errs = _control_plane_db_or_error() + if db is None: + return { + "ok": False, + "reason_code": _SESSION_UNVERIFIED, + "reasons": errs, + } + try: + rows = db.list_sessions(statuses=("active",)) + except Exception as exc: # noqa: BLE001 — surface structured + return { + "ok": False, + "reason_code": _SESSION_UNVERIFIED, + "reasons": [ + "could not read control-plane sessions to verify " + f"session_id: {_redact(str(exc))} (fail closed)" + ], + } + match = next( + (r for r in rows if str(r.get("session_id") or "") == declared), None + ) + if match is None: + return { + "ok": False, + "reason_code": _SESSION_UNVERIFIED, + "reasons": [ + f"session_id '{declared}' is not an active control-plane " + "session (fail closed)" + ], + } + row_role = (match.get("role") or "").strip().lower() + row_profile = (match.get("profile") or "").strip() + if row_role and row_role != "author": + return { + "ok": False, + "reason_code": _SESSION_UNVERIFIED, + "reasons": [ + f"session_id '{declared}' is recorded for role " + f"'{row_role}', not author (fail closed)" + ], + "expected": "author", + "actual": row_role, + } + if row_profile and row_profile != profile_name: + return { + "ok": False, + "reason_code": _SESSION_UNVERIFIED, + "reasons": [ + f"session_id '{declared}' is recorded for profile " + f"'{row_profile}', not '{profile_name}' (fail closed)" + ], + "expected": row_profile, + "actual": profile_name, + } + return {"ok": True, "session_id": declared, "session_source": "declared"} + + lock = None + try: + lock = issue_lock_store.load_issue_lock( + remote=remote, + org=org or "", + repo=repo or "", + issue_number=int(issue_number), + ) + except Exception: # noqa: BLE001 — absent/unreadable lock is not fatal here + lock = None + lock_session = issue_lock_store.lease_task_session_id(lock) if lock else "" + if lock_session: + lease = (lock or {}).get("work_lease") or {} + claimant = lease.get("claimant") or {} + lock_identity = (claimant.get("username") or "").strip() + lock_profile = (claimant.get("profile") or "").strip() + if (lock_identity and lock_identity != identity) or ( + lock_profile and lock_profile != profile_name + ): + return { + "ok": False, + "reason_code": _SESSION_LOCK_OWNER_MISMATCH, + "reasons": [ + f"issue #{int(issue_number)} is locked by " + f"'{lock_identity or 'unknown'}' ({lock_profile or 'unknown'}), " + f"not '{identity}' ({profile_name}) (fail closed)" + ], + "expected": f"{lock_identity}/{lock_profile}", + "actual": f"{identity}/{profile_name}", + } + return { + "ok": True, + "session_id": lock_session, + "session_source": "issue_lock", + } + + if (assignment_id or "").strip() or (lease_id or "").strip(): + return { + "ok": False, + "reason_code": _SESSION_REQUIRED, + "reasons": [ + "assignment_id/lease_id were supplied but no owning workflow " + "session could be established (fail closed); pass the " + "session_id that holds the lease, or bind the issue lock first" + ], + } + + return { + "ok": True, + "session_id": issue_lock_store.mint_task_session_id( + issue_lock_store.AUTHOR_ISSUE_WORK_LEASE + ), + "session_source": "minted_task_key", + } + + def _authenticated_actor(host: str) -> dict: """Resolve the authenticated actor's stable identity (#709 F7 review 438). @@ -9988,6 +10275,24 @@ def gitea_commit_files( } +def _author_mutation_block(reasons: list[str], **extra) -> dict: + """Uniform fail-closed shape for an author mutation refused after a reviewer stop. + + #943: referenced by ``gitea_bootstrap_author_issue_worktree`` and never + defined, so the reviewer-stop refusal path raised ``NameError`` instead of + returning its refusal. Mirrors the inline shape the other author mutations + return for the same ``check_author_mutation_after_reviewer_stop`` block. + """ + payload = { + "success": False, + "performed": False, + "outcome": "REFUSED", + "reasons": reasons, + } + payload.update(extra) + return payload + + def _publication_block(reasons: list[str], **extra) -> dict: """Uniform fail-closed shape for publication refusals (#812 AC20).""" payload = { @@ -10244,6 +10549,7 @@ def gitea_bootstrap_author_issue_worktree( branch_name: str | None = None, worktree_path: str | None = None, idempotency_key: str | None = None, + session_id: str | None = None, remote: str = "dadeschools", host: str | None = None, org: str | None = None, @@ -10265,6 +10571,14 @@ def gitea_bootstrap_author_issue_worktree( branch_name: Optional custom branch name (must match issue- pattern). worktree_path: Optional custom worktree path under branches/. idempotency_key: Optional key for idempotent replay/resume. + session_id: Optional workflow session that owns the assignment/lease. + Required when assignment_id/lease_id are supplied, because ownership + of an allocated lease is compared by session identifier and cannot + be derived from the daemon process (#943 review 622 B1). Verified + against the control-plane sessions table; an unverifiable value is + refused rather than trusted. Omit for an unallocated bootstrap: the + owning session is then taken from an existing issue lock, or a fresh + per-task key is minted. remote: Known instance — 'dadeschools' or 'prgs'. host: Override Gitea host. org: Override Org. @@ -10303,6 +10617,44 @@ def gitea_bootstrap_author_issue_worktree( h, o, r = _resolve(remote, host, org, repo) canonical_root = _canonical_local_git_root() + # One authority snapshot supplies both halves of the claimant pair (F3), and + # an unresolvable profile becomes a structured refusal rather than a silent + # fallback (F4). + authority = _active_mutation_authority(h) + if not authority.get("ok"): + return _author_mutation_block( + authority.get("reasons") or ["mutation authority unresolved"], + reason_code=authority.get("reason_code"), + retryable=False, + transport_survives=True, + expected=authority.get("expected"), + actual=authority.get("actual"), + issue_number=int(issue_number), + ) + + # The owning workflow session — never process- or PID-derived (B1). + session = _resolve_owner_workflow_session( + issue_number=issue_number, + assignment_id=assignment_id, + lease_id=lease_id, + session_id=session_id, + identity=authority["identity"], + profile_name=authority["profile_name"], + remote=remote, + org=o, + repo=r, + ) + if not session.get("ok"): + return _author_mutation_block( + session.get("reasons") or ["owning workflow session unresolved"], + reason_code=session.get("reason_code"), + retryable=False, + transport_survives=True, + expected=session.get("expected"), + actual=session.get("actual"), + issue_number=int(issue_number), + ) + import author_issue_bootstrap return author_issue_bootstrap.bootstrap_author_issue_worktree( @@ -10318,9 +10670,9 @@ def gitea_bootstrap_author_issue_worktree( host=h, org=o, repo=r, - active_identity=_active_username(), - active_profile=_active_profile_name(), - owner_session=_current_session_id(), + active_identity=authority["identity"], + active_profile=authority["profile_name"], + owner_session=session["session_id"], dry_run=dry_run, ) diff --git a/tests/test_issue_943_runtime_context_helpers.py b/tests/test_issue_943_runtime_context_helpers.py new file mode 100644 index 0000000..8bbe5af --- /dev/null +++ b/tests/test_issue_943_runtime_context_helpers.py @@ -0,0 +1,747 @@ +"""Regression: author bootstrap runtime authority and session ownership (#943). + +Two rounds of defects live here. + +**Round 1 (#943 as filed).** ``gitea_bootstrap_author_issue_worktree`` passed +four values down to the bootstrap service that were never defined: +``_active_username``, ``_active_profile_name``, ``_current_session_id`` and +``_author_mutation_block``. Every call — dry-run included — raised +``NameError`` while evaluating the arguments, before the service was entered. + +**Round 2 (review 622 on PR #944).** The first fix defined all four but made +``_current_session_id`` mint ``--`` once per process. The MCP +daemon outlives every task it serves, so that value conflates sequential author +tasks and can never equal the control-plane session that owns an +allocator-created lease: ``_verify_assignment_and_lease_ids`` refused the whole +allocated path with ``lease_session_mismatch``. The reviewed round also read the +identity from the pinned session context while reading the profile from the live +profile, so a rebind could produce a mixed claimant pair, and it swallowed every +``get_profile()`` exception. + +These tests therefore drive real state, not mocks of internals: a temporary +control-plane SQLite database and a temporary issue-lock directory, both +redirected through the same environment variables production uses +(``GITEA_CONTROL_PLANE_DB``, ``GITEA_ISSUE_LOCK_DIR``). The ownership gate that +runs is the real one. + +``test_every_global_referenced_by_the_wrapper_resolves`` remains: it is what +found ``_author_mutation_block``, and it generalises to the next missing +reference. It supplements the runtime coverage below rather than standing in for +it. +""" + +from __future__ import annotations + +import ast +import builtins +import os +import re +import subprocess +import tempfile +import unittest +from unittest import mock + +import author_issue_bootstrap as aib +import control_plane_db +import create_issue_bootstrap as cib +import gitea_mcp_server as gms +import issue_lock_store +import workflow_scope_guard + +BOOTSTRAP_TASK = "bootstrap_author_issue_worktree" +WRAPPER_NAME = "gitea_bootstrap_author_issue_worktree" +RUNTIME_HELPERS = ( + "_active_mutation_authority", + "_active_username", + "_active_profile_name", + "_resolve_owner_workflow_session", + "_author_mutation_block", +) +ORG = "Scaled-Tech-Consulting" +REPO = "Gitea-Tools" +IDENTITY = "jcwalker3" +PROFILE = "prgs-author" + +# A per-task ownership key must carry no process identifier (#790). +TASK_KEY_RE = re.compile(r"^author_issue_work-[0-9a-f]{16}$") + + +def _make_control_repo(tmp: str) -> tuple[str, str]: + """Create a clean control checkout on master and return (path, head).""" + repo = os.path.join(tmp, "repo") + os.makedirs(os.path.join(repo, "branches")) + subprocess.check_call( + ["git", "init", "-b", "master", repo], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + subprocess.check_call( + [ + "git", "-C", repo, + "-c", "user.email=t@t", "-c", "user.name=t", + "commit", "--allow-empty", "-m", "init", + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + head = subprocess.check_output( + ["git", "-C", repo, "rev-parse", "HEAD"], text=True + ).strip() + return repo, head + + +def _wrapper_ast() -> ast.FunctionDef: + """Return the AST of the bootstrap wrapper as it exists on disk.""" + path = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "gitea_mcp_server.py", + ) + with open(path, encoding="utf-8") as fh: + tree = ast.parse(fh.read()) + for node in ast.walk(tree): + if isinstance(node, ast.FunctionDef) and node.name == WRAPPER_NAME: + return node + raise AssertionError(f"{WRAPPER_NAME} not found in gitea_mcp_server.py") + + +class _IsolatedControlPlane(unittest.TestCase): + """Temp control-plane DB and temp issue-lock dir, via production env vars.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.tmp = self._tmp.name + self.db_path = os.path.join(self.tmp, "control-plane.sqlite3") + self.lock_dir = os.path.join(self.tmp, "issue-locks") + self.journals = os.path.join(self.tmp, "journals") + os.makedirs(self.lock_dir) + os.makedirs(self.journals) + env = mock.patch.dict( + os.environ, + { + control_plane_db.DB_PATH_ENV: self.db_path, + issue_lock_store.LOCK_DIR_ENV: self.lock_dir, + }, + ) + env.start() + self.addCleanup(env.stop) + self.db = control_plane_db.ControlPlaneDB(self.db_path) + + def _allocate(self, session_id: str, *, issue: int = 943): + """Create a real assignment + lease owned by *session_id*.""" + self.db.upsert_session( + session_id=session_id, role="author", profile=PROFILE, pid=os.getpid() + ) + res = self.db.assign_and_lease( + session_id=session_id, role="author", remote="prgs", + org=ORG, repo=REPO, kind="issue", number=issue, + ) + self.assertEqual(res.outcome, "assigned", res) + return res.assignment_id, res.lease_id + + def _authority(self): + """A resolved authority pair, as the wrapper would compute it.""" + return {"ok": True, "identity": IDENTITY, "profile_name": PROFILE} + + def _resolve_session(self, **over): + kwargs = dict( + issue_number=943, + assignment_id=None, + lease_id=None, + session_id=None, + identity=IDENTITY, + profile_name=PROFILE, + remote="prgs", + org=ORG, + repo=REPO, + ) + kwargs.update(over) + return gms._resolve_owner_workflow_session(**kwargs) + + +class OwnershipGateTests(_IsolatedControlPlane): + """B2: the allocator-driven ownership path, against a real control plane.""" + + def setUp(self): + super().setUp() + self.repo, self.head = _make_control_repo(self.tmp) + + def _bootstrap(self, **over): + kwargs = dict( + issue_number=943, + canonical_repo_root=self.repo, + expected_base_sha=self.head, + branch_name="fix/issue-943-runtime-context-helpers", + remote="prgs", + org=ORG, + repo=REPO, + active_identity=IDENTITY, + active_profile=PROFILE, + lock_dir=self.journals, + idempotency_key="test-943", + dry_run=True, + ) + kwargs.update(over) + return aib.bootstrap_author_issue_worktree(**kwargs) + + def test_true_owning_session_passes_the_ownership_gate(self): + """The canonical owner reaches and completes the service.""" + session = "prgs-author-task-a" + assignment_id, lease_id = self._allocate(session) + res = self._bootstrap( + assignment_id=assignment_id, lease_id=lease_id, owner_session=session + ) + self.assertTrue(res.get("success"), res) + self.assertTrue(res.get("dry_run")) + self.assertEqual(res.get("base_sha"), self.head) + + def test_different_session_is_refused(self): + session = "prgs-author-task-a" + assignment_id, lease_id = self._allocate(session) + res = self._bootstrap( + assignment_id=assignment_id, + lease_id=lease_id, + owner_session="prgs-author-task-b", + ) + self.assertFalse(res.get("success")) + self.assertEqual(res.get("reason_code"), "lease_session_mismatch") + + def test_process_derived_session_would_be_refused(self): + """The reviewed round-1 value shape can never own an allocated lease.""" + session = "prgs-author-task-a" + assignment_id, lease_id = self._allocate(session) + round_one_value = f"{PROFILE}-{os.getpid()}-deadbeef" + self.assertNotEqual(round_one_value, session) + res = self._bootstrap( + assignment_id=assignment_id, + lease_id=lease_id, + owner_session=round_one_value, + ) + self.assertFalse(res.get("success")) + self.assertEqual(res.get("reason_code"), "lease_session_mismatch") + + def test_unknown_lease_fails_closed(self): + session = "prgs-author-task-a" + assignment_id, _ = self._allocate(session) + res = self._bootstrap( + assignment_id=assignment_id, + lease_id="lease-does-not-exist", + owner_session=session, + ) + self.assertFalse(res.get("success")) + self.assertEqual(res.get("reason_code"), "unknown_lease_id") + + def test_released_lease_fails_closed(self): + session = "prgs-author-task-a" + assignment_id, lease_id = self._allocate(session) + self.db.release_lease(lease_id, session_id=session) + res = self._bootstrap( + assignment_id=assignment_id, lease_id=lease_id, owner_session=session + ) + self.assertFalse(res.get("success")) + self.assertEqual(res.get("reason_code"), "lease_not_live") + + def test_force_expired_lease_fails_closed(self): + session = "prgs-author-task-a" + assignment_id, lease_id = self._allocate(session) + self.db.force_expire_lease(lease_id, reason="test") + res = self._bootstrap( + assignment_id=assignment_id, lease_id=lease_id, owner_session=session + ) + self.assertFalse(res.get("success")) + self.assertEqual(res.get("reason_code"), "lease_not_live") + + def test_replacement_lease_does_not_inherit_prior_ownership(self): + """A second task's lease is not ownable by the first task's session.""" + first = "prgs-author-task-a" + assignment_a, lease_a = self._allocate(first) + self.db.release_lease(lease_a, session_id=first) + second = "prgs-author-task-b" + assignment_b, lease_b = self._allocate(second) + self.assertNotEqual(lease_a, lease_b) + res = self._bootstrap( + assignment_id=assignment_b, lease_id=lease_b, owner_session=first + ) + self.assertFalse(res.get("success")) + self.assertEqual(res.get("reason_code"), "lease_session_mismatch") + + def test_assignment_lease_identifier_mismatch_fails_closed(self): + session = "prgs-author-task-a" + _, lease_id = self._allocate(session) + res = self._bootstrap( + assignment_id="asn-not-the-recorded-one", + lease_id=lease_id, + owner_session=session, + ) + self.assertFalse(res.get("success")) + self.assertEqual(res.get("reason_code"), "assignment_lease_mismatch") + + def test_lease_id_without_assignment_id_fails_closed(self): + session = "prgs-author-task-a" + _, lease_id = self._allocate(session) + res = self._bootstrap(lease_id=lease_id, owner_session=session) + self.assertFalse(res.get("success")) + self.assertEqual(res.get("reason_code"), "incomplete_assignment_lease_ids") + + def test_dry_run_with_valid_allocator_bindings_leaves_no_durable_state(self): + session = "prgs-author-task-a" + assignment_id, lease_id = self._allocate(session) + res = self._bootstrap( + assignment_id=assignment_id, lease_id=lease_id, owner_session=session + ) + self.assertTrue(res.get("success"), res) + + branches = subprocess.check_output( + ["git", "-C", self.repo, "branch", "--list"], text=True + ) + self.assertNotIn("issue-943", branches) + worktrees = subprocess.check_output( + ["git", "-C", self.repo, "worktree", "list"], text=True + ) + self.assertNotIn("issue-943", worktrees) + self.assertFalse( + os.path.exists( + os.path.join(self.repo, "branches", + "fix-issue-943-runtime-context-helpers") + ) + ) + journal = res.get("phase_journal") or {} + self.assertFalse(journal.get("completed")) + self.assertFalse(any((journal.get("artifacts_created") or {}).values())) + # The dry run must not have created an issue lock in the isolated dir. + self.assertEqual(os.listdir(self.lock_dir), []) + + def test_apply_reaches_the_intended_transition_with_valid_bindings(self): + session = "prgs-author-task-a" + assignment_id, lease_id = self._allocate(session) + res = self._bootstrap( + assignment_id=assignment_id, + lease_id=lease_id, + owner_session=session, + dry_run=False, + ) + self.assertTrue(res.get("success"), res) + self.assertNotEqual(res.get("dry_run"), True) + branches = subprocess.check_output( + ["git", "-C", self.repo, "branch", "--list"], text=True + ) + self.assertIn("issue-943", branches) + self.assertTrue(os.path.isdir(res.get("worktree_path") or "")) + + +class WorkflowSessionResolutionTests(_IsolatedControlPlane): + """B1: the wrapper resolves the owning session, never a process identifier.""" + + def test_declared_session_is_verified_against_the_control_plane(self): + session = "prgs-author-task-a" + assignment_id, lease_id = self._allocate(session) + res = self._resolve_session( + session_id=session, assignment_id=assignment_id, lease_id=lease_id + ) + self.assertTrue(res.get("ok"), res) + self.assertEqual(res.get("session_id"), session) + self.assertEqual(res.get("session_source"), "declared") + + def test_unknown_declared_session_is_refused_not_trusted(self): + res = self._resolve_session(session_id="prgs-author-not-a-session") + self.assertFalse(res.get("ok")) + self.assertEqual(res.get("reason_code"), "workflow_session_unverified") + + def test_declared_session_for_another_role_is_refused(self): + self.db.upsert_session( + session_id="prgs-reviewer-x", role="reviewer", profile="prgs-reviewer", + pid=os.getpid(), + ) + res = self._resolve_session(session_id="prgs-reviewer-x") + self.assertFalse(res.get("ok")) + self.assertEqual(res.get("reason_code"), "workflow_session_unverified") + + def test_declared_session_for_another_profile_is_refused(self): + self.db.upsert_session( + session_id="other-profile-session", role="author", + profile="prgs-controller", pid=os.getpid(), + ) + res = self._resolve_session(session_id="other-profile-session") + self.assertFalse(res.get("ok")) + self.assertEqual(res.get("reason_code"), "workflow_session_unverified") + + def test_allocated_work_without_a_session_is_refused(self): + """Supplying a lease is not itself evidence of ownership.""" + session = "prgs-author-task-a" + assignment_id, lease_id = self._allocate(session) + res = self._resolve_session(assignment_id=assignment_id, lease_id=lease_id) + self.assertFalse(res.get("ok")) + self.assertEqual( + res.get("reason_code"), "workflow_session_required_for_allocated_work" + ) + + def test_existing_issue_lock_supplies_its_per_task_session(self): + lock_session = issue_lock_store.mint_task_session_id( + issue_lock_store.AUTHOR_ISSUE_WORK_LEASE + ) + path = issue_lock_store.lock_file_path( + remote="prgs", org=ORG, repo=REPO, issue_number=943, + lock_dir=self.lock_dir, + ) + issue_lock_store.write_lock_file( + path, + { + "issue_number": 943, + "branch_name": "fix/issue-943-runtime-context-helpers", + "work_lease": { + "task_session_id": lock_session, + "claimant": {"username": IDENTITY, "profile": PROFILE}, + }, + }, + ) if hasattr(issue_lock_store, "write_lock_file") else _write_json( + path, + { + "issue_number": 943, + "branch_name": "fix/issue-943-runtime-context-helpers", + "work_lease": { + "task_session_id": lock_session, + "claimant": {"username": IDENTITY, "profile": PROFILE}, + }, + }, + ) + res = self._resolve_session() + self.assertTrue(res.get("ok"), res) + self.assertEqual(res.get("session_id"), lock_session) + self.assertEqual(res.get("session_source"), "issue_lock") + + def test_issue_lock_owned_by_another_identity_is_refused(self): + path = issue_lock_store.lock_file_path( + remote="prgs", org=ORG, repo=REPO, issue_number=943, + lock_dir=self.lock_dir, + ) + _write_json( + path, + { + "issue_number": 943, + "work_lease": { + "task_session_id": "author_issue_work-" + "0" * 16, + "claimant": {"username": "someone-else", "profile": PROFILE}, + }, + }, + ) + res = self._resolve_session() + self.assertFalse(res.get("ok")) + self.assertEqual(res.get("reason_code"), "issue_lock_owner_mismatch") + + def test_unallocated_bootstrap_mints_a_per_task_key(self): + res = self._resolve_session() + self.assertTrue(res.get("ok"), res) + self.assertEqual(res.get("session_source"), "minted_task_key") + self.assertRegex(res["session_id"], TASK_KEY_RE) + + def test_minted_key_contains_no_process_identifier(self): + res = self._resolve_session() + self.assertNotIn(str(os.getpid()), res["session_id"]) + self.assertNotIn(PROFILE, res["session_id"]) + + def test_sequential_tasks_on_one_daemon_do_not_share_ownership(self): + """The round-1 defect: one identifier per process for every task.""" + first = self._resolve_session()["session_id"] + second = self._resolve_session()["session_id"] + third = self._resolve_session()["session_id"] + self.assertNotEqual(first, second) + self.assertNotEqual(second, third) + self.assertEqual(len({first, second, third}), 3) + + def test_no_process_lifetime_cache_remains(self): + self.assertFalse(hasattr(gms, "_ACTIVE_SESSION_ID")) + self.assertFalse(hasattr(gms, "_current_session_id")) + + +class MutationAuthorityTests(unittest.TestCase): + """F3/F4: one coherent authority pair, drift detected, no silent fallback.""" + + def _ctx(self, **over): + base = {"identity": IDENTITY, "profile_name": PROFILE} + base.update(over) + return base + + def test_matching_live_and_pinned_authority_resolves(self): + with mock.patch.object(gms, "get_profile", + return_value={"profile_name": PROFILE}), \ + mock.patch.object(gms, "_authenticated_username", + return_value=IDENTITY), \ + mock.patch.object(gms.session_ctx, "get_session_context", + return_value=self._ctx()): + res = gms._active_mutation_authority("gitea.prgs.cc") + self.assertTrue(res.get("ok"), res) + self.assertEqual(res["identity"], IDENTITY) + self.assertEqual(res["profile_name"], PROFILE) + + def test_identity_drift_fails_closed(self): + with mock.patch.object(gms, "get_profile", + return_value={"profile_name": PROFILE}), \ + mock.patch.object(gms, "_authenticated_username", + return_value="someone-else"), \ + mock.patch.object(gms.session_ctx, "get_session_context", + return_value=self._ctx()): + res = gms._active_mutation_authority("gitea.prgs.cc") + self.assertFalse(res.get("ok")) + self.assertEqual(res.get("reason_code"), "authority_identity_drift") + self.assertEqual(res.get("expected"), IDENTITY) + self.assertEqual(res.get("actual"), "someone-else") + + def test_profile_drift_fails_closed(self): + with mock.patch.object(gms, "get_profile", + return_value={"profile_name": "prgs-controller"}), \ + mock.patch.object(gms, "_authenticated_username", + return_value=IDENTITY), \ + mock.patch.object(gms.session_ctx, "get_session_context", + return_value=self._ctx()): + res = gms._active_mutation_authority("gitea.prgs.cc") + self.assertFalse(res.get("ok")) + self.assertEqual(res.get("reason_code"), "authority_profile_drift") + + def test_identity_and_profile_never_come_from_different_snapshots(self): + """Round 2's mixed pair: pinned identity plus live profile.""" + with mock.patch.object(gms, "get_profile", + return_value={"profile_name": "prgs-controller"}), \ + mock.patch.object(gms, "_authenticated_username", + return_value="new-identity"), \ + mock.patch.object(gms.session_ctx, "get_session_context", + return_value=self._ctx()): + res = gms._active_mutation_authority("gitea.prgs.cc") + self.assertFalse(res.get("ok")) + self.assertIsNone(gms._active_username("gitea.prgs.cc")) + self.assertIsNone(gms._active_profile_name("gitea.prgs.cc")) + + def test_unresolvable_profile_is_a_structured_refusal_not_a_fallback(self): + """F4: no bare-except fallback to a previously pinned profile name.""" + with mock.patch.object(gms, "get_profile", + side_effect=RuntimeError("profile disabled")), \ + mock.patch.object(gms, "_authenticated_username", + return_value=IDENTITY), \ + mock.patch.object(gms.session_ctx, "get_session_context", + return_value=self._ctx()): + res = gms._active_mutation_authority("gitea.prgs.cc") + self.assertFalse(res.get("ok")) + self.assertEqual(res.get("reason_code"), "authority_profile_unresolved") + self.assertNotEqual(res.get("profile_name"), PROFILE) + + def test_malformed_profile_without_name_fails_closed(self): + with mock.patch.object(gms, "get_profile", return_value={}), \ + mock.patch.object(gms, "_authenticated_username", + return_value=IDENTITY), \ + mock.patch.object(gms.session_ctx, "get_session_context", + return_value=None): + res = gms._active_mutation_authority("gitea.prgs.cc") + self.assertFalse(res.get("ok")) + self.assertEqual(res.get("reason_code"), "authority_profile_unresolved") + + def test_unresolved_identity_fails_closed(self): + for value in (None, "", " "): + with self.subTest(identity=value): + with mock.patch.object(gms, "get_profile", + return_value={"profile_name": PROFILE}), \ + mock.patch.object(gms, "_authenticated_username", + return_value=value), \ + mock.patch.object(gms.session_ctx, "get_session_context", + return_value=None): + res = gms._active_mutation_authority("gitea.prgs.cc") + self.assertFalse(res.get("ok")) + self.assertEqual( + res.get("reason_code"), "authority_identity_unresolved" + ) + + def test_missing_host_cannot_yield_an_identity(self): + with mock.patch.object(gms, "get_profile", + return_value={"profile_name": PROFILE}), \ + mock.patch.object(gms.session_ctx, "get_session_context", + return_value=None): + res = gms._active_mutation_authority(None) + self.assertFalse(res.get("ok")) + self.assertEqual(res.get("reason_code"), "authority_identity_unresolved") + + def test_expected_username_is_never_substituted_for_authentication(self): + with mock.patch.object( + gms, "get_profile", + return_value={"profile_name": PROFILE, "username": IDENTITY}, + ), mock.patch.object(gms, "_authenticated_username", return_value=None), \ + mock.patch.object(gms.session_ctx, "get_session_context", + return_value={"expected_username": IDENTITY}): + self.assertIsNone(gms._active_username("gitea.prgs.cc")) + + def test_accessors_share_one_snapshot(self): + with mock.patch.object(gms, "get_profile", + return_value={"profile_name": PROFILE}), \ + mock.patch.object(gms, "_authenticated_username", + return_value=IDENTITY), \ + mock.patch.object(gms.session_ctx, "get_session_context", + return_value=self._ctx()): + self.assertEqual(gms._active_username("gitea.prgs.cc"), IDENTITY) + self.assertEqual(gms._active_profile_name("gitea.prgs.cc"), PROFILE) + + +class AuthorMutationBlockTests(unittest.TestCase): + """Preserved: the structured refusal shape review 622 confirmed correct.""" + + def test_matches_the_sibling_refusal_shape(self): + res = gms._author_mutation_block(["stopped"]) + self.assertIs(res["success"], False) + self.assertIs(res["performed"], False) + self.assertEqual(res["outcome"], "REFUSED") + self.assertEqual(res["reasons"], ["stopped"]) + + def test_carries_reason_code_and_transport_fields(self): + res = gms._author_mutation_block( + ["nope"], reason_code="authority_identity_drift", + retryable=False, transport_survives=True, + expected="a", actual="b", issue_number=943, + ) + self.assertEqual(res["reason_code"], "authority_identity_drift") + self.assertIs(res["retryable"], False) + self.assertIs(res["transport_survives"], True) + self.assertEqual((res["expected"], res["actual"]), ("a", "b")) + self.assertEqual(res["issue_number"], 943) + self.assertIs(res["success"], False) + + +class RuntimeHelperResolutionTests(unittest.TestCase): + """Every runtime helper the wrapper references is defined and callable. + + Supplements the runtime coverage above; it does not replace it. + """ + + def test_named_helpers_are_defined_and_callable(self): + for name in RUNTIME_HELPERS: + with self.subTest(helper=name): + self.assertTrue(hasattr(gms, name), f"{name} is not defined") + self.assertTrue(callable(getattr(gms, name))) + + def test_every_global_referenced_by_the_wrapper_resolves(self): + """The generalised form of the round-1 defect: an unresolvable global.""" + fn = _wrapper_ast() + bound: set[str] = {a.arg for a in fn.args.args} + bound |= {a.arg for a in fn.args.kwonlyargs} + if fn.args.vararg: + bound.add(fn.args.vararg.arg) + if fn.args.kwarg: + bound.add(fn.args.kwarg.arg) + for node in ast.walk(fn): + if isinstance(node, ast.Name) and isinstance( + node.ctx, (ast.Store, ast.Del) + ): + bound.add(node.id) + elif isinstance(node, (ast.Import, ast.ImportFrom)): + for alias in node.names: + bound.add((alias.asname or alias.name).split(".")[0]) + elif isinstance(node, ast.ExceptHandler) and node.name: + bound.add(node.name) + + unresolved = sorted( + node.id + for node in ast.walk(fn) + if isinstance(node, ast.Name) + and isinstance(node.ctx, ast.Load) + and node.id not in bound + and not hasattr(gms, node.id) + and not hasattr(builtins, node.id) + ) + self.assertEqual( + unresolved, [], + f"{WRAPPER_NAME} references undefined globals: {unresolved}", + ) + + def test_wrapper_wires_the_authority_and_session_resolvers(self): + fn = _wrapper_ast() + called = { + node.func.id + for node in ast.walk(fn) + if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) + } + self.assertIn("_active_mutation_authority", called) + self.assertIn("_resolve_owner_workflow_session", called) + self.assertIn("_author_mutation_block", called) + + def test_wrapper_accepts_an_optional_session_id(self): + """ABI addition stays backward compatible: optional, defaulting to None.""" + fn = _wrapper_ast() + names = [a.arg for a in fn.args.args] + self.assertIn("session_id", names) + offset = len(names) - len(fn.args.defaults) + default = fn.args.defaults[names.index("session_id") - offset] + self.assertIsInstance(default, ast.Constant) + self.assertIsNone(default.value) + + +class Issue941ScopeGuardNotRegressedTests(unittest.TestCase): + """Preserved: PR #942's bootstrap-scope wiring still holds.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.repo, self.head = _make_control_repo(self._tmp.name) + + def _assessment(self, task: str = BOOTSTRAP_TASK) -> dict: + return aib.assess_author_issue_bootstrap( + workspace_path=self.repo, + canonical_repo_root=self.repo, + current_branch="master", + head_sha=self.head, + porcelain_status="", + remote_master_sha=self.head, + task=task, + ) + + def test_bootstrap_task_still_permitted_from_clean_control_checkout(self): + res = workflow_scope_guard.assess_root_source_mutation( + workspace_path=self.repo, + canonical_repo_root=self.repo, + role_kind="author", + mutation_task=BOOTSTRAP_TASK, + porcelain_status="", + bootstrap_assessment=self._assessment(), + ) + self.assertFalse(res.get("block"), res) + self.assertNotEqual( + res.get("blocker_kind"), workflow_scope_guard.BLOCKER_MISSING_WORKTREE + ) + + def test_bootstrap_task_still_blocked_without_evidence(self): + res = workflow_scope_guard.assess_root_source_mutation( + workspace_path=self.repo, + canonical_repo_root=self.repo, + role_kind="author", + mutation_task=BOOTSTRAP_TASK, + porcelain_status="", + ) + self.assertTrue(res.get("block")) + self.assertEqual( + res.get("blocker_kind"), workflow_scope_guard.BLOCKER_MISSING_WORKTREE + ) + + def test_ordinary_author_mutation_still_blocked_from_control_checkout(self): + res = workflow_scope_guard.assess_root_source_mutation( + workspace_path=self.repo, + canonical_repo_root=self.repo, + role_kind="author", + mutation_task="commit_files", + porcelain_status="", + bootstrap_assessment=self._assessment(), + ) + self.assertTrue(res.get("block")) + self.assertEqual( + res.get("blocker_kind"), workflow_scope_guard.BLOCKER_MISSING_WORKTREE + ) + + def test_create_issue_bootstrap_unchanged(self): + self.assertTrue(cib.is_create_issue_task("create_issue")) + self.assertFalse(cib.is_create_issue_task(BOOTSTRAP_TASK)) + + +def _write_json(path: str, payload: dict) -> None: + """Write an issue-lock file directly, for lock-precedence tests.""" + import json + + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as fh: + json.dump(payload, fh) + + +if __name__ == "__main__": + unittest.main()