From b4c9f558901699e29fac77b4f748565bcdb5f272 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 28 Jul 2026 02:08:21 -0400 Subject: [PATCH] test(author): execute the native #953 tool functions and compensation paths (#953) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review 632 F4. Every prior reference to `gitea_recover_incomplete_bootstrap_lock` and `gitea_inspect_issue_lock_contract` under `tests/` was a string literal — a `tool=` argument, an assertion on returned prose, or a docs substring check — and the suite reconstructed the recovery sequence by hand from `assess_bootstrap_lock_recovery`, `build_canonical_issue_lock`, `build_recovery_record`, and `bind_session_lock`. A hand-written sequence validates the decision layer but cannot see a divergence between itself and the tool body, which is exactly how F1 and F2 — both call-site defects — survived 61 passing cases. The new cases drive the registered functions against a real `git init` repository, a real durable lock file, and config-backed profiles: * `NamespaceMutationWallOnRecovery` — the gate is reached with this task and `author_role_exclusive=True`; its return value aborts the tool rather than being computed and discarded; the author namespace succeeds and produces a canonical lock; a reviewer namespace is refused with `namespace_block` even when the claimant data would otherwise match, and emits the standard BLOCKED audit record; merger is refused; a mismatched claimant profile is refused by the exact-owner layer with the namespace wall explicitly clear; a mismatched head is refused; every refusal leaves the lock bytes, generation, branch, worktree, and an unrelated lock untouched. A subtest matrix asserts the role-kind wall admits `author` and refuses reviewer, merger, limited, and mixed. * `InspectionToolExecutes` — the registered read-only tool reports the contract and the recovery preview while leaving lock bytes, mtime, HEAD, and porcelain status unchanged, and reports an absent lock without creating one. * `Ac7PostCompensationGuidance` — drives the real bootstrap to its AC7 refusal with a forced partial lock and asserts the returned action against the state the rollback actually left: complete cleanup directs to a bootstrap retry and that retry is then executed and succeeds, leaving exactly one canonical lock and one branch; partial cleanup with a surviving lock, and with a surviving branch and worktree, each get their own executable action; a rollback that never completed is distinguished from both; no recommendation names a deleted artifact; unrelated locks are byte-identical afterwards. Two cases cover `release_session_lock` directly — that the rollback now really removes the lock, and that it refuses a lock owned by another session. * `NativeEndToEndBootstrapToCreatePr` — bootstrap, inspect, heartbeat, legitimate divergence (commit and push), pre-mutation ownership re-check, and the unchanged #447 create-PR provenance guard, in one sequence against a real origin. `gitea_lock_issue` is patched to fail the test if anything reaches for it, so the bootstrap lock is proved to carry the whole cycle unrepaired. * `DeadProvenanceConstantRemoved` — the removed constant stays removed and the sanctioned source set stays unwidened. `_NativeToolBase` clears `role_session_router` route state per test: the sticky reviewer-stop marker is process-global and, now that this task is registered in `AUTHOR_TASKS`, an earlier suite leaving it set would make the first gate refuse before the namespace gate under test is reached. Also documents both new tools in `docs/mcp-tool-inventory.md`, so this branch adds no drift to `test_documented_inventory_equals_registered_tools`; the failure reason there is now identical to the pinned base's. Suite: 61 -> 92 cases. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/mcp-tool-inventory.md | 2 + .../test_issue_953_bootstrap_lock_contract.py | 965 ++++++++++++++++++ 2 files changed, 967 insertions(+) diff --git a/docs/mcp-tool-inventory.md b/docs/mcp-tool-inventory.md index 799d0d0..e93d359 100644 --- a/docs/mcp-tool-inventory.md +++ b/docs/mcp-tool-inventory.md @@ -103,6 +103,7 @@ that gates each call, not which tools exist. - `gitea_get_shell_health` - `gitea_heartbeat_issue_lock` - `gitea_heartbeat_reviewer_pr_lease` +- `gitea_inspect_issue_lock_contract` - `gitea_inspect_workflow_lease` - `gitea_issue_irrecoverable_provenance_authorization` - `gitea_list_dependency_edges` @@ -134,6 +135,7 @@ that gates each call, not which tools exist. - `gitea_record_pre_review_command` - `gitea_record_shell_spawn_outcome` - `gitea_record_stable_branch_push_attempt` +- `gitea_recover_incomplete_bootstrap_lock` - `gitea_release_merger_pr_lease` - `gitea_release_reviewer_pr_lease` - `gitea_release_workflow_lease` diff --git a/tests/test_issue_953_bootstrap_lock_contract.py b/tests/test_issue_953_bootstrap_lock_contract.py index 64e51ad..04fc741 100644 --- a/tests/test_issue_953_bootstrap_lock_contract.py +++ b/tests/test_issue_953_bootstrap_lock_contract.py @@ -12,19 +12,29 @@ The #949-shaped regression reproduces that lock *shape*; it never touches the real issue #949 branch, worktree, lock, issue, or head. """ +import contextlib import os import subprocess import sys import tempfile import unittest from datetime import datetime, timedelta, timezone +from unittest import mock sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 + +import author_issue_bootstrap # noqa: E402 import author_lock_contract # noqa: E402 import bootstrap_lock_recovery # noqa: E402 +import gitea_audit # noqa: E402 import issue_lock_provenance # noqa: E402 import issue_lock_store # noqa: E402 +import mcp_server # noqa: E402 +import role_namespace_gate # noqa: E402 +import role_session_router # noqa: E402 ISSUE = 9530 BRANCH = f"fix/issue-{ISSUE}-canonical-contract" @@ -964,6 +974,961 @@ class BootstrapWiring(unittest.TestCase): self.assertIn("gitea_recover_incomplete_bootstrap_lock", action) +class _NativeToolBase(unittest.TestCase): + """A real git worktree, a real durable lock, and the real registered tools. + + Review 632 F4: every previous reference to the two new tools under + ``tests/`` was a string literal, and the suite reconstructed the recovery + sequence by hand. A hand-written sequence cannot see a divergence between + itself and the tool body — which is exactly how F1 and F2, both call-site + defects, survived 61 passing cases. These call the registered functions. + """ + + TOOL_ISSUE = 9531 + TOOL_BRANCH = "fix/issue-9531-native-tool-path" + TOOL_ORG = "Example-Org" + TOOL_REPO = "Example-Repo" + AUTHOR_PROFILE = "test-author-prgs" + REVIEWER_PROFILE = "test-reviewer-prgs" + MERGER_PROFILE = "test-merger-prgs" + TOOL_IDENTITY = "example-native-user" + + def setUp(self): + self.lock_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.lock_dir.cleanup) + self.repo = tempfile.mkdtemp(prefix="issue953-native-") + self.addCleanup( + lambda: subprocess.run(["rm", "-rf", self.repo], check=False) + ) + self._init_repo() + self.remotes = mock.patch.dict( + mcp_server.REMOTES, + { + "prgs": { + "host": "gitea.prgs.cc", + "org": self.TOOL_ORG, + "repo": self.TOOL_REPO, + } + }, + ) + self.remotes.start() + self.addCleanup(mock.patch.stopall) + mcp_server._IDENTITY_CACHE.clear() + # The sticky reviewer-stop route is process-global and outlives whatever + # test set it. These cases assert on the *namespace* gate, so the gate + # ahead of it must start clean or it refuses first for another reason. + role_session_router.clear_route_state() + self.addCleanup(role_session_router.clear_route_state) + + def _git(self, *args): + return subprocess.run( + ["git", "-C", self.repo, *args], + capture_output=True, + text=True, + check=True, + ) + + def _init_repo(self): + self._git("init", "-q", "-b", "master") + self._git("config", "user.email", "test@example.com") + self._git("config", "user.name", "Test") + with open(os.path.join(self.repo, "seed.txt"), "w") as handle: + handle.write("seed\n") + self._git("add", "seed.txt") + self._git("commit", "-q", "-m", "seed") + self._git("checkout", "-q", "-b", self.TOOL_BRANCH) + # The state recovery exists for: the branch already carries pushed work. + with open(os.path.join(self.repo, "impl.txt"), "w") as handle: + handle.write("implementation\n") + self._git("add", "impl.txt") + self._git("commit", "-q", "-m", "implementation") + self.head = self._git("rev-parse", "HEAD").stdout.strip() + self.worktree = os.path.realpath(self.repo) + + def _lock_path(self): + return issue_lock_store.lock_file_path( + remote=REMOTE, + org=self.TOOL_ORG, + repo=self.TOOL_REPO, + issue_number=self.TOOL_ISSUE, + lock_dir=self.lock_dir.name, + ) + + def write_incomplete_lock(self): + """The exact malformed shape the old bootstrap left behind.""" + lock = bootstrap_shaped_lock( + issue_number=self.TOOL_ISSUE, + branch=self.TOOL_BRANCH, + branch_name=self.TOOL_BRANCH, + worktree_path=self.worktree, + org=self.TOOL_ORG, + repo=self.TOOL_REPO, + claimant={ + "username": self.TOOL_IDENTITY, + "profile": self.AUTHOR_PROFILE, + }, + ) + path = self._lock_path() + lock["lock_file_path"] = path + issue_lock_store.save_lock_file(path, lock) + return path + + def _env(self, profile_name): + env = shared_mutation_env( + profile_name, + include_example_repo=True, + GITEA_ISSUE_LOCK_DIR=self.lock_dir.name, + ) + env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + return env + + def call_recovery( + self, + *, + profile_name=None, + identity=None, + claimant_profile=None, + expected_head=None, + audit_sink=None, + namespace_patch=None, + **kwargs, + ): + """Invoke the registered recovery tool itself, not its assessors.""" + profile_name = profile_name or self.AUTHOR_PROFILE + env = self._env(profile_name) + patches = [ + mock.patch( + "mcp_server._work_lease_claimant", + return_value={ + "username": identity or self.TOOL_IDENTITY, + "profile": claimant_profile or profile_name, + }, + ), + mock.patch("mcp_server.get_auth_header", return_value="token x"), + mock.patch( + "mcp_server._canonical_local_git_root", return_value=self.worktree + ), + mock.patch.dict(os.environ, env, clear=True), + ] + if audit_sink is not None: + patches.append( + mock.patch( + "mcp_server._audit", + side_effect=lambda *a, **kw: audit_sink.append((a, kw)), + ) + ) + if namespace_patch is not None: + patches.append(namespace_patch) + with contextlib.ExitStack() as stack: + for patch in patches: + stack.enter_context(patch) + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + return mcp_server.gitea_recover_incomplete_bootstrap_lock( + issue_number=kwargs.pop("issue_number", self.TOOL_ISSUE), + branch_name=kwargs.pop("branch_name", self.TOOL_BRANCH), + worktree_path=kwargs.pop("worktree_path", self.worktree), + expected_head=expected_head or self.head, + remote="prgs", + **kwargs, + ) + + def call_inspection(self, *, profile_name=None, **kwargs): + """Invoke the registered read-only inspection tool itself.""" + env = self._env(profile_name or self.AUTHOR_PROFILE) + with mock.patch( + "mcp_server._work_lease_claimant", + return_value={ + "username": self.TOOL_IDENTITY, + "profile": profile_name or self.AUTHOR_PROFILE, + }, + ), mock.patch( + "mcp_server.get_auth_header", return_value="token x" + ), mock.patch( + "mcp_server._canonical_local_git_root", return_value=self.worktree + ), mock.patch.dict( + os.environ, env, clear=True + ): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + return mcp_server.gitea_inspect_issue_lock_contract( + issue_number=kwargs.pop("issue_number", self.TOOL_ISSUE), + remote="prgs", + **kwargs, + ) + + +class NamespaceMutationWallOnRecovery(_NativeToolBase): + """Review 632 F1: the namespace/session wall on the new author mutation. + + ``gitea_recover_incomplete_bootstrap_lock`` writes the same durable lock as + ``gitea_recover_dirty_orphaned_issue_worktree`` and must carry the same + third gate. Exact-owner claimant comparison inside + ``assess_bootstrap_lock_recovery`` is a later layer, not a substitute: it + refuses without a namespace evaluation and without a BLOCKED audit record. + """ + + def test_the_recovery_tool_calls_the_namespace_mutation_gate(self): + """The gate must be reached, with this task, before any write.""" + self.write_incomplete_lock() + seen = [] + + def _spy(task, **kwargs): + seen.append((task, kwargs)) + return None + + self.call_recovery( + namespace_patch=mock.patch( + "mcp_server._namespace_mutation_block", side_effect=_spy + ) + ) + + self.assertEqual(len(seen), 1, seen) + task, kwargs = seen[0] + self.assertEqual(task, "recover_incomplete_bootstrap_lock") + self.assertTrue(kwargs.get("author_role_exclusive")) + + def test_the_gate_return_value_is_consumed_and_returned(self): + """A gate refusal must abort the tool, not be computed and discarded.""" + path = self.write_incomplete_lock() + before = issue_lock_store.read_lock_file(path) + refusal = {"success": False, "performed": False, "namespace_block": True} + + result = self.call_recovery( + namespace_patch=mock.patch( + "mcp_server._namespace_mutation_block", return_value=refusal + ) + ) + + self.assertIs(result, refusal) + self.assertEqual(issue_lock_store.read_lock_file(path), before) + + def test_correct_author_namespace_succeeds(self): + path = self.write_incomplete_lock() + result = self.call_recovery() + self.assertTrue(result.get("success"), result) + self.assertTrue(result.get("performed"), result) + written = issue_lock_store.read_lock_file(path) + self.assertTrue( + author_lock_contract.assess_lock_contract(written)["canonical"], written + ) + + def test_reviewer_namespace_is_rejected_with_matching_claimant_data(self): + """Identity that would satisfy the owner check must not be a way in.""" + path = self.write_incomplete_lock() + before = issue_lock_store.read_lock_file(path) + + result = self.call_recovery( + profile_name=self.REVIEWER_PROFILE, + claimant_profile=self.AUTHOR_PROFILE, + ) + + self.assertFalse(result.get("success"), result) + self.assertFalse(result.get("performed"), result) + self.assertTrue(result.get("namespace_block"), result) + self.assertEqual(result.get("mcp_namespace"), "gitea-reviewer") + # Not the later exact-owner refusal — the namespace layer stopped it. + self.assertNotEqual(result.get("refusal_code"), "foreign_claimant") + self.assertEqual(issue_lock_store.read_lock_file(path), before) + + def test_reviewer_rejection_emits_the_standard_blocked_audit(self): + self.write_incomplete_lock() + audit = [] + self.call_recovery(profile_name=self.REVIEWER_PROFILE, audit_sink=audit) + + blocked = [ + (args, kwargs) + for args, kwargs in audit + if kwargs.get("result") == gitea_audit.BLOCKED + ] + self.assertTrue(blocked, audit) + args, kwargs = blocked[0] + self.assertEqual(args[0], "recover_incomplete_bootstrap_lock") + self.assertEqual( + kwargs.get("mutation_task"), "recover_incomplete_bootstrap_lock" + ) + + def test_merger_profile_is_rejected(self): + """gitea.issue.comment is held by every role; the wall cannot rely on it.""" + path = self.write_incomplete_lock() + before = issue_lock_store.read_lock_file(path) + result = self.call_recovery(profile_name=self.MERGER_PROFILE) + self.assertFalse(result.get("success"), result) + self.assertFalse(result.get("performed"), result) + # Specifically the namespace/role wall, not some later refusal. + self.assertTrue(result.get("namespace_block"), result) + self.assertTrue( + any( + "recover_incomplete_bootstrap_lock' blocked" in reason + for reason in result.get("reasons") or [] + ), + result, + ) + self.assertIsNone(result.get("refusal_code"), result) + self.assertEqual(issue_lock_store.read_lock_file(path), before) + + def test_wrong_profile_for_the_claimant_is_rejected(self): + """A matching username under a different profile is still foreign.""" + path = self.write_incomplete_lock() + before = issue_lock_store.read_lock_file(path) + result = self.call_recovery(claimant_profile="test-author-dadeschools") + self.assertFalse(result.get("success"), result) + self.assertFalse(result.get("performed"), result) + # The namespace wall passes here (the session *is* author-bound); this + # must be the exact-owner layer refusing the mismatched profile. + self.assertIsNone(result.get("namespace_block"), result) + self.assertIn( + result.get("refusal_code"), ("foreign_claimant", "healthy_foreign_lock"), result + ) + self.assertEqual(issue_lock_store.read_lock_file(path), before) + + def test_mismatched_head_is_rejected_without_mutation(self): + path = self.write_incomplete_lock() + before = issue_lock_store.read_lock_file(path) + result = self.call_recovery(expected_head=OTHER_HEAD) + self.assertFalse(result.get("success"), result) + self.assertFalse(result.get("mutation_performed"), result) + self.assertEqual(issue_lock_store.read_lock_file(path), before) + + def test_rejection_leaves_branch_worktree_and_unrelated_locks_untouched(self): + unrelated = issue_lock_store.lock_file_path( + remote=REMOTE, + org=self.TOOL_ORG, + repo=self.TOOL_REPO, + issue_number=self.TOOL_ISSUE + 41, + lock_dir=self.lock_dir.name, + ) + issue_lock_store.save_lock_file( + unrelated, bootstrap_shaped_lock(issue_number=self.TOOL_ISSUE + 41) + ) + unrelated_before = issue_lock_store.read_lock_file(unrelated) + + path = self.write_incomplete_lock() + before = issue_lock_store.read_lock_file(path) + head_before = self._git("rev-parse", "HEAD").stdout.strip() + + self.call_recovery(profile_name=self.REVIEWER_PROFILE) + + self.assertEqual(issue_lock_store.read_lock_file(path), before) + self.assertEqual(issue_lock_store.read_lock_file(unrelated), unrelated_before) + self.assertEqual(self._git("rev-parse", "HEAD").stdout.strip(), head_before) + self.assertTrue(os.path.isdir(self.worktree)) + self.assertEqual(self._git("status", "--porcelain").stdout.strip(), "") + + def test_the_gate_routes_this_task_as_author_required(self): + """Without a router entry the namespace check silently allows everything.""" + self.assertEqual( + role_session_router.required_role_for_task( + "recover_incomplete_bootstrap_lock" + ), + "author", + ) + ok, reasons = role_namespace_gate.check_author_mutation_namespace( + "recover_incomplete_bootstrap_lock", + { + "profile_name": "prgs-reviewer", + "allowed_operations": ["gitea.read", "gitea.pr.approve"], + "forbidden_operations": [], + }, + ) + self.assertFalse(ok, reasons) + + def test_role_kind_wall_admits_author_and_refuses_every_other_role(self): + cases = { + "author": (["gitea.pr.create", "gitea.branch.push"], True), + "reviewer": (["gitea.pr.approve"], False), + "merger": (["gitea.pr.merge"], False), + "limited": (["gitea.issue.comment", "gitea.read"], False), + "mixed": (["gitea.pr.approve", "gitea.pr.create"], False), + } + for label, (ops, expected) in cases.items(): + with self.subTest(role=label): + ok, _ = role_namespace_gate.check_author_role_kind( + "recover_incomplete_bootstrap_lock", + { + "profile_name": f"prgs-{label}", + "allowed_operations": ops, + "forbidden_operations": [], + }, + ) + self.assertEqual(ok, expected) + + +class InspectionToolExecutes(_NativeToolBase): + """AC16 proved against the registered tool, not only its assessors.""" + + def test_the_registered_inspection_tool_reports_the_contract(self): + self.write_incomplete_lock() + result = self.call_inspection() + self.assertTrue(result.get("success"), result) + self.assertTrue(result.get("read_only")) + self.assertTrue(result.get("lock_present")) + self.assertFalse(result["lock_contract"]["canonical"]) + + def test_the_registered_inspection_tool_mutates_nothing(self): + path = self.write_incomplete_lock() + before = issue_lock_store.read_lock_file(path) + mtime_before = os.path.getmtime(path) + head_before = self._git("rev-parse", "HEAD").stdout.strip() + + result = self.call_inspection( + branch_name=self.TOOL_BRANCH, worktree_path=self.worktree + ) + + self.assertFalse(result.get("mutation_performed")) + self.assertFalse(result.get("performed")) + self.assertIn("recovery_preview", result) + self.assertEqual(issue_lock_store.read_lock_file(path), before) + self.assertEqual(os.path.getmtime(path), mtime_before) + self.assertEqual(self._git("rev-parse", "HEAD").stdout.strip(), head_before) + self.assertEqual(self._git("status", "--porcelain").stdout.strip(), "") + + def test_inspection_reports_an_absent_lock_without_creating_one(self): + result = self.call_inspection() + self.assertTrue(result.get("success"), result) + self.assertFalse(result.get("lock_present")) + self.assertFalse(os.path.exists(self._lock_path())) + + +class Ac7PostCompensationGuidance(unittest.TestCase): + """Review 632 F2: the returned action must fit the post-rollback state. + + The AC7 refusal runs ``run_compensating_recovery`` first, which releases the + lock and removes the branch and worktree. Recommending incomplete-lock + recovery for those exact artifacts hands the author ``no_durable_lock`` and + then ``worktree_invalid`` — the unexecutable-guidance failure class #953 + exists to remove, reintroduced on the new fail-closed path. + """ + + ISSUE_NUMBER = 9532 + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.repo = os.path.join(self.tmp.name, "repo") + os.makedirs(self.repo) + self._git("init", "-q", "-b", "master") + self._git("config", "user.email", "test@example.com") + self._git("config", "user.name", "Test") + with open(os.path.join(self.repo, "README.md"), "w") as handle: + handle.write("# seed\n") + self._git("add", "README.md") + self._git("commit", "-q", "-m", "seed") + self.master_sha = self._git("rev-parse", "HEAD").stdout.strip() + os.makedirs(os.path.join(self.repo, "branches"), exist_ok=True) + self.lock_dir = os.path.join(self.tmp.name, "locks") + os.makedirs(self.lock_dir, exist_ok=True) + self.journal_dir = os.path.join(self.tmp.name, "journals") + os.makedirs(self.journal_dir, exist_ok=True) + self._journal_env = mock.patch.dict( + os.environ, {"GITEA_BOOTSTRAP_JOURNAL_DIR": self.journal_dir} + ) + self._journal_env.start() + self.addCleanup(self._journal_env.stop) + + def _git(self, *args): + return subprocess.run( + ["git", "-C", self.repo, *args], + capture_output=True, + text=True, + check=True, + ) + + def _branch_names(self): + out = subprocess.run( + ["git", "-C", self.repo, "branch", "--format=%(refname:short)"], + capture_output=True, + text=True, + check=False, + ) + return [line.strip() for line in out.stdout.splitlines() if line.strip()] + + def _lock_files(self): + """Durable issue locks only — not phase journals or session pointers.""" + found = [] + for path in issue_lock_store.iter_lock_files(self.lock_dir): + record = issue_lock_store.read_lock_file(path) or {} + if "lock_generation" in record: + found.append(os.path.basename(path)) + return sorted(found) + + def _run_bootstrap(self, *, key, partial=False, extra_patches=()): + """Drive the real bootstrap; optionally force a partial written lock.""" + with contextlib.ExitStack() as stack: + if partial: + real_build = author_lock_contract.build_canonical_issue_lock + + def _partial(**kwargs): + record = real_build(**kwargs) + # The #949 shape: claimant hoisted to the top level, no + # work_lease, no provenance, no expiry. + return { + "remote": record["remote"], + "org": record["org"], + "repo": record["repo"], + "issue_number": record["issue_number"], + "branch": record["branch"], + "branch_name": record["branch_name"], + "worktree_path": record["worktree_path"], + "claimant": dict(record["work_lease"]["claimant"]), + "lease_id": None, + "owner_session": record.get("owner_session"), + } + + stack.enter_context( + mock.patch.object( + author_issue_bootstrap.author_lock_contract, + "build_canonical_issue_lock", + side_effect=_partial, + ) + ) + for patch in extra_patches: + stack.enter_context(patch) + return author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=self.ISSUE_NUMBER, + canonical_repo_root=self.repo, + expected_base_sha=self.master_sha, + idempotency_key=key, + remote="prgs", + lock_dir=self.lock_dir, + owner_session=f"session-{key}", + active_identity=IDENTITY, + active_profile=PROFILE, + ) + + def test_ac7_failure_then_complete_compensation_directs_to_retry_bootstrap(self): + result = self._run_bootstrap(key="ac7-complete", partial=True) + + self.assertFalse(result.get("success"), result) + self.assertEqual(result.get("reason_code"), "incomplete_issue_lock_contract") + self.assertFalse(result.get("implementation_allowed")) + state = result["post_compensation_state"] + self.assertEqual(state["cleanup_state"], author_lock_contract.CLEANUP_COMPLETE) + self.assertEqual(state["surviving_artifacts"], []) + + action = result["exact_next_action"] + self.assertIn("gitea_bootstrap_author_issue_worktree", action) + self.assertIn("Do not call gitea_recover_incomplete_bootstrap_lock", action) + + def test_the_returned_retry_action_is_executable(self): + """Follow the advice literally and it must succeed.""" + first = self._run_bootstrap(key="ac7-retry-1", partial=True) + self.assertIn( + "gitea_bootstrap_author_issue_worktree", first["exact_next_action"] + ) + + second = self._run_bootstrap(key="ac7-retry-2") + self.assertTrue(second.get("success"), second) + self.assertTrue(second.get("implementation_allowed")) + self.assertTrue(second["lock_contract"]["canonical"], second["lock_contract"]) + self.assertTrue(second.get("task_session_id")) + + def test_retry_after_complete_compensation_leaves_exactly_one_lock(self): + self._run_bootstrap(key="ac7-single-1", partial=True) + self.assertEqual(self._lock_files(), []) + + second = self._run_bootstrap(key="ac7-single-2") + self.assertTrue(second.get("success"), second) + locks = self._lock_files() + self.assertEqual(len(locks), 1, locks) + written = issue_lock_store.read_lock_file(second["lock_state"]) + self.assertTrue(author_lock_contract.assess_lock_contract(written)["canonical"]) + branches = [b for b in self._branch_names() if b != "master"] + self.assertEqual(len(branches), 1, branches) + + def test_no_recommendation_names_an_artifact_the_rollback_deleted(self): + result = self._run_bootstrap(key="ac7-no-ghosts", partial=True) + action = result["exact_next_action"] + state = result["post_compensation_state"] + + self.assertFalse(state["branch_present"]) + self.assertFalse(state["worktree_present"]) + self.assertFalse(state["lock_present"]) + self.assertNotIn(result["worktree_path"], action) + self.assertNotIn(f"branch '{result['branch_name']}'", action) + self.assertFalse(os.path.isdir(result["worktree_path"])) + self.assertNotIn(result["branch_name"], self._branch_names()) + + def test_partial_compensation_with_a_surviving_lock_is_not_reported_complete(self): + def _boom(**kwargs): + raise RuntimeError("lock release failed") + + outcome = self._run_bootstrap( + key="ac7-lock-survives", + partial=True, + extra_patches=[ + mock.patch.object( + author_issue_bootstrap.issue_lock_store, + "release_session_lock", + side_effect=_boom, + ) + ], + ) + + state = outcome["post_compensation_state"] + self.assertTrue(state["lock_present"], state) + self.assertEqual(state["cleanup_state"], author_lock_contract.CLEANUP_PARTIAL) + self.assertIn("lock", state["surviving_artifacts"]) + self.assertTrue(state["failed_rollback_steps"], state) + action = outcome["exact_next_action"] + self.assertIn("failed step", action) + self.assertIn("gitea_inspect_issue_lock_contract", action) + # The advice must not send the author at artifacts the rollback removed. + self.assertNotIn("re-run gitea_bootstrap_author_issue_worktree", action) + + def test_partial_compensation_with_surviving_branch_and_worktree(self): + """A worktree dirty at rollback time is preserved, and so is its branch.""" + real_assess = author_lock_contract.assess_lock_contract + + def _dirty_then_report(lock): + verdict = real_assess(lock) + path = (lock or {}).get("worktree_path") + if path and os.path.isdir(path): + with open(os.path.join(path, "uncommitted.txt"), "w") as handle: + handle.write("author bytes\n") + return verdict + + outcome = self._run_bootstrap( + key="ac7-wt-survives", + partial=True, + extra_patches=[ + mock.patch.object( + author_issue_bootstrap.author_lock_contract, + "assess_lock_contract", + side_effect=_dirty_then_report, + ) + ], + ) + + state = outcome["post_compensation_state"] + self.assertEqual(state["cleanup_state"], author_lock_contract.CLEANUP_PARTIAL) + self.assertTrue(state["worktree_present"], state) + self.assertTrue(state["branch_present"], state) + self.assertTrue(os.path.isdir(outcome["worktree_path"])) + self.assertIn(outcome["branch_name"], self._branch_names()) + self.assertIn("gitea_lock_issue", outcome["exact_next_action"]) + self.assertIn(outcome["branch_name"], outcome["exact_next_action"]) + + def test_compensation_failure_is_distinguished_from_partial_cleanup(self): + outcome = self._run_bootstrap( + key="ac7-comp-failed", + partial=True, + extra_patches=[ + mock.patch.object( + author_issue_bootstrap, + "run_compensating_recovery", + return_value={ + "executed": False, + "rolled_back": [], + "reason": "boom", + }, + ) + ], + ) + + state = outcome["post_compensation_state"] + self.assertEqual(state["cleanup_state"], author_lock_contract.CLEANUP_FAILED) + action = outcome["exact_next_action"] + self.assertIn("did not complete", action) + self.assertIn("gitea_inspect_issue_lock_contract", action) + self.assertNotIn("re-run gitea_bootstrap_author_issue_worktree", action) + + def test_a_failed_rollback_step_is_recorded_even_when_nothing_survives(self): + """A step that errored is still reported, and cleanup is still complete.""" + state = author_lock_contract.assess_post_compensation_state( + { + "executed": True, + "rolled_back": ["lease_release_failed:lease-1:RuntimeError"], + }, + lock_present=False, + worktree_present=False, + branch_present=False, + ) + self.assertEqual(state["cleanup_state"], author_lock_contract.CLEANUP_COMPLETE) + self.assertEqual( + state["failed_rollback_steps"], + ["lease_release_failed:lease-1:RuntimeError"], + ) + + def test_the_compensation_lock_release_actually_removes_the_lock(self): + """The rollback's lock half was dead code before #953 review 632 F2.""" + self.assertTrue(hasattr(issue_lock_store, "release_session_lock")) + result = self._run_bootstrap(key="ac7-release-real", partial=True) + self.assertIn( + f"lock:issue-{self.ISSUE_NUMBER}", + result["compensating_recovery"]["rolled_back"], + ) + self.assertEqual(self._lock_files(), []) + + def test_release_refuses_a_lock_owned_by_a_different_session(self): + created = self._run_bootstrap(key="ac7-foreign-release") + self.assertTrue(created.get("success"), created) + with self.assertRaises(FileNotFoundError): + issue_lock_store.release_session_lock( + issue_number=self.ISSUE_NUMBER, + session="session-somebody-else", + lock_dir=self.lock_dir, + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + ) + self.assertEqual(len(self._lock_files()), 1, self._lock_files()) + + def test_ac7_refusal_never_reports_implementation_ready(self): + result = self._run_bootstrap(key="ac7-never-ready", partial=True) + self.assertFalse(result.get("success")) + self.assertFalse(result.get("implementation_allowed")) + self.assertNotIn( + "proceed with author implementation", result["exact_next_action"] + ) + + def test_ac7_refusal_touches_no_unrelated_lock(self): + unrelated_path = issue_lock_store.lock_file_path( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + issue_number=self.ISSUE_NUMBER + 63, + lock_dir=self.lock_dir, + ) + issue_lock_store.save_lock_file( + unrelated_path, bootstrap_shaped_lock(issue_number=self.ISSUE_NUMBER + 63) + ) + before = issue_lock_store.read_lock_file(unrelated_path) + mtime_before = os.path.getmtime(unrelated_path) + + self._run_bootstrap(key="ac7-isolation", partial=True) + + self.assertEqual(issue_lock_store.read_lock_file(unrelated_path), before) + self.assertEqual(os.path.getmtime(unrelated_path), mtime_before) + + def test_surviving_lock_and_worktree_direct_to_target_specific_recovery(self): + """The one state in which incomplete-lock recovery *is* executable.""" + state = author_lock_contract.assess_post_compensation_state( + {"executed": True, "rolled_back": []}, + lock_present=True, + worktree_present=True, + branch_present=True, + ) + action = author_lock_contract.post_compensation_action( + state, + issue_number=self.ISSUE_NUMBER, + branch_name=BRANCH, + worktree_path="/scratch/wt", + missing_fields=["work_lease"], + ) + self.assertEqual(state["cleanup_state"], author_lock_contract.CLEANUP_PARTIAL) + self.assertIn("gitea_recover_incomplete_bootstrap_lock", action) + self.assertIn(BRANCH, action) + self.assertIn("/scratch/wt", action) + + +class NativeEndToEndBootstrapToCreatePr(unittest.TestCase): + """AC17/AC18 driven through the real tools, with no gitea_lock_issue repair. + + bootstrap → inspect → heartbeat/renew → legitimate divergence → downstream + validation → create_pr provenance. The lock the real bootstrap writes must + carry the whole sequence on its own; repairing it with the older + ``gitea_lock_issue`` path would prove nothing about the new contract, so + that path is patched to fail the test if anything reaches for it. + """ + + ISSUE_NUMBER = 9533 + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.origin = os.path.join(self.tmp.name, "origin.git") + self.repo = os.path.join(self.tmp.name, "repo") + subprocess.run( + ["git", "init", "-q", "--bare", self.origin], check=True, capture_output=True + ) + subprocess.run( + ["git", "init", "-q", "-b", "master", self.repo], + check=True, + capture_output=True, + ) + self._git("config", "user.email", "author@example.invalid") + self._git("config", "user.name", "Example Author") + with open(os.path.join(self.repo, "README.md"), "w") as handle: + handle.write("base\n") + self._git("add", "README.md") + self._git("commit", "-q", "-m", "base commit") + self._git("remote", "add", "origin", self.origin) + self._git("push", "-q", "-u", "origin", "master") + self.master_sha = self._git("rev-parse", "HEAD").stdout.strip() + + self.lock_dir = os.path.join(self.tmp.name, "locks") + os.makedirs(self.lock_dir, exist_ok=True) + self.journal_dir = os.path.join(self.tmp.name, "journals") + os.makedirs(self.journal_dir, exist_ok=True) + self._journal_env = mock.patch.dict( + os.environ, {"GITEA_BOOTSTRAP_JOURNAL_DIR": self.journal_dir} + ) + self._journal_env.start() + self.addCleanup(self._journal_env.stop) + self.remotes = mock.patch.dict( + mcp_server.REMOTES, + { + "prgs": { + "host": "gitea.prgs.cc", + "org": "Scaled-Tech-Consulting", + "repo": "Gitea-Tools", + } + }, + ) + self.remotes.start() + self.addCleanup(mock.patch.stopall) + mcp_server._IDENTITY_CACHE.clear() + + def _git(self, *args, cwd=None): + return subprocess.run( + ["git", "-C", cwd or self.repo, *args], + check=True, + capture_output=True, + text=True, + ) + + def _inspect(self, worktree, branch): + env = shared_mutation_env( + "test-author-prgs", GITEA_ISSUE_LOCK_DIR=self.lock_dir + ) + env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir + with mock.patch( + "mcp_server._work_lease_claimant", + return_value={"username": IDENTITY, "profile": PROFILE}, + ), mock.patch( + "mcp_server.get_auth_header", return_value="token x" + ), mock.patch( + "mcp_server._canonical_local_git_root", return_value=self.repo + ), mock.patch.dict( + os.environ, env, clear=True + ): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir + return mcp_server.gitea_inspect_issue_lock_contract( + issue_number=self.ISSUE_NUMBER, + branch_name=branch, + worktree_path=worktree, + remote="prgs", + ) + + def test_bootstrap_inspect_heartbeat_diverge_and_create_pr_provenance(self): + def _forbidden(*args, **kwargs): + raise AssertionError( + "gitea_lock_issue must not be needed to repair a bootstrap lock" + ) + + with mock.patch.object(mcp_server, "gitea_lock_issue", side_effect=_forbidden): + # 1. Bootstrap through the real function. + result = author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=self.ISSUE_NUMBER, + canonical_repo_root=self.repo, + expected_base_sha=self.master_sha, + idempotency_key="e2e-953", + remote="prgs", + lock_dir=self.lock_dir, + owner_session="session-e2e-953", + active_identity=IDENTITY, + active_profile=PROFILE, + ) + self.assertTrue(result.get("success"), result) + self.assertTrue(result.get("implementation_allowed")) + branch = result["branch_name"] + worktree = result["worktree_path"] + token = result["task_session_id"] + self.assertTrue(token) + + # 2. Inspect through the registered read-only tool. + inspected = self._inspect(worktree, branch) + self.assertTrue(inspected.get("success"), inspected) + self.assertTrue(inspected["lock_contract"]["canonical"], inspected) + self.assertTrue(inspected["lock_contract"]["heartbeatable"]) + self.assertTrue(inspected["lock_contract"]["create_pr_eligible"]) + self.assertFalse(inspected.get("mutation_performed")) + + # 3. Heartbeat/renew while still base-equivalent. + before = issue_lock_store.heartbeat_session_lock( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + issue_number=self.ISSUE_NUMBER, + branch_name=branch, + worktree_path=worktree, + identity=IDENTITY, + profile=PROFILE, + task_session_id=token, + lock_dir=self.lock_dir, + ) + self.assertTrue(before["success"], before.get("reasons")) + + # 4. Legitimate divergence: implement, commit, push. + with open(os.path.join(worktree, "feature.py"), "w") as handle: + handle.write("VALUE = 1\n") + self._git("add", "feature.py", cwd=worktree) + self._git("commit", "-q", "-m", "feat: implement", cwd=worktree) + self._git("push", "-q", "-u", "origin", branch, cwd=worktree) + head = self._git("rev-parse", "HEAD", cwd=worktree).stdout.strip() + remote_head = subprocess.run( + ["git", "-C", self.origin, "rev-parse", f"refs/heads/{branch}"], + capture_output=True, + text=True, + check=True, + ).stdout.strip() + self.assertEqual(head, remote_head) + + # 5. Downstream validation after divergence. + after = issue_lock_store.heartbeat_session_lock( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + issue_number=self.ISSUE_NUMBER, + branch_name=branch, + worktree_path=worktree, + identity=IDENTITY, + profile=PROFILE, + task_session_id=token, + lock_dir=self.lock_dir, + ) + self.assertTrue(after["success"], after.get("reasons")) + # The pre-mutation ownership re-check (#438) accepts the diverged + # branch without any repair step. + diverged_lock = issue_lock_store.read_lock_file(result["lock_state"]) + proof = issue_lock_store.verify_lock_for_mutation( + diverged_lock, + issue_number=self.ISSUE_NUMBER, + branch_name=branch, + worktree_path=worktree, + ) + self.assertTrue(proof["proven"], proof["reasons"]) + + # 6. The unchanged #447 create_pr provenance guard accepts it. + final = issue_lock_store.read_lock_file(result["lock_state"]) + verdict = issue_lock_provenance.assess_lock_file_for_create_pr(final) + self.assertTrue(verdict["proven"], verdict["reasons"]) + + # The branch was never rewound to satisfy any gate. + merge_base = self._git("merge-base", "master", branch, cwd=worktree).stdout.strip() + self.assertEqual(merge_base, self.master_sha) + + +class DeadProvenanceConstantRemoved(unittest.TestCase): + """Review 632 F3: no second lock source may appear to exist.""" + + def test_no_recovery_source_constant_is_exported(self): + self.assertFalse( + hasattr(author_lock_contract, "SOURCE_BOOTSTRAP_LOCK_RECOVERY") + ) + + def test_bootstrap_source_is_still_the_sanctioned_lock_issue_source(self): + self.assertEqual( + author_lock_contract.SOURCE_BOOTSTRAP, + issue_lock_provenance.SOURCE_LOCK_ISSUE, + ) + + def test_the_sanctioned_source_set_is_still_not_widened(self): + self.assertNotIn( + "gitea_recover_incomplete_bootstrap_lock", + issue_lock_provenance.SANCTIONED_LOCK_SOURCES, + ) + + class CapabilityRegistration(unittest.TestCase): """The new operations are registered and role-gated."""