feat: replace global issue lock with keyed persistent store (Closes #443 )

Store per remote/org/repo/issue locks under GITEA_ISSUE_LOCK_DIR with atomic writes and per-session binding. Integrate own-branch adoption for lock recovery, update worktree-start and cleanup reconcile, and add tests documenting the ban on manual global lock seeding. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 16:45:37 -04:00
13 changed files with 17 additions and 835 deletions
@@ -293,43 +293,6 @@ adoption rebinds the session when the issue's exact branch already exists (#442)
 `gitea_create_pr` resolves the durable keyed lock by session pointer or by
 matching `head` branch without unsafe manual seeding.
 **Issue-lock recovery (#447):** Do not manually seed, restore, or delete
 `/tmp/gitea_issue_lock.json` as a normal recovery path. That file is global
 shared state and manual writes can clobber another session's live lease. Use
 `sanctioned recovery` instead:
 1. `gitea_lock_issue` on a clean `branches/` worktree (normal path).
 2. Own-branch adoption via #442 when the issue's exact branch is already pushed.
 3. Operator override only when explicitly authorized — record
   `External-state mutations` and `operator override proof` in the final report.
 `gitea_create_pr` rejects lock files that lack sanctioned `lock_provenance`
 metadata. Final-report validation blocks handoffs that hide lock read/write/delete
 under `External-state mutations: none` or mix author PR creation with reviewer
 approval in one run. See also #438 (global lock redesign).
 ### Non-destructive lock recovery after push (#440)
 When work is pushed but the in-memory MCP session lock is lost (server restart,
 crashed process, or stale session pointer):
 1. **Do not delete the remote branch.** Branch deletion is not a normal recovery
   step and can destroy the only copy of unmerged work.
 2. **Re-run `gitea_lock_issue`** with the same `issue_number`, exact
   `branch_name`, and active `branches/` worktree. Own-branch adoption
   reacquires the lease when the remote branch is the caller's exact branch and
   no open PR or competing same-issue branch exists.
 3. **Call `gitea_create_pr`** with the same `head` branch. The server resolves
   the durable keyed lock file even when the session pointer was cleared at
   restart.
 4. **Stop fail-closed** when another actor owns a competing branch, an open PR
   already exists, or a live foreign lease blocks takeover — never adopt their
   branch.
 Branch ownership uses structured evidence
 `(fix|feat|docs|chore)/issue-<n>-<desc>` — not broad substring matches on
 `issue-<n>`.
 Remote branches matching the issue number are also treated as active work unless
 the recovery review proves the branch is abandoned or superseded. Never delete
 or clean up a branch when it has an active lease, dirty worktree, open PR, or is
@@ -11,7 +11,6 @@ import inspect
 import re
 from typing import Any, Callable
 import issue_lock_provenance
 from review_proofs import (
    HANDOFF_HEADING,
    assess_controller_handoff,
@@ -871,54 +870,6 @@ def _rule_reviewer_mutation_ledger(
    )
 def _rule_shared_issue_lock_external_state(report_text: str) -> list[dict[str, str]]:
    result = issue_lock_provenance.assess_issue_lock_external_state_report(report_text)
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "shared.issue_lock_external_state",
        result.get("reasons") or [],
        field="External-state mutations",
        severity="block",
        safe_next_action=(
            "disclose gitea_issue_lock.json read/write/delete under "
            "External-state mutations; never claim none after lock seeding"
        ),
    )
 def _rule_shared_manual_lock_pr_override(report_text: str) -> list[dict[str, str]]:
    result = issue_lock_provenance.assess_manual_lock_pr_without_override(report_text)
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "shared.manual_lock_pr_override",
        result.get("reasons") or [],
        field="External-state mutations",
        severity="block",
        safe_next_action=(
            "use gitea_lock_issue or #442 adoption instead of manual lock seeding; "
            "if operator override was authorized, cite override proof"
        ),
    )
 def _rule_shared_author_reviewer_same_run(report_text: str) -> list[dict[str, str]]:
    result = issue_lock_provenance.assess_author_reviewer_same_run_report(report_text)
    if result.get("proven"):
        return []
    return _findings_from_reasons(
        "shared.author_reviewer_same_run",
        result.get("reasons") or [],
        field="Review mutations",
        severity="block",
        safe_next_action=(
            "split author PR creation and reviewer approval across separate "
            "sessions and handoffs"
        ),
    )
 def _rule_reviewer_review_mutation(
    report_text: str,
    *,
@@ -938,17 +889,10 @@ def _rule_reviewer_review_mutation(
    )
 _SHARED_ISSUE_LOCK_RULES = (
    _rule_shared_issue_lock_external_state,
    _rule_shared_manual_lock_pr_override,
    _rule_shared_author_reviewer_same_run,
 )
 _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
    "review_pr": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_reviewer_legacy_workspace_mutations,
        _rule_reviewer_vague_mutations_none,
        _rule_reviewer_mutation_categories,
@@ -969,7 +913,6 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
    "reconcile_already_landed": [
        _rule_reconcile_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_reconcile_stale_author_fields,
        _rule_reconcile_eligible_reviewed,
        _rule_reconcile_linked_issue_live,
@@ -981,30 +924,25 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
    "author_issue": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_reviewer_vague_mutations_none,
    ],
    "work_issue": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_reviewer_vague_mutations_none,
    ],
    "issue_filing": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
    ],
    "inventory": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
        _rule_reconcile_pagination_proof,
    ],
    "issue_selection": [
        _rule_shared_controller_handoff,
        _rule_shared_email_disclosure,
        *_SHARED_ISSUE_LOCK_RULES,
    ],
 }
@@ -538,10 +538,8 @@ import task_capability_map  # noqa: E402
 import review_proofs  # noqa: E402
 import agent_temp_artifacts
 import issue_lock_worktree  # noqa: E402
 import issue_lock_provenance  # noqa: E402
 import issue_lock_store  # noqa: E402
 import issue_lock_adoption  # noqa: E402
 import issue_branch_ownership  # noqa: E402
 import already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
@@ -1287,11 +1285,11 @@ def gitea_lock_issue(
        worktree_path: Author scratch-clone path to validate (defaults to
            GITEA_AUTHOR_WORKTREE or the MCP server project root).
    """
-    # 1. Enforce canonical issue branch ownership (#440)
+    # 1. Enforce branch name includes issue number
-    if not issue_branch_ownership.branch_belongs_to_issue(branch_name, issue_number):
+    expected_pattern = f"issue-{issue_number}"
    if expected_pattern not in branch_name:
        raise ValueError(
-            f"Branch name '{branch_name}' must match "
+            f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)"
            f"(fix|feat|docs|chore)/issue-{issue_number}-<desc> (fail closed)"
        )
    blocked = _profile_permission_block(
@@ -1381,10 +1379,6 @@ def gitea_lock_issue(
        "repo": r,
        "worktree_path": resolved_worktree,
        "work_lease": work_lease,
        "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
            tool="gitea_lock_issue",
            claimant=work_lease.get("claimant"),
        ),
    }
    lock_file_path = _save_issue_lock(data)
@@ -1512,14 +1506,6 @@ def gitea_create_pr(
    # ── Issue Lock Validation (Issue #194 / #196 / #443) ──
    lock_data = _resolve_issue_lock_for_pr(remote=remote, org=o, repo=r, head=head)
    lock_provenance_check = issue_lock_provenance.assess_lock_file_for_create_pr(
        lock_data
    )
    if lock_provenance_check["block"]:
        raise RuntimeError(
            issue_lock_provenance.format_lock_provenance_error(lock_provenance_check)
        )
    locked_issue = lock_data.get("issue_number")
    locked_branch = lock_data.get("branch_name")
    locked_worktree = lock_data.get("worktree_path")
@@ -1,28 +0,0 @@
 """Structured issue/branch ownership evidence (#440).
 Author branches must follow ``(fix|feat|docs|chore)/issue-<n>-<desc>``. Duplicate-work
 and lock-recovery gates use this parser instead of broad substring matching on
 ``issue-<n>`` so unrelated branch names cannot false-positive.
 """
 from __future__ import annotations
 import re
 ISSUE_BRANCH_RE = re.compile(
    r"^(?P<prefix>fix|feat|docs|chore)/issue-(?P<num>\d+)(?:-|$)"
 )
 def parse_issue_branch(branch_name: str) -> int | None:
    """Return the issue number encoded in a canonical author branch, else None."""
    match = ISSUE_BRANCH_RE.match((branch_name or "").strip())
    if not match:
        return None
    return int(match.group("num"))
 def branch_belongs_to_issue(branch_name: str, issue_number: int) -> bool:
    """True when ``branch_name`` is a canonical branch for ``issue_number``."""
    parsed = parse_issue_branch(branch_name)
    return parsed is not None and parsed == issue_number
@@ -1,4 +1,4 @@
-"""Own-branch lock adoption / recovery for ``gitea_lock_issue`` (#440 / #442 / #443).
+"""Own-branch lock adoption / recovery for ``gitea_lock_issue`` (#442 / #443).
 When an issue's own already-pushed branch exists, lock reacquisition must be
 allowed (adoption) instead of being treated as #400 duplicate competing work.
@@ -14,8 +14,6 @@ this module additionally records whether they passed for proof purposes.
 from __future__ import annotations
 import issue_branch_ownership
 ADOPT = "adopt_existing_branch"
 BLOCK_COMPETING = "block_competing_branch"
 NO_MATCH = "no_matching_branch"
@@ -42,12 +40,13 @@ def assess_own_branch_adoption(
    existing_branches,
 ) -> dict:
    """Decide whether an existing matching branch is adoptable."""
    marker = f"issue-{issue_number}"
    requested = (requested_branch or "").strip()
    matches: list[tuple[str, str | None]] = []
    for entry in existing_branches or []:
        name = _branch_name(entry).strip()
-        if issue_branch_ownership.branch_belongs_to_issue(name, issue_number):
+        if marker in name:
            matches.append((name, _branch_sha(entry)))
    competing = sorted({name for name, _ in matches if name != requested})
@@ -1,269 +0,0 @@
 """Issue-lock provenance and external-state disclosure (#447).
 Sanctioned locks are written only by ``gitea_lock_issue`` (or adoption recovery
 #442). Manual seeding of ``/tmp/gitea_issue_lock.json`` is unsafe and must be
 blocked at PR creation unless explicit operator override proof is recorded.
 """
 from __future__ import annotations
 import os
 import re
 from datetime import datetime, timezone
 ISSUE_LOCK_FILE = os.environ.get("GITEA_ISSUE_LOCK_FILE", "/tmp/gitea_issue_lock.json")
 SOURCE_LOCK_ISSUE = "gitea_lock_issue"
 SOURCE_LOCK_ADOPTION = "gitea_lock_issue_adoption"
 SOURCE_OPERATOR_OVERRIDE = "operator_override"
 SANCTIONED_LOCK_SOURCES = frozenset({
    SOURCE_LOCK_ISSUE,
    SOURCE_LOCK_ADOPTION,
    SOURCE_OPERATOR_OVERRIDE,
 })
 _OPERATOR_OVERRIDE_ENV = "GITEA_ISSUE_LOCK_OPERATOR_OVERRIDE"
 _ISSUE_LOCK_PATH_RE = re.compile(
    r"(?:/tmp/)?gitea_issue_lock\.json",
    re.IGNORECASE,
 )
 _LOCK_SEED_RE = re.compile(
    r"(?:seed(?:ed|ing)?|restor(?:e|ed|ing)|wrote|written|write|programmatically|"
    r"hand[- ]forg|manual(?:ly)?).{0,80}gitea_issue_lock",
    re.IGNORECASE | re.DOTALL,
 )
 _LOCK_REMOVE_RE = re.compile(
    r"(?:\brm\b|remove|deleted?|unlink).{0,80}gitea_issue_lock",
    re.IGNORECASE | re.DOTALL,
 )
 _LOCK_READ_RE = re.compile(
    r"(?:read|loaded?|parsed?).{0,80}gitea_issue_lock",
    re.IGNORECASE | re.DOTALL,
 )
 _EXTERNAL_NONE_RE = re.compile(
    r"external[- ]state mutations\s*:\s*none\b",
    re.IGNORECASE,
 )
 _EXTERNAL_FIELD_RE = re.compile(
    r"external[- ]state mutations\s*:\s*(.+)$",
    re.IGNORECASE | re.MULTILINE,
 )
 _CLEANUP_ONLY_RE = re.compile(
    r"cleanup mutations\s*:\s*(?:none|lock removed|removed issue lock)",
    re.IGNORECASE,
 )
 _PR_CREATED_RE = re.compile(
    r"(?:\bgitea_create_pr\b|PR\s*#\s*\d+\s+created|created\s+PR\s*#|opened\s+PR\s*#|"
    r"PR\s+creation\s+(?:succeeded|complete))",
    re.IGNORECASE,
 )
 _REVIEW_APPROVE_RE = re.compile(
    r"(?:submitted\s+(?:['\"]approve['\"]|approve\s+review)|"
    r"review decision\s*:\s*approve|approved\s+PR\s*#|gitea_review_pr.*approve)",
    re.IGNORECASE,
 )
 _OVERRIDE_PROOF_RE = re.compile(
    r"operator[- ]override\s+proof\s*:\s*(.+)$",
    re.IGNORECASE | re.MULTILINE,
 )
 def _utc_now_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
 def build_sanctioned_lock_provenance(
    *,
    tool: str,
    source: str = SOURCE_LOCK_ISSUE,
    claimant: dict | None = None,
    adoption: dict | None = None,
 ) -> dict:
    """Return provenance metadata stored with a sanctioned lock write."""
    entry = {
        "source": source,
        "written_at": _utc_now_iso(),
        "written_by_tool": tool,
        "lock_file_path": ISSUE_LOCK_FILE,
    }
    if claimant:
        entry["claimant"] = claimant
    if adoption:
        entry["adoption"] = adoption
    return entry
 def operator_override_requested() -> bool:
    return os.environ.get(_OPERATOR_OVERRIDE_ENV, "").strip().lower() in {
        "1",
        "true",
        "yes",
    }
 def build_operator_override_provenance(*, reason: str, claimant: dict | None = None) -> dict:
    text = (reason or "").strip()
    if not text:
        raise ValueError(
            "operator override requires a non-empty override reason (fail closed)"
        )
    entry = build_sanctioned_lock_provenance(
        tool="operator_override",
        source=SOURCE_OPERATOR_OVERRIDE,
        claimant=claimant,
    )
    entry["override_reason"] = text
    return entry
 def assess_lock_file_for_create_pr(lock_data: dict | None) -> dict:
    """Fail closed when lock file lacks sanctioned provenance (#447)."""
    data = lock_data if isinstance(lock_data, dict) else {}
    reasons: list[str] = []
    provenance = data.get("lock_provenance")
    if not isinstance(provenance, dict):
        reasons.append(
            "issue lock file lacks sanctioned lock_provenance; manual seeding is "
            "not a normal recovery path — call gitea_lock_issue or use #442 adoption"
        )
        return _provenance_result(False, reasons, provenance)
    source = str(provenance.get("source") or "").strip()
    if source not in SANCTIONED_LOCK_SOURCES:
        reasons.append(
            f"issue lock provenance source '{source or '(missing)'}' is not sanctioned"
        )
    if source == SOURCE_OPERATOR_OVERRIDE and not str(
        provenance.get("override_reason") or ""
    ).strip():
        reasons.append(
            "operator_override lock provenance requires override_reason proof"
        )
    if not data.get("work_lease"):
        reasons.append("issue lock file missing work_lease metadata")
    if not str(provenance.get("written_by_tool") or "").strip():
        reasons.append("issue lock provenance missing written_by_tool")
    proven = not reasons
    return _provenance_result(proven, reasons, provenance)
 def _provenance_result(proven: bool, reasons: list[str], provenance: dict | None) -> dict:
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "lock_provenance": provenance,
    }
 def format_lock_provenance_error(assessment: dict) -> str:
    reasons = "; ".join(assessment.get("reasons") or ["unknown lock provenance violation"])
    return f"Issue lock provenance guard (#447): {reasons} (fail closed)"
 def _lock_activity_detected(text: str) -> dict[str, bool]:
    body = text or ""
    return {
        "seed_or_restore": bool(_LOCK_SEED_RE.search(body)),
        "remove": bool(_LOCK_REMOVE_RE.search(body)),
        "read": bool(_LOCK_READ_RE.search(body)),
    }
 def _external_state_discloses_lock(text: str) -> bool:
    match = _EXTERNAL_FIELD_RE.search(text or "")
    if not match:
        return False
    value = (match.group(1) or "").strip().lower()
    if value in {"", "none", "n/a"}:
        return False
    return "lock" in value or "gitea_issue_lock" in value or "issue-lock" in value
 def assess_issue_lock_external_state_report(report_text: str) -> dict:
    """Require explicit external-state disclosure for issue-lock mutations (#447)."""
    text = report_text or ""
    activity = _lock_activity_detected(text)
    if not any(activity.values()):
        return {"proven": True, "block": False, "reasons": [], "activity": activity}
    reasons: list[str] = []
    disclosed = _external_state_discloses_lock(text)
    if activity["seed_or_restore"] and _EXTERNAL_NONE_RE.search(text):
        reasons.append(
            "report mentions seeding/restoring gitea_issue_lock.json but claims "
            "External-state mutations: none"
        )
    elif activity["seed_or_restore"] and not disclosed:
        reasons.append(
            "report mentions issue-lock file activity but External-state mutations "
            "does not disclose read/write of gitea_issue_lock.json"
        )
    if activity["remove"]:
        if _EXTERNAL_NONE_RE.search(text):
            reasons.append(
                "report mentions removing gitea_issue_lock.json but claims "
                "External-state mutations: none"
            )
        elif not disclosed and _CLEANUP_ONLY_RE.search(text):
            reasons.append(
                "report removes issue lock but classifies it as cleanup only; "
                "record under External-state mutations"
            )
        elif not disclosed:
            reasons.append(
                "report mentions deleting issue lock without External-state "
                "mutation disclosure"
            )
    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "activity": activity,
    }
 def assess_manual_lock_pr_without_override(report_text: str) -> dict:
    """Block reports that created a PR via manual lock seed without override proof."""
    text = report_text or ""
    seeded = bool(_LOCK_SEED_RE.search(text))
    created = bool(_PR_CREATED_RE.search(text))
    if not (seeded and created):
        return {"proven": True, "block": False, "reasons": []}
    if _OVERRIDE_PROOF_RE.search(text):
        return {"proven": True, "block": False, "reasons": []}
    return {
        "proven": False,
        "block": True,
        "reasons": [
            "report created/opened a PR after manual issue-lock seeding without "
            "operator override proof"
        ],
    }
 def assess_author_reviewer_same_run_report(report_text: str) -> dict:
    """Reviewer handoff must not create and approve the same PR in one run (#447)."""
    text = report_text or ""
    if not (_PR_CREATED_RE.search(text) and _REVIEW_APPROVE_RE.search(text)):
        return {"proven": True, "block": False, "reasons": []}
    return {
        "proven": False,
        "block": True,
        "reasons": [
            "report mixes author-side PR creation and reviewer approval in one "
            "final handoff; split author and reviewer sessions"
        ],
    }
@@ -747,12 +747,6 @@ Use only precise categories:
 * External-state mutations:
 * Read-only diagnostics:
 Issue-lock file (`/tmp/gitea_issue_lock.json`) read/write/delete is always an
 external-state mutation. Never claim `External-state mutations: none` after
 seeding, restoring, or removing that file. Manual lock seeding is not a normal
 recovery path (#447); use `gitea_lock_issue` or the #442 adoption recovery path
 instead. Link broader redesign: #438.
 `git fetch`, `git remote update`, and any command that updates refs must be listed under `Git ref mutations`, not read-only diagnostics.
 If `git reset --hard`, checkout, clean, worktree add/remove, merge simulation, merge abort, or similar commands occurred, report them under `Worktree/index mutations`.
@@ -67,18 +67,9 @@ class TestCommitPayloads(unittest.TestCase):
        self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name)
        import issue_lock_store
        import issue_lock_provenance
        self._lock_dir = tempfile.TemporaryDirectory()
        os.environ["GITEA_ISSUE_LOCK_DIR"] = self._lock_dir.name
        work_lease = {
            "operation_type": "author_issue_work",
            "issue_number": 263,
            "branch": "feat/issue-263-native-commit-payloads",
            "claimant": {"username": "test-user", "profile": "test-author"},
            "expires_at": "2999-01-01T00:00:00Z",
        }
        self.lock_data = {
            "issue_number": 263,
            "branch_name": "feat/issue-263-native-commit-payloads",
@@ -86,11 +77,10 @@ class TestCommitPayloads(unittest.TestCase):
            "org": "Example-Org",
            "repo": "Example-Repo",
            "worktree_path": self.locked_worktree_path,
-            "work_lease": work_lease,
+            "work_lease": {
-            "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
+                "operation_type": "author_issue_work",
-                tool="gitea_lock_issue",
+                "expires_at": "2999-01-01T00:00:00Z",
-                claimant=work_lease.get("claimant"),
+            },
            ),
        }
        self.lock_file_path = issue_lock_store.bind_session_lock(self.lock_data)
@@ -1,36 +0,0 @@
 """Structured issue branch ownership (#440)."""
 import sys
 import unittest
 from pathlib import Path
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 import issue_branch_ownership as ibo  # noqa: E402
 class TestIssueBranchOwnership(unittest.TestCase):
    def test_canonical_branch_parses_issue_number(self):
        self.assertEqual(
            ibo.parse_issue_branch("feat/issue-420-server-code-parity"),
            420,
        )
    def test_prefix_variants_supported(self):
        for prefix in ("fix", "feat", "docs", "chore"):
            branch = f"{prefix}/issue-440-lock-recovery"
            self.assertTrue(ibo.branch_belongs_to_issue(branch, 440))
    def test_substring_false_positive_rejected(self):
        self.assertFalse(
            ibo.branch_belongs_to_issue("feat/my-issue-420-backport", 420)
        )
        self.assertFalse(ibo.branch_belongs_to_issue("release/issue-420-hotfix", 420))
    def test_different_issue_number_rejected(self):
        self.assertFalse(
            ibo.branch_belongs_to_issue("feat/issue-421-server-code-parity", 420)
        )
 if __name__ == "__main__":
    unittest.main()
@@ -43,15 +43,6 @@ class TestAssessOwnBranchAdoption(unittest.TestCase):
        )
        self.assertEqual(result["outcome"], NO_MATCH)
    def test_substring_only_branch_name_is_ignored(self):
        result = assess_own_branch_adoption(
            issue_number=420,
            requested_branch=REQ,
            existing_branches=[{"name": "feat/my-issue-420-backport"}],
        )
        self.assertEqual(result["outcome"], NO_MATCH)
        self.assertFalse(result["block"])
 class TestBuildAdoptionProof(unittest.TestCase):
    def test_proof_has_required_fields(self):
@@ -1,130 +0,0 @@
 """Tests for issue-lock provenance and external-state disclosure (#447)."""
 from __future__ import annotations
 import sys
 import unittest
 from pathlib import Path
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 import issue_lock_provenance as ilp  # noqa: E402
 from final_report_validator import assess_final_report_validator  # noqa: E402
 def _sanctioned_lock(**overrides):
    work_lease = {
        "operation_type": "author_issue_work",
        "issue_number": 447,
        "branch": "feat/issue-447-lock-provenance",
        "claimant": {"username": "jcwalker3", "profile": "prgs-author"},
        "expires_at": "2999-01-01T00:00:00Z",
    }
    data = {
        "issue_number": 447,
        "branch_name": "feat/issue-447-lock-provenance",
        "work_lease": work_lease,
        "lock_provenance": ilp.build_sanctioned_lock_provenance(
            tool="gitea_lock_issue",
            claimant=work_lease["claimant"],
        ),
    }
    data.update(overrides)
    return data
 class TestLockProvenanceForCreatePr(unittest.TestCase):
    def test_sanctioned_lock_passes(self):
        result = ilp.assess_lock_file_for_create_pr(_sanctioned_lock())
        self.assertTrue(result["proven"])
        self.assertFalse(result["block"])
    def test_manual_seed_without_provenance_blocked(self):
        result = ilp.assess_lock_file_for_create_pr(
            {"issue_number": 420, "branch_name": "feat/x", "work_lease": {}}
        )
        self.assertTrue(result["block"])
        self.assertIn("lock_provenance", result["reasons"][0])
    def test_operator_override_requires_reason(self):
        result = ilp.assess_lock_file_for_create_pr(
            _sanctioned_lock(
                lock_provenance=ilp.build_sanctioned_lock_provenance(
                    tool="operator_override",
                    source=ilp.SOURCE_OPERATOR_OVERRIDE,
                )
            )
        )
        self.assertTrue(result["block"])
 class TestExternalStateReportRules(unittest.TestCase):
    def test_seed_with_external_none_blocked(self):
        report = (
            "Restored /tmp/gitea_issue_lock.json to unblock PR creation.\n"
            "- External-state mutations: none\n"
        )
        result = ilp.assess_issue_lock_external_state_report(report)
        self.assertTrue(result["block"])
    def test_seed_with_disclosure_passes(self):
        report = (
            "Restored /tmp/gitea_issue_lock.json after MCP restart.\n"
            "- External-state mutations: wrote /tmp/gitea_issue_lock.json\n"
        )
        result = ilp.assess_issue_lock_external_state_report(report)
        self.assertTrue(result["proven"])
    def test_remove_claimed_as_cleanup_only_blocked(self):
        report = (
            "rm /tmp/gitea_issue_lock.json after PR creation.\n"
            "- Cleanup mutations: lock removed\n"
            "- External-state mutations: none\n"
        )
        result = ilp.assess_issue_lock_external_state_report(report)
        self.assertTrue(result["block"])
    def test_manual_lock_pr_without_override_blocked(self):
        report = (
            "Programmatically seeded gitea_issue_lock.json then gitea_create_pr.\n"
            "PR #444 created.\n"
        )
        result = ilp.assess_manual_lock_pr_without_override(report)
        self.assertTrue(result["block"])
    def test_author_reviewer_same_run_blocked(self):
        report = (
            "gitea_create_pr opened PR #444.\n"
            "Submitted approve review on PR #444.\n"
        )
        result = ilp.assess_author_reviewer_same_run_report(report)
        self.assertTrue(result["block"])
 class TestFinalReportValidatorIntegration(unittest.TestCase):
    def test_work_issue_blocks_hidden_lock_mutation(self):
        report = (
            "## Controller Handoff\n"
            "- Task: work issue #420\n"
            "- External-state mutations: none\n"
            "Restored /tmp/gitea_issue_lock.json before PR creation.\n"
        )
        result = assess_final_report_validator(report, "work_issue")
        rule_ids = {f["rule_id"] for f in result["findings"]}
        self.assertIn("shared.issue_lock_external_state", rule_ids)
        self.assertTrue(result["blocked"])
    def test_review_pr_blocks_create_and_approve(self):
        report = (
            "## Controller Handoff\n"
            "- Task: review PR #444\n"
            "- Review decision: approve\n"
            "Created PR #444 via gitea_create_pr earlier in this run.\n"
        )
        result = assess_final_report_validator(report, "review_pr")
        rule_ids = {f["rule_id"] for f in result["findings"]}
        self.assertIn("shared.author_reviewer_same_run", rule_ids)
 if __name__ == "__main__":
    unittest.main()
@@ -1,170 +0,0 @@
 """End-to-end lock recovery scenarios for issue #440."""
 import os
 import sys
 import tempfile
 import unittest
 from pathlib import Path
 from unittest.mock import patch
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 import issue_lock_adoption  # noqa: E402
 import issue_lock_store as ils  # noqa: E402
 import mcp_server  # noqa: E402
 from mcp_server import gitea_create_pr, gitea_lock_issue  # noqa: E402
 PRGS_REPO = mcp_server.REMOTES["prgs"]["repo"]
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 BRANCH = "feat/issue-420-server-code-parity"
 ISSUE_WRITE_ENV = {
    "GITEA_ALLOWED_OPERATIONS": (
        "gitea.issue.create,gitea.issue.close,gitea.issue.comment"
    ),
 }
 CREATE_PR_ENV = {
    "GITEA_PROFILE_NAME": "author-test",
    "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.create,gitea.branch.push",
    "GITEA_FORBIDDEN_OPERATIONS": "gitea.pr.approve,gitea.pr.merge,gitea.pr.review",
 }
 def _clean_master_git_state_for_lock():
    return {
        "current_branch": "master",
        "porcelain_status": "",
        "base_equivalent": True,
        "inspected_git_root": "/tmp/repo",
        "base_branch": "master",
    }
 def _lock_record(**overrides):
    import issue_lock_provenance
    work_lease = {
        "operation_type": "author_issue_work",
        "expires_at": "2999-01-01T00:00:00Z",
    }
    record = {
        "issue_number": 420,
        "branch_name": BRANCH,
        "remote": "prgs",
        "org": "Scaled-Tech-Consulting",
        "repo": PRGS_REPO,
        "worktree_path": os.path.realpath(os.getcwd()),
        "work_lease": work_lease,
        "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
            tool="gitea_lock_issue",
            claimant=work_lease.get("claimant"),
        ),
    }
    record.update(overrides)
    return record
 class TestIssueLockRecovery(unittest.TestCase):
    def setUp(self):
        self._tmpdir = tempfile.TemporaryDirectory()
        self.addCleanup(self._tmpdir.cleanup)
        self.lock_dir = self._tmpdir.name
    def test_substring_branch_does_not_trigger_adoption(self):
        result = issue_lock_adoption.assess_own_branch_adoption(
            issue_number=420,
            requested_branch=BRANCH,
            existing_branches=[{"name": "feat/my-issue-420-backport"}],
        )
        self.assertEqual(result["outcome"], issue_lock_adoption.NO_MATCH)
    def test_competing_actor_branch_blocks_adoption(self):
        result = issue_lock_adoption.assess_own_branch_adoption(
            issue_number=420,
            requested_branch=BRANCH,
            existing_branches=[{"name": "feat/issue-420-other-work"}],
        )
        self.assertTrue(result["block"])
    @patch(
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value=_clean_master_git_state_for_lock(),
    )
    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    @patch(
        "mcp_server.issue_duplicate_context_fetcher",
        return_value=([], [], {"status": "not_claimed"}),
    )
    def test_lock_recovery_after_restart_adopts_own_branch(self, _dup_fetcher, _auth, mock_api, _git):
        mock_api.return_value = [{"name": BRANCH, "commit": {"id": "934688a"}}]
        env = {**ISSUE_WRITE_ENV, "GITEA_ISSUE_LOCK_DIR": self.lock_dir}
        with patch.dict(os.environ, env, clear=True):
            with patch("os.getpid", return_value=9999):
                res = gitea_lock_issue(
                    issue_number=420,
                    branch_name=BRANCH,
                    remote="prgs",
                )
        self.assertTrue(res["success"])
        self.assertIn("adoption", res)
        found = ils.find_lock_for_branch(
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo=PRGS_REPO,
            branch_name=BRANCH,
            lock_dir=self.lock_dir,
        )
        self.assertEqual(found["branch_name"], BRANCH)
    @patch("mcp_server.api_request")
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_resolves_durable_lock_after_session_loss(self, _auth, _role, mock_api):
        mock_api.return_value = {"number": 421, "html_url": "https://example/pr/421"}
        worktree = os.path.realpath(os.getcwd())
        path = ils.lock_file_path(
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo=PRGS_REPO,
            issue_number=420,
            lock_dir=self.lock_dir,
        )
        ils.save_lock_file(path, _lock_record(worktree_path=worktree))
        env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": self.lock_dir}
        with patch.dict(os.environ, env, clear=True):
            with patch("os.getpid", return_value=8888):
                self.assertIsNone(ils.read_session_issue_lock(lock_dir=self.lock_dir))
                res = gitea_create_pr(
                    title=f"feat: server parity Closes #420",
                    head=BRANCH,
                    remote="prgs",
                    worktree_path=worktree,
                )
        self.assertEqual(res["number"], 421)
    @patch(
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value=_clean_master_git_state_for_lock(),
    )
    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    @patch(
        "mcp_server.issue_duplicate_context_fetcher",
        return_value=([{
            "number": 99,
            "head": {"ref": BRANCH},
            "title": "WIP",
            "body": "",
        }], [], {"status": "not_claimed"}),
    )
    def test_open_pr_blocks_recovery_lock(self, _dup_fetcher, _auth, mock_api, _git):
        env = {**ISSUE_WRITE_ENV, "GITEA_ISSUE_LOCK_DIR": self.lock_dir}
        with patch.dict(os.environ, env, clear=True):
            with self.assertRaises(ValueError) as ctx:
                gitea_lock_issue(issue_number=420, branch_name=BRANCH, remote="prgs")
        self.assertIn("open PR #99 already covers issue", str(ctx.exception))
 if __name__ == "__main__":
    unittest.main()
@@ -99,19 +99,7 @@ CREATE_PR_ENV = {
    ),
 }
 ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
 def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
    import issue_lock_provenance
    work_lease = {
        "operation_type": "author_issue_work",
        "issue_number": issue_number,
        "branch": branch_name,
        "claimant": {"username": "test-user", "profile": "test-author"},
        "expires_at": "2999-01-01T00:00:00Z",
    }
    record = {
        "issue_number": issue_number,
        "branch_name": branch_name,
@@ -119,11 +107,10 @@ def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
        "org": "Scaled-Tech-Consulting",
        "repo": "Gitea-Tools",
        "worktree_path": "/tmp/test-worktree",
-        "work_lease": work_lease,
+        "work_lease": {
-        "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
+            "operation_type": "author_issue_work",
-            tool="gitea_lock_issue",
+            "expires_at": "2999-01-01T00:00:00Z",
-            claimant=work_lease.get("claimant"),
+        },
        ),
    }
    record.update(overrides)
    return record
@@ -3146,7 +3133,7 @@ class TestIssueLocking(unittest.TestCase):
    def test_lock_issue_mismatch_branch_fails(self):
        with self.assertRaises(ValueError) as ctx:
            gitea_lock_issue(issue_number=196, branch_name="feat/issue-195-mutations", remote="prgs")
-        self.assertIn("must match", str(ctx.exception))
+        self.assertIn("must contain locked issue pattern", str(ctx.exception))
    @patch(
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
@@ -3203,6 +3190,7 @@ class TestIssueLocking(unittest.TestCase):
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_lock_issue_adopts_exact_own_branch(self, _auth, mock_api, _git_state):
        branch = "feat/issue-196-mutations"
        self.mock_dup_fetcher.return_value = ([], [branch], {"status": "not_claimed"})
        mock_api.return_value = [{"name": branch, "commit": {"id": "abc123"}}]
        res = gitea_lock_issue(issue_number=196, branch_name=branch, remote="prgs")
        self.assertTrue(res["success"])
@@ -3425,40 +3413,6 @@ class TestIssueLocking(unittest.TestCase):
                )
        self.assertIn("does not match locked worktree", str(ctx.exception))
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_manual_lock_seed_blocked(self, _auth, _role):
        worktree = os.path.realpath(os.getcwd())
        env = self._create_pr_env()
        with patch.dict(os.environ, env, clear=True):
            issue_lock_store.save_lock_file(
                issue_lock_store.lock_file_path(
                    remote="prgs",
                    org="Scaled-Tech-Consulting",
                    repo=mcp_server.REMOTES["prgs"]["repo"],
                    issue_number=447,
                    lock_dir=self._lock_dir.name,
                ),
                _sample_issue_lock(
                    issue_number=447,
                    branch_name="feat/issue-447-lock-provenance",
                    remote="prgs",
                    org="Scaled-Tech-Consulting",
                    repo=mcp_server.REMOTES["prgs"]["repo"],
                    worktree_path=worktree,
                    lock_provenance=None,
                ),
            )
            with self.assertRaises(RuntimeError) as ctx:
                gitea_create_pr(
                    title="feat: lock provenance Closes #447",
                    head="feat/issue-447-lock-provenance",
                    remote="prgs",
                    worktree_path=worktree,
                )
        self.assertIn("lock provenance", str(ctx.exception).lower())
    @patch("mcp_server.api_request")
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))