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

Store per remote/org/repo/issue locks under GITEA_ISSUE_LOCK_DIR with
atomic writes and per-session binding. Integrate own-branch adoption for
lock recovery, update worktree-start and cleanup reconcile, and add tests
documenting the ban on manual global lock seeding.

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

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
-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.
+records an `author_issue_work` lease in a keyed lock file under
+`GITEA_ISSUE_LOCK_DIR` (default `~/.cache/gitea-tools/issue-locks`), one file
+per `remote` + `org` + `repo` + `issue_number`. The current MCP session binds
+its active lock through a per-process pointer so concurrent repos/issues never
+share one overwrite-prone slot (#443).
+
+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.
 
 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

gitea_mcp_server.py

@@ -538,6 +538,8 @@ import task_capability_map  # noqa: E402
 import review_proofs  # noqa: E402
 import agent_temp_artifacts
 import issue_lock_worktree  # noqa: E402
+import issue_lock_store  # noqa: E402
+import issue_lock_adoption  # noqa: E402
 import already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
@@ -548,8 +550,9 @@ 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,
-# consumed by gitea_create_pr and scripts/worktree-start.
+# Keyed issue-lock storage (#443): per remote/org/repo/issue files under
+# 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"
@@ -581,15 +584,59 @@ def _parse_work_lease_timestamp(value: str | None) -> datetime | None:
         return None
 
 
-def _load_existing_issue_lock() -> dict | None:
-    if not os.path.exists(ISSUE_LOCK_FILE):
-        return None
+def _load_existing_issue_lock(
+    *,
+    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, encoding="utf-8") as f:
-            data = json.load(f)
-        return data if isinstance(data, dict) else None
-    except Exception:
-        return None
+        return issue_lock_store.bind_session_lock(data)
+    except Exception as e:
+        raise RuntimeError(f"Could not write issue lock file: {e}") from e
 
 
 def _work_lease_claimant(host: str | None) -> dict:
@@ -679,6 +726,19 @@ def _branch_entry_name(branch: dict | str) -> str:
     return str(branch.get("name") or branch.get("ref") or "")
 
 
+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
@@ -1126,8 +1186,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(
-        _load_existing_issue_lock(),
+    h, o, r = _resolve(remote, host, org, repo)
+    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,
@@ -1152,7 +1213,6 @@ def gitea_lock_issue(
         )
 
     # 2. Check if the issue already has an open PR (reuse protection)
-    h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/pulls?state=open"
 
@@ -1186,13 +1246,24 @@ def gitea_lock_issue(
         branches = api_get_all(branch_url, auth)
     except Exception as e:
         raise RuntimeError(f"Could not list branches to verify issue lock: {e}")
-    for branch in branches:
-        name = _branch_entry_name(branch)
-        if expected_pattern in name:
-            raise ValueError(
-                f"Issue #{issue_number} already has matching branch '{name}' "
-                "(fail closed)"
-            )
+    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,
@@ -1210,11 +1281,7 @@ def gitea_lock_issue(
         "work_lease": work_lease,
     }
 
-    try:
-        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}")
+    lock_file_path = _save_issue_lock(data)
 
     agent_artifacts = agent_temp_artifacts.find_agent_temp_artifacts_from_porcelain(
         git_state.get("porcelain_status") or ""
@@ -1229,7 +1296,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): "
@@ -1288,15 +1370,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) ──
-    if not os.path.exists(ISSUE_LOCK_FILE):
-        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)")
+    # ── Issue Lock Validation (Issue #194 / #196 / #443) ──
+    lock_data = _resolve_issue_lock_for_pr(remote=remote, org=o, repo=r, head=head)
 
     locked_issue = lock_data.get("issue_number")
     locked_branch = lock_data.get("branch_name")
@@ -2792,13 +2867,7 @@ def _prepare_commit_payload_files(files: list[dict]) -> tuple[list[dict], list[d
     processed_files = []
     source_proofs = []
 
-    lock_data = {}
-    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
+    lock_data = issue_lock_store.read_session_issue_lock() or {}
 
     locked_worktree = lock_data.get("worktree_path")
     if locked_worktree:

issue_lock_adoption.py

@@ -0,0 +1,116 @@
+"""Own-branch lock adoption / recovery for ``gitea_lock_issue`` (#442 / #443).
+
+When an issue's own already-pushed branch exists, lock reacquisition must be
+allowed (adoption) instead of being treated as #400 duplicate competing work.
+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
+
+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."""
+    marker = f"issue-{issue_number}"
+    requested = (requested_branch or "").strip()
+
+    matches: list[tuple[str, str | None]] = []
+    for entry in existing_branches or []:
+        name = _branch_name(entry).strip()
+        if marker in name:
+            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 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
+    if path:
+        return issue_lock_store.read_lock_file(path.strip())
+    return issue_lock_store.read_session_issue_lock()
 
 
 def has_active_issue_lock(branch: str, lock_path: str | None = None) -> bool:
-    lock = read_issue_lock(lock_path)
-    if not lock:
-        return False
-    return (lock.get("branch_name") or "").strip() == (branch or "").strip()
+    if lock_path:
+        lock = issue_lock_store.read_lock_file(lock_path.strip())
+        if not lock:
+            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]:

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
-    echo "Error: Issue lock file '/tmp/gitea_issue_lock.json' is missing. You must lock exactly one issue before branch creation (fail closed)." >&2
+  locked_branch=$(python3 -c "
+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"
 

tests/test_agent_temp_artifacts.py

@@ -74,18 +74,20 @@ 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.api_get_all", return_value=[])
     @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_commit_payloads.py

@@ -66,7 +66,10 @@ 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
+
+        self._lock_dir = tempfile.TemporaryDirectory()
+        os.environ["GITEA_ISSUE_LOCK_DIR"] = self._lock_dir.name
         self.lock_data = {
             "issue_number": 263,
             "branch_name": "feat/issue-263-native-commit-payloads",
@@ -74,9 +77,12 @@ class TestCommitPayloads(unittest.TestCase):
             "org": "Example-Org",
             "repo": "Example-Repo",
             "worktree_path": self.locked_worktree_path,
+            "work_lease": {
+                "operation_type": "author_issue_work",
+                "expires_at": "2999-01-01T00:00:00Z",
+            },
         }
-        with open(self.lock_file_path, "w", encoding="utf-8") as fh:
-            fh.write(json.dumps(self.lock_data))
+        self.lock_file_path = issue_lock_store.bind_session_lock(self.lock_data)
 
         # Reset preflight status to bypass/pass verification in tests
         self.orig_whoami_called = mcp_server._preflight_whoami_called
@@ -93,8 +99,7 @@ class TestCommitPayloads(unittest.TestCase):
 
         self._dir.cleanup()
         self.locked_worktree_dir.cleanup()
-        if os.path.exists(self.lock_file_path):
-            os.remove(self.lock_file_path)
+        self._lock_dir.cleanup()
 
     def _env(self, profile: str) -> dict:
         return {
@@ -103,6 +108,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_lock_adoption.py

@@ -0,0 +1,68 @@
+"""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)
+
+
+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_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_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"
 
@@ -97,9 +99,6 @@ CREATE_PR_ENV = {
     ),
 }
 
-ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
-
-
 def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
     record = {
         "issue_number": issue_number,
@@ -107,11 +106,27 @@ 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": {
+            "operation_type": "author_issue_work",
+            "expires_at": "2999-01-01T00:00:00Z",
+        },
     }
     record.update(overrides)
     return record
 
 
+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
 # ---------------------------------------------------------------------------
@@ -170,18 +185,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)
-    @patch("builtins.open")
-    def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role):
-        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
+    def test_creates_pr(self, _auth, mock_api, _role):
+        worktree = os.path.realpath(os.getcwd())
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
-            result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
+        with tempfile.TemporaryDirectory() as lock_dir:
+            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")
@@ -191,30 +209,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)
-    @patch("builtins.open")
-    def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role):
-        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
+    def test_create_pr_reveal_opt_in_includes_url(self, _auth, mock_api, _role):
+        worktree = os.path.realpath(os.getcwd())
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
-        env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
-        with patch.dict(os.environ, env, clear=True):
-            result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
+        with tempfile.TemporaryDirectory() as lock_dir:
+            env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
+            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)
-    @patch("builtins.open")
-    def test_create_pr_locked_issue_mismatch_fails(self, mock_open, mock_exists, _auth, _role):
-        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
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
-            with self.assertRaises(ValueError) as ctx:
-                gitea_create_pr(title="feat: X Closes #999", head="feat/x", base="main")
+    def test_create_pr_locked_issue_mismatch_fails(self, _auth, _role):
+        worktree = os.path.realpath(os.getcwd())
+        with tempfile.TemporaryDirectory() as lock_dir:
+            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,
+                )
+                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")
 
 
 # ---------------------------------------------------------------------------
@@ -3042,13 +3072,23 @@ 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()
 
     def tearDown(self):
         self._env_patcher.stop()
-        if os.path.exists(ISSUE_LOCK_FILE):
-            os.remove(ISSUE_LOCK_FILE)
+        self._lock_dir.cleanup()
+
+    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",
@@ -3067,9 +3107,8 @@ 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))
-        with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
-            lock = json.load(f)
+        self.assertIn("lock_file_path", res)
+        lock = issue_lock_store.read_lock_file(res["lock_file_path"])
         self.assertIn("worktree_path", lock)
         self.assertIn("work_lease", lock)
 
@@ -3125,34 +3164,85 @@ class TestIssueLocking(unittest.TestCase):
         ]
         with self.assertRaises(ValueError) as ctx:
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
-        self.assertIn("already has matching branch", str(ctx.exception))
+        self.assertIn("not the requested branch", str(ctx.exception))
 
-    def test_lock_issue_blocks_active_same_operation_lease(self):
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
-            json.dump({
+    @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)
+    def test_lock_issue_adopts_exact_own_branch(self, _auth, mock_api, _git_state):
+        branch = "feat/issue-196-mutations"
+        mock_api.side_effect = [
+            [],
+            [{"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):
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
-            json.dump({
+    @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_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))
@@ -3219,9 +3309,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):
-            os.remove(ISSUE_LOCK_FILE)
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+        with patch.dict(os.environ, self._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))
@@ -3230,37 +3318,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:
-            json.dump(_sample_issue_lock(
-                issue_number=196, branch_name="feat/issue-196-mutations"), f)
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+        worktree = os.path.realpath(os.getcwd())
+        _bind_test_lock(
+            issue_number=196,
+            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:
-            json.dump(_sample_issue_lock(
-                issue_number=196, branch_name="feat/issue-196-mutations"), f)
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+        worktree = os.path.realpath(os.getcwd())
+        _bind_test_lock(
+            issue_number=196,
+            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:
-            json.dump(_sample_issue_lock(
-                issue_number=196, branch_name="feat/issue-196-mutations"), f)
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+        worktree = os.path.realpath(os.getcwd())
+        _bind_test_lock(
+            issue_number=196,
+            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",
@@ -3268,13 +3383,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:
-            json.dump(_sample_issue_lock(
-                issue_number=249,
-                branch_name="feat/issue-249-issue-lock-scratch-worktree",
-                worktree_path=scratch,
-            ), f)
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+        _bind_test_lock(
+            issue_number=249,
+            branch_name="feat/issue-249-issue-lock-scratch-worktree",
+            worktree_path=scratch,
+            remote="prgs",
+        )
+        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",
@@ -3291,13 +3406,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:
-            json.dump(_sample_issue_lock(
-                issue_number=249,
-                branch_name="feat/issue-249-issue-lock-scratch-worktree",
-                worktree_path=scratch,
-            ), f)
-        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+        _bind_test_lock(
+            issue_number=249,
+            branch_name="feat/issue-249-issue-lock-scratch-worktree",
+            worktree_path=scratch,
+            remote="prgs",
+        )
+        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_webui_queue_dashboard.py

@@ -12,12 +12,14 @@ from starlette.testclient import TestClient
 from webui.app import create_app
 from webui.queue_loader import (
     PaginationMeta,
+    QueueSnapshot,
     _classify_issue,
     _classify_pr,
     _extract_linked_issue,
     load_queue_snapshot,
     snapshot_to_dict,
 )
+from webui.queue_views import render_queue_page
 
 _RECENT = datetime.now(timezone.utc).isoformat()
 _STALE = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat()
@@ -209,36 +211,77 @@ class TestQueueRoutes(unittest.TestCase):
         self.assertIsNotNone(snapshot.fetch_error)
         self.assertEqual(len(snapshot.prs), 0)
 
-    def test_queue_fail_closed_hides_empty_state_copy(self):
-        with mock.patch("webui.queue_loader.get_auth_header", return_value=None):
-            snapshot = load_queue_snapshot()
-        with mock.patch("webui.app.load_queue_snapshot", return_value=snapshot):
-            response = self.client.get("/queue")
-        self.assertEqual(response.status_code, 200)
-        self.assertIn("Queue unavailable", response.text)
-        self.assertIn("not fetched", response.text.lower())
-        self.assertNotIn("No open items.", response.text)
-        self.assertNotIn("Open pull requests", response.text)
-        self.assertNotIn("pages_fetched", response.text)
 
-    def test_queue_successful_empty_inventory_shows_empty_state(self):
-        def _empty_prs(*_args, **_kwargs):
-            return [], _mock_pagination(0)
+def _empty_pagination() -> PaginationMeta:
+    return PaginationMeta(
+        page=1,
+        per_page=50,
+        returned_count=0,
+        has_more=False,
+        is_final_page=True,
+        inventory_complete=True,
+        pages_fetched=1,
+    )
 
-        def _empty_issues(*_args, **_kwargs):
-            return [], _mock_pagination(0)
 
+def _empty_fetch(*_args, **_kwargs):
+    return [], _empty_pagination()
+
+
+class TestQueueFailClosedUx(unittest.TestCase):
+    """Regression tests for #458 fail-closed empty-state copy."""
+
+    def test_fail_closed_view_suppresses_empty_queue_copy(self):
+        snapshot = QueueSnapshot(
+            project_id="gitea-tools",
+            repo_label="Scaled-Tech-Consulting/Gitea-Tools",
+            prs=(),
+            issues=(),
+            pr_pagination=None,
+            issue_pagination=None,
+            fetch_error="Gitea credentials unavailable for gitea.prgs.cc",
+        )
+        html = render_queue_page(snapshot)
+        self.assertIn("Queue unavailable", html)
+        self.assertIn("Not loaded", html)
+        self.assertNotIn("No open items.", html)
+        self.assertIn("pagination:</strong> unavailable", html)
+
+    def test_fail_closed_route_does_not_show_empty_queue_copy(self):
+        client = TestClient(create_app())
         snapshot = load_queue_snapshot(
-            fetch_prs=_empty_prs,
-            fetch_issues=_empty_issues,
+            fetch_prs=_empty_fetch,
+            fetch_issues=_empty_fetch,
+        )
+        snapshot = QueueSnapshot(
+            project_id=snapshot.project_id,
+            repo_label=snapshot.repo_label,
+            prs=(),
+            issues=(),
+            pr_pagination=None,
+            issue_pagination=None,
+            fetch_error="Gitea credentials unavailable for gitea.prgs.cc",
         )
         with mock.patch("webui.app.load_queue_snapshot", return_value=snapshot):
-            response = self.client.get("/queue")
+            response = client.get("/queue")
         self.assertEqual(response.status_code, 200)
+        self.assertIn("Queue unavailable", response.text)
+        self.assertNotIn("No open items.", response.text)
+
+    def test_successful_empty_inventory_shows_empty_copy(self):
+        snapshot = load_queue_snapshot(
+            fetch_prs=_empty_fetch,
+            fetch_issues=_empty_fetch,
+        )
         self.assertIsNone(snapshot.fetch_error)
-        self.assertEqual(response.text.count("No open items."), 2)
-        self.assertIn("pages_fetched", response.text)
-        self.assertIn("pagination (complete)", response.text)
+        self.assertEqual(len(snapshot.prs), 0)
+        self.assertEqual(len(snapshot.issues), 0)
+        self.assertTrue(snapshot.pr_pagination.inventory_complete)
+
+        html = render_queue_page(snapshot)
+        self.assertNotIn("Queue unavailable", html)
+        self.assertEqual(html.count("No open items."), 2)
+        self.assertIn("pagination (complete)", html)
 
 
 if __name__ == "__main__":

tests/test_worktrees.py

@@ -20,33 +20,50 @@ def run(script, *args):
             branch = arg
             break
 
-    lock_file = Path("/tmp/gitea_issue_lock.json")
-    created_lock = False
+    lock_dir_ctx = None
+    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"
-        }), encoding="utf-8")
-        created_lock = True
+            "repo": "Gitea-Tools",
+            "worktree_path": "/tmp/test-worktree",
+            "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():
-            lock_file.unlink()
+        if lock_dir_ctx is not None:
+            lock_dir_ctx.cleanup()
 
 
 class TestWorktreeStart(unittest.TestCase):

webui/queue_views.py

@@ -38,7 +38,13 @@ def _queue_table(
     title: str,
     items: tuple[QueueItem, ...],
     columns: tuple[tuple[str, str], ...],
+    fetch_failed: bool = False,
 ) -> str:
+    if fetch_failed:
+        return (
+            f"<h3>{html.escape(title)}</h3>"
+            "<p class='muted'>Not loaded — queue fetch did not complete.</p>"
+        )
     if not items:
         return f"<h3>{html.escape(title)}</h3><p class='muted'>No open items.</p>"
 
@@ -72,22 +78,12 @@ def render_queue_page(snapshot: QueueSnapshot) -> str:
             f'<div class="stub"><p><strong>Queue unavailable:</strong> '
             f"{html.escape(snapshot.fetch_error)}</p></div>"
         )
-        inventory_block = (
-            "<p class='muted'><strong>Inventory:</strong> not fetched "
-            "(fail closed — queue tables omitted).</p>"
-        )
-        return (
-            "<h2>Live queue</h2>"
-            f"<p class='meta'>Repository: <code>{html.escape(snapshot.repo_label)}</code> "
-            f"· project <code>{html.escape(snapshot.project_id)}</code></p>"
-            f"{error_block}"
-            f"{inventory_block}"
-            "<p class='muted'>Read-only MVP — claims, reviews, and merges stay in Gitea/MCP tools.</p>"
-        )
 
+    fetch_failed = bool(snapshot.fetch_error)
     pr_section = _queue_table(
         title="Open pull requests",
         items=snapshot.prs,
+        fetch_failed=fetch_failed,
         columns=(
             ("number", "#"),
             ("title", "Title"),
@@ -100,6 +96,7 @@ def render_queue_page(snapshot: QueueSnapshot) -> str:
     issue_section = _queue_table(
         title="Open issues",
         items=snapshot.issues,
+        fetch_failed=fetch_failed,
         columns=(
             ("number", "#"),
             ("title", "Title"),
@@ -113,6 +110,7 @@ def render_queue_page(snapshot: QueueSnapshot) -> str:
         "<h2>Live queue</h2>"
         f"<p class='meta'>Repository: <code>{html.escape(snapshot.repo_label)}</code> "
         f"· project <code>{html.escape(snapshot.project_id)}</code></p>"
+        f"{error_block}"
         f"{_pagination_html('PR', snapshot.pr_pagination)}"
         f"{pr_section}"
         f"{_pagination_html('Issue', snapshot.issue_pagination)}"