"""One canonical bootstrap/lock contract and its recovery path (#953). Covers the defect in which ``gitea_bootstrap_author_issue_worktree`` reported a lock as created, wrote a shape no downstream reader accepts, and then directed the author to implement — after which heartbeat, re-lock, exact-owner renewal, and the #447 create-PR guard all refuse simultaneously and no sanctioned recovery path remains eligible. Every fixture here is synthetic and isolated: locks are written into temporary directories and the git repositories are created per-test with ``git init``. The #949-shaped regression reproduces that lock *shape*; it never touches the real issue #949 branch, worktree, lock, issue, or head. """ import os import subprocess import sys import tempfile import unittest from datetime import datetime, timedelta, timezone sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) import author_lock_contract # noqa: E402 import bootstrap_lock_recovery # noqa: E402 import issue_lock_provenance # noqa: E402 import issue_lock_store # noqa: E402 ISSUE = 9530 BRANCH = f"fix/issue-{ISSUE}-canonical-contract" IDENTITY = "example-author-user" PROFILE = "example-author-profile" FOREIGN_IDENTITY = "example-other-user" FOREIGN_PROFILE = "example-other-profile" REMOTE = "prgs" ORG = "ExampleOrg" REPO = "ExampleRepo" HEAD = "a" * 40 OTHER_HEAD = "b" * 40 def _ts(delta_minutes: int = 0) -> str: return ( (datetime.now(timezone.utc) + timedelta(minutes=delta_minutes)) .replace(microsecond=0) .isoformat() .replace("+00:00", "Z") ) def bootstrap_shaped_lock(**overrides): """The exact malformed shape #949 was left in by the old bootstrap. Claimant at the lock top level, ``lease_id`` null, and no ``work_lease``, ``lock_provenance``, or ``expires_at``. """ lock = { "remote": REMOTE, "org": ORG, "repo": REPO, "issue_number": ISSUE, "branch": BRANCH, "branch_name": BRANCH, "worktree_path": "/scratch/wt-9530", "owner_session": "author_issue_work-deadbeefdeadbeef", "claimant": {"username": IDENTITY, "profile": PROFILE}, "assignment_id": None, "lease_id": None, "expected_base_sha": OTHER_HEAD, "lock_generation": 1, } lock.update(overrides) return lock def canonical_lock(worktree="/scratch/wt-9530", **overrides): lock = author_lock_contract.build_canonical_issue_lock( issue_number=ISSUE, branch_name=BRANCH, worktree_path=worktree, remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE, tool="gitea_lock_issue", ) lock["lock_generation"] = 2 lock["session_pid"] = os.getpid() lock.update(overrides) return lock class CanonicalContractShape(unittest.TestCase): """AC1, AC6, AC13: one contract, emitted with every required field.""" def test_bootstrap_builder_emits_the_full_canonical_schema(self): lock = author_lock_contract.build_canonical_issue_lock( issue_number=ISSUE, branch_name=BRANCH, worktree_path="/scratch/wt", remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE, tool="gitea_bootstrap_author_issue_worktree", source=author_lock_contract.SOURCE_BOOTSTRAP, ) for field in author_lock_contract.REQUIRED_LOCK_FIELDS: self.assertIn(field, lock, f"canonical lock missing {field}") for field in author_lock_contract.REQUIRED_WORK_LEASE_FIELDS: self.assertIn(field, lock["work_lease"], f"work_lease missing {field}") self.assertEqual( lock["work_lease"]["claimant"], {"username": IDENTITY, "profile": PROFILE}, ) self.assertEqual( lock["lock_provenance"]["written_by_tool"], "gitea_bootstrap_author_issue_worktree", ) def test_bootstrap_and_lock_issue_produce_the_same_contract(self): """AC13: the two writers must not disagree about what a lock is.""" from_bootstrap = author_lock_contract.build_canonical_issue_lock( issue_number=ISSUE, branch_name=BRANCH, worktree_path="/scratch/wt", remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE, tool="gitea_bootstrap_author_issue_worktree", source=author_lock_contract.SOURCE_BOOTSTRAP, ) from_lock_issue = author_lock_contract.build_canonical_issue_lock( issue_number=ISSUE, branch_name=BRANCH, worktree_path="/scratch/wt", remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE, tool="gitea_lock_issue", ) self.assertEqual(sorted(from_bootstrap.keys()), sorted(from_lock_issue.keys())) self.assertEqual( sorted(from_bootstrap["work_lease"].keys()), sorted(from_lock_issue["work_lease"].keys()), ) for lock in (from_bootstrap, from_lock_issue): self.assertTrue( author_lock_contract.assess_lock_contract(lock)["canonical"] ) def test_no_successful_build_returns_a_null_ownership_token(self): """AC6: the fencing token every later check keys on is never null.""" lock = author_lock_contract.build_canonical_issue_lock( issue_number=ISSUE, branch_name=BRANCH, worktree_path="/scratch/wt", remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE, tool="gitea_bootstrap_author_issue_worktree", ) self.assertTrue(lock["work_lease"]["task_session_id"]) self.assertIsNotNone(lock["work_lease"]["expires_at"]) def test_provenance_cannot_be_supplied_by_a_caller(self): """Safety: provenance is minted server-side, never accepted.""" import inspect as _inspect params = _inspect.signature( author_lock_contract.build_canonical_issue_lock ).parameters self.assertNotIn("lock_provenance", params) self.assertNotIn("provenance", params) class MalformedAndPartialLocks(unittest.TestCase): """AC7, AC12, AC19: malformed, partial, and missing-expiration locks.""" def test_bootstrap_shaped_lock_is_reported_as_not_canonical(self): assessment = author_lock_contract.assess_lock_contract(bootstrap_shaped_lock()) self.assertFalse(assessment["canonical"]) self.assertIn("work_lease", assessment["missing_fields"]) self.assertIn("lock_provenance", assessment["missing_fields"]) def test_missing_fields_are_reported_structurally_and_by_name(self): """AC7: the refusal names what is missing, not just that it failed.""" assessment = author_lock_contract.assess_lock_contract(bootstrap_shaped_lock()) message = author_lock_contract.format_contract_refusal(assessment) self.assertIn("work_lease", message) self.assertIn("lock_provenance", message) self.assertIsInstance(assessment["missing_fields"], list) def test_missing_expiration_is_explicit_not_never_expiring(self): """AC12: the bug — absent expiry read as 'not yet expired'.""" lock = bootstrap_shaped_lock() # The pre-existing reader still reports "not expired" for this lock... self.assertFalse(issue_lock_store.is_lease_expired(lock)) # ...so the contract states the real fact explicitly instead. state = author_lock_contract.expiration_state(lock) self.assertEqual(state["state"], author_lock_contract.EXPIRATION_MISSING) self.assertIsNone(state["expired"]) assessment = author_lock_contract.assess_lock_contract(lock) self.assertTrue( any("neither expirable nor renewable" in r for r in assessment["reasons"]) ) def test_missing_expiration_lock_is_recoverable_rather_than_stranded(self): """AC12: it must not be non-expiring *and* ineligible for every path.""" assessment = bootstrap_lock_recovery.assess_bootstrap_lock_recovery( bootstrap_shaped_lock(worktree_path="/scratch/wt-9530"), issue_number=ISSUE, branch_name=BRANCH, worktree_path="/scratch/wt-9530", remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE, observed_head=HEAD, declared_head=HEAD, worktree_exists=True, worktree_registered=True, current_branch=BRANCH, ) self.assertTrue(assessment["recovery_sanctioned"], assessment["reasons"]) def test_partial_lock_missing_only_provenance_is_not_canonical(self): lock = canonical_lock() lock.pop("lock_provenance") assessment = author_lock_contract.assess_lock_contract(lock) self.assertFalse(assessment["canonical"]) self.assertFalse(assessment["create_pr_eligible"]) def test_unparseable_expiration_is_named_rather_than_silently_ignored(self): lock = canonical_lock() lock["work_lease"]["expires_at"] = "not-a-timestamp" state = author_lock_contract.expiration_state(lock) self.assertEqual(state["state"], author_lock_contract.EXPIRATION_UNPARSEABLE) def test_expired_lock_is_reported_as_expired(self): lock = canonical_lock() lock["work_lease"]["expires_at"] = _ts(-60) state = author_lock_contract.expiration_state(lock) self.assertEqual(state["state"], author_lock_contract.EXPIRATION_RECORDED) self.assertTrue(state["expired"]) def test_absent_lock_reports_absent_contract(self): assessment = author_lock_contract.assess_lock_contract(None) self.assertEqual(assessment["contract"], author_lock_contract.CONTRACT_ABSENT) self.assertIn( "gitea_lock_issue", author_lock_contract.recommended_action(assessment) ) class ClaimantCompatibility(unittest.TestCase): """AC2, AC13, AC14: legacy and canonical claimant placement.""" def test_claimant_is_read_from_the_legacy_top_level_placement(self): recorded = author_lock_contract.lock_claimant(bootstrap_shaped_lock()) self.assertEqual(recorded, {"username": IDENTITY, "profile": PROFILE}) def test_claimant_is_read_from_the_canonical_work_lease_placement(self): recorded = author_lock_contract.lock_claimant(canonical_lock()) self.assertEqual(recorded, {"username": IDENTITY, "profile": PROFILE}) def test_work_lease_placement_wins_over_a_stale_top_level_copy(self): """An upgraded lock must not be re-read from its stale legacy copy.""" lock = canonical_lock() lock["claimant"] = {"username": FOREIGN_IDENTITY, "profile": FOREIGN_PROFILE} self.assertEqual( author_lock_contract.lock_claimant(lock), {"username": IDENTITY, "profile": PROFILE}, ) def test_ownership_check_accepts_the_legacy_placement(self): """AC2: the exact refusal that made a fresh bootstrap lock un-heartbeatable.""" refusals = issue_lock_store._ownership_refusals( bootstrap_shaped_lock(worktree_path="/scratch/wt-9530"), issue_number=ISSUE, branch_name=BRANCH, worktree_path="/scratch/wt-9530", identity=IDENTITY, profile=PROFILE, ) self.assertNotIn( "lock does not record both a claimant username and profile", refusals ) self.assertEqual(refusals, []) def test_ownership_check_still_refuses_a_mismatched_claimant(self): """Tolerating the placement must not tolerate the wrong owner.""" refusals = issue_lock_store._ownership_refusals( bootstrap_shaped_lock(worktree_path="/scratch/wt-9530"), issue_number=ISSUE, branch_name=BRANCH, worktree_path="/scratch/wt-9530", identity=FOREIGN_IDENTITY, profile=PROFILE, ) self.assertTrue(any("does not match active identity" in r for r in refusals)) def test_ownership_check_still_refuses_a_lock_with_no_claimant_at_all(self): lock = bootstrap_shaped_lock(worktree_path="/scratch/wt-9530") lock.pop("claimant") refusals = issue_lock_store._ownership_refusals( lock, issue_number=ISSUE, branch_name=BRANCH, worktree_path="/scratch/wt-9530", identity=IDENTITY, profile=PROFILE, ) self.assertIn( "lock does not record both a claimant username and profile", refusals ) class CreatePrProvenanceGuardPreserved(unittest.TestCase): """AC4 and the safety requirement that #447 is not weakened.""" def test_canonical_bootstrap_lock_passes_the_447_guard(self): lock = author_lock_contract.build_canonical_issue_lock( issue_number=ISSUE, branch_name=BRANCH, worktree_path="/scratch/wt", remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE, tool="gitea_bootstrap_author_issue_worktree", source=author_lock_contract.SOURCE_BOOTSTRAP, ) verdict = issue_lock_provenance.assess_lock_file_for_create_pr(lock) self.assertTrue(verdict["proven"], verdict["reasons"]) def test_the_old_bootstrap_shape_is_still_rejected_by_the_447_guard(self): """The guard must keep failing closed on a lock with no provenance.""" verdict = issue_lock_provenance.assess_lock_file_for_create_pr( bootstrap_shaped_lock() ) self.assertFalse(verdict["proven"]) self.assertTrue(verdict["block"]) def test_guard_still_rejects_an_unsanctioned_provenance_source(self): lock = canonical_lock() lock["lock_provenance"]["source"] = "hand_written_by_caller" verdict = issue_lock_provenance.assess_lock_file_for_create_pr(lock) self.assertFalse(verdict["proven"]) def test_guard_still_rejects_provenance_without_work_lease(self): lock = canonical_lock() lock.pop("work_lease") verdict = issue_lock_provenance.assess_lock_file_for_create_pr(lock) self.assertFalse(verdict["proven"]) def test_sanctioned_source_set_was_not_widened(self): """Bootstrap satisfies the guard; it does not get its own exemption.""" self.assertEqual( author_lock_contract.SOURCE_BOOTSTRAP, issue_lock_provenance.SOURCE_LOCK_ISSUE, ) class RecoveryOwnershipVerification(unittest.TestCase): """AC10, AC11: what recovery proves before it changes lock state.""" def _assess(self, lock=None, **overrides): kwargs = dict( issue_number=ISSUE, branch_name=BRANCH, worktree_path="/scratch/wt-9530", remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE, observed_head=HEAD, declared_head=HEAD, worktree_exists=True, worktree_registered=True, current_branch=BRANCH, ) kwargs.update(overrides) target = ( lock if lock is not None else bootstrap_shaped_lock(worktree_path="/scratch/wt-9530") ) return bootstrap_lock_recovery.assess_bootstrap_lock_recovery(target, **kwargs) def test_exact_owner_recovery_is_sanctioned(self): self.assertTrue(self._assess()["recovery_sanctioned"]) def test_mismatched_repository_is_refused(self): result = self._assess(repo="OtherRepo") self.assertFalse(result["recovery_sanctioned"]) self.assertEqual( result["refusal_code"], bootstrap_lock_recovery.REFUSAL_BINDING_MISMATCH ) def test_mismatched_org_is_refused(self): self.assertFalse(self._assess(org="OtherOrg")["recovery_sanctioned"]) def test_mismatched_remote_is_refused(self): self.assertFalse(self._assess(remote="dadeschools")["recovery_sanctioned"]) def test_mismatched_issue_is_refused(self): self.assertFalse(self._assess(issue_number=ISSUE + 1)["recovery_sanctioned"]) def test_mismatched_branch_is_refused(self): self.assertFalse( self._assess(branch_name="fix/issue-9530-other")["recovery_sanctioned"] ) def test_mismatched_worktree_is_refused(self): self.assertFalse( self._assess(worktree_path="/scratch/elsewhere")["recovery_sanctioned"] ) def test_missing_worktree_is_refused(self): result = self._assess(worktree_exists=False) self.assertFalse(result["recovery_sanctioned"]) self.assertEqual( result["refusal_code"], bootstrap_lock_recovery.REFUSAL_WORKTREE_INVALID ) def test_unregistered_worktree_is_refused(self): self.assertFalse(self._assess(worktree_registered=False)["recovery_sanctioned"]) def test_worktree_on_a_different_branch_is_refused(self): self.assertFalse(self._assess(current_branch="master")["recovery_sanctioned"]) def test_head_mismatch_is_refused(self): result = self._assess(declared_head=OTHER_HEAD) self.assertFalse(result["recovery_sanctioned"]) self.assertEqual( result["refusal_code"], bootstrap_lock_recovery.REFUSAL_HEAD_MISMATCH ) def test_unresolvable_identity_is_refused(self): self.assertFalse(self._assess(identity="")["recovery_sanctioned"]) def test_unresolvable_profile_is_refused(self): self.assertFalse(self._assess(profile="")["recovery_sanctioned"]) def test_matching_username_alone_does_not_prove_ownership(self): """Safety: a matching username with the wrong profile is still foreign.""" self.assertFalse(self._assess(profile=FOREIGN_PROFILE)["recovery_sanctioned"]) def test_healthy_foreign_owned_lock_cannot_be_recovered(self): """AC11: the foreign-takeover refusal.""" foreign = canonical_lock(worktree="/scratch/wt-9530") foreign["work_lease"]["claimant"] = { "username": FOREIGN_IDENTITY, "profile": FOREIGN_PROFILE, } foreign["session_pid"] = os.getpid() # alive → healthy result = self._assess(lock=foreign) self.assertFalse(result["recovery_sanctioned"]) self.assertEqual( result["refusal_code"], bootstrap_lock_recovery.REFUSAL_HEALTHY_FOREIGN ) def test_foreign_owned_incomplete_lock_is_also_refused(self): """A foreign lock is refused whether or not it is healthy.""" foreign = bootstrap_shaped_lock( worktree_path="/scratch/wt-9530", claimant={"username": FOREIGN_IDENTITY, "profile": FOREIGN_PROFILE}, ) result = self._assess(lock=foreign) self.assertFalse(result["recovery_sanctioned"]) self.assertIn( result["refusal_code"], { bootstrap_lock_recovery.REFUSAL_FOREIGN_CLAIMANT, bootstrap_lock_recovery.REFUSAL_HEALTHY_FOREIGN, }, ) def test_healthy_same_owner_canonical_lock_is_left_alone(self): """Nothing to recover: rewriting would invalidate a live heartbeat token.""" result = self._assess(lock=canonical_lock(worktree="/scratch/wt-9530")) self.assertFalse(result["recovery_sanctioned"]) self.assertEqual( result["refusal_code"], bootstrap_lock_recovery.REFUSAL_ALREADY_CANONICAL ) def test_absent_lock_is_refused(self): result = self._assess(lock={}) self.assertFalse(result["recovery_sanctioned"]) self.assertEqual(result["refusal_code"], bootstrap_lock_recovery.REFUSAL_NO_LOCK) def test_recovery_never_requires_base_equivalence(self): """AC9: the branch carries commits; that must not be disqualifying.""" import inspect as _inspect params = _inspect.signature( bootstrap_lock_recovery.assess_bootstrap_lock_recovery ).parameters self.assertNotIn("base_equivalent", params) self.assertNotIn("expected_base_sha", params) def test_recovery_accepts_no_caller_supplied_authorization(self): """Safety: no caller-manufactured provenance or authorization.""" import inspect as _inspect params = _inspect.signature( bootstrap_lock_recovery.assess_bootstrap_lock_recovery ).parameters for forbidden in ( "recovery_sanctioned", "lock_provenance", "provenance", "operator_override", "authorized", ): self.assertNotIn(forbidden, params) class RecoveryAuditTrail(unittest.TestCase): """AC10: auditable ownership and generation transition.""" def _assessment(self, **overrides): lock = bootstrap_shaped_lock(worktree_path="/scratch/wt-9530", **overrides) return bootstrap_lock_recovery.assess_bootstrap_lock_recovery( lock, issue_number=ISSUE, branch_name=BRANCH, worktree_path="/scratch/wt-9530", remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE, observed_head=HEAD, declared_head=HEAD, worktree_exists=True, worktree_registered=True, current_branch=BRANCH, ) def test_recovery_record_preserves_both_sides_of_the_transition(self): record = bootstrap_lock_recovery.build_recovery_record( self._assessment(), recovered_at=_ts(), new_task_session_id="task-new" ) self.assertEqual(record["recovery_kind"], "incomplete_bootstrap_lock") self.assertEqual(record["prior_generation"], 1) self.assertEqual( record["prior_owner_session"], "author_issue_work-deadbeefdeadbeef" ) self.assertEqual(record["replacement_task_session_id"], "task-new") self.assertEqual(record["preserved_head"], HEAD) self.assertFalse(record["branch_reset"]) self.assertFalse(record["base_equivalence_required"]) self.assertIn("work_lease", record["prior_missing_fields"]) def test_expected_generation_is_reported_for_compare_and_swap(self): self.assertEqual( self._assessment(lock_generation=7)["expected_generation"], 7 ) class RecoveryIsolationAndPersistence(unittest.TestCase): """AC8, AC16: recovery upgrades only its target and inspection mutates nothing.""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.lock_dir = self.tmp.name def _write(self, lock, **kwargs): return issue_lock_store.bind_session_lock( lock, lock_dir=self.lock_dir, **kwargs ) def test_recovery_upgrades_the_target_lock_to_canonical(self): path = self._write(bootstrap_shaped_lock(worktree_path="/scratch/wt-9530")) before = issue_lock_store.read_lock_file(path) self.assertFalse(author_lock_contract.assess_lock_contract(before)["canonical"]) self._write( author_lock_contract.build_canonical_issue_lock( issue_number=ISSUE, branch_name=BRANCH, worktree_path="/scratch/wt-9530", remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE, tool="gitea_recover_incomplete_bootstrap_lock", ) ) after = issue_lock_store.read_lock_file(path) self.assertTrue(author_lock_contract.assess_lock_contract(after)["canonical"]) self.assertGreater(after["lock_generation"], before["lock_generation"]) def test_recovery_does_not_touch_an_unrelated_lock(self): """A failure or success must affect only the exact target lock.""" other_issue = ISSUE + 77 other_path = self._write( bootstrap_shaped_lock( issue_number=other_issue, branch_name=f"fix/issue-{other_issue}-unrelated", worktree_path="/scratch/wt-other", ) ) other_before = issue_lock_store.read_lock_file(other_path) target_path = self._write( bootstrap_shaped_lock(worktree_path="/scratch/wt-9530") ) self._write( author_lock_contract.build_canonical_issue_lock( issue_number=ISSUE, branch_name=BRANCH, worktree_path="/scratch/wt-9530", remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE, tool="gitea_recover_incomplete_bootstrap_lock", ) ) self.assertNotEqual(target_path, other_path) self.assertEqual(issue_lock_store.read_lock_file(other_path), other_before) def test_inspection_performs_no_mutation(self): """AC16: assessing a lock must not rewrite it.""" path = self._write(bootstrap_shaped_lock(worktree_path="/scratch/wt-9530")) before = issue_lock_store.read_lock_file(path) mtime_before = os.path.getmtime(path) author_lock_contract.assess_lock_contract(before) bootstrap_lock_recovery.assess_bootstrap_lock_recovery( before, issue_number=ISSUE, branch_name=BRANCH, worktree_path="/scratch/wt-9530", remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE, observed_head=HEAD, declared_head=HEAD, worktree_exists=True, worktree_registered=True, current_branch=BRANCH, ) self.assertEqual(issue_lock_store.read_lock_file(path), before) self.assertEqual(os.path.getmtime(path), mtime_before) class HeartbeatOnFreshAndRecoveredLocks(unittest.TestCase): """AC2, AC3: heartbeat and renewal against real durable locks.""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.lock_dir = self.tmp.name self.worktree = os.path.join(self.tmp.name, "wt") os.makedirs(self.worktree, exist_ok=True) def _canonical(self, tool="gitea_bootstrap_author_issue_worktree"): return author_lock_contract.build_canonical_issue_lock( issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.worktree, remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE, tool=tool, ) def _heartbeat(self, token): return issue_lock_store.heartbeat_session_lock( remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.worktree, identity=IDENTITY, profile=PROFILE, task_session_id=token, lock_dir=self.lock_dir, ) def test_a_freshly_built_canonical_lock_can_be_heartbeated_immediately(self): """AC2: the property a bootstrap lock never had.""" lock = self._canonical() issue_lock_store.bind_session_lock(lock, lock_dir=self.lock_dir) outcome = self._heartbeat(lock["work_lease"]["task_session_id"]) self.assertTrue(outcome["success"], outcome.get("reasons")) self.assertTrue(outcome["performed"]) def test_heartbeat_does_not_change_ownership_or_create_a_second_lock(self): lock = self._canonical() path = issue_lock_store.bind_session_lock(lock, lock_dir=self.lock_dir) self._heartbeat(lock["work_lease"]["task_session_id"]) after = issue_lock_store.read_lock_file(path) self.assertEqual( author_lock_contract.lock_claimant(after), {"username": IDENTITY, "profile": PROFILE}, ) # Count issue locks only: bind_session_lock also writes a # session-.json pointer, which is pre-existing behaviour and not a # second claim on the issue. locks = [ f for f in os.listdir(self.lock_dir) if f.endswith(".json") and not f.startswith("session-") ] self.assertEqual(len(locks), 1, f"expected exactly one issue lock, got {locks}") def test_heartbeat_refuses_a_foreign_task_session_token(self): lock = self._canonical(tool="gitea_lock_issue") issue_lock_store.bind_session_lock(lock, lock_dir=self.lock_dir) outcome = self._heartbeat("author_issue_work-someoneelse") self.assertFalse(outcome["success"]) class BootstrapToCreatePrRegression(unittest.TestCase): """AC17, AC18: the exact #949 sequence, against isolated fixtures only. This reproduces the *shape* of the #949 failure — bootstrap, implement, commit, push, create PR — in a throwaway git repository. It never reads or writes the real issue #949 branch, worktree, lock, issue, or head. """ def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.lock_dir = os.path.join(self.tmp.name, "locks") os.makedirs(self.lock_dir, exist_ok=True) self.origin = os.path.join(self.tmp.name, "origin.git") self.repo = os.path.join(self.tmp.name, "repo") subprocess.run( ["git", "init", "--bare", self.origin], check=True, capture_output=True ) subprocess.run(["git", "init", 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", "-m", "base commit") self._git("branch", "-M", "master") self._git("remote", "add", "origin", self.origin) self._git("push", "-u", "origin", "master") def _git(self, *args): return subprocess.run( ["git", "-C", self.repo, *args], check=True, capture_output=True, text=True ) def _heartbeat(self, token): return issue_lock_store.heartbeat_session_lock( remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.repo, identity=IDENTITY, profile=PROFILE, task_session_id=token, lock_dir=self.lock_dir, ) def _implement_and_commit(self): with open(os.path.join(self.repo, "feature.py"), "w") as handle: handle.write("VALUE = 1\n") self._git("add", "feature.py") self._git("commit", "-m", "feat: implement the issue") def test_bootstrap_implement_commit_push_create_pr_completes(self): # 1. Bootstrap: branch, worktree, and a canonical lock. self._git("checkout", "-b", BRANCH) lock = author_lock_contract.build_canonical_issue_lock( issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.repo, remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE, tool="gitea_bootstrap_author_issue_worktree", source=author_lock_contract.SOURCE_BOOTSTRAP, ) path = issue_lock_store.bind_session_lock(lock, lock_dir=self.lock_dir) token = lock["work_lease"]["task_session_id"] # The lock is usable the moment bootstrap returns. self.assertTrue( author_lock_contract.assess_lock_contract( issue_lock_store.read_lock_file(path) )["canonical"] ) # 2. Renewal is available *before* the branch diverges (AC3). before = self._heartbeat(token) self.assertTrue(before["success"], before.get("reasons")) # 3. Implement and commit — the branch now carries real work. self._implement_and_commit() head = self._git("rev-parse", "HEAD").stdout.strip() # 4. Push. self._git("push", "-u", "origin", BRANCH) 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. Renewal still available *after* commits (AC3) — no base-equivalence. after = self._heartbeat(token) self.assertTrue(after["success"], after.get("reasons")) # 6. create_pr's #447 provenance guard accepts the lock (AC4). final = issue_lock_store.read_lock_file(path) verdict = issue_lock_provenance.assess_lock_file_for_create_pr(final) self.assertTrue(verdict["proven"], verdict["reasons"]) # AC18: completed with no manual lock edit, no branch rewind, no # fallback transport. The base commit is still an ancestor of head. merge_base = self._git("merge-base", "master", BRANCH).stdout.strip() master_head = self._git("rev-parse", "master").stdout.strip() self.assertEqual(merge_base, master_head) def test_the_old_bootstrap_shape_reproduces_the_949_dead_end(self): """The regression must actually fail without the fix.""" self._git("checkout", "-b", BRANCH) self._implement_and_commit() legacy = bootstrap_shaped_lock(worktree_path=self.repo) issue_lock_store.bind_session_lock(legacy, lock_dir=self.lock_dir) # create_pr refuses — the #447 guard, unchanged. self.assertFalse( issue_lock_provenance.assess_lock_file_for_create_pr(legacy)["proven"] ) # And it is never classified as expired, so renewal never engages. self.assertFalse(issue_lock_store.is_lease_expired(legacy)) self.assertEqual( author_lock_contract.expiration_state(legacy)["state"], author_lock_contract.EXPIRATION_MISSING, ) def test_recovery_of_a_committed_branch_preserves_the_commits(self): """AC9: recovery must not rewind a branch that carries pushed work.""" self._git("checkout", "-b", BRANCH) self._implement_and_commit() self._git("push", "-u", "origin", BRANCH) head_before = self._git("rev-parse", "HEAD").stdout.strip() legacy = bootstrap_shaped_lock(worktree_path=self.repo) path = issue_lock_store.bind_session_lock(legacy, lock_dir=self.lock_dir) assessment = bootstrap_lock_recovery.assess_bootstrap_lock_recovery( issue_lock_store.read_lock_file(path), issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.repo, remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE, observed_head=head_before, declared_head=head_before, worktree_exists=True, worktree_registered=True, current_branch=BRANCH, ) self.assertTrue(assessment["recovery_sanctioned"], assessment["reasons"]) recovered = author_lock_contract.build_canonical_issue_lock( issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.repo, remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE, tool="gitea_recover_incomplete_bootstrap_lock", ) recovered["bootstrap_lock_recovery"] = ( bootstrap_lock_recovery.build_recovery_record( assessment, recovered_at=_ts(), new_task_session_id=recovered["work_lease"]["task_session_id"], ) ) issue_lock_store.bind_session_lock( recovered, lock_dir=self.lock_dir, expected_generation=assessment["expected_generation"], recovery_sanctioned=True, ) # The branch head is untouched and the lock is now canonical. self.assertEqual(self._git("rev-parse", "HEAD").stdout.strip(), head_before) final = issue_lock_store.read_lock_file(path) self.assertTrue(author_lock_contract.assess_lock_contract(final)["canonical"]) self.assertTrue( issue_lock_provenance.assess_lock_file_for_create_pr(final)["proven"] ) self.assertEqual( final["bootstrap_lock_recovery"]["preserved_head"], head_before ) self.assertFalse(final["bootstrap_lock_recovery"]["branch_reset"]) class BootstrapWiring(unittest.TestCase): """AC1, AC5, AC7, AC15: what the bootstrap tool itself now does.""" def _source(self): import author_issue_bootstrap with open(author_issue_bootstrap.__file__) as handle: return handle.read() def test_bootstrap_builds_through_the_canonical_contract(self): source = self._source() self.assertIn("author_lock_contract.build_canonical_issue_lock", source) self.assertIn("author_lock_contract.assess_lock_contract", source) def test_bootstrap_fails_closed_on_an_incomplete_written_lock(self): """AC7: partial lock creation stops before implementation begins.""" source = self._source() self.assertIn("incomplete_issue_lock_contract", source) self.assertIn('"implementation_allowed": False', source) def test_next_action_for_a_canonical_lock_directs_to_implementation(self): """AC5: executable under the state actually returned.""" assessment = author_lock_contract.assess_lock_contract(canonical_lock()) self.assertIn("work_issue", author_lock_contract.recommended_action(assessment)) def test_next_action_for_an_incomplete_lock_forbids_implementation(self): """AC5, AC15: never direct an author into the unrecoverable state.""" assessment = author_lock_contract.assess_lock_contract(bootstrap_shaped_lock()) action = author_lock_contract.recommended_action(assessment) self.assertIn("Do not begin implementation", action) self.assertIn("gitea_recover_incomplete_bootstrap_lock", action) class CapabilityRegistration(unittest.TestCase): """The new operations are registered and role-gated.""" def test_recovery_is_registered_as_an_author_operation(self): import task_capability_map self.assertEqual( task_capability_map.required_permission( "recover_incomplete_bootstrap_lock" ), "gitea.issue.comment", ) self.assertEqual( task_capability_map.required_role("recover_incomplete_bootstrap_lock"), "author", ) def test_inspection_requires_only_read_permission(self): """AC16: a read-only surface must not carry a mutating permission.""" import task_capability_map self.assertEqual( task_capability_map.required_permission("inspect_issue_lock_contract"), "gitea.read", ) class WorkflowDocumentation(unittest.TestCase): """AC20: the canonical ordering and recovery path are documented.""" def test_author_workflow_documents_ordering_and_recovery(self): root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) doc = os.path.join(root, "docs", "author-issue-lock-contract.md") self.assertTrue(os.path.exists(doc), f"missing {doc}") with open(doc) as handle: text = handle.read() self.assertIn("gitea_recover_incomplete_bootstrap_lock", text) self.assertIn("gitea_lock_issue", text) self.assertIn("gitea_bootstrap_author_issue_worktree", text) self.assertIn("before writing any implementation", text.lower()) if __name__ == "__main__": unittest.main()