feat: non-destructive branch recovery after lost issue locks (Closes #440)

Implements the #420 recovery umbrella: keyed persistent issue locks with
own-branch adoption, structured branch ownership parsing, create_pr lock
resolution after MCP restart, and workflow docs forbidding destructive
branch deletion as the default recovery path.

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

docs/llm-workflow-runbooks.md

@@ -274,12 +274,26 @@ 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 (#440). Branch ownership
+uses structured ``(fix|feat|docs|chore)/issue-<n>-`` parsing — not broad
+substring matches like ``issue-420`` inside ``issue-4200``.
 
 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

docs/webui-local-dev.md

@@ -0,0 +1,89 @@
+# Internal web UI — local development (#426)
+
+Read-only MVP skeleton for the MCP Control Plane operator console. Gitea,
+MCP capability gates, and `skills/llm-project-workflow/` remain the source of
+truth; this UI only provides route stubs and layout.
+
+## Prerequisites
+
+- Python 3.11+ with project dependencies installed (`pip install -r requirements.txt`)
+- No secrets in repo, config, or client bundle
+
+## Start the server
+
+From the repository root (or an issue worktree):
+
+```bash
+./scripts/run-webui
+```
+
+Or directly:
+
+```bash
+python3 -m webui
+```
+
+Optional environment variables:
+
+| Variable | Default | Purpose |
+|----------|---------|---------|
+| `WEBUI_HOST` | `127.0.0.1` | Bind address (keep local for MVP) |
+| `WEBUI_PORT` | `8765` | Listen port |
+
+## Routes (MVP)
+
+| Path | Description |
+|------|-------------|
+| `/` | Home / operator overview |
+| `/health` | JSON liveness (`status`, `service`, `mode`, `timestamp`) |
+| `/queue` | Live PR and issue queue dashboard (#429) |
+| `/api/queue` | JSON queue export with pagination metadata |
+| `/projects` | Project registry list (#427) |
+| `/projects/{id}` | Project detail + onboarding checklist |
+| `/api/projects` | JSON registry export |
+| `/prompts` | Prompt library with per-prompt copy buttons (#428) |
+| `/api/prompts` | JSON prompt export with workflow hashes |
+| `/runtime` | Stub — MCP runtime health (#430) |
+| `/audit` | Stub — report audit paste (#431) |
+| `/worktrees` | Stub — hygiene dashboard (#432) |
+| `/leases` | Stub — lease visibility (#433) |
+
+All routes are GET-only. POST/PUT/PATCH/DELETE return `405` with
+`read-only-mvp`.
+
+## Project registry (#427)
+
+Versioned registry file: `webui/data/projects.registry.json` (schema version `1`).
+
+Override path with `WEBUI_PROJECT_REGISTRY` when operators keep a machine-local
+copy outside git. The registry stores repo identity, remotes, profile names,
+workflow/schema path references, and onboarding checklist steps — never tokens
+or credentials.
+
+Seed entry: **Gitea-Tools** on `https://gitea.prgs.cc` with `prgs-author`,
+`prgs-reviewer`, and `prgs-reconciler` profiles.
+
+## Prompt library (#428)
+
+Prompts are generated at load time from canonical workflow files under
+`skills/llm-project-workflow/workflows/`. SHA-256 hashes are computed from
+`WEBUI_REPO_ROOT` (defaults to the repository root). Prompt bodies are short
+copy/paste starters; canonical workflow files remain the only full policy
+source.
+
+## Live queue dashboard (#429)
+
+`/queue` loads open PRs and issues for the default registry project (seed:
+**Gitea-Tools** on `https://gitea.prgs.cc`) using existing `gitea_auth` read
+credentials. The UI surfaces pagination proof (returned count, pages fetched,
+`has_more`, `inventory_complete`) and classification badges (`claimed`,
+`blocked`, `in-review`, `duplicate`) when evidence exists.
+
+If credentials are missing or the fetch fails, the page shows an explicit error
+instead of an empty queue (fail closed).
+
+## Tests
+
+```bash
+pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py -q
+```

gitea_mcp_server.py

@@ -440,10 +440,48 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
             "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
         )
 
-    if worktree_path:
-        dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(worktree_path)))
+    workspace = author_mutation_worktree.resolve_mutation_workspace(
+        worktree_path,
+        PROJECT_ROOT,
+        active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
+        author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
+    )
+    real_workspace = os.path.realpath(workspace)
+    real_root = os.path.realpath(PROJECT_ROOT)
+
+    if real_workspace != real_root:
+        if not _preflight_in_test_mode():
+            if not os.path.exists(real_workspace):
+                raise RuntimeError(
+                    f"Branches-only mutation guard (#274): worktree path '{workspace}' does not exist (fail closed)"
+                )
+            if not os.path.isdir(real_workspace):
+                raise RuntimeError(
+                    f"Branches-only mutation guard (#274): worktree path '{workspace}' is not a directory (fail closed)"
+                )
+            try:
+                res = subprocess.run(
+                    ["git", "-C", real_workspace, "rev-parse", "--git-common-dir"],
+                    capture_output=True,
+                    text=True,
+                    check=True,
+                )
+                common_dir = os.path.realpath(res.stdout.strip())
+                expected_dir = os.path.realpath(os.path.join(real_root, ".git"))
+                if common_dir != expected_dir:
+                    raise RuntimeError(
+                        f"Branches-only mutation guard (#274): worktree '{workspace}' does not belong to the target repository '{PROJECT_ROOT}' (fail closed)"
+                    )
+            except Exception as e:
+                if isinstance(e, RuntimeError):
+                    raise e
+                raise RuntimeError(
+                    f"Branches-only mutation guard (#274): worktree '{workspace}' is not a valid git repository (fail closed)"
+                )
+
+        dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(workspace)))
         if dirty_files:
-            details = _preflight_workspace_details(worktree_path, dirty_files)
+            details = _preflight_workspace_details(workspace, dirty_files)
             raise RuntimeError(
                 "Pre-flight order violation: Active task workspace has tracked "
                 "file edits before mutation (fail closed). "
@@ -499,7 +537,10 @@ import role_namespace_gate  # noqa: E402
 import task_capability_map  # noqa: E402
 import review_proofs  # noqa: E402
 import agent_temp_artifacts
+import issue_branch_ownership  # noqa: E402
 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
@@ -510,8 +551,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"
@@ -543,15 +585,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:
@@ -641,6 +727,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
@@ -972,6 +1071,7 @@ def gitea_create_issue(
     repo: str | None = None,
     allow_duplicate_override: bool = False,
     split_from_issue: int | None = None,
+    worktree_path: str | None = None,
 ) -> dict:
     """Create a new issue on a Gitea repository.
 
@@ -984,6 +1084,7 @@ def gitea_create_issue(
         repo: Override the repository name.
         allow_duplicate_override: Operator-approved split after duplicate found.
         split_from_issue: Existing duplicate issue number when overriding.
+        worktree_path: Optional path to verify branches-only guard.
 
     Returns:
         dict with 'number' of the created issue ('url' only with the reveal opt-in).
@@ -1010,7 +1111,7 @@ def gitea_create_issue(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, worktree_path=worktree_path)
     base = repo_api_url(h, o, r)
     open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
     closed_issues = api_get_all(
@@ -1086,8 +1187,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,
@@ -1112,7 +1214,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"
 
@@ -1126,7 +1227,7 @@ def gitea_lock_issue(
         pr_title = pr.get("title", "")
         pr_body = pr.get("body", "")
 
-        if expected_pattern in pr_head:
+        if issue_branch_ownership.branch_tracks_issue(pr_head, issue_number):
             raise ValueError(
                 f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}, branch '{pr_head}') (fail closed)"
             )
@@ -1146,13 +1247,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,
@@ -1170,11 +1282,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 ""
@@ -1189,7 +1297,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): "
@@ -1248,15 +1371,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")
@@ -2752,13 +2868,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_branch_ownership.py

@@ -0,0 +1,31 @@
+"""Structured issue/branch ownership evidence (#440).
+
+Author branches must follow ``(fix|feat|docs|chore)/issue-<n>-<desc>``. Duplicate-work
+and recovery gates use this parser instead of broad substring checks like
+``issue-420`` inside unrelated names (for example ``issue-4200``).
+"""
+
+from __future__ import annotations
+
+import re
+
+IMPLEMENTATION_BRANCH_RE = re.compile(
+    r"^(?:fix|feat|docs|chore)/issue-(\d+)(?:-.+)?$"
+)
+
+
+def parse_tracked_issue_number(branch_name: str) -> int | None:
+    """Return the issue number encoded in a canonical author branch name."""
+    text = (branch_name or "").strip()
+    if not text:
+        return None
+    match = IMPLEMENTATION_BRANCH_RE.match(text)
+    if not match:
+        return None
+    return int(match.group(1))
+
+
+def branch_tracks_issue(branch_name: str, issue_number: int) -> bool:
+    """True when ``branch_name`` structurally belongs to ``issue_number``."""
+    parsed = parse_tracked_issue_number(branch_name)
+    return parsed == issue_number if parsed is not None else False

issue_lock_adoption.py

@@ -0,0 +1,117 @@
+"""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
+
+import issue_branch_ownership
+
+ADOPT = "adopt_existing_branch"
+BLOCK_COMPETING = "block_competing_branch"
+NO_MATCH = "no_matching_branch"
+
+
+def _branch_name(entry) -> str:
+    if isinstance(entry, dict):
+        return str(entry.get("name") or "")
+    return str(entry or "")
+
+
+def _branch_sha(entry) -> str | None:
+    if isinstance(entry, dict):
+        sha = entry.get("commit_sha")
+        if sha:
+            return str(sha)
+    return None
+
+
+def assess_own_branch_adoption(
+    *,
+    issue_number: int,
+    requested_branch: str,
+    existing_branches,
+) -> dict:
+    """Decide whether an existing matching branch is adoptable."""
+    requested = (requested_branch or "").strip()
+
+    matches: list[tuple[str, str | None]] = []
+    for entry in existing_branches or []:
+        name = _branch_name(entry).strip()
+        if issue_branch_ownership.branch_tracks_issue(name, issue_number):
+            matches.append((name, _branch_sha(entry)))
+
+    competing = sorted({name for name, _ in matches if name != requested})
+    exact = [(name, sha) for name, sha in matches if name == requested]
+
+    if competing:
+        return {
+            "outcome": BLOCK_COMPETING,
+            "adopt": False,
+            "block": True,
+            "reason": (
+                f"issue #{issue_number} already has matching branch(es) "
+                f"{competing} that are not the requested branch "
+                f"'{requested}' (fail closed)"
+            ),
+            "matched_branch": None,
+            "matched_head_sha": None,
+            "competing_branches": competing,
+        }
+
+    if exact:
+        name, sha = exact[0]
+        return {
+            "outcome": ADOPT,
+            "adopt": True,
+            "block": False,
+            "reason": (
+                f"existing branch '{name}' is the exact requested branch for "
+                f"issue #{issue_number}; adopting it for lock recovery"
+            ),
+            "matched_branch": name,
+            "matched_head_sha": sha,
+            "competing_branches": [],
+        }
+
+    return {
+        "outcome": NO_MATCH,
+        "adopt": False,
+        "block": False,
+        "reason": f"no existing branch matches issue #{issue_number}",
+        "matched_branch": None,
+        "matched_head_sha": None,
+        "competing_branches": [],
+    }
+
+
+def build_adoption_proof(
+    *,
+    issue_number: int,
+    branch_name: str,
+    assessment: dict,
+    open_pr_checked: bool,
+    competing_lock_checked: bool,
+    lock_file_path: str,
+    lock_file_status: str,
+) -> dict:
+    """Assemble the proof block returned by ``gitea_lock_issue`` on adoption."""
+    return {
+        "issue_number": issue_number,
+        "branch_name": branch_name,
+        "branch_head_commit": assessment.get("matched_head_sha"),
+        "adoption_reason": assessment.get("reason"),
+        "no_existing_pr_proof": bool(open_pr_checked),
+        "no_competing_live_lock_proof": bool(competing_lock_checked),
+        "lock_file_path": lock_file_path,
+        "lock_file_status": lock_file_status,
+    }

issue_lock_store.py

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

merged_cleanup_reconcile.py

@@ -13,9 +13,9 @@ import subprocess
 from typing import Any
 
 from reviewer_worktree import parse_dirty_tracked_files
+import issue_lock_store
 
 PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})
-ISSUE_LOCK_FILE = os.environ.get("GITEA_ISSUE_LOCK_FILE", "/tmp/gitea_issue_lock.json")
 CLOSES_FIXES_RE = re.compile(r"\b(?:closes|fixes)\s+#(\d+)\b", re.IGNORECASE)
 
 
@@ -37,22 +37,18 @@ def resolve_worktree_path(project_root: str, branch: str) -> str:
 
 
 def read_issue_lock(path: str | None = None) -> dict[str, Any] | None:
-    lock_path = (path or ISSUE_LOCK_FILE).strip()
-    if 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/run-webui

@@ -0,0 +1,6 @@
+#!/usr/bin/env bash
+set -euo pipefail
+script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+repo_root="$(cd "$script_dir/.." && pwd)"
+cd "$repo_root"
+exec python3 -m webui "$@"

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"
 

skills/llm-project-workflow/workflows/work-issue.md

@@ -309,6 +309,15 @@ Do not implement unclaimed work.
 
 If the claim/lock gates are broken, produce a recovery handoff.
 
+### Lost lock recovery after push (non-destructive)
+
+If the MCP server restarts after the issue branch was pushed but before PR creation:
+
+* **Do not** delete the remote branch as the normal recovery path.
+* Call `gitea_lock_issue` again with the same issue number and exact branch name. Own-branch adoption rebinds the session when open-PR, competing-lock, and worktree safety checks pass.
+* Call `gitea_create_pr` afterward. Durable keyed locks resolve from the session pointer or by matching the PR `head` branch without manual lock seeding.
+* If adoption is blocked by an open PR, a competing live lease, or a different same-issue branch, stop and produce a recovery handoff.
+
 Create a tooling issue only if this run is explicitly authorized to switch to issue-creation mode and exact `create_issue` capability is proven.
 
 Report:

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_create_issue_workspace_guard.py

@@ -0,0 +1,145 @@
+import os
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch, MagicMock
+
+# Ensure we import from the repo root
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import gitea_mcp_server as srv
+
+FAKE_AUTH = {"Authorization": "token test-token"}
+# Stable control checkout (parent of branches/), not the MCP server worktree root.
+CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3])
+PROJECT_ROOT = srv.PROJECT_ROOT
+
+
+class TestCreateIssueWorkspaceGuard(unittest.TestCase):
+
+    def setUp(self):
+        # Reset preflight flags
+        srv._preflight_whoami_called = True
+        srv._preflight_capability_called = True
+        srv._preflight_resolved_role = "author"
+        srv._preflight_whoami_violation = False
+        srv._preflight_capability_violation = False
+        
+        # Disable early return in verify_preflight_purity for testing
+        self._orig_in_test = srv._preflight_in_test_mode
+        srv._preflight_in_test_mode = lambda: False
+
+    def tearDown(self):
+        srv._preflight_in_test_mode = self._orig_in_test
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server.api_get_all", return_value=[])
+    @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
+    def test_create_issue_stable_checkout_rejected(self, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
+        # Without worktree_path/env hints, workspace resolves to PROJECT_ROOT. When that
+        # path is the stable control checkout (not under branches/), mutation must fail.
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with self.assertRaises(RuntimeError) as ctx:
+                srv.gitea_create_issue(title="Test issue", body="body text")
+        self.assertIn("stable control checkout", str(ctx.exception))
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server.api_get_all", return_value=[])
+    @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
+    @patch("os.path.exists", return_value=True)
+    @patch("os.path.isdir", return_value=True)
+    @patch("subprocess.run")
+    def test_create_issue_valid_worktree_succeeds(self, mock_run, mock_isdir, mock_exists, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
+        # Mock subprocess.run for git --git-common-dir to return PROJECT_ROOT/.git
+        mock_res = MagicMock()
+        mock_res.stdout = f"{CONTROL_CHECKOUT_ROOT}/.git\n"
+        mock_run.return_value = mock_res
+
+        mock_api.return_value = {"number": 42, "html_url": "https://gitea.example.com/issues/42"}
+
+        # Provide a valid branches path under the control checkout root
+        valid_path = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "feat-issue-1")
+        
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""):
+                res = srv.gitea_create_issue(
+                    title="Test issue", body="body", worktree_path=valid_path
+                )
+            
+        self.assertEqual(res["number"], 42)
+        mock_api.assert_called_once()
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server.api_get_all", return_value=[])
+    @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
+    def test_create_issue_missing_worktree_fails_closed(self, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
+        # Path under branches/ but doesn't exist
+        missing_path = os.path.join(
+            CONTROL_CHECKOUT_ROOT, "branches", "nonexistent-worktree-path-999"
+        )
+        
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with self.assertRaises(RuntimeError) as ctx:
+                srv.gitea_create_issue(
+                    title="Test issue", body="body", worktree_path=missing_path
+                )
+        self.assertIn("does not exist (fail closed)", str(ctx.exception))
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server.api_get_all", return_value=[])
+    @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
+    @patch("os.path.exists", return_value=True)
+    @patch("os.path.isdir", return_value=True)
+    @patch("subprocess.run")
+    def test_create_issue_wrong_repo_fails_closed(self, mock_run, mock_isdir, mock_exists, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
+        # Mock subprocess.run for git --git-common-dir to return a different path
+        mock_res = MagicMock()
+        mock_res.stdout = "/Users/jasonwalker/Development/some-other-repo/.git\n"
+        mock_run.return_value = mock_res
+
+        wrong_repo_path = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "feat-issue-1")
+
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with self.assertRaises(RuntimeError) as ctx:
+                srv.gitea_create_issue(
+                    title="Test issue", body="body", worktree_path=wrong_repo_path
+                )
+        self.assertIn("does not belong to the target repository", str(ctx.exception))
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    def test_create_issue_fails_without_whoami_preflight(self, _ns, _prof, _auth):
+        srv._preflight_whoami_called = False
+        with self.assertRaises(RuntimeError) as ctx:
+            srv.gitea_create_issue(title="Test issue", body="body")
+        self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    def test_create_issue_fails_without_capability_preflight(self, _ns, _prof, _auth):
+        srv._preflight_capability_called = False
+        with self.assertRaises(RuntimeError) as ctx:
+            srv.gitea_create_issue(title="Test issue", body="body")
+        self.assertIn("Task capability (gitea_resolve_task_capability) has not been resolved", str(ctx.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_branch_ownership.py

@@ -0,0 +1,35 @@
+"""Unit tests for structured issue/branch ownership parsing (#440)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from issue_branch_ownership import (  # noqa: E402
+    branch_tracks_issue,
+    parse_tracked_issue_number,
+)
+
+
+class TestIssueBranchOwnership(unittest.TestCase):
+    def test_parses_canonical_author_branch(self):
+        self.assertEqual(
+            parse_tracked_issue_number("feat/issue-420-server-code-parity"),
+            420,
+        )
+
+    def test_issue_4200_does_not_track_issue_420(self):
+        self.assertEqual(parse_tracked_issue_number("feat/issue-4200-unrelated"), 4200)
+        self.assertFalse(branch_tracks_issue("feat/issue-4200-unrelated", 420))
+
+    def test_branch_tracks_issue_exact_match(self):
+        self.assertTrue(
+            branch_tracks_issue("fix/issue-440-branch-recovery", 440)
+        )
+
+    def test_unrelated_branch_does_not_track(self):
+        self.assertFalse(branch_tracks_issue("feat/issue-999-other", 440))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_lock_adoption.py

@@ -0,0 +1,76 @@
+"""Unit tests for own-branch lock adoption decision (#442 / #443)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from issue_lock_adoption import (  # noqa: E402
+    ADOPT,
+    BLOCK_COMPETING,
+    NO_MATCH,
+    assess_own_branch_adoption,
+    build_adoption_proof,
+)
+
+REQ = "feat/issue-420-server-code-parity"
+
+
+class TestAssessOwnBranchAdoption(unittest.TestCase):
+    def test_exact_own_branch_is_adopted(self):
+        result = assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=REQ,
+            existing_branches=[{"name": REQ, "commit_sha": "934688a"}],
+        )
+        self.assertEqual(result["outcome"], ADOPT)
+        self.assertTrue(result["adopt"])
+
+    def test_different_branch_same_issue_blocks(self):
+        result = assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=REQ,
+            existing_branches=[{"name": "feat/issue-420-other-work"}],
+        )
+        self.assertEqual(result["outcome"], BLOCK_COMPETING)
+        self.assertTrue(result["block"])
+
+    def test_no_matching_branch_is_normal_path(self):
+        result = assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=REQ,
+            existing_branches=[{"name": "feat/issue-999-unrelated"}],
+        )
+        self.assertEqual(result["outcome"], NO_MATCH)
+
+    def test_issue_number_substring_collision_is_ignored(self):
+        result = assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=REQ,
+            existing_branches=[{"name": "feat/issue-4200-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_recovery.py

@@ -0,0 +1,152 @@
+"""Integration tests for non-destructive lock/PR recovery (#440)."""
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import issue_lock_adoption  # noqa: E402
+import issue_lock_store as ils  # noqa: E402
+import mcp_server  # noqa: E402
+from mcp_server import gitea_create_pr, gitea_lock_issue  # noqa: E402
+from tests.test_mcp_server import CREATE_PR_ENV, ISSUE_WRITE_ENV  # noqa: E402
+
+FAKE_AUTH = "token fake"
+BRANCH = "feat/issue-420-server-code-parity"
+
+
+def _lock_record(**overrides):
+    record = {
+        "issue_number": 420,
+        "branch_name": BRANCH,
+        "remote": "prgs",
+        "org": "Scaled-Tech-Consulting",
+        "repo": mcp_server.REMOTES["prgs"]["repo"],
+        "worktree_path": "/tmp/wt-420",
+        "work_lease": {
+            "operation_type": "author_issue_work",
+            "expires_at": "2999-01-01T00:00:00Z",
+        },
+    }
+    record.update(overrides)
+    return record
+
+
+def _clean_git_state():
+    return {
+        "current_branch": BRANCH,
+        "porcelain_status": "",
+        "base_equivalent": True,
+        "inspected_git_root": os.getcwd(),
+        "base_branch": "master",
+    }
+
+
+class TestIssueLockRecovery(unittest.TestCase):
+    def setUp(self):
+        self._tmpdir = tempfile.TemporaryDirectory()
+        self._lock_dir = self._tmpdir.name
+        self._env = {
+            **ISSUE_WRITE_ENV,
+            "GITEA_ISSUE_LOCK_DIR": self._lock_dir,
+        }
+        self._create_pr_env = {
+            **CREATE_PR_ENV,
+            "GITEA_ISSUE_LOCK_DIR": self._lock_dir,
+        }
+
+    def tearDown(self):
+        self._tmpdir.cleanup()
+
+    def test_adoption_after_restart_without_session_pointer(self):
+        with patch.dict(os.environ, self._env, clear=True):
+            with mock.patch("os.getpid", return_value=9999):
+                self.assertIsNone(ils.read_session_issue_lock())
+
+            with mock.patch(
+                "mcp_server.issue_lock_worktree.read_worktree_git_state",
+                return_value=_clean_git_state(),
+            ), mock.patch("mcp_server.api_get_all") as mock_api, mock.patch(
+                "mcp_server.get_auth_header", return_value=FAKE_AUTH
+            ):
+                mock_api.side_effect = [
+                    [],
+                    [{"name": BRANCH, "commit": {"id": "934688a"}}],
+                ]
+                res = gitea_lock_issue(
+                    issue_number=420,
+                    branch_name=BRANCH,
+                    remote="prgs",
+                    worktree_path=os.path.realpath(os.getcwd()),
+                )
+        self.assertTrue(res["success"])
+        self.assertIn("adoption", res)
+        self.assertEqual(res["adoption"]["branch_head_commit"], "934688a")
+
+    @mock.patch(
+        "mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+        return_value=(True, []),
+    )
+    @mock.patch("mcp_server.api_request")
+    @mock.patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_resolves_keyed_lock_after_restart(self, _auth, mock_api, _role):
+        mock_api.return_value = {"number": 501, "html_url": "https://example/pr/501"}
+        worktree = os.path.realpath(os.getcwd())
+        with patch.dict(os.environ, self._create_pr_env, clear=True):
+            path = ils.lock_file_path(
+                remote="prgs",
+                org=mcp_server.REMOTES["prgs"]["org"],
+                repo=mcp_server.REMOTES["prgs"]["repo"],
+                issue_number=420,
+                lock_dir=self._lock_dir,
+            )
+            ils.save_lock_file(path, _lock_record(worktree_path=worktree))
+
+            with mock.patch("os.getpid", return_value=4242):
+                self.assertIsNone(ils.read_session_issue_lock())
+
+            res = gitea_create_pr(
+                title="feat: recovery Closes #420",
+                head=BRANCH,
+                base="master",
+                remote="prgs",
+                worktree_path=worktree,
+            )
+        self.assertEqual(res["number"], 501)
+
+    def test_open_pr_blocks_adoption(self):
+        with patch.dict(os.environ, self._env, clear=True):
+            with mock.patch(
+                "mcp_server.issue_lock_worktree.read_worktree_git_state",
+                return_value=_clean_git_state(),
+            ), mock.patch("mcp_server.api_get_all") as mock_api, mock.patch(
+                "mcp_server.get_auth_header", return_value=FAKE_AUTH
+            ):
+                mock_api.side_effect = [
+                    [{"number": 99, "head": {"ref": BRANCH}, "title": "", "body": ""}],
+                    [{"name": BRANCH, "commit": {"id": "934688a"}}],
+                ]
+                with self.assertRaises(ValueError) as ctx:
+                    gitea_lock_issue(
+                        issue_number=420,
+                        branch_name=BRANCH,
+                        remote="prgs",
+                        worktree_path=os.path.realpath(os.getcwd()),
+                    )
+        self.assertIn("already tied to an open PR", str(ctx.exception))
+
+    def test_structured_ownership_ignores_issue_4200_collision(self):
+        result = issue_lock_adoption.assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=BRANCH,
+            existing_branches=[{"name": "feat/issue-4200-unrelated"}],
+        )
+        self.assertEqual(result["outcome"], issue_lock_adoption.NO_MATCH)
+
+
+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_project_registry.py

@@ -0,0 +1,111 @@
+"""Tests for web UI project registry (#427)."""
+import json
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from starlette.testclient import TestClient
+
+from webui.app import create_app
+from webui.project_registry import (
+    default_registry_path,
+    load_registry,
+    project_to_dict,
+)
+
+
+class TestProjectRegistryLoader(unittest.TestCase):
+    def test_default_registry_loads_gitea_tools(self):
+        registry = load_registry()
+        self.assertEqual(registry.version, 1)
+        self.assertEqual(len(registry.projects), 1)
+        project = registry.projects[0]
+        self.assertEqual(project.id, "gitea-tools")
+        self.assertEqual(project.repo_name, "Gitea-Tools")
+        self.assertEqual(project.gitea_owner, "Scaled-Tech-Consulting")
+        self.assertEqual(project.remote_host, "https://gitea.prgs.cc")
+        self.assertEqual(project.profiles["author"], "prgs-author")
+        self.assertEqual(project.profiles["reviewer"], "prgs-reviewer")
+        self.assertEqual(project.profiles["reconciler"], "prgs-reconciler")
+        self.assertIn("skill", project.workflow_paths)
+        self.assertGreaterEqual(len(project.onboarding_checklist), 4)
+
+    def test_registry_rejects_credential_keys(self):
+        payload = {
+            "version": 1,
+            "projects": [
+                {
+                    "id": "bad",
+                    "repo_name": "Bad",
+                    "gitea_owner": "Org",
+                    "remote_host": "https://gitea.example.invalid",
+                    "default_branch": "main",
+                    "local_checkout_path": ".",
+                    "profiles": {
+                        "author": "a",
+                        "reviewer": "r",
+                        "reconciler": "c",
+                    },
+                    "workflow_paths": {"skill": "skills/x.md"},
+                    "api_token": "secret",
+                }
+            ],
+        }
+        with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
+            json.dump(payload, handle)
+            path = Path(handle.name)
+        try:
+            with self.assertRaises(ValueError):
+                load_registry(path)
+        finally:
+            path.unlink(missing_ok=True)
+
+    def test_default_registry_path_points_at_packaged_data(self):
+        path = default_registry_path()
+        self.assertTrue(path.name == "projects.registry.json")
+        self.assertTrue(path.parent.name == "data")
+
+
+class TestProjectRegistryRoutes(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+
+    def test_projects_page_lists_gitea_tools(self):
+        response = self.client.get("/projects")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Gitea-Tools", response.text)
+        self.assertIn("Scaled-Tech-Consulting", response.text)
+        self.assertIn("prgs-author", response.text)
+        self.assertNotIn("child issue", response.text.lower())
+
+    def test_project_detail_renders_checklist(self):
+        response = self.client.get("/projects/gitea-tools")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Onboarding checklist", response.text)
+        self.assertIn("Configure execution profiles", response.text)
+        self.assertIn("branches/", response.text)
+
+    def test_project_detail_404(self):
+        response = self.client.get("/projects/unknown-repo")
+        self.assertEqual(response.status_code, 404)
+
+    def test_api_projects_json(self):
+        response = self.client.get("/api/projects")
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+        self.assertEqual(data["version"], 1)
+        self.assertEqual(len(data["projects"]), 1)
+        self.assertEqual(data["projects"][0]["id"], "gitea-tools")
+        self.assertIn("onboarding_checklist", data["projects"][0])
+
+    def test_project_to_dict_is_json_safe(self):
+        registry = load_registry()
+        encoded = json.dumps(project_to_dict(registry.projects[0]))
+        self.assertIn("gitea-tools", encoded)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_webui_prompt_library.py

@@ -0,0 +1,90 @@
+"""Tests for web UI prompt library (#428)."""
+import json
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from starlette.testclient import TestClient
+
+from webui.app import create_app
+from webui.prompt_library import find_prompt, library_to_dict, load_prompt_library, prompt_to_dict
+
+REQUIRED_PROMPT_SLUGS = frozenset({
+    "review-pr",
+    "work-issue",
+    "create-issue",
+    "comment-issue",
+    "cleanup",
+    "audit",
+    "onboarding",
+})
+
+
+class TestPromptLibraryLoader(unittest.TestCase):
+    def test_library_loads_required_prompts(self):
+        entries = load_prompt_library()
+        slugs = {entry.slug for entry in entries}
+        self.assertEqual(slugs, REQUIRED_PROMPT_SLUGS)
+
+    def test_workflow_hashes_present(self):
+        review = find_prompt("review-pr")
+        self.assertIsNotNone(review)
+        assert review is not None
+        self.assertTrue(review.workflow_hash)
+        self.assertEqual(len(review.workflow_hash), 64)
+
+    def test_prompt_text_is_short(self):
+        for entry in load_prompt_library():
+            self.assertLess(len(entry.prompt_text), 400)
+            self.assertNotIn("Do not improvise around the gates", entry.prompt_text)
+
+
+class TestPromptLibraryRoutes(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+
+    def test_prompts_page_lists_all_entries(self):
+        response = self.client.get("/prompts")
+        self.assertEqual(response.status_code, 200)
+        for label in (
+            "Review PR",
+            "Work issue",
+            "Create issue",
+            "Comment on issue",
+            "Post-merge cleanup",
+            "Reconciliation audit",
+            "Project onboarding",
+        ):
+            self.assertIn(label, response.text)
+        self.assertIn("Copy prompt", response.text)
+        self.assertIn("sha256:", response.text)
+        self.assertNotIn("child issue", response.text.lower())
+
+    def test_prompt_detail_route(self):
+        response = self.client.get("/prompts/work-issue")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("work-issue.md", response.text)
+        self.assertIn("Copy prompt", response.text)
+
+    def test_prompt_detail_404(self):
+        self.assertEqual(self.client.get("/prompts/missing").status_code, 404)
+
+    def test_api_prompts_json(self):
+        response = self.client.get("/api/prompts")
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+        self.assertEqual(data["count"], 7)
+        review = next(item for item in data["prompts"] if item["slug"] == "review-pr")
+        self.assertIn("workflow_hash", review)
+        self.assertIn("review-merge-pr.md", review["workflow_path"])
+
+    def test_prompt_to_dict_roundtrip(self):
+        entry = load_prompt_library()[0]
+        encoded = json.dumps(prompt_to_dict(entry))
+        self.assertIn("workflow_path", encoded)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_webui_queue_dashboard.py

@@ -0,0 +1,288 @@
+"""Tests for web UI live queue dashboard (#429)."""
+import sys
+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))
+
+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()
+
+_SAMPLE_PRS = [
+    {
+        "number": 100,
+        "title": "feat: queue dashboard (Closes #429)",
+        "body": "",
+        "mergeable": True,
+        "updated_at": _RECENT,
+        "head": {"ref": "feat/issue-429", "sha": "abc123def456"},
+        "base": {"ref": "master"},
+    },
+    {
+        "number": 99,
+        "title": "fix: conflict",
+        "body": "Closes #50",
+        "mergeable": False,
+        "updated_at": _STALE,
+        "head": {"ref": "fix/issue-50", "sha": "deadbeef0001"},
+        "base": {"ref": "master"},
+    },
+    {
+        "number": 98,
+        "title": "feat: duplicate A (Closes #60)",
+        "body": "",
+        "mergeable": True,
+        "updated_at": _RECENT,
+        "head": {"ref": "feat/issue-60-a", "sha": "111111111111"},
+        "base": {"ref": "master"},
+    },
+    {
+        "number": 97,
+        "title": "feat: duplicate B (Closes #60)",
+        "body": "",
+        "mergeable": True,
+        "updated_at": _RECENT,
+        "head": {"ref": "feat/issue-60-b", "sha": "222222222222"},
+        "base": {"ref": "master"},
+    },
+]
+
+_SAMPLE_ISSUES = [
+    {
+        "number": 429,
+        "title": "Web UI queue dashboard",
+        "state": "open",
+        "labels": [{"name": "status:in-progress"}],
+        "assignee": None,
+        "updated_at": _RECENT,
+    },
+    {
+        "number": 60,
+        "title": "Duplicate PR target",
+        "state": "open",
+        "labels": [],
+        "assignee": None,
+        "updated_at": _RECENT,
+    },
+    {
+        "number": 50,
+        "title": "Blocked PR target",
+        "state": "open",
+        "labels": [],
+        "assignee": None,
+        "updated_at": _STALE,
+    },
+]
+
+
+def _mock_pagination(count: int) -> PaginationMeta:
+    return PaginationMeta(
+        page=1,
+        per_page=50,
+        returned_count=count,
+        has_more=False,
+        is_final_page=True,
+        inventory_complete=True,
+        pages_fetched=1,
+    )
+
+
+def _mock_fetch_prs(*_args, **_kwargs):
+    return list(_SAMPLE_PRS), _mock_pagination(len(_SAMPLE_PRS))
+
+
+def _mock_fetch_issues(*_args, **_kwargs):
+    return list(_SAMPLE_ISSUES), _mock_pagination(len(_SAMPLE_ISSUES))
+
+
+class TestQueueClassification(unittest.TestCase):
+    def test_extract_linked_issue_from_title(self):
+        self.assertEqual(
+            _extract_linked_issue("feat: X (Closes #429)", ""),
+            429,
+        )
+
+    def test_classify_pr_blocked_and_stale(self):
+        pr = _SAMPLE_PRS[1]
+        badges = _classify_pr(
+            pr,
+            issue_claimed=set(),
+            issue_to_prs={50: [99]},
+        )
+        self.assertIn("blocked", badges)
+        self.assertIn("stale", badges)
+
+    def test_classify_pr_duplicate(self):
+        badges = _classify_pr(
+            _SAMPLE_PRS[2],
+            issue_claimed=set(),
+            issue_to_prs={60: [98, 97]},
+        )
+        self.assertIn("duplicate", badges)
+        self.assertIn("in-review", badges)
+
+    def test_classify_issue_claimed_and_duplicate(self):
+        claimed = _classify_issue(_SAMPLE_ISSUES[0], linked_prs=[])
+        self.assertIn("claimed", claimed)
+        duplicate = _classify_issue(_SAMPLE_ISSUES[1], linked_prs=[98, 97])
+        self.assertIn("duplicate", duplicate)
+
+
+class TestQueueLoader(unittest.TestCase):
+    def test_snapshot_with_mock_fetch(self):
+        snapshot = load_queue_snapshot(
+            fetch_prs=_mock_fetch_prs,
+            fetch_issues=_mock_fetch_issues,
+        )
+        self.assertEqual(snapshot.project_id, "gitea-tools")
+        self.assertIsNone(snapshot.fetch_error)
+        self.assertEqual(len(snapshot.prs), 4)
+        self.assertEqual(len(snapshot.issues), 3)
+        self.assertTrue(snapshot.pr_pagination.inventory_complete)
+        self.assertTrue(snapshot.issue_pagination.inventory_complete)
+
+        pr100 = next(p for p in snapshot.prs if p.number == 100)
+        self.assertEqual(pr100.extra["linked_issue"], "429")
+        self.assertIn("claimed", pr100.badges)
+
+    def test_snapshot_dict_export(self):
+        snapshot = load_queue_snapshot(
+            fetch_prs=_mock_fetch_prs,
+            fetch_issues=_mock_fetch_issues,
+        )
+        data = snapshot_to_dict(snapshot)
+        self.assertEqual(data["project_id"], "gitea-tools")
+        self.assertIsNone(data["fetch_error"])
+        self.assertEqual(len(data["prs"]), 4)
+        self.assertTrue(data["pagination"]["prs"]["inventory_complete"])
+
+
+class TestQueueRoutes(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+        self._patch = mock.patch(
+            "webui.app.load_queue_snapshot",
+            return_value=load_queue_snapshot(
+                fetch_prs=_mock_fetch_prs,
+                fetch_issues=_mock_fetch_issues,
+            ),
+        )
+        self._patch.start()
+
+    def tearDown(self):
+        self._patch.stop()
+
+    def test_queue_page_renders_tables_and_pagination(self):
+        response = self.client.get("/queue")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Live queue", response.text)
+        self.assertIn("Open pull requests", response.text)
+        self.assertIn("Open issues", response.text)
+        self.assertIn("pages_fetched", response.text)
+        self.assertIn("Gitea-Tools", response.text)
+        self.assertNotIn("child issue", response.text.lower())
+
+    def test_api_queue_json(self):
+        response = self.client.get("/api/queue")
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+        self.assertEqual(data["repo"], "Scaled-Tech-Consulting/Gitea-Tools")
+        self.assertEqual(data["pagination"]["issues"]["returned_count"], 3)
+
+    def test_queue_fail_closed_without_credentials(self):
+        with mock.patch("webui.queue_loader.get_auth_header", return_value=None):
+            snapshot = load_queue_snapshot()
+        self.assertIsNotNone(snapshot.fetch_error)
+        self.assertEqual(len(snapshot.prs), 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_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")
+        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(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__":
+    unittest.main()

tests/test_webui_skeleton.py

@@ -0,0 +1,73 @@
+"""Tests for internal web UI skeleton (#426)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from starlette.testclient import TestClient
+
+from webui.app import create_app
+
+
+class TestWebuiSkeleton(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+
+    def test_health_returns_json(self):
+        response = self.client.get("/health")
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+        self.assertEqual(data["status"], "ok")
+        self.assertEqual(data["service"], "mcp-control-plane-webui")
+        self.assertEqual(data["mode"], "read-only-mvp")
+        self.assertIn("timestamp", data)
+
+    def test_home_renders(self):
+        response = self.client.get("/")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Operator console", response.text)
+        self.assertIn("Read-only MVP", response.text)
+
+    def test_route_stubs_render(self):
+        for path in ("/runtime", "/audit"):
+            with self.subTest(path=path):
+                response = self.client.get(path)
+                self.assertEqual(response.status_code, 200)
+                self.assertIn("child issue", response.text.lower())
+
+    def test_prompts_is_implemented(self):
+        response = self.client.get("/prompts")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Prompt library", response.text)
+
+    def test_projects_is_implemented(self):
+        response = self.client.get("/projects")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Gitea-Tools", response.text)
+
+    def test_extra_stub_routes(self):
+        for path in ("/worktrees", "/leases"):
+            with self.subTest(path=path):
+                self.assertEqual(self.client.get(path).status_code, 200)
+
+    def test_post_is_rejected(self):
+        response = self.client.post("/health")
+        self.assertEqual(response.status_code, 405)
+        self.assertEqual(response.json()["error"], "read-only-mvp")
+
+    def test_queue_route_renders(self):
+        response = self.client.get("/queue")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Live queue", response.text)
+
+    def test_nav_links_on_all_pages(self):
+        for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit"):
+            with self.subTest(path=path):
+                text = self.client.get(path).text
+                for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit"):
+                    self.assertIn(f'href="{href}"', text)
+
+
+if __name__ == "__main__":
+    unittest.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/__init__.py

@@ -0,0 +1,5 @@
+"""Internal MCP Control Plane web UI (read-only MVP skeleton, #426)."""
+
+from webui.app import create_app
+
+__all__ = ["create_app"]

webui/__main__.py

@@ -0,0 +1,19 @@
+"""Run the internal web UI: ``python -m webui``."""
+
+from __future__ import annotations
+
+import os
+
+import uvicorn
+
+from webui.app import create_app
+
+
+def main() -> None:
+    host = os.environ.get("WEBUI_HOST", "127.0.0.1")
+    port = int(os.environ.get("WEBUI_PORT", "8765"))
+    uvicorn.run(create_app(), host=host, port=port, log_level="info")
+
+
+if __name__ == "__main__":
+    main()

webui/app.py

@@ -0,0 +1,180 @@
+"""Starlette application for the internal read-only web UI (#426)."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import HTMLResponse, JSONResponse, Response
+from starlette.routing import Route
+
+from webui.layout import render_page
+from webui.project_registry import find_project, load_registry, registry_to_dict
+from webui.project_views import render_project_detail, render_projects_list
+from webui.prompt_library import find_prompt, library_to_dict
+from webui.prompt_views import render_prompt_detail, render_prompts_page
+from webui.queue_loader import load_queue_snapshot, snapshot_to_dict
+from webui.queue_views import render_queue_page
+
+_READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
+
+
+def _stub_page(title: str, description: str) -> HTMLResponse:
+    body = (
+        f"<h2>{title}</h2>"
+        f'<div class="stub"><p>{description}</p>'
+        "<p>Implementation tracked in a child issue of #425.</p></div>"
+    )
+    return HTMLResponse(render_page(title=title, body_html=body))
+
+
+async def home(_request: Request) -> HTMLResponse:
+    body = (
+        "<h2>Operator console</h2>"
+        "<p>Local entry point for MCP Control Plane operational views.</p>"
+        "<ul>"
+        "<li><strong>Queue</strong> — live PR and issue dashboard (#429)</li>"
+        "<li><strong>Projects</strong> — registry and onboarding (#427)</li>"
+        "<li><strong>Prompts</strong> — canonical workflow prompt library (#428)</li>"
+        "<li><strong>Runtime</strong> — MCP health and stale-runtime detection (#430)</li>"
+        "<li><strong>Audit</strong> — final-report paste and validator preview (#431)</li>"
+        "<li><strong>Worktrees</strong> — branch hygiene dashboard (#432)</li>"
+        "<li><strong>Leases</strong> — collision and lease visibility (#433)</li>"
+        "</ul>"
+    )
+    return HTMLResponse(render_page(title="Home", body_html=body))
+
+
+async def health(_request: Request) -> JSONResponse:
+    return JSONResponse({
+        "status": "ok",
+        "service": "mcp-control-plane-webui",
+        "mode": "read-only-mvp",
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+    })
+
+
+async def queue(_request: Request) -> HTMLResponse:
+    snapshot = load_queue_snapshot()
+    return HTMLResponse(render_page(title="Queue", body_html=render_queue_page(snapshot)))
+
+
+async def api_queue(_request: Request) -> JSONResponse:
+    return JSONResponse(snapshot_to_dict(load_queue_snapshot()))
+
+
+async def projects(_request: Request) -> HTMLResponse:
+    registry = load_registry()
+    return HTMLResponse(render_projects_list(registry))
+
+
+async def project_detail(request: Request) -> HTMLResponse:
+    project_id = request.path_params["project_id"]
+    registry = load_registry()
+    project = find_project(registry, project_id)
+    if project is None:
+        return HTMLResponse(
+            render_page(
+                title="Project not found",
+                body_html=(
+                    "<h2>Project not found</h2>"
+                    f"<p>No registry entry for <code>{project_id}</code>.</p>"
+                    '<p><a href="/projects">← All projects</a></p>'
+                ),
+            ),
+            status_code=404,
+        )
+    return HTMLResponse(render_project_detail(project))
+
+
+async def api_projects(_request: Request) -> JSONResponse:
+    registry = load_registry()
+    return JSONResponse(registry_to_dict(registry))
+
+
+async def prompts(_request: Request) -> HTMLResponse:
+    return HTMLResponse(render_prompts_page())
+
+
+async def prompt_detail(request: Request) -> HTMLResponse:
+    prompt_id = request.path_params["prompt_id"]
+    prompt = find_prompt(prompt_id)
+    if prompt is None:
+        return HTMLResponse(
+            render_page(
+                title="Prompt not found",
+                body_html=(
+                    "<h2>Prompt not found</h2>"
+                    f"<p>No library entry for <code>{prompt_id}</code>.</p>"
+                    '<p><a href="/prompts">← All prompts</a></p>'
+                ),
+            ),
+            status_code=404,
+        )
+    return HTMLResponse(render_prompt_detail(prompt))
+
+
+async def api_prompts(_request: Request) -> JSONResponse:
+    return JSONResponse(library_to_dict())
+
+
+async def runtime(_request: Request) -> HTMLResponse:
+    return _stub_page(
+        "Runtime",
+        "Runtime health will report MCP profile, preflight, and stale-server signals.",
+    )
+
+
+async def audit(_request: Request) -> HTMLResponse:
+    return _stub_page(
+        "Audit",
+        "Report audit will accept pasted final reports and run validator previews.",
+    )
+
+
+async def worktrees(_request: Request) -> HTMLResponse:
+    return _stub_page(
+        "Worktrees",
+        "Worktree hygiene will summarize branches/ session folders and cleanup risk.",
+    )
+
+
+async def leases(_request: Request) -> HTMLResponse:
+    return _stub_page(
+        "Leases",
+        "Lease visibility will show active issue and reviewer PR leases.",
+    )
+
+
+async def method_not_allowed(request: Request, _exc: Exception) -> Response:
+    if request.method not in _READ_ONLY_METHODS:
+        return JSONResponse(
+            {"error": "read-only-mvp", "detail": f"{request.method} not permitted"},
+            status_code=405,
+        )
+    return JSONResponse({"error": "not_found"}, status_code=404)
+
+
+def create_app() -> Starlette:
+    """Build the read-only MVP Starlette app."""
+    return Starlette(
+        debug=False,
+        routes=[
+            Route("/", home, methods=["GET"]),
+            Route("/health", health, methods=["GET"]),
+            Route("/queue", queue, methods=["GET"]),
+            Route("/api/queue", api_queue, methods=["GET"]),
+            Route("/projects", projects, methods=["GET"]),
+            Route("/projects/{project_id}", project_detail, methods=["GET"]),
+            Route("/api/projects", api_projects, methods=["GET"]),
+            Route("/prompts", prompts, methods=["GET"]),
+            Route("/prompts/{prompt_id}", prompt_detail, methods=["GET"]),
+            Route("/api/prompts", api_prompts, methods=["GET"]),
+            Route("/runtime", runtime, methods=["GET"]),
+            Route("/audit", audit, methods=["GET"]),
+            Route("/worktrees", worktrees, methods=["GET"]),
+            Route("/leases", leases, methods=["GET"]),
+        ],
+        exception_handlers={405: method_not_allowed},
+    )

webui/data/projects.registry.json

@@ -0,0 +1,49 @@
+{
+  "version": 1,
+  "projects": [
+    {
+      "id": "gitea-tools",
+      "repo_name": "Gitea-Tools",
+      "gitea_owner": "Scaled-Tech-Consulting",
+      "remote_host": "https://gitea.prgs.cc",
+      "default_branch": "master",
+      "local_checkout_path": ".",
+      "profiles": {
+        "author": "prgs-author",
+        "reviewer": "prgs-reviewer",
+        "reconciler": "prgs-reconciler"
+      },
+      "workflow_paths": {
+        "skill": "skills/llm-project-workflow/SKILL.md",
+        "work_issue": "skills/llm-project-workflow/workflows/work-issue.md",
+        "review_merge": "skills/llm-project-workflow/workflows/review-merge-pr.md"
+      },
+      "schema_paths": {
+        "mcp_config_v2": "gitea-mcp.v2-contexts.example.json",
+        "mcp_config_v1": "gitea-mcp.example.json"
+      },
+      "onboarding_checklist": [
+        {
+          "id": "profiles",
+          "title": "Configure execution profiles",
+          "description": "Install author, reviewer, and reconciler MCP profiles (prgs-author, prgs-reviewer, prgs-reconciler) in separate namespaces. Tokens stay in keychain — never in this registry."
+        },
+        {
+          "id": "mcp_config",
+          "title": "Wire MCP v2 contexts",
+          "description": "Copy and customize gitea-mcp.v2-contexts.example.json for your machine. Map this repo path under projects with default_owner Scaled-Tech-Consulting and default_repo Gitea-Tools."
+        },
+        {
+          "id": "wiki_gate",
+          "title": "Wiki publication readiness",
+          "description": "For wiki-tracked work, satisfy the live Gitea Wiki proof gate (#224) before closing issues. See docs/wiki/Safety-and-Gates.md."
+        },
+        {
+          "id": "branches_layout",
+          "title": "Isolate work under branches/",
+          "description": "All LLM task edits happen in worktrees under branches/. Main checkout stays clean; use skills/llm-project-workflow templates for start-issue and review flows."
+        }
+      ]
+    }
+  ]
+}

webui/layout.py

@@ -0,0 +1,176 @@
+"""Shared HTML layout for the internal web UI."""
+
+from __future__ import annotations
+
+NAV_ITEMS = (
+    ("/", "Home"),
+    ("/queue", "Queue"),
+    ("/projects", "Projects"),
+    ("/prompts", "Prompts"),
+    ("/runtime", "Runtime"),
+    ("/audit", "Audit"),
+    ("/worktrees", "Worktrees"),
+    ("/leases", "Leases"),
+)
+
+MVP_NOTICE = (
+    "Read-only MVP — Gitea, MCP tools, and canonical workflows remain the "
+    "source of truth. No mutation endpoints."
+)
+
+
+def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
+    nav_links = "".join(
+        f'<a href="{href}">{label}</a>' for href, label in NAV_ITEMS
+    )
+    return f"""<!DOCTYPE html>
+<html lang="en">
+<head>
+  <meta charset="utf-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1">
+  <title>{title} · MCP Control Plane</title>
+  <style>
+    :root {{
+      --bg: #0f1419;
+      --surface: #1a2332;
+      --text: #e7ecf3;
+      --muted: #8b9cb3;
+      --accent: #5b9fd4;
+      --border: #2a3648;
+    }}
+    * {{ box-sizing: border-box; }}
+    body {{
+      margin: 0;
+      font-family: system-ui, -apple-system, sans-serif;
+      background: var(--bg);
+      color: var(--text);
+      line-height: 1.5;
+    }}
+    header {{
+      background: var(--surface);
+      border-bottom: 1px solid var(--border);
+      padding: 0.75rem 1.25rem;
+    }}
+    header h1 {{
+      margin: 0 0 0.5rem;
+      font-size: 1.1rem;
+      font-weight: 600;
+    }}
+    nav {{
+      display: flex;
+      flex-wrap: wrap;
+      gap: 0.75rem 1rem;
+    }}
+    nav a {{
+      color: var(--accent);
+      text-decoration: none;
+      font-size: 0.9rem;
+    }}
+    nav a:hover {{ text-decoration: underline; }}
+    main {{
+      max-width: 52rem;
+      margin: 0 auto;
+      padding: 1.5rem 1.25rem 2.5rem;
+    }}
+    .notice {{
+      color: var(--muted);
+      font-size: 0.85rem;
+      margin-bottom: 1.25rem;
+      padding: 0.65rem 0.85rem;
+      border: 1px solid var(--border);
+      border-radius: 6px;
+      background: var(--surface);
+    }}
+    h2 {{ margin-top: 0; font-size: 1.35rem; }}
+    p {{ color: var(--muted); }}
+    .stub {{
+      border-left: 3px solid var(--accent);
+      padding-left: 0.85rem;
+      margin: 1rem 0;
+    }}
+    .meta {{ font-size: 0.85rem; }}
+    table.registry, table.detail {{
+      width: 100%;
+      border-collapse: collapse;
+      margin: 1rem 0;
+      font-size: 0.9rem;
+    }}
+    table.registry th, table.registry td,
+    table.detail th, table.detail td {{
+      text-align: left;
+      padding: 0.45rem 0.6rem;
+      border-bottom: 1px solid var(--border);
+    }}
+    table.registry th, table.detail th {{
+      color: var(--muted);
+      font-weight: 500;
+    }}
+    code {{
+      font-family: ui-monospace, monospace;
+      font-size: 0.85em;
+      color: var(--text);
+    }}
+    ol.checklist {{
+      padding-left: 1.25rem;
+      margin: 0.5rem 0 1.5rem;
+    }}
+    ol.checklist li {{ margin-bottom: 0.85rem; }}
+    ol.checklist p {{ margin: 0.25rem 0 0; font-size: 0.9rem; }}
+    .prompt-card {{
+      margin: 1.25rem 0 1.75rem;
+      padding: 1rem 1.1rem;
+      border: 1px solid var(--border);
+      border-radius: 8px;
+      background: var(--surface);
+    }}
+    .prompt-card h3 {{ margin: 0 0 0.5rem; font-size: 1.05rem; }}
+    pre.prompt-text {{
+      white-space: pre-wrap;
+      word-break: break-word;
+      margin: 0.75rem 0;
+      padding: 0.75rem 0.85rem;
+      border-radius: 6px;
+      background: var(--bg);
+      border: 1px solid var(--border);
+      font-size: 0.9rem;
+      color: var(--text);
+    }}
+    .copy-btn {{
+      background: var(--accent);
+      color: #0b1219;
+      border: none;
+      border-radius: 6px;
+      padding: 0.4rem 0.85rem;
+      font-size: 0.85rem;
+      cursor: pointer;
+    }}
+    .copy-btn:hover {{ filter: brightness(1.08); }}
+    .muted {{ color: var(--muted); }}
+    .badges {{ display: inline-flex; flex-wrap: wrap; gap: 0.35rem; margin-left: 0.5rem; }}
+    .badge {{
+      font-size: 0.72rem;
+      padding: 0.1rem 0.45rem;
+      border-radius: 999px;
+      border: 1px solid var(--border);
+      color: var(--muted);
+      text-transform: lowercase;
+    }}
+    .badge-claimed {{ color: #8fd19e; border-color: #3d6b4a; }}
+    .badge-blocked {{ color: #f0a8a8; border-color: #7a3b3b; }}
+    .badge-in-review {{ color: #9ec8f0; border-color: #3d5f7a; }}
+    .badge-duplicate {{ color: #e0c27a; border-color: #6b5730; }}
+    .badge-stale {{ color: #c9b8e8; border-color: #5a4a78; }}
+  </style>
+  {extra_head}
+</head>
+<body>
+  <header>
+    <h1>MCP Control Plane</h1>
+    <nav>{nav_links}</nav>
+  </header>
+  <main>
+    <p class="notice">{MVP_NOTICE}</p>
+    {body_html}
+  </main>
+</body>
+</html>"""

webui/project_registry.py

@@ -0,0 +1,196 @@
+"""Load and validate the web UI project registry (#427)."""
+
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+_FORBIDDEN_EXACT_KEYS = frozenset({
+    "token",
+    "password",
+    "secret",
+    "credential",
+    "auth",
+    "api_key",
+    "api-key",
+})
+_FORBIDDEN_KEY_PREFIXES = ("auth_", "api_key_", "api-key_")
+_FORBIDDEN_KEY_SUFFIXES = ("_token", "_secret", "_password", "_credential", "_auth")
+
+
+def _is_forbidden_key(key: str) -> bool:
+    lowered = key.lower()
+    if lowered in _FORBIDDEN_EXACT_KEYS:
+        return True
+    return (
+        lowered.startswith(_FORBIDDEN_KEY_PREFIXES)
+        or lowered.endswith(_FORBIDDEN_KEY_SUFFIXES)
+    )
+
+_REQUIRED_PROJECT_FIELDS = (
+    "id",
+    "repo_name",
+    "gitea_owner",
+    "remote_host",
+    "default_branch",
+    "local_checkout_path",
+    "profiles",
+    "workflow_paths",
+)
+
+_REQUIRED_PROFILE_ROLES = ("author", "reviewer", "reconciler")
+
+
+@dataclass(frozen=True)
+class OnboardingStep:
+    id: str
+    title: str
+    description: str
+
+
+@dataclass(frozen=True)
+class ProjectRecord:
+    id: str
+    repo_name: str
+    gitea_owner: str
+    remote_host: str
+    default_branch: str
+    local_checkout_path: str
+    profiles: dict[str, str]
+    workflow_paths: dict[str, str]
+    schema_paths: dict[str, str]
+    onboarding_checklist: tuple[OnboardingStep, ...]
+
+
+@dataclass(frozen=True)
+class ProjectRegistry:
+    version: int
+    projects: tuple[ProjectRecord, ...]
+    source_path: Path
+
+
+def default_registry_path() -> Path:
+    override = os.environ.get("WEBUI_PROJECT_REGISTRY", "").strip()
+    if override:
+        return Path(override).expanduser().resolve()
+    return (Path(__file__).resolve().parent / "data" / "projects.registry.json").resolve()
+
+
+def _reject_credential_keys(obj: Any, *, path: str = "") -> None:
+    if isinstance(obj, dict):
+        for key, value in obj.items():
+            key_path = f"{path}.{key}" if path else key
+            if _is_forbidden_key(key):
+                raise ValueError(f"registry must not store credentials ({key_path})")
+            _reject_credential_keys(value, path=key_path)
+    elif isinstance(obj, list):
+        for index, item in enumerate(obj):
+            _reject_credential_keys(item, path=f"{path}[{index}]")
+
+
+def _parse_onboarding(raw: list[dict[str, Any]] | None) -> tuple[OnboardingStep, ...]:
+    if not raw:
+        return ()
+    steps: list[OnboardingStep] = []
+    for item in raw:
+        steps.append(
+            OnboardingStep(
+                id=str(item["id"]),
+                title=str(item["title"]),
+                description=str(item["description"]),
+            )
+        )
+    return tuple(steps)
+
+
+def _parse_project(raw: dict[str, Any]) -> ProjectRecord:
+    missing = [field for field in _REQUIRED_PROJECT_FIELDS if field not in raw]
+    if missing:
+        raise ValueError(f"project missing required fields: {', '.join(missing)}")
+
+    profiles = raw["profiles"]
+    if not isinstance(profiles, dict):
+        raise ValueError("profiles must be an object")
+    for role in _REQUIRED_PROFILE_ROLES:
+        if role not in profiles or not profiles[role]:
+            raise ValueError(f"profiles.{role} is required")
+
+    workflow_paths = raw["workflow_paths"]
+    if not isinstance(workflow_paths, dict) or not workflow_paths:
+        raise ValueError("workflow_paths must be a non-empty object")
+
+    schema_paths = raw.get("schema_paths") or {}
+    if not isinstance(schema_paths, dict):
+        raise ValueError("schema_paths must be an object when present")
+
+    return ProjectRecord(
+        id=str(raw["id"]),
+        repo_name=str(raw["repo_name"]),
+        gitea_owner=str(raw["gitea_owner"]),
+        remote_host=str(raw["remote_host"]),
+        default_branch=str(raw["default_branch"]),
+        local_checkout_path=str(raw["local_checkout_path"]),
+        profiles={role: str(profiles[role]) for role in _REQUIRED_PROFILE_ROLES},
+        workflow_paths={key: str(value) for key, value in workflow_paths.items()},
+        schema_paths={key: str(value) for key, value in schema_paths.items()},
+        onboarding_checklist=_parse_onboarding(raw.get("onboarding_checklist")),
+    )
+
+
+def load_registry(path: Path | None = None) -> ProjectRegistry:
+    """Load the versioned project registry from disk."""
+    source = (path or default_registry_path()).resolve()
+    raw_text = source.read_text(encoding="utf-8")
+    payload = json.loads(raw_text)
+    if not isinstance(payload, dict):
+        raise ValueError("registry root must be an object")
+
+    version = payload.get("version")
+    if version != 1:
+        raise ValueError(f"unsupported registry version: {version!r}")
+
+    _reject_credential_keys(payload)
+
+    projects_raw = payload.get("projects")
+    if not isinstance(projects_raw, list) or not projects_raw:
+        raise ValueError("projects must be a non-empty array")
+
+    projects = tuple(_parse_project(item) for item in projects_raw)
+    return ProjectRegistry(version=version, projects=projects, source_path=source)
+
+
+def project_to_dict(project: ProjectRecord) -> dict[str, Any]:
+    """Serialize a project for JSON API responses."""
+    return {
+        "id": project.id,
+        "repo_name": project.repo_name,
+        "gitea_owner": project.gitea_owner,
+        "remote_host": project.remote_host,
+        "default_branch": project.default_branch,
+        "local_checkout_path": project.local_checkout_path,
+        "profiles": dict(project.profiles),
+        "workflow_paths": dict(project.workflow_paths),
+        "schema_paths": dict(project.schema_paths),
+        "onboarding_checklist": [
+            {"id": step.id, "title": step.title, "description": step.description}
+            for step in project.onboarding_checklist
+        ],
+    }
+
+
+def registry_to_dict(registry: ProjectRegistry) -> dict[str, Any]:
+    return {
+        "version": registry.version,
+        "source_path": str(registry.source_path),
+        "projects": [project_to_dict(project) for project in registry.projects],
+    }
+
+
+def find_project(registry: ProjectRegistry, project_id: str) -> ProjectRecord | None:
+    for project in registry.projects:
+        if project.id == project_id:
+            return project
+    return None

webui/project_views.py

@@ -0,0 +1,93 @@
+"""HTML views for project registry pages (#427)."""
+
+from __future__ import annotations
+
+import html
+
+from webui.layout import render_page
+from webui.project_registry import ProjectRecord, ProjectRegistry
+
+
+def _escape(text: str) -> str:
+    return html.escape(text, quote=True)
+
+
+def render_projects_list(registry: ProjectRegistry) -> str:
+    rows = []
+    for project in registry.projects:
+        rows.append(
+            "<tr>"
+            f"<td><a href=\"/projects/{_escape(project.id)}\">{_escape(project.repo_name)}</a></td>"
+            f"<td>{_escape(project.gitea_owner)}</td>"
+            f"<td>{_escape(project.remote_host)}</td>"
+            f"<td>{_escape(project.default_branch)}</td>"
+            f"<td><code>{_escape(project.profiles['author'])}</code></td>"
+            "</tr>"
+        )
+    table = (
+        "<table class=\"registry\">"
+        "<thead><tr>"
+        "<th>Repository</th><th>Owner</th><th>Remote</th>"
+        "<th>Branch</th><th>Author profile</th>"
+        "</tr></thead>"
+        f"<tbody>{''.join(rows)}</tbody></table>"
+    )
+    body = (
+        "<h2>Projects</h2>"
+        "<p>Configured repositories managed by the MCP Control Plane.</p>"
+        f"<p class=\"meta\">Registry: <code>{_escape(str(registry.source_path))}</code> "
+        f"(version {registry.version})</p>"
+        f"{table}"
+        "<p><a href=\"/api/projects\">JSON API</a></p>"
+    )
+    return render_page(title="Projects", body_html=body)
+
+
+def render_project_detail(project: ProjectRecord) -> str:
+    profile_rows = "".join(
+        f"<tr><th>{_escape(role)}</th><td><code>{_escape(name)}</code></td></tr>"
+        for role, name in project.profiles.items()
+    )
+    workflow_rows = "".join(
+        f"<tr><th>{_escape(key)}</th><td><code>{_escape(path)}</code></td></tr>"
+        for key, path in project.workflow_paths.items()
+    )
+    schema_rows = "".join(
+        f"<tr><th>{_escape(key)}</th><td><code>{_escape(path)}</code></td></tr>"
+        for key, path in project.schema_paths.items()
+    )
+    checklist_items = []
+    for index, step in enumerate(project.onboarding_checklist, start=1):
+        checklist_items.append(
+            "<li>"
+            f"<strong>{index}. {_escape(step.title)}</strong>"
+            f"<p>{_escape(step.description)}</p>"
+            "</li>"
+        )
+    checklist_html = (
+        "<ol class=\"checklist\">" + "".join(checklist_items) + "</ol>"
+        if checklist_items
+        else "<p>No onboarding steps defined.</p>"
+    )
+    body = (
+        f"<h2>{_escape(project.repo_name)}</h2>"
+        "<p><a href=\"/projects\">← All projects</a></p>"
+        "<h3>Identity</h3>"
+        "<table class=\"detail\">"
+        f"<tr><th>Registry id</th><td><code>{_escape(project.id)}</code></td></tr>"
+        f"<tr><th>Gitea owner</th><td>{_escape(project.gitea_owner)}</td></tr>"
+        f"<tr><th>Remote host</th><td>{_escape(project.remote_host)}</td></tr>"
+        f"<tr><th>Default branch</th><td><code>{_escape(project.default_branch)}</code></td></tr>"
+        f"<tr><th>Local checkout</th><td><code>{_escape(project.local_checkout_path)}</code></td></tr>"
+        "</table>"
+        "<h3>Profiles</h3>"
+        f"<table class=\"detail\">{profile_rows}</table>"
+        "<h3>Workflow paths</h3>"
+        f"<table class=\"detail\">{workflow_rows}</table>"
+        "<h3>Schema paths</h3>"
+        f"<table class=\"detail\">{schema_rows}</table>"
+        "<h3>Onboarding checklist</h3>"
+        "<p class=\"meta\">Read-only MVP — complete these steps outside the UI.</p>"
+        f"{checklist_html}"
+    )
+    return render_page(title=project.repo_name, body_html=body)

webui/prompt_library.py

@@ -0,0 +1,197 @@
+"""Canonical workflow prompt library for the internal web UI (#428)."""
+
+from __future__ import annotations
+
+import hashlib
+import os
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+_WORKFLOW_ROOT = Path("skills/llm-project-workflow/workflows")
+
+_DEFAULT_PROMPT_RE = re.compile(
+    r"\*\*Default task prompt:\*\*\s*\n+>\s*(.+?)(?=\n\n|\nDo not improvise)",
+    re.DOTALL,
+)
+_FRONTMATTER_TASK_MODE_RE = re.compile(r"^task_mode:\s*(\S+)", re.MULTILINE)
+
+
+@dataclass(frozen=True)
+class PromptEntry:
+    slug: str
+    label: str
+    prompt_text: str
+    workflow_path: str
+    task_mode: str | None
+    workflow_hash: str | None
+    source_note: str
+
+
+def _repo_root() -> Path:
+    override = (os.environ.get("WEBUI_REPO_ROOT") or "").strip()
+    if override:
+        return Path(override).resolve()
+    return Path(__file__).resolve().parent.parent
+
+
+def _workflow_file(path: str) -> Path:
+    return _repo_root() / path
+
+
+def _sha256_hex(content: str) -> str:
+    return hashlib.sha256(content.encode("utf-8")).hexdigest()
+
+
+def _read_workflow(path: str) -> tuple[str, str]:
+    file_path = _workflow_file(path)
+    text = file_path.read_text(encoding="utf-8")
+    return text, _sha256_hex(text)
+
+
+def _extract_default_prompt(markdown: str) -> str | None:
+    match = _DEFAULT_PROMPT_RE.search(markdown)
+    if not match:
+        return None
+    lines = [line.strip() for line in match.group(1).splitlines()]
+    return " ".join(line for line in lines if line)
+
+
+def _extract_task_mode(markdown: str) -> str | None:
+    match = _FRONTMATTER_TASK_MODE_RE.search(markdown)
+    return match.group(1) if match else None
+
+
+def _entry_from_workflow(
+    *,
+    slug: str,
+    label: str,
+    workflow_path: str,
+    prompt_override: str | None = None,
+    source_note: str = "",
+) -> PromptEntry:
+    markdown, digest = _read_workflow(workflow_path)
+    prompt_text = prompt_override or _extract_default_prompt(markdown)
+    if not prompt_text:
+        raise ValueError(f"No default task prompt found in {workflow_path}")
+    return PromptEntry(
+        slug=slug,
+        label=label,
+        prompt_text=prompt_text,
+        workflow_path=workflow_path,
+        task_mode=_extract_task_mode(markdown),
+        workflow_hash=digest,
+        source_note=source_note,
+    )
+
+
+def _static_entry(
+    *,
+    slug: str,
+    label: str,
+    prompt_text: str,
+    workflow_path: str,
+    source_note: str,
+) -> PromptEntry:
+    path = _workflow_file(workflow_path)
+    digest = _sha256_hex(path.read_text(encoding="utf-8")) if path.is_file() else None
+    markdown = path.read_text(encoding="utf-8") if path.is_file() else ""
+    return PromptEntry(
+        slug=slug,
+        label=label,
+        prompt_text=prompt_text,
+        workflow_path=workflow_path,
+        task_mode=_extract_task_mode(markdown) if markdown else None,
+        workflow_hash=digest,
+        source_note=source_note,
+    )
+
+
+def load_prompt_library() -> tuple[PromptEntry, ...]:
+    """Load operator prompts derived from canonical workflows."""
+    entries = (
+        _entry_from_workflow(
+            slug="review-pr",
+            label="Review PR",
+            workflow_path=str(_WORKFLOW_ROOT / "review-merge-pr.md"),
+        ),
+        _entry_from_workflow(
+            slug="work-issue",
+            label="Work issue",
+            workflow_path=str(_WORKFLOW_ROOT / "work-issue.md"),
+        ),
+        _entry_from_workflow(
+            slug="create-issue",
+            label="Create issue",
+            workflow_path=str(_WORKFLOW_ROOT / "create-issue.md"),
+        ),
+        _static_entry(
+            slug="comment-issue",
+            label="Comment on issue",
+            workflow_path=str(_WORKFLOW_ROOT / "create-issue.md"),
+            prompt_text=(
+                "Comment on the target Gitea issue only if exact comment_issue "
+                "capability is proven. Load the canonical create-issue workflow "
+                "first and follow §16 (comment-on-existing issue rule). Include "
+                "specific evidence; do not duplicate existing comments."
+            ),
+            source_note="Derived from create-issue.md §16; full policy remains in the workflow file.",
+        ),
+        _static_entry(
+            slug="cleanup",
+            label="Post-merge cleanup",
+            workflow_path="skills/llm-project-workflow/templates/worktree-cleanup.md",
+            prompt_text=(
+                "Task: clean up branch/worktree for PR #<pr> / issue #<n> after merge. "
+                "Confirm the merge on remote master before any deletion; never "
+                "force-remove a dirty worktree."
+            ),
+            source_note="Full cleanup steps live in templates/worktree-cleanup.md.",
+        ),
+        _entry_from_workflow(
+            slug="audit",
+            label="Reconciliation audit",
+            workflow_path=str(_WORKFLOW_ROOT / "reconcile-landed-pr.md"),
+        ),
+        _static_entry(
+            slug="onboarding",
+            label="Project onboarding",
+            workflow_path="skills/llm-project-workflow/SKILL.md",
+            prompt_text=(
+                "Onboard this repository into the MCP Control Plane: prove identity "
+                "and task capability, configure author/reviewer/reconciler profiles "
+                "in separate namespaces, then complete the checklist at /projects. "
+                "Canonical router: skills/llm-project-workflow/SKILL.md."
+            ),
+            source_note="Checklist details live in webui/data/projects.registry.json and /projects.",
+        ),
+    )
+    return entries
+
+
+def find_prompt(slug: str) -> PromptEntry | None:
+    for entry in load_prompt_library():
+        if entry.slug == slug:
+            return entry
+    return None
+
+
+def prompt_to_dict(entry: PromptEntry) -> dict[str, Any]:
+    return {
+        "slug": entry.slug,
+        "label": entry.label,
+        "prompt_text": entry.prompt_text,
+        "workflow_path": entry.workflow_path,
+        "task_mode": entry.task_mode,
+        "workflow_hash": entry.workflow_hash,
+        "source_note": entry.source_note,
+    }
+
+
+def library_to_dict() -> dict[str, Any]:
+    entries = load_prompt_library()
+    return {
+        "count": len(entries),
+        "prompts": [prompt_to_dict(entry) for entry in entries],
+    }

webui/prompt_views.py

@@ -0,0 +1,87 @@
+"""HTML views for the prompt library (#428)."""
+
+from __future__ import annotations
+
+import html
+
+from webui.layout import render_page
+from webui.prompt_library import PromptEntry, load_prompt_library
+
+
+def _escape(text: str) -> str:
+    return html.escape(text, quote=True)
+
+
+def _render_prompt_card(entry: PromptEntry) -> str:
+    hash_short = (
+        f"<code>{_escape(entry.workflow_hash[:12])}</code>"
+        if entry.workflow_hash
+        else "<span class=\"muted\">n/a</span>"
+    )
+    task_mode = (
+        f"<code>{_escape(entry.task_mode)}</code>"
+        if entry.task_mode
+        else "<span class=\"muted\">n/a</span>"
+    )
+    note = (
+        f'<p class="meta">{_escape(entry.source_note)}</p>'
+        if entry.source_note
+        else ""
+    )
+    prompt_id = f"prompt-{entry.slug}"
+    return (
+        f'<section class="prompt-card" id="{_escape(entry.slug)}">'
+        f"<h3>{_escape(entry.label)}</h3>"
+        f'<p class="meta">Workflow: <code>{_escape(entry.workflow_path)}</code> · '
+        f"task_mode: {task_mode} · sha256: {hash_short}</p>"
+        f'<pre class="prompt-text" id="{prompt_id}">{_escape(entry.prompt_text)}</pre>'
+        f'<button type="button" class="copy-btn" data-copy-target="{prompt_id}">'
+        "Copy prompt</button>"
+        f"{note}"
+        "</section>"
+    )
+
+
+PROMPT_PAGE_SCRIPT = """
+<script>
+document.querySelectorAll('.copy-btn').forEach((btn) => {
+  btn.addEventListener('click', async () => {
+    const targetId = btn.getAttribute('data-copy-target');
+    const node = document.getElementById(targetId);
+    if (!node) return;
+    const text = node.textContent || '';
+    try {
+      await navigator.clipboard.writeText(text);
+      const prior = btn.textContent;
+      btn.textContent = 'Copied';
+      setTimeout(() => { btn.textContent = prior; }, 1200);
+    } catch (_err) {
+      btn.textContent = 'Copy failed';
+    }
+  });
+});
+</script>
+"""
+
+
+def render_prompts_page() -> str:
+    entries = load_prompt_library()
+    cards = "".join(_render_prompt_card(entry) for entry in entries)
+    body = (
+        "<h2>Prompt library</h2>"
+        "<p>Short copy/paste task prompts derived from canonical workflows. "
+        "Full policy remains in the cited workflow files — not duplicated here.</p>"
+        f"{cards}"
+        "<p><a href=\"/api/prompts\">JSON API</a></p>"
+        f"{PROMPT_PAGE_SCRIPT}"
+    )
+    return render_page(title="Prompts", body_html=body)
+
+
+def render_prompt_detail(entry: PromptEntry) -> str:
+    body = (
+        f"<p><a href=\"/prompts\">← All prompts</a></p>"
+        f"{_render_prompt_card(entry)}"
+        f"{PROMPT_PAGE_SCRIPT}"
+    )
+    return render_page(title=entry.label, body_html=body)

webui/queue_loader.py

@@ -0,0 +1,385 @@
+"""Load live Gitea PR/issue queue state for the web UI dashboard (#429)."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, Callable
+from urllib.parse import urlparse
+
+from gitea_auth import api_fetch_page, get_auth_header, repo_api_url
+
+from webui.project_registry import ProjectRecord, load_registry
+
+_CLOSES_RE = re.compile(r"(?:closes|fixes|resolves)\s+#(\d+)", re.I)
+_ISSUE_REF_RE = re.compile(r"#(\d+)")
+_STALE_DAYS = 14
+
+
+@dataclass(frozen=True)
+class PaginationMeta:
+    page: int
+    per_page: int
+    returned_count: int
+    has_more: bool
+    is_final_page: bool
+    inventory_complete: bool
+    pages_fetched: int
+
+
+@dataclass(frozen=True)
+class QueueItem:
+    number: int
+    title: str
+    badges: tuple[str, ...]
+    extra: dict[str, str]
+
+
+@dataclass(frozen=True)
+class QueueSnapshot:
+    project_id: str
+    repo_label: str
+    prs: tuple[QueueItem, ...]
+    issues: tuple[QueueItem, ...]
+    pr_pagination: PaginationMeta | None
+    issue_pagination: PaginationMeta | None
+    fetch_error: str | None = None
+
+
+def _host_from_url(remote_host: str) -> str:
+    parsed = urlparse(remote_host.strip())
+    return parsed.netloc or remote_host.strip().rstrip("/")
+
+
+def _extract_linked_issue(title: str | None, body: str | None = None) -> int | None:
+    for text in (title, body):
+        if not text:
+            continue
+        match = _CLOSES_RE.search(text)
+        if match:
+            return int(match.group(1))
+    if body:
+        refs = _ISSUE_REF_RE.findall(body)
+        if refs:
+            return int(refs[0])
+    return None
+
+
+def _is_stale(updated_at: str | None) -> bool:
+    if not updated_at:
+        return False
+    try:
+        normalized = updated_at.replace("Z", "+00:00")
+        parsed = datetime.fromisoformat(normalized)
+        if parsed.tzinfo is None:
+            parsed = parsed.replace(tzinfo=timezone.utc)
+        age = datetime.now(timezone.utc) - parsed.astimezone(timezone.utc)
+        return age.days >= _STALE_DAYS
+    except ValueError:
+        return False
+
+
+def _classify_pr(
+    pr: dict,
+    *,
+    issue_claimed: set[int],
+    issue_to_prs: dict[int, list[int]],
+) -> tuple[str, ...]:
+    badges: list[str] = []
+    mergeable = pr.get("mergeable")
+    if mergeable is False:
+        badges.append("blocked")
+    linked = _extract_linked_issue(pr.get("title"), pr.get("body"))
+    if linked is not None:
+        if linked in issue_claimed:
+            badges.append("claimed")
+        if len(issue_to_prs.get(linked, [])) > 1:
+            badges.append("duplicate")
+    if _is_stale(pr.get("updated_at")):
+        badges.append("stale")
+    if mergeable is True and "blocked" not in badges:
+        badges.append("in-review")
+    if not badges:
+        badges.append("open")
+    return tuple(dict.fromkeys(badges))
+
+
+def _classify_issue(
+    issue: dict,
+    *,
+    linked_prs: list[int],
+) -> tuple[str, ...]:
+    badges: list[str] = []
+    labels = [lb.get("name", "") for lb in issue.get("labels", [])]
+    if "status:in-progress" in labels:
+        badges.append("claimed")
+    if len(linked_prs) > 1:
+        badges.append("duplicate")
+    elif linked_prs and "claimed" not in badges:
+        badges.append("in-review")
+    if _is_stale(issue.get("updated_at")):
+        badges.append("stale")
+    if not badges:
+        badges.append("open")
+    return tuple(dict.fromkeys(badges))
+
+
+def _format_pr_item(pr: dict, badges: tuple[str, ...]) -> QueueItem:
+    head = pr.get("head") or {}
+    base = pr.get("base") or {}
+    mergeable = pr.get("mergeable")
+    merge_label = (
+        "mergeable" if mergeable is True else "conflicted" if mergeable is False else "unknown"
+    )
+    linked = _extract_linked_issue(pr.get("title"), pr.get("body"))
+    return QueueItem(
+        number=int(pr["number"]),
+        title=str(pr.get("title") or ""),
+        badges=badges,
+        extra={
+            "branch": f"{head.get('ref', '?')} → {base.get('ref', '?')}",
+            "head_sha": str(head.get("sha") or "")[:12],
+            "mergeable": merge_label,
+            "linked_issue": str(linked) if linked is not None else "",
+        },
+    )
+
+
+def _format_issue_item(issue: dict, badges: tuple[str, ...]) -> QueueItem:
+    labels = ", ".join(lb.get("name", "") for lb in issue.get("labels", []))
+    assignee = (issue.get("assignee") or {}).get("login", "")
+    return QueueItem(
+        number=int(issue["number"]),
+        title=str(issue.get("title") or ""),
+        badges=badges,
+        extra={
+            "labels": labels or "—",
+            "assignee": assignee or "unassigned",
+            "state": str(issue.get("state") or ""),
+        },
+    )
+
+
+def _pagination_from_pages(
+    *,
+    per_page: int,
+    pages_fetched: int,
+    returned_count: int,
+    is_final_page: bool,
+) -> PaginationMeta:
+    return PaginationMeta(
+        page=1,
+        per_page=per_page,
+        returned_count=returned_count,
+        has_more=not is_final_page,
+        is_final_page=is_final_page,
+        inventory_complete=is_final_page,
+        pages_fetched=pages_fetched,
+    )
+
+
+def _fetch_prs(
+    host: str,
+    org: str,
+    repo: str,
+    auth: str,
+    *,
+    per_page: int = 50,
+) -> tuple[list[dict], PaginationMeta]:
+    url = f"{repo_api_url(host, org, repo)}/pulls?state=open"
+    all_raw: list[dict] = []
+    pages_fetched = 0
+    is_final = False
+    page = 1
+    while pages_fetched < 20:
+        raw_page, meta = api_fetch_page(url, auth, page=page, limit=per_page)
+        pages_fetched += 1
+        all_raw.extend(raw_page)
+        is_final = bool(meta["is_final_page"])
+        if is_final:
+            break
+        page += 1
+    pagination = _pagination_from_pages(
+        per_page=per_page,
+        pages_fetched=pages_fetched,
+        returned_count=len(all_raw),
+        is_final_page=is_final,
+    )
+    return all_raw, pagination
+
+
+def _fetch_issues(
+    host: str,
+    org: str,
+    repo: str,
+    auth: str,
+    *,
+    per_page: int = 50,
+) -> tuple[list[dict], PaginationMeta]:
+    url = f"{repo_api_url(host, org, repo)}/issues?state=open&type=issues"
+    all_raw: list[dict] = []
+    page = 1
+    pages_fetched = 0
+    is_final = False
+    while pages_fetched < 20:
+        raw_page, meta = api_fetch_page(url, auth, page=page, limit=per_page)
+        pages_fetched += 1
+        all_raw.extend(raw_page)
+        is_final = bool(meta["is_final_page"])
+        if is_final:
+            break
+        page += 1
+    pagination = _pagination_from_pages(
+        per_page=per_page,
+        pages_fetched=pages_fetched,
+        returned_count=len(all_raw),
+        is_final_page=is_final,
+    )
+    return all_raw, pagination
+
+
+def load_queue_snapshot(
+    project_id: str | None = None,
+    *,
+    fetch_prs: Callable[..., tuple[list[dict], PaginationMeta]] | None = None,
+    fetch_issues: Callable[..., tuple[list[dict], PaginationMeta]] | None = None,
+) -> QueueSnapshot:
+    """Load open PR and issue queues for a registry project (default: first entry)."""
+    registry = load_registry()
+    project: ProjectRecord | None = None
+    if project_id:
+        for entry in registry.projects:
+            if entry.id == project_id:
+                project = entry
+                break
+    else:
+        project = registry.projects[0] if registry.projects else None
+
+    if project is None:
+        return QueueSnapshot(
+            project_id=project_id or "",
+            repo_label="",
+            prs=(),
+            issues=(),
+            pr_pagination=None,
+            issue_pagination=None,
+            fetch_error="project not found in registry",
+        )
+
+    host = _host_from_url(project.remote_host)
+    pr_fetch = fetch_prs or _fetch_prs
+    issue_fetch = fetch_issues or _fetch_issues
+    using_live_fetch = fetch_prs is None or fetch_issues is None
+    auth = get_auth_header(host) if using_live_fetch else "test-auth"
+    if using_live_fetch and not auth:
+        return QueueSnapshot(
+            project_id=project.id,
+            repo_label=f"{project.gitea_owner}/{project.repo_name}",
+            prs=(),
+            issues=(),
+            pr_pagination=None,
+            issue_pagination=None,
+            fetch_error=(
+                f"Gitea credentials unavailable for {host}; "
+                "queue cannot be loaded (fail closed — not showing empty queue)"
+            ),
+        )
+
+    try:
+        raw_prs, pr_pagination = pr_fetch(
+            host, project.gitea_owner, project.repo_name, auth
+        )
+        raw_issues, issue_pagination = issue_fetch(
+            host, project.gitea_owner, project.repo_name, auth
+        )
+    except Exception as exc:  # noqa: BLE001 — surface operator-visible fetch errors
+        return QueueSnapshot(
+            project_id=project.id,
+            repo_label=f"{project.gitea_owner}/{project.repo_name}",
+            prs=(),
+            issues=(),
+            pr_pagination=None,
+            issue_pagination=None,
+            fetch_error=f"Gitea fetch failed: {exc}",
+        )
+
+    issue_to_prs: dict[int, list[int]] = {}
+    for pr in raw_prs:
+        linked = _extract_linked_issue(pr.get("title"), pr.get("body"))
+        if linked is not None:
+            issue_to_prs.setdefault(linked, []).append(int(pr["number"]))
+
+    issue_claimed = {
+        int(i["number"])
+        for i in raw_issues
+        if any(lb.get("name") == "status:in-progress" for lb in i.get("labels", []))
+    }
+
+    pr_items = tuple(
+        _format_pr_item(
+            pr,
+            _classify_pr(
+                pr,
+                issue_claimed=issue_claimed,
+                issue_to_prs=issue_to_prs,
+            ),
+        )
+        for pr in sorted(raw_prs, key=lambda p: int(p["number"]), reverse=True)
+    )
+    issue_items = tuple(
+        _format_issue_item(
+            issue,
+            _classify_issue(
+                issue,
+                linked_prs=issue_to_prs.get(int(issue["number"]), []),
+            ),
+        )
+        for issue in sorted(raw_issues, key=lambda i: int(i["number"]), reverse=True)
+    )
+
+    return QueueSnapshot(
+        project_id=project.id,
+        repo_label=f"{project.gitea_owner}/{project.repo_name}",
+        prs=pr_items,
+        issues=issue_items,
+        pr_pagination=pr_pagination,
+        issue_pagination=issue_pagination,
+    )
+
+
+def snapshot_to_dict(snapshot: QueueSnapshot) -> dict[str, Any]:
+    """JSON-serializable export for /api/queue."""
+
+    def _page(meta: PaginationMeta | None) -> dict[str, Any] | None:
+        if meta is None:
+            return None
+        return {
+            "page": meta.page,
+            "per_page": meta.per_page,
+            "returned_count": meta.returned_count,
+            "has_more": meta.has_more,
+            "is_final_page": meta.is_final_page,
+            "inventory_complete": meta.inventory_complete,
+            "pages_fetched": meta.pages_fetched,
+        }
+
+    def _item(item: QueueItem) -> dict[str, Any]:
+        return {
+            "number": item.number,
+            "title": item.title,
+            "badges": list(item.badges),
+            **item.extra,
+        }
+
+    return {
+        "project_id": snapshot.project_id,
+        "repo": snapshot.repo_label,
+        "fetch_error": snapshot.fetch_error,
+        "prs": [_item(p) for p in snapshot.prs],
+        "issues": [_item(i) for i in snapshot.issues],
+        "pagination": {
+            "prs": _page(snapshot.pr_pagination),
+            "issues": _page(snapshot.issue_pagination),
+        },
+    }

webui/queue_views.py

@@ -0,0 +1,119 @@
+"""HTML views for the live Gitea queue dashboard (#429)."""
+
+from __future__ import annotations
+
+import html
+
+from webui.queue_loader import PaginationMeta, QueueItem, QueueSnapshot
+
+
+def _badge_html(badges: tuple[str, ...]) -> str:
+    if not badges:
+        return ""
+    chips = "".join(
+        f'<span class="badge badge-{html.escape(b)}">{html.escape(b)}</span>'
+        for b in badges
+    )
+    return f'<span class="badges">{chips}</span>'
+
+
+def _pagination_html(label: str, meta: PaginationMeta | None) -> str:
+    if meta is None:
+        return (
+            f"<p class='muted'><strong>{html.escape(label)} pagination:</strong> "
+            "unavailable</p>"
+        )
+    status = "complete" if meta.inventory_complete else "partial"
+    more = "yes" if meta.has_more else "no"
+    return (
+        f"<p class='meta'><strong>{html.escape(label)} pagination ({status}):</strong> "
+        f"returned {meta.returned_count} · per_page {meta.per_page} · "
+        f"pages_fetched {meta.pages_fetched} · has_more {more} · "
+        f"final_page {'yes' if meta.is_final_page else 'no'}</p>"
+    )
+
+
+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>"
+
+    headers = "".join(f"<th>{html.escape(label)}</th>" for _, label in columns)
+    rows = []
+    for item in items:
+        cells = []
+        for key, _ in columns:
+            if key == "number":
+                cells.append(f"<td>#{item.number}</td>")
+            elif key == "title":
+                cells.append(
+                    f"<td>{html.escape(item.title)}{_badge_html(item.badges)}</td>"
+                )
+            else:
+                cells.append(f"<td>{html.escape(item.extra.get(key, ''))}</td>")
+        rows.append("<tr>" + "".join(cells) + "</tr>")
+
+    body = (
+        f"<h3>{html.escape(title)}</h3>"
+        f"<table class='registry'><thead><tr>{headers}</tr></thead>"
+        f"<tbody>{''.join(rows)}</tbody></table>"
+    )
+    return body
+
+
+def render_queue_page(snapshot: QueueSnapshot) -> str:
+    error_block = ""
+    if snapshot.fetch_error:
+        error_block = (
+            f'<div class="stub"><p><strong>Queue unavailable:</strong> '
+            f"{html.escape(snapshot.fetch_error)}</p></div>"
+        )
+
+    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"),
+            ("branch", "Branch"),
+            ("head_sha", "Head"),
+            ("mergeable", "Mergeable"),
+            ("linked_issue", "Linked issue"),
+        ),
+    )
+    issue_section = _queue_table(
+        title="Open issues",
+        items=snapshot.issues,
+        fetch_failed=fetch_failed,
+        columns=(
+            ("number", "#"),
+            ("title", "Title"),
+            ("labels", "Labels"),
+            ("assignee", "Assignee"),
+            ("state", "State"),
+        ),
+    )
+
+    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"{_pagination_html('PR', snapshot.pr_pagination)}"
+        f"{pr_section}"
+        f"{_pagination_html('Issue', snapshot.issue_pagination)}"
+        f"{issue_section}"
+        "<p class='muted'>Read-only MVP — claims, reviews, and merges stay in Gitea/MCP tools.</p>"
+    )