feat: harden issue lock store against concurrent session clobbering (Closes #438)

Replace read-modify-write on the global /tmp/gitea_issue_lock.json pointer with
atomic per-issue locks under GITEA_ISSUE_LOCK_DIR using flock serialization,
PID/session metadata, fail-closed stale recovery, and pre-mutation freshness
re-checks in gitea_create_pr. Queue inventory now surfaces live_issue_locks.

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

gitea_mcp_server.py

@@ -538,6 +538,7 @@ 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 already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
@@ -582,14 +583,22 @@ def _parse_work_lease_timestamp(value: str | None) -> datetime | None:
 
 
 def _load_existing_issue_lock() -> dict | None:
-    if not os.path.exists(ISSUE_LOCK_FILE):
-        return None
-    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.read_lock_file(ISSUE_LOCK_FILE)
+
+
+def _load_issue_lock_for_scope(
+    *,
+    issue_number: int,
+    remote: str,
+    org: str,
+    repo: str,
+) -> dict | None:
+    return issue_lock_store.resolve_lock_for_issue(
+        issue_number=issue_number,
+        remote=remote,
+        org=org,
+        repo=repo,
+    ) or _load_existing_issue_lock()
 
 
 def _work_lease_claimant(host: str | None) -> dict:
@@ -1126,8 +1135,14 @@ def gitea_lock_issue(
     resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
         worktree_path, PROJECT_ROOT
     )
+    h, o, r = _resolve(remote, host, org, repo)
     active_lease_block = _active_work_lease_block(
-        _load_existing_issue_lock(),
+        _load_issue_lock_for_scope(
+            issue_number=issue_number,
+            remote=remote,
+            org=o,
+            repo=r,
+        ),
         issue_number=issue_number,
         branch_name=branch_name,
         worktree_path=resolved_worktree,
@@ -1152,7 +1167,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"
 
@@ -1200,25 +1214,36 @@ def gitea_lock_issue(
         worktree_path=resolved_worktree,
         host=h,
     )
-    data = {
-        "issue_number": issue_number,
-        "branch_name": branch_name,
-        "remote": remote,
-        "org": o,
-        "repo": r,
-        "worktree_path": resolved_worktree,
-        "work_lease": work_lease,
-    }
-
     try:
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
-            json.dump(data, f)
+        acquisition = issue_lock_store.acquire_issue_lock(
+            issue_number=issue_number,
+            branch_name=branch_name,
+            remote=remote,
+            org=o,
+            repo=r,
+            worktree_path=resolved_worktree,
+            work_lease=work_lease,
+            claimant=_work_lease_claimant(h),
+        )
+    except RuntimeError:
+        raise
     except Exception as e:
-        raise RuntimeError(f"Could not write issue lock file: {e}")
+        raise RuntimeError(f"Could not acquire issue lock: {e}") from e
 
     agent_artifacts = agent_temp_artifacts.find_agent_temp_artifacts_from_porcelain(
         git_state.get("porcelain_status") or ""
     )
+    competing = [
+        entry
+        for entry in issue_lock_store.list_live_locks()
+        if entry.get("issue_number") != issue_number
+    ]
+    lock_proof = issue_lock_store.format_lock_proof(
+        acquisition["record"],
+        freshness=acquisition["freshness"],
+        competing_live_locks=competing,
+        released=False,
+    )
     result = {
         "success": True,
         "message": (
@@ -1229,6 +1254,11 @@ def gitea_lock_issue(
         "branch_name": branch_name,
         "worktree_path": resolved_worktree,
         "work_lease": work_lease,
+        "lock_path": acquisition["lock_path"],
+        "session_id": acquisition["session_id"],
+        "lock_freshness": acquisition["freshness"],
+        "legacy_pointer": acquisition["legacy_pointer"],
+        "lock_proof": lock_proof,
     }
     if agent_artifacts:
         result["warnings"] = [
@@ -1288,15 +1318,19 @@ 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):
+    # ── Issue Lock Validation (Issue #194 / #196 / #438) ──
+    lock_data = _load_existing_issue_lock()
+    if not lock_data:
         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)")
+    scoped_lock = issue_lock_store.resolve_lock_for_issue(
+        issue_number=int(lock_data.get("issue_number") or 0),
+        remote=lock_data.get("remote"),
+        org=lock_data.get("org"),
+        repo=lock_data.get("repo"),
+    )
+    if scoped_lock:
+        lock_data = scoped_lock
 
     locked_issue = lock_data.get("issue_number")
     locked_branch = lock_data.get("branch_name")
@@ -1313,6 +1347,10 @@ def gitea_create_pr(
             f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)"
         )
 
+    ownership = issue_lock_store.verify_lock_for_mutation(lock_data)
+    if ownership["block"]:
+        raise ValueError(ownership["reasons"][0])
+
     # Check for forbidden terms anywhere in title/body
     forbidden_terms = ["equivalent", "related", "same as"]
     text_to_check = f"{title} {body}".lower()
@@ -2792,13 +2830,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 = _load_existing_issue_lock() or {}
 
     locked_worktree = lock_data.get("worktree_path")
     if locked_worktree:
@@ -6169,6 +6201,15 @@ def gitea_reconcile_issue_claims(
         heartbeat_lease_minutes=heartbeat_lease_minutes,
         reclaim_after_minutes=reclaim_after_minutes,
     )
+    live_locks = issue_lock_store.list_live_locks()
+    inventory["live_issue_locks"] = live_locks
+    inventory["live_issue_lock_numbers"] = sorted(
+        {
+            int(entry["issue_number"])
+            for entry in live_locks
+            if entry.get("issue_number") is not None
+        }
+    )
     inventory["cleanup_plan"] = issue_claim_heartbeat.build_cleanup_plan(inventory)
     inventory["success"] = True
     inventory["performed"] = False