fix: hide empty-state copy on queue credential fail-closed (Closes #458 )

When fetch_error is set, render only the unavailable banner and an explicit not-fetched inventory note. Suppress queue tables, pagination proof, and "No open items." copy that implied a successful empty fetch. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 16:07:20 -04:00
9 changed files with 84 additions and 904 deletions
@@ -274,20 +274,12 @@ 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`
-acquires an atomic per-issue lock under `GITEA_ISSUE_LOCK_DIR` (default
+records an `author_issue_work` lease in the issue-lock payload with issue
-`/tmp/gitea_issue_locks/`) and updates the legacy session pointer at
+number, optional PR number, branch, worktree path, claimant identity/profile,
 `GITEA_ISSUE_LOCK_FILE` only when safe. The payload records issue number,
 branch, repo scope, worktree path, claimant identity/profile, PID, session id,
 created timestamp, expiry timestamp, and last heartbeat timestamp. An active
-same-issue/same-operation lease blocks duplicate work. An expired or dead-PID
+same-issue/same-operation lease blocks duplicate work. An expired lease still
-lease still blocks takeover until a recovery review records why the prior work
+blocks takeover until a recovery review records why the prior work is abandoned,
-is abandoned, completed, or unsafe to continue.
+completed, or unsafe to continue.
 Author/reviewer/reconciler final reports must include **Issue lock proof** with:
 lock acquired, lock owner, lock freshness, no competing live lock (when
 applicable), and whether the lock was released or intentionally retained.
 `gitea_lock_issue` returns a canonical `lock_proof` string for handoffs.
 `gitea_list_claim_inventory` exposes `live_issue_locks` for queue visibility.
 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
@@ -538,7 +538,6 @@ 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
@@ -583,22 +582,14 @@ def _parse_work_lease_timestamp(value: str | None) -> datetime | None:
 def _load_existing_issue_lock() -> dict | None:
-    return issue_lock_store.read_lock_file(ISSUE_LOCK_FILE)
+    if not os.path.exists(ISSUE_LOCK_FILE):
-
+        return None
-
+    try:
-def _load_issue_lock_for_scope(
+        with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
-    *,
+            data = json.load(f)
-    issue_number: int,
+        return data if isinstance(data, dict) else None
-    remote: str,
+    except Exception:
-    org: str,
+        return None
    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:
@@ -1135,14 +1126,8 @@ 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_issue_lock_for_scope(
+        _load_existing_issue_lock(),
            issue_number=issue_number,
            remote=remote,
            org=o,
            repo=r,
        ),
        issue_number=issue_number,
        branch_name=branch_name,
        worktree_path=resolved_worktree,
@@ -1167,6 +1152,7 @@ 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"
@@ -1214,36 +1200,25 @@ 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:
-        acquisition = issue_lock_store.acquire_issue_lock(
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
-            issue_number=issue_number,
+            json.dump(data, f)
            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 acquire issue lock: {e}") from 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 ""
    )
    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": (
@@ -1254,11 +1229,6 @@ 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"] = [
@@ -1318,19 +1288,15 @@ 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 / #438) ──
+    # ── Issue Lock Validation (Issue #194 / #196) ──
-    lock_data = _load_existing_issue_lock()
+    if not os.path.exists(ISSUE_LOCK_FILE):
    if not lock_data:
        raise RuntimeError("Issue lock is missing (fail closed). Call gitea_lock_issue first.")
-    scoped_lock = issue_lock_store.resolve_lock_for_issue(
+    try:
-        issue_number=int(lock_data.get("issue_number") or 0),
+        with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
-        remote=lock_data.get("remote"),
+            lock_data = json.load(f)
-        org=lock_data.get("org"),
+    except Exception as e:
-        repo=lock_data.get("repo"),
+        raise RuntimeError(f"Could not read issue lock file: {e} (fail closed)")
    )
    if scoped_lock:
        lock_data = scoped_lock
    locked_issue = lock_data.get("issue_number")
    locked_branch = lock_data.get("branch_name")
@@ -1347,10 +1313,6 @@ 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()
@@ -2830,7 +2792,13 @@ def _prepare_commit_payload_files(files: list[dict]) -> tuple[list[dict], list[d
    processed_files = []
    source_proofs = []
-    lock_data = _load_existing_issue_lock() or {}
+    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
    locked_worktree = lock_data.get("worktree_path")
    if locked_worktree:
@@ -6201,15 +6169,6 @@ 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
@@ -1,483 +0,0 @@
 """Atomic per-issue lock store (#438).
 Distinct issues use separate lock files under ``GITEA_ISSUE_LOCK_DIR`` so
 concurrent author sessions do not clobber each other. Acquisition is
 serialized per issue with ``fcntl.flock`` and fail-closed stale handling.
 """
 from __future__ import annotations
 import errno
 import fcntl
 import json
 import os
 import tempfile
 import uuid
 from contextlib import contextmanager
 from datetime import datetime, timedelta, timezone
 from typing import Any
 DEFAULT_LOCK_DIR = os.environ.get("GITEA_ISSUE_LOCK_DIR", "/tmp/gitea_issue_locks")
 LEGACY_LOCK_FILE = os.environ.get("GITEA_ISSUE_LOCK_FILE", "/tmp/gitea_issue_lock.json")
 LOCK_VERSION = 1
 class LockContentionError(RuntimeError):
    """Raised when an exclusive per-issue lock cannot be acquired."""
 def _sanitize(value: str) -> str:
    text = (value or "").strip()
    if not text:
        return "unknown"
    return "".join(c if c.isalnum() or c in "-_" else "-" for c in text)
 def lock_scope_key(remote: str, org: str, repo: str, issue_number: int) -> str:
    return "_".join(
        (
            _sanitize(remote),
            _sanitize(org),
            _sanitize(repo),
            f"issue-{issue_number}",
        )
    )
 def issue_lock_path(
    remote: str,
    org: str,
    repo: str,
    issue_number: int,
    *,
    lock_dir: str | None = None,
 ) -> str:
    base = (lock_dir or DEFAULT_LOCK_DIR).strip()
    return os.path.join(base, f"{lock_scope_key(remote, org, repo, issue_number)}.json")
 def flock_path(json_path: str) -> str:
    return f"{json_path}.lock"
 def _iso(value: datetime) -> str:
    return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
 def parse_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 is_process_alive(pid: int | None) -> bool:
    if not pid or pid <= 0:
        return False
    try:
        os.kill(int(pid), 0)
        return True
    except OSError as exc:
        return exc.errno != errno.ESRCH
    except (TypeError, ValueError):
        return False
 def read_lock_file(path: str) -> dict[str, Any] | None:
    if not os.path.exists(path):
        return None
    try:
        with open(path, encoding="utf-8") as handle:
            data = json.load(handle)
        return data if isinstance(data, dict) else None
    except Exception:
        return None
 def atomic_write_json(path: str, data: dict[str, Any]) -> None:
    directory = os.path.dirname(path) or "."
    os.makedirs(directory, exist_ok=True)
    fd, tmp_path = tempfile.mkstemp(dir=directory, prefix=".lock-", suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            json.dump(data, handle)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(tmp_path, path)
    finally:
        if os.path.exists(tmp_path):
            os.remove(tmp_path)
@contextmanager
 def _exclusive_file_lock(lock_path: str):
    os.makedirs(os.path.dirname(lock_path) or ".", exist_ok=True)
    fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
    try:
        try:
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError as exc:
            raise LockContentionError(
                f"could not acquire exclusive lock on '{lock_path}'"
            ) from exc
        yield fd
    finally:
        try:
            fcntl.flock(fd, fcntl.LOCK_UN)
        finally:
            os.close(fd)
 def _lease_expires_at(lock_data: dict[str, Any] | None) -> datetime | None:
    if not lock_data:
        return None
    lease = lock_data.get("work_lease")
    if isinstance(lease, dict):
        return parse_timestamp(lease.get("expires_at"))
    return parse_timestamp(lock_data.get("expires_at"))
 def assess_lock_freshness(
    lock_data: dict[str, Any] | None,
    *,
    now: datetime | None = None,
 ) -> dict[str, Any]:
    """Classify a lock as live, expired, stale, or absent."""
    current = now or datetime.now(timezone.utc)
    if not lock_data:
        return {
            "status": "absent",
            "live": False,
            "stale": False,
            "reason": "no lock record",
        }
    expires_at = _lease_expires_at(lock_data)
    heartbeat_at = parse_timestamp(lock_data.get("last_heartbeat_at"))
    if heartbeat_at is None and isinstance(lock_data.get("work_lease"), dict):
        heartbeat_at = parse_timestamp(lock_data["work_lease"].get("last_heartbeat_at"))
    pid = lock_data.get("pid")
    pid_alive = is_process_alive(pid) if pid is not None else False
    if expires_at and expires_at <= current:
        return {
            "status": "expired",
            "live": False,
            "stale": True,
            "reason": f"lease expired at {expires_at.isoformat()}",
            "pid_alive": pid_alive,
        }
    if pid is not None and not pid_alive:
        return {
            "status": "stale",
            "live": False,
            "stale": True,
            "reason": f"owner pid {pid} is not alive",
            "pid_alive": False,
        }
    return {
        "status": "live",
        "live": True,
        "stale": False,
        "reason": "lock heartbeat and lease are fresh",
        "pid_alive": pid_alive,
        "heartbeat_at": _iso(heartbeat_at) if heartbeat_at else None,
        "expires_at": _iso(expires_at) if expires_at else None,
    }
 def _same_owner(
    existing: dict[str, Any],
    *,
    branch_name: str,
    worktree_path: str,
    claimant: dict[str, Any] | None,
 ) -> bool:
    same_branch = existing.get("branch_name") == branch_name
    same_worktree = os.path.realpath(str(existing.get("worktree_path") or "")) == os.path.realpath(
        worktree_path
    )
    if not (same_branch and same_worktree):
        return False
    if claimant and isinstance(existing.get("claimant"), dict):
        return (
            existing["claimant"].get("profile") == claimant.get("profile")
            and existing["claimant"].get("username") == claimant.get("username")
        )
    return same_branch and same_worktree
 def update_legacy_session_pointer(record: dict[str, Any]) -> dict[str, Any]:
    """Update the legacy global pointer without clobbering another live session."""
    existing = read_lock_file(LEGACY_LOCK_FILE)
    if existing:
        freshness = assess_lock_freshness(existing)
        if freshness["status"] == "live":
            different_issue = existing.get("issue_number") != record.get("issue_number")
            different_pid = existing.get("pid") != os.getpid()
            if different_issue and different_pid and is_process_alive(existing.get("pid")):
                return {
                    "updated": False,
                    "reason": (
                        "retained global pointer for live issue "
                        f"#{existing.get('issue_number')} pid={existing.get('pid')}; "
                        f"per-issue lock at '{record.get('lock_path')}' is authoritative"
                    ),
                }
    atomic_write_json(LEGACY_LOCK_FILE, record)
    return {"updated": True, "path": LEGACY_LOCK_FILE}
 def acquire_issue_lock(
    *,
    issue_number: int,
    branch_name: str,
    remote: str,
    org: str,
    repo: str,
    worktree_path: str,
    work_lease: dict[str, Any],
    claimant: dict[str, Any] | None = None,
    lock_dir: str | None = None,
    allow_stale_recovery: bool = False,
 ) -> dict[str, Any]:
    """Acquire an atomic per-issue lock; fail closed on live competing ownership."""
    path = issue_lock_path(remote, org, repo, issue_number, lock_dir=lock_dir)
    sentinel = flock_path(path)
    now = datetime.now(timezone.utc)
    session_id = str(uuid.uuid4())
    resolved_worktree = os.path.realpath(worktree_path)
    try:
        with _exclusive_file_lock(sentinel):
            existing = read_lock_file(path)
            freshness = assess_lock_freshness(existing, now=now)
            if existing and freshness["live"]:
                if not _same_owner(
                    existing,
                    branch_name=branch_name,
                    worktree_path=resolved_worktree,
                    claimant=claimant,
                ):
                    owner = existing.get("claimant") or {}
                    raise RuntimeError(
                        f"Issue #{issue_number} already has an active lock owned by "
                        f"pid={existing.get('pid')} session={existing.get('session_id')} "
                        f"profile={owner.get('profile')} on branch "
                        f"'{existing.get('branch_name')}' (fail closed)"
                    )
            elif existing and freshness["stale"]:
                if not allow_stale_recovery:
                    raise RuntimeError(
                        f"Issue #{issue_number} has a stale lock ({freshness['reason']}). "
                        "Recovery review is required before takeover (fail closed)"
                    )
            record: dict[str, Any] = {
                "lock_version": LOCK_VERSION,
                "issue_number": issue_number,
                "branch_name": branch_name,
                "remote": remote,
                "org": org,
                "repo": repo,
                "worktree_path": resolved_worktree,
                "work_lease": work_lease,
                "pid": os.getpid(),
                "session_id": session_id,
                "claimant": claimant or {},
                "created_at": _iso(now),
                "last_heartbeat_at": _iso(now),
                "lock_path": path,
            }
            atomic_write_json(path, record)
            pointer = update_legacy_session_pointer(record)
    except LockContentionError as exc:
        competing = read_lock_file(path)
        if competing:
            freshness = assess_lock_freshness(competing, now=now)
            owner = competing.get("claimant") or {}
            raise RuntimeError(
                f"Issue #{issue_number} lock contention: {exc}; competing owner "
                f"pid={competing.get('pid')} session={competing.get('session_id')} "
                f"profile={owner.get('profile')} status={freshness['status']} (fail closed)"
            ) from exc
        raise RuntimeError(f"Issue #{issue_number} lock contention: {exc} (fail closed)") from exc
    freshness = assess_lock_freshness(record, now=now)
    return {
        "acquired": True,
        "lock_path": path,
        "session_id": session_id,
        "freshness": freshness,
        "legacy_pointer": pointer,
        "lock_proof": format_lock_proof(record, freshness=freshness),
        "record": record,
    }
 def resolve_lock_for_issue(
    *,
    issue_number: int,
    remote: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    lock_dir: str | None = None,
 ) -> dict[str, Any] | None:
    """Load the per-issue lock, falling back to the legacy global pointer."""
    if remote and org and repo:
        scoped = read_lock_file(
            issue_lock_path(remote, org, repo, issue_number, lock_dir=lock_dir)
        )
        if scoped:
            return scoped
    legacy = read_lock_file(LEGACY_LOCK_FILE)
    if legacy and legacy.get("issue_number") == issue_number:
        return legacy
    return None
 def resolve_lock_for_branch(
    branch_name: str,
    *,
    lock_dir: str | None = None,
 ) -> dict[str, Any] | None:
    """Resolve a lock using the issue number embedded in a branch name."""
    import re
    match = re.search(r"issue-(\d+)", branch_name or "", re.IGNORECASE)
    if not match:
        legacy = read_lock_file(LEGACY_LOCK_FILE)
        if legacy and legacy.get("branch_name") == branch_name:
            return legacy
        return None
    issue_number = int(match.group(1))
    base = (lock_dir or DEFAULT_LOCK_DIR).strip()
    if os.path.isdir(base):
        suffix = f"_issue-{issue_number}.json"
        for name in os.listdir(base):
            if name.endswith(suffix):
                record = read_lock_file(os.path.join(base, name))
                if record and record.get("branch_name") == branch_name:
                    return record
    legacy = read_lock_file(LEGACY_LOCK_FILE)
    if legacy and legacy.get("issue_number") == issue_number:
        return legacy
    return None
 def verify_lock_for_mutation(
    lock_data: dict[str, Any] | None,
    *,
    issue_number: int | None = None,
    branch_name: str | None = None,
    worktree_path: str | None = None,
 ) -> dict[str, Any]:
    """Re-check lock ownership immediately before a mutation (#438)."""
    reasons: list[str] = []
    if not lock_data:
        return {"proven": False, "block": True, "reasons": ["issue lock is missing (fail closed)"]}
    freshness = assess_lock_freshness(lock_data)
    if not freshness["live"]:
        reasons.append(f"issue lock is not live: {freshness['reason']} (fail closed)")
    if issue_number is not None and lock_data.get("issue_number") != issue_number:
        reasons.append(
            f"issue lock targets #{lock_data.get('issue_number')}, expected #{issue_number} (fail closed)"
        )
    if branch_name is not None and lock_data.get("branch_name") != branch_name:
        reasons.append(
            f"issue lock branch '{lock_data.get('branch_name')}' does not match "
            f"'{branch_name}' (fail closed)"
        )
    if worktree_path is not None:
        locked = os.path.realpath(str(lock_data.get("worktree_path") or ""))
        declared = os.path.realpath(worktree_path)
        if locked != declared:
            reasons.append(
                f"issue lock worktree '{locked}' does not match declared '{declared}' (fail closed)"
            )
    return {
        "proven": not reasons,
        "block": bool(reasons),
        "reasons": reasons,
        "freshness": freshness,
        "lock_proof": format_lock_proof(lock_data, freshness=freshness),
    }
 def list_live_locks(*, lock_dir: str | None = None, now: datetime | None = None) -> list[dict[str, Any]]:
    """Return live per-issue locks for queue visibility."""
    base = (lock_dir or DEFAULT_LOCK_DIR).strip()
    if not os.path.isdir(base):
        return []
    live: list[dict[str, Any]] = []
    for name in sorted(os.listdir(base)):
        if not name.endswith(".json"):
            continue
        record = read_lock_file(os.path.join(base, name))
        if not record:
            continue
        freshness = assess_lock_freshness(record, now=now)
        if freshness["live"]:
            live.append(
                {
                    "issue_number": record.get("issue_number"),
                    "branch_name": record.get("branch_name"),
                    "remote": record.get("remote"),
                    "org": record.get("org"),
                    "repo": record.get("repo"),
                    "worktree_path": record.get("worktree_path"),
                    "pid": record.get("pid"),
                    "session_id": record.get("session_id"),
                    "claimant": record.get("claimant"),
                    "freshness": freshness,
                    "lock_path": record.get("lock_path") or os.path.join(base, name),
                }
            )
    return live
 def format_lock_proof(
    lock_data: dict[str, Any] | None,
    *,
    freshness: dict[str, Any] | None = None,
    competing_live_locks: list[dict[str, Any]] | None = None,
    released: bool | None = None,
 ) -> str:
    """Canonical issue-lock proof string for final reports."""
    if not lock_data:
        return "issue lock proof: not acquired"
    fresh = freshness or assess_lock_freshness(lock_data)
    owner = lock_data.get("claimant") or {}
    parts = [
        "issue lock proof:",
        f"acquired issue #{lock_data.get('issue_number')}",
        f"branch {lock_data.get('branch_name')}",
        f"owner {owner.get('profile') or 'unknown'}",
        f"pid {lock_data.get('pid')}",
        f"session {lock_data.get('session_id')}",
        f"freshness {fresh.get('status')}",
    ]
    if competing_live_locks is not None:
        parts.append(
            "no competing live lock"
            if not competing_live_locks
            else f"competing live locks {len(competing_live_locks)}"
        )
    if released is True:
        parts.append("lock released")
    elif released is False:
        parts.append("lock retained")
    return "; ".join(parts)
@@ -13,7 +13,6 @@ 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")
@@ -50,10 +49,6 @@ def read_issue_lock(path: str | None = None) -> dict[str, Any] | None:
 def has_active_issue_lock(branch: str, lock_path: str | None = None) -> bool:
    lock = issue_lock_store.resolve_lock_for_branch(branch)
    if lock and (lock.get("branch_name") or "").strip() == (branch or "").strip():
        freshness = issue_lock_store.assess_lock_freshness(lock)
        return freshness["live"]
    lock = read_issue_lock(lock_path)
    if not lock:
        return False
@@ -37,27 +37,14 @@ 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
-  locked_branch=$(PYTHONPATH="$repo_root" python3 - <<'PY' "$branch")
+  if [[ ! -f "/tmp/gitea_issue_lock.json" ]]; then
-import sys
+    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 issue_lock_store
 branch = sys.argv[1]
 lock = issue_lock_store.resolve_lock_for_branch(branch)
 if not lock:
    print("", end="")
    sys.exit(1)
 print(lock.get("branch_name", ""), end="")
 PY
  )
  if [[ -z "$locked_branch" ]]; then
    echo "Error: No issue lock found for branch '$branch'. Lock the 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
@@ -81,6 +68,8 @@ 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"
@@ -1,206 +0,0 @@
 """Tests for atomic per-issue lock store (#438)."""
 from __future__ import annotations
 import os
 import sys
 import tempfile
 import threading
 import unittest
 from datetime import datetime, timedelta, timezone
 from unittest.mock import patch
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 import issue_lock_store  # noqa: E402
 def _lease(**overrides):
    now = datetime.now(timezone.utc)
    lease = {
        "operation_type": "author_issue_work",
        "issue_number": 438,
        "branch": "feat/issue-438-lock-hardening",
        "worktree_path": "/tmp/wt",
        "expires_at": (now + timedelta(hours=4)).isoformat().replace("+00:00", "Z"),
        "last_heartbeat_at": now.isoformat().replace("+00:00", "Z"),
    }
    lease.update(overrides)
    return lease
 class TestIssueLockStore(unittest.TestCase):
    def setUp(self):
        self.tempdir = tempfile.mkdtemp(prefix="issue-lock-store-")
        self.legacy = os.path.join(self.tempdir, "legacy.json")
        self.addCleanup(self._cleanup)
    def _cleanup(self):
        for root, _dirs, files in os.walk(self.tempdir, topdown=False):
            for name in files:
                try:
                    os.remove(os.path.join(root, name))
                except OSError:
                    pass
            try:
                os.rmdir(root)
            except OSError:
                pass
    def _acquire(self, issue_number=438, branch="feat/issue-438-lock-hardening", **kwargs):
        worktree_path = kwargs.pop("worktree_path", "/tmp/wt")
        return issue_lock_store.acquire_issue_lock(
            issue_number=issue_number,
            branch_name=branch,
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            worktree_path=worktree_path,
            work_lease=_lease(issue_number=issue_number, branch=branch),
            claimant={"profile": "prgs-author", "username": "jcwalker3"},
            lock_dir=self.tempdir,
            **kwargs,
        )
    def test_acquire_writes_per_issue_lock_and_legacy_pointer(self):
        with patch.object(issue_lock_store, "LEGACY_LOCK_FILE", self.legacy):
            result = self._acquire()
        path = issue_lock_store.issue_lock_path("prgs", "Scaled-Tech-Consulting", "Gitea-Tools", 438, lock_dir=self.tempdir)
        self.assertTrue(os.path.exists(path))
        self.assertTrue(result["acquired"])
        self.assertIn("lock_proof", result)
        with open(self.legacy, encoding="utf-8") as handle:
            legacy = __import__("json").load(handle)
        self.assertEqual(legacy["issue_number"], 438)
        self.assertEqual(legacy["session_id"], result["session_id"])
    def test_concurrent_acquire_same_issue_only_one_wins(self):
        barrier = threading.Barrier(2)
        results: list[dict | Exception] = []
        def worker():
            barrier.wait()
            try:
                results.append(self._acquire())
            except Exception as exc:  # noqa: BLE001
                results.append(exc)
        threads = [threading.Thread(target=worker) for _ in range(2)]
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join()
        successes = [item for item in results if isinstance(item, dict)]
        failures = [item for item in results if isinstance(item, Exception)]
        self.assertEqual(len(successes), 1)
        self.assertEqual(len(failures), 1)
        failure_text = str(failures[0]).lower()
        self.assertTrue(
            "active lock" in failure_text or "lock contention" in failure_text,
            failures[0],
        )
    def test_distinct_issues_acquire_without_contention(self):
        first = self._acquire(issue_number=438, branch="feat/issue-438-lock-hardening")
        second = self._acquire(
            issue_number=440,
            branch="feat/issue-440-recovery",
            worktree_path="/tmp/other-wt",
        )
        self.assertTrue(first["acquired"])
        self.assertTrue(second["acquired"])
        self.assertNotEqual(first["session_id"], second["session_id"])
    def test_stale_lock_requires_explicit_recovery(self):
        path = issue_lock_store.issue_lock_path("prgs", "Scaled-Tech-Consulting", "Gitea-Tools", 438, lock_dir=self.tempdir)
        stale = {
            "lock_version": 1,
            "issue_number": 438,
            "branch_name": "feat/issue-438-old",
            "remote": "prgs",
            "org": "Scaled-Tech-Consulting",
            "repo": "Gitea-Tools",
            "worktree_path": "/tmp/old",
            "pid": 999999,
            "session_id": "stale-session",
            "work_lease": _lease(expires_at="2000-01-01T00:00:00Z"),
            "last_heartbeat_at": "2000-01-01T00:00:00Z",
            "lock_path": path,
        }
        issue_lock_store.atomic_write_json(path, stale)
        with self.assertRaises(RuntimeError) as ctx:
            self._acquire()
        self.assertIn("Recovery review is required", str(ctx.exception))
        recovered = self._acquire(allow_stale_recovery=True)
        self.assertTrue(recovered["acquired"])
    def test_verify_lock_for_mutation_blocks_stale_lock(self):
        record = {
            "issue_number": 438,
            "branch_name": "feat/issue-438-lock-hardening",
            "worktree_path": "/tmp/wt",
            "pid": 999999,
            "work_lease": _lease(expires_at="2000-01-01T00:00:00Z"),
            "last_heartbeat_at": "2000-01-01T00:00:00Z",
        }
        result = issue_lock_store.verify_lock_for_mutation(
            record,
            issue_number=438,
            branch_name="feat/issue-438-lock-hardening",
            worktree_path="/tmp/wt",
        )
        self.assertTrue(result["block"])
        self.assertIn("not live", result["reasons"][0])
    def test_list_live_locks_excludes_stale_records(self):
        path = issue_lock_store.issue_lock_path("prgs", "Scaled-Tech-Consulting", "Gitea-Tools", 438, lock_dir=self.tempdir)
        issue_lock_store.atomic_write_json(
            path,
            {
                "issue_number": 438,
                "branch_name": "feat/issue-438-lock-hardening",
                "remote": "prgs",
                "org": "Scaled-Tech-Consulting",
                "repo": "Gitea-Tools",
                "worktree_path": "/tmp/wt",
                "pid": os.getpid(),
                "session_id": "live",
                "work_lease": _lease(),
                "last_heartbeat_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
                "lock_path": path,
            },
        )
        stale_path = issue_lock_store.issue_lock_path("prgs", "Scaled-Tech-Consulting", "Gitea-Tools", 440, lock_dir=self.tempdir)
        issue_lock_store.atomic_write_json(
            stale_path,
            {
                "issue_number": 440,
                "branch_name": "feat/issue-440-recovery",
                "remote": "prgs",
                "org": "Scaled-Tech-Consulting",
                "repo": "Gitea-Tools",
                "worktree_path": "/tmp/wt",
                "pid": 999999,
                "session_id": "stale",
                "work_lease": _lease(issue_number=440, branch="feat/issue-440-recovery", expires_at="2000-01-01T00:00:00Z"),
                "last_heartbeat_at": "2000-01-01T00:00:00Z",
                "lock_path": stale_path,
            },
        )
        live = issue_lock_store.list_live_locks(lock_dir=self.tempdir)
        self.assertEqual([entry["issue_number"] for entry in live], [438])
    def test_resolve_lock_for_branch_reads_per_issue_file(self):
        self._acquire()
        resolved = issue_lock_store.resolve_lock_for_branch(
            "feat/issue-438-lock-hardening",
            lock_dir=self.tempdir,
        )
        self.assertIsNotNone(resolved)
        self.assertEqual(resolved["issue_number"], 438)
 if __name__ == "__main__":
    unittest.main()
@@ -3042,38 +3042,13 @@ class TestIssueLocking(unittest.TestCase):
    """Test issue locking and PR gating constraints."""
    def setUp(self):
-        import shutil
+        self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
        import tempfile
        self._lock_dir = tempfile.mkdtemp(prefix="mcp-issue-lock-")
        self._session_lock = os.path.join(self._lock_dir, "session.json")
        env = {
            **ISSUE_WRITE_ENV,
            "GITEA_ISSUE_LOCK_DIR": self._lock_dir,
            "GITEA_ISSUE_LOCK_FILE": self._session_lock,
        }
        self._env_patcher = patch.dict(os.environ, env, clear=True)
        self._env_patcher.start()
        self._patchers = [
            patch("mcp_server.ISSUE_LOCK_FILE", self._session_lock),
            patch("mcp_server.issue_lock_store.DEFAULT_LOCK_DIR", self._lock_dir),
            patch("mcp_server.issue_lock_store.LEGACY_LOCK_FILE", self._session_lock),
            patch("tests.test_mcp_server.ISSUE_LOCK_FILE", self._session_lock),
            patch(
                "mcp_server.issue_lock_worktree.read_worktree_git_state",
                return_value=_clean_master_git_state_for_lock(),
            ),
        ]
        for patcher in self._patchers:
            patcher.start()
    def tearDown(self):
        import shutil
        for patcher in reversed(getattr(self, "_patchers", [])):
            patcher.stop()
        self._env_patcher.stop()
-        shutil.rmtree(getattr(self, "_lock_dir", ""), ignore_errors=True)
+        if os.path.exists(ISSUE_LOCK_FILE):
            os.remove(ISSUE_LOCK_FILE)
    @patch(
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
@@ -12,14 +12,12 @@ 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()
@@ -211,77 +209,36 @@ 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):
-def _empty_pagination() -> PaginationMeta:
+        with mock.patch("webui.queue_loader.get_auth_header", return_value=None):
-    return PaginationMeta(
+            snapshot = load_queue_snapshot()
        page=1,
        per_page=50,
        returned_count=0,
        has_more=False,
        is_final_page=True,
        inventory_complete=True,
        pages_fetched=1,
    )
 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_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 = client.get("/queue")
+            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_issues(*_args, **_kwargs):
            return [], _mock_pagination(0)
    def test_successful_empty_inventory_shows_empty_copy(self):
        snapshot = load_queue_snapshot(
-            fetch_prs=_empty_fetch,
+            fetch_prs=_empty_prs,
-            fetch_issues=_empty_fetch,
+            fetch_issues=_empty_issues,
        )
        with mock.patch("webui.app.load_queue_snapshot", return_value=snapshot):
            response = self.client.get("/queue")
        self.assertEqual(response.status_code, 200)
        self.assertIsNone(snapshot.fetch_error)
-        self.assertEqual(len(snapshot.prs), 0)
+        self.assertEqual(response.text.count("No open items."), 2)
-        self.assertEqual(len(snapshot.issues), 0)
+        self.assertIn("pages_fetched", response.text)
-        self.assertTrue(snapshot.pr_pagination.inventory_complete)
+        self.assertIn("pagination (complete)", response.text)
        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__":
@@ -38,13 +38,7 @@ 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>"
@@ -78,12 +72,22 @@ 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"),
@@ -96,7 +100,6 @@ 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"),
@@ -110,7 +113,6 @@ 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)}"