Gitea-Tools/tests/test_issue_953_bootstrap_lock_contract.py

"""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-<pid>.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", "[email protected]")
        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()