resolve conflicts for PR #467

Adds structured issue-branch ownership parsing, wires it into lock adoption
and gitea_lock_issue validation, documents the restart recovery workflow, and
adds regression tests for adoption, durable create_pr resolution, and open-PR
blocking without remote branch deletion.

Built on keyed persistent lock store and own-branch adoption (#443 / #442).

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

audit_reconciliation_mode.py

@@ -1,344 +0,0 @@
-"""Audit vs cleanup phase gates for reconciliation workflows (#419).
-
-Audit/reconciliation tasks are read-only unless a separate cleanup phase is
-explicitly authorized with exact capability proof, safety proof, and
-before/after snapshots. Cleanup mutations must be classified in final reports;
-audit reports must not claim ``no mutations`` when cleanup occurred.
-"""
-
-from __future__ import annotations
-
-import re
-from typing import Any
-
-RECONCILE_WORKFLOW_PATH = "workflows/reconcile-landed-pr.md"
-
-PHASE_AUDIT = "audit"
-PHASE_CLEANUP = "cleanup"
-
-# Tasks that enter audit phase on capability resolution (read-only default).
-AUDIT_PHASE_TASKS = frozenset({
-    "reconcile-landed-pr",
-    "reconcile_landed_pr",
-    "reconcile_issue_claims",
-    "reconcile_merged_cleanups",
-})
-
-# Mutation tasks forbidden during audit phase (fail closed).
-AUDIT_FORBIDDEN_TASKS = frozenset({
-    "delete_branch",
-    "create_branch",
-    "push_branch",
-    "create_pr",
-    "commit_files",
-    "gitea_commit_files",
-    "mark_issue",
-    "lock_issue",
-    "claim_issue",
-    "close_pr",
-    "close_issue",
-    "create_issue",
-    "merge_pr",
-    "review_pr",
-    "submit_pr_review",
-    "comment_pr",
-    "comment_issue",
-    "set_issue_labels",
-})
-
-# Shell/git commands audit phase must not run.
-AUDIT_FORBIDDEN_COMMAND_RE = re.compile(
-    r"(?:^|\s)(?:git\s+(?:push|branch\s+-D|worktree\s+remove)|"
-    r"gitea_delete_branch|delete_remote_branch)",
-    re.IGNORECASE,
-)
-
-_NO_MUTATIONS_RE = re.compile(
-    r"(?:no\s+mutations|mutations\s*:\s*none|no\s+unsafe\s+mutation)",
-    re.IGNORECASE,
-)
-_CLEANUP_OCCURRED_RE = re.compile(
-    r"(?:delete_remote_branch|remove_local_worktree|git\s+branch\s+-D|"
-    r"git\s+worktree\s+remove|remote branch.*deleted|worktree.*removed|"
-    r"cleanup\s+phase\s*:\s*(?!none\b)\S)",
-    re.IGNORECASE,
-)
-_EXTERNAL_STATE_RE = re.compile(
-    r"^\s*[-*]?\s*external[- ]state mutations\s*:",
-    re.IGNORECASE | re.MULTILINE,
-)
-_GIT_REF_RE = re.compile(
-    r"^\s*[-*]?\s*git ref mutations\s*:",
-    re.IGNORECASE | re.MULTILINE,
-)
-_CLEANUP_MUTATIONS_RE = re.compile(
-    r"^\s*[-*]?\s*cleanup mutations\s*:",
-    re.IGNORECASE | re.MULTILINE,
-)
-_CLEANUP_PHASE_AUTH_RE = re.compile(
-    r"^\s*[-*]?\s*cleanup phase (?:authorized|authorization)\s*:\s*true",
-    re.IGNORECASE | re.MULTILINE,
-)
-_DELETE_CAPABILITY_RE = re.compile(
-    r"^\s*[-*]?\s*delete.?branch capability(?: proven)?\s*:\s*true",
-    re.IGNORECASE | re.MULTILINE,
-)
-_BEFORE_AFTER_RE = re.compile(
-    r"^\s*[-*]?\s*before/after (?:state )?snapshot\s*:",
-    re.IGNORECASE | re.MULTILINE,
-)
-_SAFETY_PROOF_RE = re.compile(
-    r"^\s*[-*]?\s*(?:branch|worktree) safe to remove\s*:\s*true",
-    re.IGNORECASE | re.MULTILINE,
-)
-
-_session: dict[str, Any] | None = None
-
-
-def _blank_session() -> dict[str, Any]:
-    return {
-        "phase": PHASE_AUDIT,
-        "entered_from_task": None,
-        "cleanup_authorized": False,
-        "cleanup_authorization": {},
-    }
-
-
-def current_phase() -> str | None:
-    """Return active reconciliation phase or None when unset."""
-    if not _session:
-        return None
-    return _session.get("phase")
-
-
-def active_record() -> dict[str, Any] | None:
-    """Return a copy of the session record, if any."""
-    return dict(_session) if _session else None
-
-
-def clear_phase() -> None:
-    """Clear reconciliation phase state."""
-    global _session
-    _session = None
-
-
-def enter_audit_phase(task: str) -> dict[str, Any]:
-    """Enter read-only audit phase for a reconciliation task."""
-    global _session
-    normalized = (task or "").strip().lower()
-    _session = _blank_session()
-    _session["entered_from_task"] = normalized
-    return dict(_session)
-
-
-def authorize_cleanup_phase(
-    *,
-    operator_approved: bool = False,
-    workflow_authorized: bool = False,
-    delete_capability_proven: bool = False,
-    safety_proof: dict[str, Any] | None = None,
-    before_after_snapshot: dict[str, Any] | None = None,
-) -> dict[str, Any]:
-    """Authorize cleanup phase after explicit approval and safety proofs."""
-    reasons: list[str] = []
-    if not (operator_approved or workflow_authorized):
-        reasons.append(
-            "cleanup phase requires operator approval or explicit workflow "
-            "authorization"
-        )
-    if not delete_capability_proven:
-        reasons.append(
-            "cleanup phase requires exact delete_branch capability proof "
-            "(gitea.branch.delete)"
-        )
-    safety = dict(safety_proof or {})
-    if not safety.get("safe_to_delete_remote") and not safety.get(
-        "safe_to_remove_worktree"
-    ):
-        reasons.append(
-            "cleanup phase requires proof that branch/worktree is safe to remove"
-        )
-    snapshot = dict(before_after_snapshot or {})
-    if not snapshot.get("before") or not snapshot.get("after"):
-        reasons.append(
-            "cleanup phase requires before/after state snapshot"
-        )
-
-    if reasons:
-        return {
-            "authorized": False,
-            "phase": current_phase() or PHASE_AUDIT,
-            "reasons": reasons,
-            "safe_next_action": (
-                "remain in audit-only mode or supply operator approval, "
-                "delete_branch capability proof, safety proof, and "
-                "before/after snapshot before cleanup"
-            ),
-        }
-
-    global _session
-    if _session is None:
-        _session = _blank_session()
-    _session["phase"] = PHASE_CLEANUP
-    _session["cleanup_authorized"] = True
-    _session["cleanup_authorization"] = {
-        "operator_approved": operator_approved,
-        "workflow_authorized": workflow_authorized,
-        "delete_capability_proven": delete_capability_proven,
-        "safety_proof": safety,
-        "before_after_snapshot": snapshot,
-    }
-    return {
-        "authorized": True,
-        "phase": PHASE_CLEANUP,
-        "reasons": [],
-        "cleanup_authorization": dict(_session["cleanup_authorization"]),
-        "safe_next_action": "proceed with authorized cleanup mutations only",
-    }
-
-
-def check_audit_task_enters_phase(task: str) -> bool:
-    """Return whether resolving *task* should enter audit phase."""
-    return (task or "").strip().lower() in AUDIT_PHASE_TASKS
-
-
-def check_audit_mutation_allowed(task: str) -> tuple[bool, list[str]]:
-    """Fail closed when a mutation task runs during audit phase."""
-    normalized = (task or "").strip().lower()
-    phase = current_phase()
-    if phase != PHASE_AUDIT:
-        return True, []
-    if normalized in AUDIT_FORBIDDEN_TASKS:
-        return False, [
-            f"task '{normalized}' is forbidden in audit-only reconciliation "
-            "mode: switch to an explicit cleanup phase with operator approval "
-            "and exact delete_branch capability proof before cleanup mutations"
-        ]
-    return True, []
-
-
-def check_cleanup_execution_allowed() -> tuple[bool, list[str]]:
-    """Fail closed when cleanup execution is attempted without authorization."""
-    phase = current_phase()
-    if phase == PHASE_CLEANUP and (_session or {}).get("cleanup_authorized"):
-        return True, []
-    if phase is None:
-        return False, [
-            "cleanup execution requires an active reconciliation session; "
-            "resolve a reconciliation audit task first"
-        ]
-    return False, [
-        "cleanup execution forbidden in audit-only reconciliation mode; "
-        "call gitea_authorize_reconciliation_cleanup_phase with operator "
-        "approval, delete_branch capability proof, safety proof, and "
-        "before/after snapshot"
-    ]
-
-
-def classify_cleanup_mutation(action: str) -> str:
-    """Map a cleanup action to the required mutation ledger category (#419)."""
-    normalized = (action or "").strip().lower()
-    if "delete_remote" in normalized or normalized in {
-        "delete_branch",
-        "gitea_delete_branch",
-    }:
-        return "external-state"
-    if "branch" in normalized and "delete" in normalized:
-        return "git-ref"
-    if "worktree" in normalized or "remove_local" in normalized:
-        return "cleanup"
-    return "cleanup"
-
-
-def assess_audit_reconciliation_report(report_text: str) -> dict[str, Any]:
-    """Validate audit/cleanup reconciliation reports (fail closed)."""
-    text = report_text or ""
-    reasons: list[str] = []
-
-    cleanup_occurred = bool(_CLEANUP_OCCURRED_RE.search(text))
-    claims_no_mutations = bool(_NO_MUTATIONS_RE.search(text))
-
-    if cleanup_occurred and claims_no_mutations:
-        reasons.append(
-            "report claims no mutations but documents cleanup mutations; "
-            "audit-only reports must not perform cleanup and cleanup reports "
-            "must not claim no mutations"
-        )
-
-    if cleanup_occurred:
-        if not _CLEANUP_PHASE_AUTH_RE.search(text):
-            reasons.append(
-                "cleanup mutations reported without "
-                "'Cleanup phase authorized: true'"
-            )
-        if not _DELETE_CAPABILITY_RE.search(text):
-            reasons.append(
-                "cleanup mutations reported without delete_branch capability "
-                "proof"
-            )
-        if not _BEFORE_AFTER_RE.search(text):
-            reasons.append(
-                "cleanup mutations reported without before/after state snapshot"
-            )
-        if not _SAFETY_PROOF_RE.search(text):
-            reasons.append(
-                "cleanup mutations reported without branch/worktree safety proof"
-            )
-
-        if re.search(r"delete_remote|remote branch.*delet", text, re.I):
-            if not _EXTERNAL_STATE_RE.search(text):
-                reasons.append(
-                    "remote branch deletion must be classified under "
-                    "External-state mutations"
-                )
-        if re.search(r"git\s+branch\s+-D|local branch.*delet", text, re.I):
-            if not _GIT_REF_RE.search(text):
-                reasons.append(
-                    "local branch deletion must be classified under "
-                    "Git ref mutations"
-                )
-        if re.search(r"worktree.*remov|remove_local_worktree", text, re.I):
-            if not _CLEANUP_MUTATIONS_RE.search(text):
-                reasons.append(
-                    "worktree removal must be classified under Cleanup mutations"
-                )
-
-    if (
-        RECONCILE_WORKFLOW_PATH.replace("workflows/", "") in text
-        or "reconcile-landed-pr" in text.lower()
-    ):
-        if cleanup_occurred and "audit phase" in text.lower():
-            if "cleanup phase" not in text.lower():
-                reasons.append(
-                    "report mixes audit phase with cleanup mutations without "
-                    "documenting cleanup phase transition"
-                )
-
-    proven = not reasons
-    return {
-        "proven": proven,
-        "block": not proven,
-        "reasons": reasons,
-        "cleanup_occurred": cleanup_occurred,
-        "claims_no_mutations": claims_no_mutations,
-        "safe_next_action": (
-            "proceed"
-            if proven
-            else "fix audit/cleanup report: separate audit from cleanup phase, "
-            "classify mutations, and do not claim no mutations after cleanup"
-        ),
-    }
-
-
-def assess_audit_command_allowed(command: str) -> tuple[bool, list[str]]:
-    """Block shell commands that perform cleanup during audit phase."""
-    phase = current_phase()
-    if phase != PHASE_AUDIT:
-        return True, []
-    cmd = (command or "").strip()
-    if AUDIT_FORBIDDEN_COMMAND_RE.search(cmd):
-        return False, [
-            f"command forbidden in audit-only reconciliation mode: {cmd!r}; "
-            "authorize cleanup phase before branch/worktree deletion or push"
-        ]
-    return True, []

docs/llm-workflow-runbooks.md

@@ -274,12 +274,24 @@ is proven abandoned and the takeover is recorded.
 
 Gitea-Tools lease gates: `gitea_lock_issue` (fail-closed before author
 mutations), `status:in-progress`, and claim comments. `gitea_lock_issue`
-records an `author_issue_work` lease in the issue-lock payload with issue
+records an `author_issue_work` lease in a keyed lock file under
-number, optional PR number, branch, worktree path, claimant identity/profile,
+`GITEA_ISSUE_LOCK_DIR` (default `~/.cache/gitea-tools/issue-locks`), one file
-created timestamp, expiry timestamp, and last heartbeat timestamp. An active
+per `remote` + `org` + `repo` + `issue_number`. The current MCP session binds
-same-issue/same-operation lease blocks duplicate work. An expired lease still
+its active lock through a per-process pointer so concurrent repos/issues never
-blocks takeover until a recovery review records why the prior work is abandoned,
+share one overwrite-prone slot (#443).
-completed, or unsafe to continue.
+
+Each lock payload includes issue number, optional PR number, branch, worktree
+path, claimant identity/profile, created timestamp, expiry timestamp, and last
+heartbeat timestamp. An active same-issue/same-operation lease blocks duplicate
+work. An expired lease still blocks takeover until a recovery review records why
+the prior work is abandoned, completed, or unsafe to continue.
+
+**Do not manually seed `/tmp/gitea_issue_lock.json` or any lock file as a normal
+recovery path.** That global slot is deprecated and can clobber unrelated live
+leases (#438). After an MCP restart, call `gitea_lock_issue` again — own-branch
+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
@@ -296,6 +308,28 @@ metadata. Final-report validation blocks handoffs that hide lock read/write/dele
 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

final_report_validator.py

@@ -919,22 +919,6 @@ def _rule_shared_author_reviewer_same_run(report_text: str) -> list[dict[str, st
     )
 
 
-def _rule_audit_reconciliation_boundary(report_text: str) -> list[dict[str, str]]:
-    from audit_reconciliation_mode import assess_audit_reconciliation_report
-
-    result = assess_audit_reconciliation_report(report_text)
-    if result.get("proven"):
-        return []
-    return _findings_from_reasons(
-        "reconcile.audit_cleanup_boundary",
-        result.get("reasons") or [],
-        field="Audit/cleanup phase",
-        severity="block",
-        safe_next_action=result.get("safe_next_action")
-        or "separate audit from authorized cleanup and classify mutations",
-    )
-
-
 def _rule_reviewer_review_mutation(
     report_text: str,
     *,
@@ -993,7 +977,6 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_git_fetch_readonly,
         _rule_reviewer_legacy_workspace_mutations,
         _rule_reviewer_vague_mutations_none,
-        _rule_audit_reconciliation_boundary,
     ],
     "author_issue": [
         _rule_shared_controller_handoff,

gitea_mcp_server.py

@@ -539,6 +539,9 @@ 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
@@ -546,13 +549,13 @@ import issue_work_duplicate_gate  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
 import reconciler_profile  # noqa: E402
 import reconciliation_workflow  # noqa: E402
-import audit_reconciliation_mode  # noqa: E402
 import review_merge_state_machine  # noqa: E402
 import native_mcp_preference  # noqa: E402
 
 
-# Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
+# Keyed issue-lock storage (#443): per remote/org/repo/issue files under
-# consumed by gitea_create_pr and scripts/worktree-start.
+# GITEA_ISSUE_LOCK_DIR, bound to the current MCP session via a per-PID pointer.
+# Legacy global path retained only for test/doc references — do not seed manually.
 ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
 WORK_LEASE_TTL_HOURS = 4
 AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"
@@ -584,15 +587,59 @@ def _parse_work_lease_timestamp(value: str | None) -> datetime | None:
         return None
 
 
-def _load_existing_issue_lock() -> dict | None:
+def _load_existing_issue_lock(
-    if not os.path.exists(ISSUE_LOCK_FILE):
+    *,
-        return None
+    remote: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    issue_number: int | None = None,
+) -> dict | None:
+    if remote and org and repo and issue_number is not None:
+        return issue_lock_store.load_issue_lock(
+            remote=remote,
+            org=org,
+            repo=repo,
+            issue_number=issue_number,
+        )
+    return issue_lock_store.read_session_issue_lock()
+
+
+def _resolve_issue_lock_for_pr(
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    head: str,
+) -> dict:
+    lock_data = issue_lock_store.read_session_issue_lock()
+    if not lock_data:
+        lock_data = issue_lock_store.find_lock_for_branch(
+            remote=remote,
+            org=org,
+            repo=repo,
+            branch_name=head,
+        )
+    if not lock_data:
+        raise RuntimeError(
+            "Issue lock is missing (fail closed). Call gitea_lock_issue first."
+        )
+    return lock_data
+
+
+def _save_issue_lock(data: dict) -> str:
+    existing = issue_lock_store.load_issue_lock(
+        remote=str(data.get("remote") or ""),
+        org=str(data.get("org") or ""),
+        repo=str(data.get("repo") or ""),
+        issue_number=int(data.get("issue_number") or 0),
+    )
+    overwrite_block = issue_lock_store.assess_foreign_lock_overwrite(existing, data)
+    if overwrite_block:
+        raise RuntimeError(overwrite_block)
     try:
-        with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
+        return issue_lock_store.bind_session_lock(data)
-            data = json.load(f)
+    except Exception as e:
-        return data if isinstance(data, dict) else None
+        raise RuntimeError(f"Could not write issue lock file: {e}") from e
-    except Exception:
-        return None
 
 
 def _work_lease_claimant(host: str | None) -> dict:
@@ -795,6 +842,19 @@ def _enforce_locked_issue_duplicate_recheck(
     return None
 
 
+def _branch_entry_commit_sha(branch: dict | str) -> str | None:
+    """Best-effort head SHA for a Gitea branch entry (None when absent)."""
+    if not isinstance(branch, dict):
+        return None
+    commit = branch.get("commit")
+    if isinstance(commit, dict):
+        sha = commit.get("id") or commit.get("sha")
+        if sha:
+            return str(sha)
+    sha = branch.get("commit_sha")
+    return str(sha) if sha else None
+
+
 def _reveal_endpoints() -> bool:
     """Admin/debug opt-in (#120): include endpoint URLs and token source
     names in tool output. Off by default so normal LLM-facing responses
@@ -1227,11 +1287,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 branch name includes issue number
+    # 1. Enforce canonical issue branch ownership (#440)
-    expected_pattern = f"issue-{issue_number}"
+    if not issue_branch_ownership.branch_belongs_to_issue(branch_name, issue_number):
-    if expected_pattern not in branch_name:
         raise ValueError(
-            f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)"
+            f"Branch name '{branch_name}' must match "
+            f"(fix|feat|docs|chore)/issue-{issue_number}-<desc> (fail closed)"
         )
 
     blocked = _profile_permission_block(
@@ -1242,8 +1302,9 @@ def gitea_lock_issue(
     resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
         worktree_path, PROJECT_ROOT
     )
-    active_lease_block = _active_work_lease_block(
+    h, o, r = _resolve(remote, host, org, repo)
-        _load_existing_issue_lock(),
+    active_lease_block = issue_lock_store.assess_same_issue_lease_conflict(
+        _load_existing_issue_lock(remote=remote, org=o, repo=r, issue_number=issue_number),
         issue_number=issue_number,
         branch_name=branch_name,
         worktree_path=resolved_worktree,
@@ -1267,7 +1328,6 @@ def gitea_lock_issue(
             issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment)
         )
 
-    h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     duplicate_gate = _assess_issue_duplicate_gate(
         issue_number,
@@ -1283,6 +1343,30 @@ def gitea_lock_issue(
             f"duplicate work gate blocked issue #{issue_number} (fail closed)"
         ]))
 
+    branch_url = f"{repo_api_url(h, o, r)}/branches"
+    try:
+        branches = api_get_all(branch_url, auth)
+    except Exception as e:
+        raise RuntimeError(f"Could not list branches to verify issue lock: {e}")
+    existing_branch_entries = [
+        {
+            "name": _branch_entry_name(branch),
+            "commit_sha": _branch_entry_commit_sha(branch),
+        }
+        for branch in branches
+    ]
+    adoption = issue_lock_adoption.assess_own_branch_adoption(
+        issue_number=issue_number,
+        requested_branch=branch_name,
+        existing_branches=existing_branch_entries,
+    )
+    if adoption["block"]:
+        competing = ", ".join(adoption["competing_branches"])
+        raise ValueError(
+            f"Issue #{issue_number} already has matching branch '{competing}' "
+            "that is not the requested branch (fail closed)"
+        )
+
     work_lease = _build_author_issue_work_lease(
         issue_number=issue_number,
         branch_name=branch_name,
@@ -1303,11 +1387,7 @@ def gitea_lock_issue(
         ),
     }
 
-    try:
+    lock_file_path = _save_issue_lock(data)
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
-            json.dump(data, f)
-    except Exception as e:
-        raise RuntimeError(f"Could not write issue lock file: {e}")
 
     agent_artifacts = agent_temp_artifacts.find_agent_temp_artifacts_from_porcelain(
         git_state.get("porcelain_status") or ""
@@ -1322,7 +1402,22 @@ def gitea_lock_issue(
         "branch_name": branch_name,
         "worktree_path": resolved_worktree,
         "work_lease": work_lease,
+        "lock_file_path": lock_file_path,
     }
+    if adoption["adopt"]:
+        result["adoption"] = issue_lock_adoption.build_adoption_proof(
+            issue_number=issue_number,
+            branch_name=branch_name,
+            assessment=adoption,
+            open_pr_checked=True,
+            competing_lock_checked=True,
+            lock_file_path=lock_file_path,
+            lock_file_status="written",
+        )
+        result["message"] = (
+            f"Adopted existing branch '{branch_name}' and locked issue "
+            f"#{issue_number} for recovery (fail-closed check complete)."
+        )
     if agent_artifacts:
         result["warnings"] = [
             "Agent temp artifacts at repo root (delete before implementation): "
@@ -1414,15 +1509,8 @@ def gitea_create_pr(
     verify_preflight_purity(remote, worktree_path=worktree_path)
     h, o, r = _resolve(remote, host, org, repo)
 
-    # ── Issue Lock Validation (Issue #194 / #196) ──
+    # ── Issue Lock Validation (Issue #194 / #196 / #443) ──
-    if not os.path.exists(ISSUE_LOCK_FILE):
+    lock_data = _resolve_issue_lock_for_pr(remote=remote, org=o, repo=r, head=head)
-        raise RuntimeError("Issue lock is missing (fail closed). Call gitea_lock_issue first.")
-
-    try:
-        with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
-            lock_data = json.load(f)
-    except Exception as e:
-        raise RuntimeError(f"Could not read issue lock file: {e} (fail closed)")
 
     lock_provenance_check = issue_lock_provenance.assess_lock_file_for_create_pr(
         lock_data
@@ -2941,13 +3029,7 @@ def _prepare_commit_payload_files(files: list[dict]) -> tuple[list[dict], list[d
     processed_files = []
     source_proofs = []
 
-    lock_data = {}
+    lock_data = issue_lock_store.read_session_issue_lock() or {}
-    if os.path.exists(ISSUE_LOCK_FILE):
-        try:
-            with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
-                lock_data = json.load(f)
-        except Exception:
-            pass
 
     locked_worktree = lock_data.get("worktree_path")
     if locked_worktree:
@@ -3681,18 +3763,6 @@ def gitea_delete_branch(
             "permission_report": _permission_block_report("gitea.branch.delete"),
         }
 
-    audit_allowed, audit_reasons = (
-        audit_reconciliation_mode.check_audit_mutation_allowed("delete_branch")
-    )
-    if not audit_allowed:
-        return {
-            "success": False,
-            "performed": False,
-            "required_permission": "gitea.branch.delete",
-            "reasons": audit_reasons,
-            "audit_phase": audit_reconciliation_mode.current_phase(),
-        }
-
     verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -3762,18 +3832,6 @@ def gitea_reconcile_merged_cleanups(
             "execute_confirmed must be True when dry_run=False (fail closed)"
         )
 
-    if not dry_run:
-        exec_allowed, exec_reasons = (
-            audit_reconciliation_mode.check_cleanup_execution_allowed()
-        )
-        if not exec_allowed:
-            return {
-                "success": False,
-                "performed": False,
-                "reasons": exec_reasons,
-                "audit_phase": audit_reconciliation_mode.current_phase(),
-            }
-
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -3856,53 +3914,6 @@ def gitea_reconcile_merged_cleanups(
     return {"success": True, "performed": True, **report}
 
 
-@mcp.tool()
-def gitea_authorize_reconciliation_cleanup_phase(
-    operator_approved: bool = False,
-    workflow_authorized: bool = False,
-    delete_capability_proven: bool = False,
-    safe_to_delete_remote: bool = False,
-    safe_to_remove_worktree: bool = False,
-    before_state: str = "",
-    after_state: str = "",
-) -> dict:
-    """Authorize cleanup phase after audit-only reconciliation (#419).
-
-    Requires operator or workflow approval, exact delete_branch capability proof,
-    branch/worktree safety proof, and before/after state snapshots. Audit phase
-    forbids branch deletion, worktree removal, pushes, and issue/PR mutations.
-    """
-    delete_gate = _profile_operation_gate("gitea.branch.delete")
-    capability_ok = not bool(delete_gate)
-    if delete_capability_proven and delete_gate:
-        return {
-            "authorized": False,
-            "performed": False,
-            "delete_capability_verified": False,
-            "reasons": [
-                "delete_capability_proven=true but active profile lacks "
-                "gitea.branch.delete",
-            ] + delete_gate,
-            "audit_phase": audit_reconciliation_mode.current_phase(),
-        }
-    result = audit_reconciliation_mode.authorize_cleanup_phase(
-        operator_approved=operator_approved,
-        workflow_authorized=workflow_authorized,
-        delete_capability_proven=delete_capability_proven and capability_ok,
-        safety_proof={
-            "safe_to_delete_remote": safe_to_delete_remote,
-            "safe_to_remove_worktree": safe_to_remove_worktree,
-        },
-        before_after_snapshot={
-            "before": (before_state or "").strip(),
-            "after": (after_state or "").strip(),
-        },
-    )
-    result["performed"] = bool(result.get("authorized"))
-    result["delete_capability_verified"] = capability_ok
-    return result
-
-
 @mcp.tool()
 def gitea_assess_already_landed_reconciliation(
     pr_number: int,
@@ -7009,9 +7020,6 @@ def gitea_resolve_task_capability(
     if reason_msg:
         result["reason"] = reason_msg
     role_session_router.sync_route_from_capability(result)
-    if audit_reconciliation_mode.check_audit_task_enters_phase(task):
-        phase_record = audit_reconciliation_mode.enter_audit_phase(task)
-        result["reconciliation_phase"] = phase_record.get("phase")
     was_terminal = capability_stop_terminal.is_active()
     terminal = capability_stop_terminal.sync_from_capability_result(result)
     if terminal:

issue_branch_ownership.py

@@ -0,0 +1,28 @@
+"""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

issue_lock_adoption.py

@@ -0,0 +1,117 @@
+"""Own-branch lock adoption / recovery for ``gitea_lock_issue`` (#440 / #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.
+This module isolates the pure decision so it can be unit-tested apart from the
+MCP server's live Gitea calls.
+
+Adoption is granted only for the issue's *exact* requested branch. Any other
+branch that merely contains the same ``issue-<n>`` marker is competing work and
+stays fail-closed. Open-PR, competing-live-lock, capability, and worktree
+safety checks are enforced by the caller before this decision is consulted;
+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"
+
+
+def _branch_name(entry) -> str:
+    if isinstance(entry, dict):
+        return str(entry.get("name") or "")
+    return str(entry or "")
+
+
+def _branch_sha(entry) -> str | None:
+    if isinstance(entry, dict):
+        sha = entry.get("commit_sha")
+        if sha:
+            return str(sha)
+    return None
+
+
+def assess_own_branch_adoption(
+    *,
+    issue_number: int,
+    requested_branch: str,
+    existing_branches,
+) -> dict:
+    """Decide whether an existing matching branch is adoptable."""
+    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):
+            matches.append((name, _branch_sha(entry)))
+
+    competing = sorted({name for name, _ in matches if name != requested})
+    exact = [(name, sha) for name, sha in matches if name == requested]
+
+    if competing:
+        return {
+            "outcome": BLOCK_COMPETING,
+            "adopt": False,
+            "block": True,
+            "reason": (
+                f"issue #{issue_number} already has matching branch(es) "
+                f"{competing} that are not the requested branch "
+                f"'{requested}' (fail closed)"
+            ),
+            "matched_branch": None,
+            "matched_head_sha": None,
+            "competing_branches": competing,
+        }
+
+    if exact:
+        name, sha = exact[0]
+        return {
+            "outcome": ADOPT,
+            "adopt": True,
+            "block": False,
+            "reason": (
+                f"existing branch '{name}' is the exact requested branch for "
+                f"issue #{issue_number}; adopting it for lock recovery"
+            ),
+            "matched_branch": name,
+            "matched_head_sha": sha,
+            "competing_branches": [],
+        }
+
+    return {
+        "outcome": NO_MATCH,
+        "adopt": False,
+        "block": False,
+        "reason": f"no existing branch matches issue #{issue_number}",
+        "matched_branch": None,
+        "matched_head_sha": None,
+        "competing_branches": [],
+    }
+
+
+def build_adoption_proof(
+    *,
+    issue_number: int,
+    branch_name: str,
+    assessment: dict,
+    open_pr_checked: bool,
+    competing_lock_checked: bool,
+    lock_file_path: str,
+    lock_file_status: str,
+) -> dict:
+    """Assemble the proof block returned by ``gitea_lock_issue`` on adoption."""
+    return {
+        "issue_number": issue_number,
+        "branch_name": branch_name,
+        "branch_head_commit": assessment.get("matched_head_sha"),
+        "adoption_reason": assessment.get("reason"),
+        "no_existing_pr_proof": bool(open_pr_checked),
+        "no_competing_live_lock_proof": bool(competing_lock_checked),
+        "lock_file_path": lock_file_path,
+        "lock_file_status": lock_file_status,
+    }

issue_lock_store.py

@@ -0,0 +1,385 @@
+"""Keyed, persistent issue-lock storage (#443).
+
+Replaces the single global ``/tmp/gitea_issue_lock.json`` slot with per-issue
+lock files under ``GITEA_ISSUE_LOCK_DIR`` (default
+``~/.cache/gitea-tools/issue-locks``). Each MCP session binds its active lock
+via a per-process pointer file so concurrent repos/issues never clobber each
+other.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import tempfile
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+LOCK_DIR_ENV = "GITEA_ISSUE_LOCK_DIR"
+DEFAULT_LOCK_DIR = os.path.expanduser("~/.cache/gitea-tools/issue-locks")
+WORK_LEASE_TTL_HOURS = 4
+AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"
+
+_SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")
+
+
+def default_lock_dir() -> str:
+    raw = (os.environ.get(LOCK_DIR_ENV) or DEFAULT_LOCK_DIR).strip()
+    return raw or DEFAULT_LOCK_DIR
+
+
+def _sanitize_segment(value: str) -> str:
+    text = (value or "").strip()
+    if not text:
+        return "_"
+    return _SAFE_SEGMENT_RE.sub("_", text)
+
+
+def lock_key(
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    issue_number: int,
+) -> str:
+    return "-".join(
+        _sanitize_segment(part)
+        for part in (remote, org, repo, str(issue_number))
+    )
+
+
+def lock_file_path(
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    issue_number: int,
+    lock_dir: str | None = None,
+) -> str:
+    root = (lock_dir or default_lock_dir()).strip()
+    return os.path.join(root, f"{lock_key(remote=remote, org=org, repo=repo, issue_number=issue_number)}.json")
+
+
+def session_pointer_path(lock_dir: str | None = None) -> str:
+    root = (lock_dir or default_lock_dir()).strip()
+    return os.path.join(root, f"session-{os.getpid()}.json")
+
+
+def _ensure_lock_dir(lock_dir: str | None = None) -> str:
+    root = (lock_dir or default_lock_dir()).strip()
+    os.makedirs(root, mode=0o700, exist_ok=True)
+    return root
+
+
+def read_lock_file(path: str) -> dict[str, Any] | None:
+    lock_path = (path or "").strip()
+    if not lock_path or not os.path.exists(lock_path):
+        return None
+    try:
+        with open(lock_path, encoding="utf-8") as handle:
+            data = json.load(handle)
+    except (OSError, json.JSONDecodeError):
+        return None
+    return data if isinstance(data, dict) else None
+
+
+def save_lock_file(path: str, data: dict[str, Any]) -> None:
+    lock_path = (path or "").strip()
+    if not lock_path:
+        raise ValueError("lock path is required (fail closed)")
+    parent = os.path.dirname(lock_path) or "."
+    os.makedirs(parent, mode=0o700, exist_ok=True)
+    payload = json.dumps(data, indent=2, sort_keys=True) + "\n"
+    fd, temp_path = tempfile.mkstemp(prefix=".lock-", suffix=".json", dir=parent)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            handle.write(payload)
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(temp_path, lock_path)
+    finally:
+        if os.path.exists(temp_path):
+            try:
+                os.remove(temp_path)
+            except OSError:
+                pass
+
+
+def bind_session_lock(lock_data: dict[str, Any], lock_dir: str | None = None) -> str:
+    """Persist a keyed lock and bind it to the current process session."""
+    remote = str(lock_data.get("remote") or "")
+    org = str(lock_data.get("org") or "")
+    repo = str(lock_data.get("repo") or "")
+    issue_number = int(lock_data.get("issue_number") or 0)
+    if not remote or not org or not repo or issue_number <= 0:
+        raise ValueError("lock record must include remote, org, repo, and issue_number")
+
+    root = _ensure_lock_dir(lock_dir)
+    path = lock_file_path(
+        remote=remote,
+        org=org,
+        repo=repo,
+        issue_number=issue_number,
+        lock_dir=root,
+    )
+    record = dict(lock_data)
+    record["lock_file_path"] = path
+    record["session_pid"] = os.getpid()
+    save_lock_file(path, record)
+
+    pointer = {
+        "pid": os.getpid(),
+        "lock_file_path": path,
+        "issue_number": issue_number,
+        "branch_name": record.get("branch_name"),
+        "remote": remote,
+        "org": org,
+        "repo": repo,
+    }
+    save_lock_file(session_pointer_path(root), pointer)
+    return path
+
+
+def read_session_issue_lock(lock_dir: str | None = None) -> dict[str, Any] | None:
+    root = (lock_dir or default_lock_dir()).strip()
+    pointer = read_lock_file(session_pointer_path(root))
+    if not pointer:
+        return None
+    lock_path = str(pointer.get("lock_file_path") or "").strip()
+    if not lock_path:
+        return None
+    return read_lock_file(lock_path)
+
+
+def load_issue_lock(
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    issue_number: int,
+    lock_dir: str | None = None,
+) -> dict[str, Any] | None:
+    return read_lock_file(
+        lock_file_path(
+            remote=remote,
+            org=org,
+            repo=repo,
+            issue_number=issue_number,
+            lock_dir=lock_dir,
+        )
+    )
+
+
+def iter_lock_files(lock_dir: str | None = None) -> list[str]:
+    root = (lock_dir or default_lock_dir()).strip()
+    if not os.path.isdir(root):
+        return []
+    paths: list[str] = []
+    for name in os.listdir(root):
+        if not name.endswith(".json") or name.startswith("session-"):
+            continue
+        paths.append(os.path.join(root, name))
+    return sorted(paths)
+
+
+def find_lock_for_branch(
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    branch_name: str,
+    lock_dir: str | None = None,
+) -> dict[str, Any] | None:
+    target = (branch_name or "").strip()
+    if not target:
+        return None
+    for path in iter_lock_files(lock_dir):
+        lock = read_lock_file(path)
+        if not lock:
+            continue
+        if (
+            str(lock.get("remote") or "") == remote
+            and str(lock.get("org") or "") == org
+            and str(lock.get("repo") or "") == repo
+            and str(lock.get("branch_name") or "").strip() == target
+        ):
+            lock = dict(lock)
+            lock.setdefault("lock_file_path", path)
+            return lock
+    return None
+
+
+def _lease_now(now: datetime | None = None) -> datetime:
+    return now or datetime.now(timezone.utc)
+
+
+def _parse_lease_timestamp(value: str | None) -> datetime | None:
+    text = (value or "").strip()
+    if not text:
+        return None
+    try:
+        return datetime.fromisoformat(text.replace("Z", "+00:00")).astimezone(timezone.utc)
+    except ValueError:
+        return None
+
+
+def lease_expires_at(lock: dict[str, Any] | None) -> datetime | None:
+    if not lock:
+        return None
+    lease = lock.get("work_lease")
+    if not isinstance(lease, dict):
+        return None
+    return _parse_lease_timestamp(lease.get("expires_at"))
+
+
+def is_lease_expired(lock: dict[str, Any] | None, *, now: datetime | None = None) -> bool:
+    expires = lease_expires_at(lock)
+    if expires is None:
+        return False
+    return expires <= _lease_now(now)
+
+
+def is_lease_live(lock: dict[str, Any] | None, *, now: datetime | None = None) -> bool:
+    if not lock:
+        return False
+    lease = lock.get("work_lease")
+    if not isinstance(lease, dict):
+        return True
+    expires = _parse_lease_timestamp(lease.get("expires_at"))
+    if expires is None:
+        return True
+    return expires > _lease_now(now)
+
+
+def _same_realpath(left: str | None, right: str | None) -> bool:
+    if not left or not right:
+        return False
+    try:
+        return os.path.realpath(left) == os.path.realpath(right)
+    except OSError:
+        return left == right
+
+
+def assess_same_issue_lease_conflict(
+    existing_lock: dict[str, Any] | None,
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    operation_type: str = AUTHOR_ISSUE_WORK_LEASE,
+    now: datetime | None = None,
+) -> str | None:
+    """Return a fail-closed error when a competing live lease blocks acquisition."""
+    if not existing_lock:
+        return None
+
+    existing_issue = existing_lock.get("issue_number")
+    lease = existing_lock.get("work_lease")
+    existing_operation = (
+        lease.get("operation_type")
+        if isinstance(lease, dict)
+        else AUTHOR_ISSUE_WORK_LEASE
+    )
+    if existing_issue != issue_number or existing_operation != operation_type:
+        return None
+
+    existing_branch = existing_lock.get("branch_name")
+    existing_worktree = existing_lock.get("worktree_path")
+    same_owner = (
+        existing_branch == branch_name
+        and _same_realpath(str(existing_worktree or ""), worktree_path)
+    )
+    if is_lease_expired(existing_lock, now=now):
+        return (
+            f"Issue #{issue_number} has an expired {operation_type} lease on "
+            f"branch '{existing_branch}' from worktree '{existing_worktree}'. "
+            "Recovery review is required before takeover (fail closed)"
+        )
+    if same_owner:
+        return None
+    return (
+        f"Issue #{issue_number} already has an active {operation_type} lease on "
+        f"branch '{existing_branch}' from worktree '{existing_worktree}' "
+        "(fail closed)"
+    )
+
+
+def assess_foreign_lock_overwrite(
+    existing_lock: dict[str, Any] | None,
+    incoming_lock: dict[str, Any],
+    *,
+    now: datetime | None = None,
+) -> str | None:
+    """Block writes that would clobber an unrelated live lease on the same key."""
+    if not existing_lock:
+        return None
+
+    same_issue = existing_lock.get("issue_number") == incoming_lock.get("issue_number")
+    same_branch = existing_lock.get("branch_name") == incoming_lock.get("branch_name")
+    same_worktree = _same_realpath(
+        str(existing_lock.get("worktree_path") or ""),
+        str(incoming_lock.get("worktree_path") or ""),
+    )
+    if same_issue and same_branch and same_worktree:
+        return None
+    if not is_lease_live(existing_lock, now=now):
+        return None
+    return (
+        "Refusing to overwrite a live foreign issue lock "
+        f"(issue #{existing_lock.get('issue_number')}, "
+        f"branch '{existing_lock.get('branch_name')}', "
+        f"worktree '{existing_lock.get('worktree_path')}') (fail closed)"
+    )
+
+
+def find_live_lock_for_branch(
+    branch_name: str,
+    lock_dir: str | None = None,
+) -> dict[str, Any] | None:
+    target = (branch_name or "").strip()
+    if not target:
+        return None
+    for path in iter_lock_files(lock_dir):
+        lock = read_lock_file(path)
+        if not lock:
+            continue
+        if str(lock.get("branch_name") or "").strip() != target:
+            continue
+        if not is_lease_live(lock):
+            continue
+        record = dict(lock)
+        record.setdefault("lock_file_path", path)
+        return record
+    return None
+
+
+def resolve_locked_branch_for_session(
+    branch_name: str | None = None,
+    lock_dir: str | None = None,
+) -> str:
+    if branch_name:
+        lock = find_live_lock_for_branch(branch_name, lock_dir)
+        if lock:
+            return str(lock.get("branch_name") or "")
+    lock = read_session_issue_lock(lock_dir)
+    return str((lock or {}).get("branch_name") or "")
+
+
+def has_active_issue_lock(
+    branch: str,
+    *,
+    lock_dir: str | None = None,
+) -> bool:
+    target = (branch or "").strip()
+    if not target:
+        return False
+    for path in iter_lock_files(lock_dir):
+        lock = read_lock_file(path)
+        if not lock:
+            continue
+        if str(lock.get("branch_name") or "").strip() != target:
+            continue
+        if is_lease_live(lock):
+            return True
+    return False

merged_cleanup_reconcile.py

@@ -13,9 +13,9 @@ import subprocess
 from typing import Any
 
 from reviewer_worktree import parse_dirty_tracked_files
+import issue_lock_store
 
 PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})
-ISSUE_LOCK_FILE = os.environ.get("GITEA_ISSUE_LOCK_FILE", "/tmp/gitea_issue_lock.json")
 CLOSES_FIXES_RE = re.compile(r"\b(?:closes|fixes)\s+#(\d+)\b", re.IGNORECASE)
 
 
@@ -37,22 +37,18 @@ def resolve_worktree_path(project_root: str, branch: str) -> str:
 
 
 def read_issue_lock(path: str | None = None) -> dict[str, Any] | None:
-    lock_path = (path or ISSUE_LOCK_FILE).strip()
+    if path:
-    if not lock_path or not os.path.exists(lock_path):
+        return issue_lock_store.read_lock_file(path.strip())
-        return None
+    return issue_lock_store.read_session_issue_lock()
-    try:
-        with open(lock_path, encoding="utf-8") as handle:
-            data = json.load(handle)
-    except (OSError, json.JSONDecodeError):
-        return None
-    return data if isinstance(data, dict) else None
 
 
 def has_active_issue_lock(branch: str, lock_path: str | None = None) -> bool:
-    lock = read_issue_lock(lock_path)
+    if lock_path:
-    if not lock:
+        lock = issue_lock_store.read_lock_file(lock_path.strip())
-        return False
+        if not lock:
-    return (lock.get("branch_name") or "").strip() == (branch or "").strip()
+            return False
+        return (lock.get("branch_name") or "").strip() == (branch or "").strip()
+    return issue_lock_store.has_active_issue_lock(branch)
 
 
 def collect_open_pr_heads(open_prs: list[dict[str, Any]]) -> set[str]:

review_proofs.py

@@ -5115,15 +5115,6 @@ def assess_pr_queue_cleanup_report(report_text: str | None) -> dict:
     return _assess(report_text or "")
 
 
-def assess_audit_reconciliation_report(report_text: str | None) -> dict:
-    """#419: validate audit vs cleanup reconciliation report boundaries."""
-    from audit_reconciliation_mode import (
-        assess_audit_reconciliation_report as _assess,
-    )
-
-    return _assess(report_text or "")
-
-
 _GATE_PASSED_VALUE = re.compile(r"\bpassed\b", re.I)
 
 _NOT_APPLICABLE_VALUE = re.compile(

scripts/worktree-start

@@ -38,13 +38,21 @@ fi
 branch="$1"
 start_ref="${2:-prgs/master}"
 
+script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+repo_root="$(cd "$script_dir/.." && pwd)"
+
 # Enforce issue-linked, traceable branch names (issue → branch → worktree → PR).
 if [[ "$allow_unlinked" -eq 0 ]]; then
-  if [[ ! -f "/tmp/gitea_issue_lock.json" ]]; then
+  locked_branch=$(python3 -c "
-    echo "Error: Issue lock file '/tmp/gitea_issue_lock.json' is missing. You must lock exactly one issue before branch creation (fail closed)." >&2
+import sys
+sys.path.insert(0, '$repo_root')
+import issue_lock_store
+print(issue_lock_store.resolve_locked_branch_for_session('$branch'))
+")
+  if [[ -z "$locked_branch" ]]; then
+    echo "Error: No session issue lock is bound. Call gitea_lock_issue before branch creation (fail closed)." >&2
     exit 2
   fi
-  locked_branch=$(python3 -c "import json; print(json.load(open('/tmp/gitea_issue_lock.json')).get('branch_name', ''))")
   if [[ "$branch" != "$locked_branch" ]]; then
     echo "Error: Requested branch '$branch' does not match locked branch '$locked_branch' (fail closed)." >&2
     exit 2
@@ -68,8 +76,6 @@ EOF
   fi
 fi
 
-script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
-repo_root="$(cd "$script_dir/.." && pwd)"
 worktree_name="${branch//\//-}"
 worktree_path="$repo_root/branches/$worktree_name"
 

skills/llm-project-workflow/workflows/reconcile-landed-pr.md

@@ -304,40 +304,6 @@ If any required mutation capability is missing:
 * include safe next action (profile switch, human close, or dedicated reconciler
   profile)
 
-## 15A. Audit vs cleanup phase (#419)
-
-Reconciliation audits are **read-only** unless a separate cleanup phase is
-explicitly authorized.
-
-**Audit phase forbids** (``audit_reconciliation_mode.check_audit_mutation_allowed``
-fails closed):
-
-* ``gitea_delete_branch``
-* ``git branch -D``
-* ``git worktree remove``
-* pushes
-* issue/PR mutations
-* file edits
-
-Dry-run merged-cleanup reconciliation (``gitea_reconcile_merged_cleanups`` with
-``dry_run=True``) stays in audit phase. Execution requires:
-
-1. Operator approval or workflow authorization
-2. Exact ``delete_branch`` capability proof (``gitea.branch.delete``)
-3. Proof branch/worktree is safe to remove
-4. Before/after state snapshot
-
-Call ``gitea_authorize_reconciliation_cleanup_phase`` before any cleanup
-mutation. Final reports must not claim ``no mutations`` if cleanup occurred.
-Classify cleanup mutations as:
-
-* remote branch deletion → **External-state mutations**
-* local branch deletion → **Git ref mutations**
-* worktree removal → **Cleanup mutations**
-
-``audit_reconciliation_mode.assess_audit_reconciliation_report`` validates
-these boundaries in final reports.
-
 ## 16. Mutation classification
 
 Use precise mutation categories in the final report:

task_capability_map.py

@@ -104,10 +104,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.read",
         "role": "author",
     },
-    "reconciliation_cleanup": {
-        "permission": "gitea.branch.delete",
-        "role": "author",
-    },
     "work_issue": {
         "permission": "gitea.pr.create",
         "role": "author",

tests/test_agent_temp_artifacts.py

@@ -74,11 +74,14 @@ ISSUE_WRITE_ENV = {
 
 class TestIssueLockArtifactWarning(unittest.TestCase):
     def setUp(self):
-        self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
+        self._lock_dir = tempfile.TemporaryDirectory()
+        env = {**ISSUE_WRITE_ENV, "GITEA_ISSUE_LOCK_DIR": self._lock_dir.name}
+        self._env_patcher = patch.dict(os.environ, env, clear=True)
         self._env_patcher.start()
 
     def tearDown(self):
         self._env_patcher.stop()
+        self._lock_dir.cleanup()
 
     @patch(
         "mcp_server.issue_duplicate_context_fetcher",
@@ -86,9 +89,8 @@ class TestIssueLockArtifactWarning(unittest.TestCase):
     )
     @patch("mcp_server._auth", return_value="token x")
     @patch("mcp_server._resolve", return_value=("h", "o", "r"))
-    @patch("mcp_server.ISSUE_LOCK_FILE", new_callable=lambda: tempfile.mktemp())
     @patch("issue_lock_worktree.read_worktree_git_state")
-    def test_lock_success_includes_artifact_warning(self, mock_state, _lock_file, *_mocks):
+    def test_lock_success_includes_artifact_warning(self, mock_state, *_mocks):
         mock_state.return_value = {
             "current_branch": "master",
             "porcelain_status": "?? _emit_payload.py\n",

tests/test_audit_reconciliation_mode.py

@@ -1,291 +0,0 @@
-"""Tests for audit vs cleanup reconciliation mode (#419)."""
-import os
-import sys
-import unittest
-from unittest.mock import patch
-
-sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
-
-import audit_reconciliation_mode as arm
-import mcp_server
-from audit_reconciliation_mode import (
-    AUDIT_FORBIDDEN_TASKS,
-    PHASE_AUDIT,
-    PHASE_CLEANUP,
-    assess_audit_command_allowed,
-    assess_audit_reconciliation_report,
-    authorize_cleanup_phase,
-    check_audit_mutation_allowed,
-    check_cleanup_execution_allowed,
-    classify_cleanup_mutation,
-    clear_phase,
-    enter_audit_phase,
-)
-from final_report_validator import assess_final_report_validator
-from review_proofs import assess_audit_reconciliation_report as proofs_assess
-from task_capability_map import required_permission, required_role
-
-DELETE_PROFILE = {
-    "profile_name": "prgs-author-delete",
-    "allowed_operations": ["gitea.read", "gitea.branch.delete"],
-    "forbidden_operations": [],
-    "audit_label": "prgs-author-delete",
-}
-
-READ_PROFILE = {
-    "profile_name": "prgs-author",
-    "allowed_operations": ["gitea.read", "gitea.issue.comment"],
-    "forbidden_operations": ["gitea.branch.delete"],
-    "audit_label": "prgs-author",
-}
-
-READ_ENV = {
-    "GITEA_MCP_CONFIG": os.path.join(
-        os.path.dirname(__file__), "..", "profiles.json"
-    ),
-    "GITEA_MCP_PROFILE": "prgs-author",
-}
-
-
-def _authorize_cleanup(**kwargs):
-    defaults = {
-        "operator_approved": True,
-        "delete_capability_proven": True,
-        "safety_proof": {"safe_to_delete_remote": True},
-        "before_after_snapshot": {
-            "before": "remote branch exists",
-            "after": "remote branch absent",
-        },
-    }
-    defaults.update(kwargs)
-    return authorize_cleanup_phase(**defaults)
-
-
-def _cleanup_report(**overrides):
-    base = (
-        "Task mode: reconcile-landed-pr\n"
-        "Workflow source: workflows/reconcile-landed-pr.md\n"
-        "Audit phase: read-only assessment\n"
-        "Cleanup phase authorized: true\n"
-        "Delete-branch capability proven: true\n"
-        "Branch safe to remove: true\n"
-        "Before/after state snapshot: remote branch feat/x present → absent\n"
-        "External-state mutations: deleted remote branch feat/x\n"
-        "Git ref mutations: none\n"
-        "Cleanup mutations: removed worktree branches/feat-x\n"
-    )
-    for key, value in overrides.items():
-        base = base.replace(key, value)
-    return base
-
-
-class TestAuditPhaseGates(unittest.TestCase):
-    def setUp(self):
-        clear_phase()
-        enter_audit_phase("reconcile-landed-pr")
-
-    def tearDown(self):
-        clear_phase()
-
-    def test_audit_blocks_delete_branch_task(self):
-        allowed, reasons = check_audit_mutation_allowed("delete_branch")
-        self.assertFalse(allowed)
-        self.assertTrue(reasons)
-
-    def test_audit_blocks_worktree_shell_commands(self):
-        allowed, reasons = assess_audit_command_allowed(
-            "git worktree remove branches/feat-x"
-        )
-        self.assertFalse(allowed)
-        self.assertTrue(reasons)
-
-    def test_audit_blocks_local_branch_delete_command(self):
-        allowed, reasons = assess_audit_command_allowed("git branch -D feat/x")
-        self.assertFalse(allowed)
-
-    def test_audit_allows_read_tasks(self):
-        allowed, _ = check_audit_mutation_allowed("reconcile_landed_pr")
-        self.assertTrue(allowed)
-
-    def test_forbidden_set_covers_issue_and_pr_mutations(self):
-        for task in ("close_pr", "comment_issue", "commit_files", "push_branch"):
-            self.assertIn(task, AUDIT_FORBIDDEN_TASKS)
-
-
-class TestCleanupAuthorization(unittest.TestCase):
-    def setUp(self):
-        clear_phase()
-        enter_audit_phase("reconcile_merged_cleanups")
-
-    def tearDown(self):
-        clear_phase()
-
-    def test_cleanup_without_approval_blocked(self):
-        result = authorize_cleanup_phase(
-            delete_capability_proven=True,
-            safety_proof={"safe_to_delete_remote": True},
-            before_after_snapshot={"before": "a", "after": "b"},
-        )
-        self.assertFalse(result["authorized"])
-
-    def test_cleanup_without_capability_proof_blocked(self):
-        result = _authorize_cleanup(delete_capability_proven=False)
-        self.assertFalse(result["authorized"])
-
-    def test_cleanup_without_snapshot_blocked(self):
-        result = _authorize_cleanup(before_after_snapshot={"before": "", "after": ""})
-        self.assertFalse(result["authorized"])
-
-    def test_authorized_cleanup_switches_phase(self):
-        result = _authorize_cleanup()
-        self.assertTrue(result["authorized"])
-        self.assertEqual(result["phase"], PHASE_CLEANUP)
-
-    def test_cleanup_execution_allowed_only_after_authorization(self):
-        self.assertFalse(check_cleanup_execution_allowed()[0])
-        _authorize_cleanup()
-        self.assertTrue(check_cleanup_execution_allowed()[0])
-
-
-class TestReportVerifier(unittest.TestCase):
-    def test_false_no_mutations_after_cleanup_blocked(self):
-        report = (
-            "Task mode: reconcile-landed-pr\n"
-            "No mutations performed.\n"
-            "delete_remote_branch feat/dup\n"
-        )
-        result = assess_audit_reconciliation_report(report)
-        self.assertFalse(result["proven"])
-        self.assertIn("no mutations", result["reasons"][0].lower())
-
-    def test_cleanup_without_authorization_fields_blocked(self):
-        report = (
-            "Task mode: reconcile-landed-pr\n"
-            "remove_local_worktree branches/feat-x\n"
-        )
-        result = assess_audit_reconciliation_report(report)
-        self.assertFalse(result["proven"])
-
-    def test_authorized_cleanup_report_passes(self):
-        result = assess_audit_reconciliation_report(_cleanup_report())
-        self.assertTrue(result["proven"])
-
-    def test_mutation_classification_enforced(self):
-        report = (
-            "Task mode: reconcile-landed-pr\n"
-            "Cleanup phase authorized: true\n"
-            "Delete-branch capability proven: true\n"
-            "Branch safe to remove: true\n"
-            "Before/after state snapshot: present\n"
-            "remove_local_worktree branches/feat-x\n"
-        )
-        result = assess_audit_reconciliation_report(report)
-        self.assertFalse(result["proven"])
-
-    def test_proofs_export_matches_module(self):
-        report = "No mutations performed.\ndelete_remote_branch feat/dup"
-        self.assertEqual(
-            proofs_assess(report)["proven"],
-            assess_audit_reconciliation_report(report)["proven"],
-        )
-
-    def test_final_report_validator_includes_boundary_rule(self):
-        report = "No mutations performed.\ndelete_remote_branch feat/dup"
-        result = assess_final_report_validator(
-            report_text=report,
-            task_kind="reconcile_already_landed",
-        )
-        self.assertTrue(result["blocked"])
-        rule_ids = [f["rule_id"] for f in result["findings"]]
-        self.assertIn("reconcile.audit_cleanup_boundary", rule_ids)
-
-
-class TestMutationClassification(unittest.TestCase):
-    def test_remote_delete_is_external_state(self):
-        self.assertEqual(
-            classify_cleanup_mutation("delete_remote_branch"),
-            "external-state",
-        )
-
-    def test_worktree_remove_is_cleanup(self):
-        self.assertEqual(
-            classify_cleanup_mutation("remove_local_worktree"),
-            "cleanup",
-        )
-
-
-class TestMcpGates(unittest.TestCase):
-    def setUp(self):
-        clear_phase()
-        enter_audit_phase("reconcile_merged_cleanups")
-        self.mock_api = patch("mcp_server.api_request").start()
-        self.mock_auth = patch(
-            "mcp_server.get_auth_header", return_value="token test"
-        ).start()
-
-    def tearDown(self):
-        patch.stopall()
-        clear_phase()
-        mcp_server._IDENTITY_CACHE.clear()
-
-    @patch.dict(os.environ, READ_ENV, clear=True)
-    @patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
-    def test_delete_branch_blocked_in_audit_phase(self, _profile):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check("capability", resolved_role="author")
-        result = mcp_server.gitea_delete_branch(branch="feat/dup", remote="prgs")
-        self.assertFalse(result["success"])
-        self.assertEqual(result["audit_phase"], PHASE_AUDIT)
-        self.mock_api.assert_not_called()
-
-    @patch.dict(os.environ, READ_ENV, clear=True)
-    @patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
-    def test_reconcile_execute_blocked_without_cleanup_auth(self, _profile):
-        with self.assertRaises(ValueError):
-            mcp_server.gitea_reconcile_merged_cleanups(
-                dry_run=False,
-                execute_confirmed=False,
-                remote="prgs",
-            )
-
-    @patch.dict(os.environ, READ_ENV, clear=True)
-    @patch("mcp_server.get_profile", return_value=READ_PROFILE)
-    def test_cleanup_auth_fails_without_delete_capability(self, _profile):
-        result = mcp_server.gitea_authorize_reconciliation_cleanup_phase(
-            operator_approved=True,
-            delete_capability_proven=True,
-            safe_to_delete_remote=True,
-            before_state="exists",
-            after_state="gone",
-        )
-        self.assertFalse(result["authorized"])
-        self.assertFalse(result["delete_capability_verified"])
-
-    @patch.dict(os.environ, READ_ENV, clear=True)
-    @patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
-    def test_delete_branch_allowed_after_cleanup_authorization(self, _profile):
-        mcp_server.gitea_authorize_reconciliation_cleanup_phase(
-            operator_approved=True,
-            delete_capability_proven=True,
-            safe_to_delete_remote=True,
-            before_state="exists",
-            after_state="gone",
-        )
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check("capability", resolved_role="author")
-        self.mock_api.return_value = {}
-        result = mcp_server.gitea_delete_branch(branch="feat/dup", remote="prgs")
-        self.assertTrue(result["success"])
-
-
-class TestTaskCapabilityMap(unittest.TestCase):
-    def test_reconciliation_cleanup_maps_delete_permission(self):
-        self.assertEqual(
-            required_permission("reconciliation_cleanup"),
-            "gitea.branch.delete",
-        )
-        self.assertEqual(required_role("reconciliation_cleanup"), "author")
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_commit_payloads.py

@@ -66,9 +66,12 @@ class TestCommitPayloads(unittest.TestCase):
         )
         self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name)
 
-        self.lock_file_path = "/tmp/gitea_issue_lock.json"
+        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,
@@ -89,8 +92,7 @@ class TestCommitPayloads(unittest.TestCase):
                 claimant=work_lease.get("claimant"),
             ),
         }
-        with open(self.lock_file_path, "w", encoding="utf-8") as fh:
+        self.lock_file_path = issue_lock_store.bind_session_lock(self.lock_data)
-            fh.write(json.dumps(self.lock_data))
 
         # Reset preflight status to bypass/pass verification in tests
         self.orig_whoami_called = mcp_server._preflight_whoami_called
@@ -114,8 +116,7 @@ class TestCommitPayloads(unittest.TestCase):
 
         self._dir.cleanup()
         self.locked_worktree_dir.cleanup()
-        if os.path.exists(self.lock_file_path):
+        self._lock_dir.cleanup()
-            os.remove(self.lock_file_path)
 
     def _env(self, profile: str) -> dict:
         return {
@@ -124,6 +125,7 @@ class TestCommitPayloads(unittest.TestCase):
             "GITEA_TOKEN_AUTHOR": "author-pass",
             "GITEA_TEST_PORCELAIN": "",
             "GITEA_AUTHOR_WORKTREE": self.locked_worktree_path,
+            "GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
         }
 
     @patch("mcp_server.api_request")

tests/test_issue_branch_ownership.py

@@ -0,0 +1,36 @@
+"""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()

tests/test_issue_lock_adoption.py

@@ -0,0 +1,77 @@
+"""Unit tests for own-branch lock adoption decision (#442 / #443)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from issue_lock_adoption import (  # noqa: E402
+    ADOPT,
+    BLOCK_COMPETING,
+    NO_MATCH,
+    assess_own_branch_adoption,
+    build_adoption_proof,
+)
+
+REQ = "feat/issue-420-server-code-parity"
+
+
+class TestAssessOwnBranchAdoption(unittest.TestCase):
+    def test_exact_own_branch_is_adopted(self):
+        result = assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=REQ,
+            existing_branches=[{"name": REQ, "commit_sha": "934688a"}],
+        )
+        self.assertEqual(result["outcome"], ADOPT)
+        self.assertTrue(result["adopt"])
+
+    def test_different_branch_same_issue_blocks(self):
+        result = assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=REQ,
+            existing_branches=[{"name": "feat/issue-420-other-work"}],
+        )
+        self.assertEqual(result["outcome"], BLOCK_COMPETING)
+        self.assertTrue(result["block"])
+
+    def test_no_matching_branch_is_normal_path(self):
+        result = assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=REQ,
+            existing_branches=[{"name": "feat/issue-999-unrelated"}],
+        )
+        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):
+        assessment = assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=REQ,
+            existing_branches=[{"name": REQ, "commit_sha": "934688a"}],
+        )
+        proof = build_adoption_proof(
+            issue_number=420,
+            branch_name=REQ,
+            assessment=assessment,
+            open_pr_checked=True,
+            competing_lock_checked=True,
+            lock_file_path="/tmp/example-lock.json",
+            lock_file_status="written",
+        )
+        self.assertEqual(proof["branch_head_commit"], "934688a")
+        self.assertTrue(proof["no_existing_pr_proof"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_lock_recovery.py

@@ -0,0 +1,170 @@
+"""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()

tests/test_issue_lock_store.py

@@ -0,0 +1,181 @@
+"""Unit tests for keyed issue-lock storage (#443)."""
+import json
+import os
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import issue_lock_store as ils  # noqa: E402
+
+
+def _lease(expires_at: str) -> dict:
+    return {
+        "operation_type": ils.AUTHOR_ISSUE_WORK_LEASE,
+        "expires_at": expires_at,
+        "created_at": "2026-01-01T00:00:00Z",
+        "last_heartbeat_at": "2026-01-01T00:00:00Z",
+    }
+
+
+def _lock_record(**overrides) -> dict:
+    record = {
+        "issue_number": 420,
+        "branch_name": "feat/issue-420-server-code-parity",
+        "remote": "prgs",
+        "org": "Scaled-Tech-Consulting",
+        "repo": "Gitea-Tools",
+        "worktree_path": "/tmp/wt-420",
+        "work_lease": _lease("2999-01-01T00:00:00Z"),
+    }
+    record.update(overrides)
+    return record
+
+
+class TestIssueLockStore(unittest.TestCase):
+    def setUp(self):
+        self._dir = tempfile.TemporaryDirectory()
+        self.lock_dir = self._dir.name
+        self._env = mock.patch.dict(os.environ, {"GITEA_ISSUE_LOCK_DIR": self.lock_dir})
+        self._env.start()
+
+    def tearDown(self):
+        self._env.stop()
+        self._dir.cleanup()
+
+    def test_concurrent_repo_locks_do_not_overwrite(self):
+        lock_a = _lock_record(
+            issue_number=108,
+            branch_name="feat/issue-108-root-menu",
+            repo="mcp-control-plane",
+            worktree_path="/tmp/wt-108",
+        )
+        lock_b = _lock_record(
+            issue_number=420,
+            branch_name="feat/issue-420-server-code-parity",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/wt-420",
+        )
+        path_a = ils.bind_session_lock(lock_a)
+        with mock.patch("os.getpid", return_value=9999):
+            path_b = ils.bind_session_lock(lock_b)
+
+        self.assertNotEqual(path_a, path_b)
+        self.assertTrue(os.path.exists(path_a))
+        self.assertTrue(os.path.exists(path_b))
+        stored_a = ils.read_lock_file(path_a)
+        stored_b = ils.read_lock_file(path_b)
+        self.assertEqual(stored_a["issue_number"], 108)
+        self.assertEqual(stored_b["issue_number"], 420)
+
+    def test_concurrent_issue_locks_same_repo_do_not_overwrite(self):
+        lock_a = _lock_record(issue_number=427, branch_name="feat/issue-427-a")
+        lock_b = _lock_record(issue_number=428, branch_name="feat/issue-428-b")
+        path_a = ils.bind_session_lock(lock_a)
+        with mock.patch("os.getpid", return_value=4242):
+            path_b = ils.bind_session_lock(lock_b)
+
+        self.assertNotEqual(path_a, path_b)
+        self.assertEqual(ils.read_lock_file(path_a)["issue_number"], 427)
+        self.assertEqual(ils.read_lock_file(path_b)["issue_number"], 428)
+
+    def test_foreign_live_lease_blocks_overwrite(self):
+        existing = _lock_record(
+            branch_name="feat/issue-420-other",
+            worktree_path="/tmp/other",
+            work_lease=_lease("2999-01-01T00:00:00Z"),
+        )
+        path = ils.lock_file_path(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            issue_number=420,
+        )
+        ils.save_lock_file(path, existing)
+
+        incoming = _lock_record(worktree_path="/tmp/mine")
+        block = ils.assess_foreign_lock_overwrite(existing, incoming)
+        self.assertIn("live foreign issue lock", block or "")
+
+    def test_expired_lease_allows_takeover_with_conflict_check(self):
+        existing = _lock_record(
+            branch_name="feat/issue-420-other",
+            worktree_path="/tmp/other",
+            work_lease=_lease("2000-01-01T00:00:00Z"),
+        )
+        incoming = _lock_record(worktree_path="/tmp/mine")
+        self.assertIsNone(ils.assess_foreign_lock_overwrite(existing, incoming))
+        block = ils.assess_same_issue_lease_conflict(
+            existing,
+            issue_number=420,
+            branch_name="feat/issue-420-server-code-parity",
+            worktree_path="/tmp/mine",
+        )
+        self.assertIn("Recovery review is required", block or "")
+
+    def test_same_owner_lease_conflict_allows_refresh(self):
+        worktree = "/tmp/wt-420"
+        existing = _lock_record(worktree_path=worktree)
+        block = ils.assess_same_issue_lease_conflict(
+            existing,
+            issue_number=420,
+            branch_name="feat/issue-420-server-code-parity",
+            worktree_path=worktree,
+        )
+        self.assertIsNone(block)
+
+    def test_find_lock_for_branch_after_restart(self):
+        record = _lock_record()
+        path = ils.lock_file_path(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            issue_number=420,
+        )
+        ils.save_lock_file(path, record)
+
+        with mock.patch("os.getpid", return_value=5555):
+            self.assertIsNone(ils.read_session_issue_lock())
+
+        found = ils.find_lock_for_branch(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            branch_name="feat/issue-420-server-code-parity",
+        )
+        self.assertEqual(found["issue_number"], 420)
+
+    def test_has_active_issue_lock_scans_keyed_store(self):
+        ils.bind_session_lock(_lock_record())
+        self.assertTrue(
+            ils.has_active_issue_lock("feat/issue-420-server-code-parity")
+        )
+        self.assertFalse(ils.has_active_issue_lock("feat/issue-999-other"))
+
+    def test_atomic_write_preserves_unrelated_lock(self):
+        path_a = ils.lock_file_path(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            issue_number=108,
+        )
+        ils.save_lock_file(path_a, _lock_record(issue_number=108, repo="mcp-control-plane"))
+        path_b = ils.lock_file_path(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            issue_number=420,
+        )
+        ils.save_lock_file(path_b, _lock_record())
+
+        self.assertTrue(os.path.exists(path_a))
+        self.assertTrue(os.path.exists(path_b))
+        self.assertEqual(ils.read_lock_file(path_a)["issue_number"], 108)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_llm_workflow_split.py

@@ -87,14 +87,6 @@ def test_reconcile_landed_workflow_contract():
     assert "PARTIAL_RECONCILE_COMMENT_THEN_STOP" in text
     assert "RECOVERY_HANDOFF_ONLY" in text
     assert "resolve_partial_reconciliation_plan" in text
-    assert "check_audit_mutation_allowed" in text
-    assert "gitea_authorize_reconciliation_cleanup_phase" in text
-
-
-def test_audit_reconciliation_verifier_exported():
-    from review_proofs import assess_audit_reconciliation_report
-
-    assert callable(assess_audit_reconciliation_report)
 
 
 def test_create_issue_workflow_contract():

tests/test_mcp_server.py

@@ -6,6 +6,7 @@ the MCP protocol) with mocked API responses.
 import json
 import os
 import sys
+import tempfile
 import unittest
 from unittest.mock import patch, MagicMock
 
@@ -45,6 +46,7 @@ from gitea_auth import get_profile  # noqa: E402
 import gitea_config  # noqa: E402
 
 import mcp_server
+import issue_lock_store
 
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 
@@ -116,6 +118,7 @@ def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
         "remote": "dadeschools",
         "org": "Scaled-Tech-Consulting",
         "repo": "Gitea-Tools",
+        "worktree_path": "/tmp/test-worktree",
         "work_lease": work_lease,
         "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
             tool="gitea_lock_issue",
@@ -131,6 +134,17 @@ def _clear_duplicate_context_fetcher(*_args, **_kwargs):
     return [], [], {"status": "not_claimed"}
 
 
+def _bind_test_lock(**overrides) -> str:
+    remote = overrides.get("remote", "dadeschools")
+    record = _sample_issue_lock(**overrides)
+    if remote in mcp_server.REMOTES:
+        profile = mcp_server.REMOTES[remote]
+        record.setdefault("org", profile["org"])
+        record.setdefault("repo", profile["repo"])
+    record["remote"] = remote
+    return issue_lock_store.bind_session_lock(record)
+
+
 # ---------------------------------------------------------------------------
 # Create Issue
 # ---------------------------------------------------------------------------
@@ -193,18 +207,21 @@ class TestCreatePR(unittest.TestCase):
            return_value=(True, []))
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    @patch("os.path.exists", return_value=True)
+    def test_creates_pr(self, _auth, mock_api, _role, _dup_fetcher):
-    @patch("builtins.open")
+        worktree = os.path.realpath(os.getcwd())
-    def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role, _dup_fetcher):
-        lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
-        mock_open.return_value.__enter__.return_value.read.return_value = lock_json
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+        with tempfile.TemporaryDirectory() as lock_dir:
-            result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
+            env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir}
+            with patch.dict(os.environ, env, clear=True):
+                _bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree)
+                result = gitea_create_pr(
+                    title="feat: X Closes #123",
+                    head="feat/x",
+                    base="main",
+                    worktree_path=worktree,
+                )
         self.assertEqual(result["number"], 3)
         self.assertNotIn("url", result)
-        mock_exists.assert_called_with(ISSUE_LOCK_FILE)
-        mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
         payload = mock_api.call_args[0][3]
         self.assertEqual(payload["head"], "feat/x")
         self.assertEqual(payload["base"], "main")
@@ -218,30 +235,42 @@ class TestCreatePR(unittest.TestCase):
            return_value=(True, []))
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    @patch("os.path.exists", return_value=True)
+    def test_create_pr_reveal_opt_in_includes_url(self, _auth, mock_api, _role, _dup_fetcher):
-    @patch("builtins.open")
+        worktree = os.path.realpath(os.getcwd())
-    def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role, _dup_fetcher):
-        lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
-        mock_open.return_value.__enter__.return_value.read.return_value = lock_json
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
-        env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
+        with tempfile.TemporaryDirectory() as lock_dir:
-        with patch.dict(os.environ, env, clear=True):
+            env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
-            result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
+            with patch.dict(os.environ, env, clear=True):
+                _bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree)
+                result = gitea_create_pr(
+                    title="feat: X Closes #123",
+                    head="feat/x",
+                    base="main",
+                    worktree_path=worktree,
+                )
         self.assertIn("pulls/3", result["url"])
 
     @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)
-    @patch("os.path.exists", return_value=True)
+    def test_create_pr_locked_issue_mismatch_fails(self, _auth, _role):
-    @patch("builtins.open")
+        worktree = os.path.realpath(os.getcwd())
-    def test_create_pr_locked_issue_mismatch_fails(self, mock_open, mock_exists, _auth, _role):
+        with tempfile.TemporaryDirectory() as lock_dir:
-        lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
+            env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir}
-        mock_open.return_value.__enter__.return_value.read.return_value = lock_json
+            with patch.dict(os.environ, env, clear=True):
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+                _bind_test_lock(
-            with self.assertRaises(ValueError) as ctx:
+                    issue_number=123,
-                gitea_create_pr(title="feat: X Closes #999", head="feat/x", base="main")
+                    branch_name="feat/x",
+                    worktree_path=worktree,
+                )
+                with self.assertRaises(ValueError) as ctx:
+                    gitea_create_pr(
+                        title="feat: X Closes #999",
+                        head="feat/x",
+                        base="main",
+                        worktree_path=worktree,
+                    )
         self.assertIn("Closes #123", str(ctx.exception))
-        mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
 
 
 # ---------------------------------------------------------------------------
@@ -3069,7 +3098,12 @@ class TestIssueLocking(unittest.TestCase):
     """Test issue locking and PR gating constraints."""
 
     def setUp(self):
-        self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
+        self._lock_dir = tempfile.TemporaryDirectory()
+        env = {
+            **ISSUE_WRITE_ENV,
+            "GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
+        }
+        self._env_patcher = patch.dict(os.environ, env, clear=True)
         self._env_patcher.start()
         self._dup_fetcher_patcher = patch(
             "mcp_server.issue_duplicate_context_fetcher",
@@ -3080,15 +3114,21 @@ class TestIssueLocking(unittest.TestCase):
     def tearDown(self):
         self._dup_fetcher_patcher.stop()
         self._env_patcher.stop()
-        if os.path.exists(ISSUE_LOCK_FILE):
+        self._lock_dir.cleanup()
-            os.remove(ISSUE_LOCK_FILE)
+
+    def _create_pr_env(self) -> dict:
+        return {
+            **CREATE_PR_ENV,
+            "GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
+        }
 
     @patch(
         "mcp_server.issue_lock_worktree.read_worktree_git_state",
         return_value=_clean_master_git_state_for_lock(),
     )
+    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_lock_issue_success(self, _auth, _git_state):
+    def test_lock_issue_success(self, _auth, _api, _git_state):
         res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertTrue(res["success"])
         self.assertEqual(res["work_lease"]["operation_type"], "author_issue_work")
@@ -3098,16 +3138,15 @@ class TestIssueLocking(unittest.TestCase):
         self.assertIn("expires_at", res["work_lease"])
         self.assertIn("last_heartbeat_at", res["work_lease"])
         self.assertEqual(res["work_lease"]["claimant"]["profile"], "gitea-default")
-        self.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
+        self.assertIn("lock_file_path", res)
-        with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
+        lock = issue_lock_store.read_lock_file(res["lock_file_path"])
-            lock = json.load(f)
         self.assertIn("worktree_path", lock)
         self.assertIn("work_lease", lock)
 
     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 contain locked issue pattern", str(ctx.exception))
+        self.assertIn("must match", str(ctx.exception))
 
     @patch(
         "mcp_server.issue_lock_worktree.read_worktree_git_state",
@@ -3156,32 +3195,80 @@ class TestIssueLocking(unittest.TestCase):
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertIn("remote branch(es) already match issue pattern", str(ctx.exception))
 
-    def test_lock_issue_blocks_active_same_operation_lease(self):
+    @patch(
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+        "mcp_server.issue_lock_worktree.read_worktree_git_state",
-            json.dump({
+        return_value=_clean_master_git_state_for_lock(),
+    )
+    @patch("mcp_server.api_get_all")
+    @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"
+        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"])
+        self.assertIn("adoption", res)
+        self.assertEqual(res["adoption"]["branch_head_commit"], "abc123")
+
+    @patch(
+        "mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value=_clean_master_git_state_for_lock(),
+    )
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_blocks_active_same_operation_lease(self, _auth, _api, _git_state):
+        prgs_repo = mcp_server.REMOTES["prgs"]["repo"]
+        issue_lock_store.save_lock_file(
+            issue_lock_store.lock_file_path(
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo=prgs_repo,
+                issue_number=196,
+            ),
+            {
                 "issue_number": 196,
                 "branch_name": "feat/issue-196-other-work",
+                "remote": "prgs",
+                "org": "Scaled-Tech-Consulting",
+                "repo": prgs_repo,
                 "worktree_path": "/tmp/other-worktree",
                 "work_lease": {
                     "operation_type": "author_issue_work",
                     "expires_at": "2999-01-01T00:00:00Z",
                 },
-            }, f)
+            },
+        )
         with self.assertRaises(RuntimeError) as ctx:
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertIn("already has an active author_issue_work lease", str(ctx.exception))
 
-    def test_lock_issue_blocks_expired_same_operation_lease_for_recovery(self):
+    @patch(
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+        "mcp_server.issue_lock_worktree.read_worktree_git_state",
-            json.dump({
+        return_value=_clean_master_git_state_for_lock(),
+    )
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_blocks_expired_same_operation_lease_for_recovery(self, _auth, _api, _git_state):
+        prgs_repo = mcp_server.REMOTES["prgs"]["repo"]
+        issue_lock_store.save_lock_file(
+            issue_lock_store.lock_file_path(
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo=prgs_repo,
+                issue_number=196,
+            ),
+            {
                 "issue_number": 196,
                 "branch_name": "feat/issue-196-other-work",
+                "remote": "prgs",
+                "org": "Scaled-Tech-Consulting",
+                "repo": prgs_repo,
                 "worktree_path": "/tmp/other-worktree",
                 "work_lease": {
                     "operation_type": "author_issue_work",
                     "expires_at": "2000-01-01T00:00:00Z",
                 },
-            }, f)
+            },
+        )
         with self.assertRaises(RuntimeError) as ctx:
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertIn("Recovery review is required before takeover", str(ctx.exception))
@@ -3248,9 +3335,7 @@ class TestIssueLocking(unittest.TestCase):
            return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_create_pr_missing_lock_fails(self, _auth, _role):
-        if os.path.exists(ISSUE_LOCK_FILE):
+        with patch.dict(os.environ, self._create_pr_env(), clear=True):
-            os.remove(ISSUE_LOCK_FILE)
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             with self.assertRaises(RuntimeError) as ctx:
                 gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
         self.assertIn("Issue lock is missing", str(ctx.exception))
@@ -3259,37 +3344,64 @@ class TestIssueLocking(unittest.TestCase):
            return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_create_pr_branch_mismatch_fails(self, _auth, _role):
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+        worktree = os.path.realpath(os.getcwd())
-            json.dump(_sample_issue_lock(
+        _bind_test_lock(
-                issue_number=196, branch_name="feat/issue-196-mutations"), f)
+            issue_number=196,
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+            branch_name="feat/issue-196-mutations",
+            remote="prgs",
+            worktree_path=worktree,
+        )
+        with patch.dict(os.environ, self._create_pr_env(), clear=True):
             with self.assertRaises(ValueError) as ctx:
-                gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
+                gitea_create_pr(
+                    title="feat: X Closes #196",
+                    head="feat/issue-196-different",
+                    remote="prgs",
+                    worktree_path=worktree,
+                )
         self.assertIn("does not match locked branch", 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_forbidden_terms_fails(self, _auth, _role):
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+        worktree = os.path.realpath(os.getcwd())
-            json.dump(_sample_issue_lock(
+        _bind_test_lock(
-                issue_number=196, branch_name="feat/issue-196-mutations"), f)
+            issue_number=196,
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+            branch_name="feat/issue-196-mutations",
+            remote="prgs",
+            worktree_path=worktree,
+        )
+        with patch.dict(os.environ, self._create_pr_env(), clear=True):
             for term in ("equivalent to #196", "related to #196", "same as #196"):
                 with self.assertRaises(ValueError) as ctx:
-                    gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
+                    gitea_create_pr(
+                        title=f"feat: X {term}",
+                        head="feat/issue-196-mutations",
+                        remote="prgs",
+                        worktree_path=worktree,
+                    )
                 self.assertIn("contains forbidden term", 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_missing_closes_ref_fails(self, _auth, _role):
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+        worktree = os.path.realpath(os.getcwd())
-            json.dump(_sample_issue_lock(
+        _bind_test_lock(
-                issue_number=196, branch_name="feat/issue-196-mutations"), f)
+            issue_number=196,
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+            branch_name="feat/issue-196-mutations",
+            remote="prgs",
+            worktree_path=worktree,
+        )
+        with patch.dict(os.environ, self._create_pr_env(), clear=True):
             with self.assertRaises(ValueError) as ctx:
-                gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
+                gitea_create_pr(
+                    title="feat: X refs #196",
+                    head="feat/issue-196-mutations",
+                    remote="prgs",
+                    worktree_path=worktree,
+                )
         self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))
 
     @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
@@ -3297,13 +3409,13 @@ class TestIssueLocking(unittest.TestCase):
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_create_pr_worktree_mismatch_fails(self, _auth, _role):
         scratch = os.path.realpath("/tmp/gitea-tools-author-scratch/issue-249-pr")
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+        _bind_test_lock(
-            json.dump(_sample_issue_lock(
+            issue_number=249,
-                issue_number=249,
+            branch_name="feat/issue-249-issue-lock-scratch-worktree",
-                branch_name="feat/issue-249-issue-lock-scratch-worktree",
+            worktree_path=scratch,
-                worktree_path=scratch,
+            remote="prgs",
-            ), f)
+        )
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+        with patch.dict(os.environ, self._create_pr_env(), clear=True):
             with self.assertRaises(ValueError) as ctx:
                 gitea_create_pr(
                     title="feat: lock scratch worktree Closes #249",
@@ -3317,21 +3429,33 @@ class TestIssueLocking(unittest.TestCase):
            return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_create_pr_manual_lock_seed_blocked(self, _auth, _role):
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+        worktree = os.path.realpath(os.getcwd())
-            json.dump(
+        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,
                 ),
-                f,
             )
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             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())
 
@@ -3342,13 +3466,13 @@ class TestIssueLocking(unittest.TestCase):
     def test_create_pr_honors_scratch_worktree_lock(self, _auth, _role, mock_api):
         scratch = os.path.realpath("/tmp/gitea-tools-author-scratch/issue-249-e2e")
         mock_api.return_value = {"number": 250, "html_url": "https://example/pr/250"}
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+        _bind_test_lock(
-            json.dump(_sample_issue_lock(
+            issue_number=249,
-                issue_number=249,
+            branch_name="feat/issue-249-issue-lock-scratch-worktree",
-                branch_name="feat/issue-249-issue-lock-scratch-worktree",
+            worktree_path=scratch,
-                worktree_path=scratch,
+            remote="prgs",
-            ), f)
+        )
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+        with patch.dict(os.environ, self._create_pr_env(), clear=True):
             res = gitea_create_pr(
                 title="feat: issue-lock scratch worktree Closes #249",
                 head="feat/issue-249-issue-lock-scratch-worktree",

tests/test_worktrees.py

@@ -20,33 +20,50 @@ def run(script, *args):
             branch = arg
             break
 
-    lock_file = Path("/tmp/gitea_issue_lock.json")
+    lock_dir_ctx = None
-    created_lock = False
+    extra_env = os.environ.copy()
     if script == "worktree-start" and branch:
         import re
-        import json
+        import tempfile
+        import issue_lock_store
+
         m = re.search(r"issue-(\d+)", branch)
         if not m:
             m = re.search(r"pr-(\d+)", branch)
         issue_num = int(m.group(1)) if m else 999
-        lock_file.write_text(json.dumps({
+        lock_dir_ctx = tempfile.TemporaryDirectory()
+        extra_env["GITEA_ISSUE_LOCK_DIR"] = lock_dir_ctx.name
+        record = {
             "issue_number": issue_num,
             "branch_name": branch,
             "remote": "prgs",
             "org": "Scaled-Tech-Consulting",
-            "repo": "Gitea-Tools"
+            "repo": "Gitea-Tools",
-        }), encoding="utf-8")
+            "worktree_path": "/tmp/test-worktree",
-        created_lock = True
+            "work_lease": {
+                "operation_type": "author_issue_work",
+                "expires_at": "2999-01-01T00:00:00Z",
+            },
+        }
+        path = issue_lock_store.lock_file_path(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            issue_number=issue_num,
+            lock_dir=lock_dir_ctx.name,
+        )
+        issue_lock_store.save_lock_file(path, record)
 
     try:
         proc = subprocess.run(
             ["bash", str(SCRIPTS / script), *args],
             capture_output=True, text=True, cwd=str(REPO),
+            env=extra_env,
         )
         return proc.returncode, proc.stdout, proc.stderr
     finally:
-        if created_lock and lock_file.exists():
+        if lock_dir_ctx is not None:
-            lock_file.unlink()
+            lock_dir_ctx.cleanup()
 
 
 class TestWorktreeStart(unittest.TestCase):