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

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

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

author_duplicate_work_gate.py

@@ -1,198 +0,0 @@
-"""Early duplicate-work detection for author work-issue flows (#400)."""
-
-from __future__ import annotations
-
-import re
-from typing import Any
-
-from issue_claim_heartbeat import (
-    _linked_open_pr,
-    _matching_branch_names,
-    classify_issue_claim,
-)
-
-STAGES = (
-    "claim",
-    "lock",
-    "worktree",
-    "edit",
-    "commit",
-    "push",
-    "create_pr",
-)
-
-ELIGIBILITY_OPEN_PR_EXISTS = "OPEN_PR_EXISTS"
-ELIGIBILITY_DUPLICATE_BRANCH_EXISTS = "DUPLICATE_BRANCH_EXISTS"
-ELIGIBILITY_ACTIVE_CLAIM = "ACTIVE_CLAIM_BY_OTHER"
-ELIGIBILITY_CLEAR = "CLEAR"
-
-
-def assess_author_duplicate_work(
-    issue_number: int,
-    *,
-    stage: str,
-    open_prs: list[dict] | None = None,
-    branch_names: list[str] | None = None,
-    claim_entry: dict | None = None,
-    matching_branches: list[str] | None = None,
-    allow_stale_takeover: bool = False,
-) -> dict[str, Any]:
-    """Fail closed when duplicate work is detected before author mutations (#400)."""
-    stage_norm = (stage or "").strip().lower()
-    if stage_norm not in STAGES:
-        return {
-            "allowed": False,
-            "block": True,
-            "eligibility_class": "INVALID_STAGE",
-            "stage": stage_norm or None,
-            "reasons": [f"unknown duplicate-work stage {stage!r}"],
-            "safe_next_action": f"use one of: {', '.join(STAGES)}",
-        }
-
-    prs = list(open_prs or [])
-    linked_pr = _linked_open_pr(int(issue_number), prs)
-    branches = list(
-        matching_branches
-        if matching_branches is not None
-        else _matching_branch_names(int(issue_number), list(branch_names or []))
-    )
-
-    reasons: list[str] = []
-    eligibility = ELIGIBILITY_CLEAR
-
-    if linked_pr:
-        eligibility = ELIGIBILITY_OPEN_PR_EXISTS
-        reasons.append(
-            f"open PR #{linked_pr.get('number')} already covers issue "
-            f"#{issue_number}"
-        )
-
-    if branches and stage_norm in {"claim", "lock", "worktree", "edit"}:
-        if eligibility == ELIGIBILITY_CLEAR:
-            eligibility = ELIGIBILITY_DUPLICATE_BRANCH_EXISTS
-        reasons.append(
-            f"remote branch(es) already exist for issue #{issue_number}: "
-            f"{', '.join(branches)}"
-        )
-
-    entry = claim_entry or {}
-    claim_status = (entry.get("status") or "").strip()
-    if claim_status in {"active", "awaiting_review"} and stage_norm == "claim":
-        if entry.get("reclaimable") and allow_stale_takeover:
-            pass
-        elif claim_status == "active" and not entry.get("reclaimable"):
-            if eligibility == ELIGIBILITY_CLEAR:
-                eligibility = ELIGIBILITY_ACTIVE_CLAIM
-            reasons.append(
-                f"issue #{issue_number} has active claim "
-                f"(status={claim_status})"
-            )
-    elif claim_status == "awaiting_review" and stage_norm == "claim":
-        if not linked_pr:
-            reasons.append(
-                f"issue #{issue_number} is awaiting_review but no linked open PR "
-                "was supplied for duplicate-work proof"
-            )
-
-    allowed = not reasons
-    outcome = "duplicate_work_prevented" if not allowed else "clear"
-    if stage_norm == "create_pr" and not allowed:
-        outcome = "duplicate_pr_prevented"
-    elif stage_norm in {"commit", "push"} and not allowed:
-        outcome = "duplicate_push_prevented" if stage_norm == "push" else "duplicate_commit_prevented"
-
-    return {
-        "allowed": allowed,
-        "block": not allowed,
-        "eligibility_class": eligibility if not allowed else ELIGIBILITY_CLEAR,
-        "stage": stage_norm,
-        "linked_open_pr": linked_pr.get("number") if linked_pr else None,
-        "matching_branches": branches,
-        "claim_status": claim_status or None,
-        "outcome": outcome,
-        "reasons": reasons,
-        "safe_next_action": (
-            "stop without edits/commit/push/PR; produce reconciliation handoff "
-            "preserving local work only"
-            if not allowed
-            else "proceed"
-        ),
-    }
-
-
-def build_claim_entry_from_classification(classification: dict) -> dict:
-    """Map ``classify_issue_claim`` output to gate claim metadata."""
-    return {
-        "status": classification.get("status"),
-        "reclaimable": classification.get("reclaimable"),
-        "linked_open_pr": classification.get("linked_open_pr"),
-    }
-
-
-_DUPLICATE_OUTCOME_RE = re.compile(
-    r"(duplicate\s+(?:pr|branch|commit|push)\s+prevented|"
-    r"duplicate\s+work\s+not\s+prevented|reconciliation\s+handoff)",
-    re.IGNORECASE,
-)
-
-
-def assess_work_issue_duplicate_prevention_report(report_text: str) -> dict:
-    """#400: work-issue reports must state duplicate-work prevention outcome."""
-    text = report_text or ""
-    if "duplicate work" not in text.lower() and "duplicate pr" not in text.lower():
-        return {
-            "proven": True,
-            "block": False,
-            "reasons": [],
-            "safe_next_action": "proceed",
-        }
-    if _DUPLICATE_OUTCOME_RE.search(text):
-        return {
-            "proven": True,
-            "block": False,
-            "reasons": [],
-            "safe_next_action": "proceed",
-        }
-    return {
-        "proven": False,
-        "block": True,
-        "reasons": [
-            "duplicate-work discussion must name a prevention outcome "
-            "(duplicate PR/branch/commit/push prevented, or duplicate work "
-            "not prevented, or reconciliation handoff)"
-        ],
-        "safe_next_action": "state exact duplicate-work prevention class in final report",
-    }
-
-
-def classify_and_assess(
-    issue: dict,
-    *,
-    stage: str,
-    comments: list[dict] | None = None,
-    open_prs: list[dict] | None = None,
-    branch_names: list[str] | None = None,
-    allow_stale_takeover: bool = False,
-) -> dict[str, Any]:
-    """Combine claim classification with duplicate-work gate assessment."""
-    issue_number = int(issue.get("number") or 0)
-    claim = classify_issue_claim(
-        issue=issue,
-        comments=comments or [],
-        open_prs=open_prs or [],
-        branch_names=branch_names or [],
-    )
-    assessment = assess_author_duplicate_work(
-        issue_number,
-        stage=stage,
-        open_prs=open_prs,
-        branch_names=branch_names,
-        claim_entry=build_claim_entry_from_classification(claim),
-        matching_branches=claim.get("matching_branches"),
-        allow_stale_takeover=allow_stale_takeover,
-    )
-    return {
-        "issue_number": issue_number,
-        "claim": claim,
-        "duplicate_work": assessment,
-    }

docs/llm-workflow-runbooks.md

@@ -274,12 +274,24 @@ is proven abandoned and the takeover is recorded.
 
 Gitea-Tools lease gates: `gitea_lock_issue` (fail-closed before author
 mutations), `status:in-progress`, and claim comments. `gitea_lock_issue`
-records an `author_issue_work` lease in the issue-lock payload with issue
-number, optional PR number, branch, worktree path, claimant identity/profile,
-created timestamp, expiry timestamp, and last heartbeat timestamp. An active
-same-issue/same-operation lease blocks duplicate work. An expired lease still
-blocks takeover until a recovery review records why the prior work is abandoned,
-completed, or unsafe to continue.
+records an `author_issue_work` lease in a keyed lock file under
+`GITEA_ISSUE_LOCK_DIR` (default `~/.cache/gitea-tools/issue-locks`), one file
+per `remote` + `org` + `repo` + `issue_number`. The current MCP session binds
+its active lock through a per-process pointer so concurrent repos/issues never
+share one overwrite-prone slot (#443).
+
+Each lock payload includes issue number, optional PR number, branch, worktree
+path, claimant identity/profile, created timestamp, expiry timestamp, and last
+heartbeat timestamp. An active same-issue/same-operation lease blocks duplicate
+work. An expired lease still blocks takeover until a recovery review records why
+the prior work is abandoned, completed, or unsafe to continue.
+
+**Do not manually seed `/tmp/gitea_issue_lock.json` or any lock file as a normal
+recovery path.** That global slot is deprecated and can clobber unrelated live
+leases (#438). After an MCP restart, call `gitea_lock_issue` again — own-branch
+adoption rebinds the session when the issue's exact branch already exists (#442).
+`gitea_create_pr` resolves the durable keyed lock by session pointer or by
+matching `head` branch without unsafe manual seeding.
 
 Remote branches matching the issue number are also treated as active work unless
 the recovery review proves the branch is abandoned or superseded. Never delete

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). "
@@ -500,10 +538,12 @@ import task_capability_map  # noqa: E402
 import review_proofs  # noqa: E402
 import agent_temp_artifacts
 import issue_lock_worktree  # noqa: E402
+import issue_lock_store  # noqa: E402
+import issue_lock_adoption  # noqa: E402
 import already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
-import author_duplicate_work_gate  # noqa: E402
+import issue_work_duplicate_gate  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
 import reconciler_profile  # noqa: E402
 import reconciliation_workflow  # noqa: E402
@@ -511,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"
@@ -544,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:
@@ -642,6 +727,132 @@ def _branch_entry_name(branch: dict | str) -> str:
     return str(branch.get("name") or branch.get("ref") or "")
 
 
+def _live_fetch_issue_duplicate_context(
+    h: str,
+    o: str,
+    r: str,
+    auth: str,
+    issue_number: int,
+) -> tuple[list[dict], list[str], dict]:
+    """Live open PRs, remote branch names, and claim state for one issue."""
+    base = repo_api_url(h, o, r)
+    open_prs = api_get_all(f"{base}/pulls?state=open", auth)
+    branches = api_get_all(f"{base}/branches", auth)
+    branch_names = [_branch_entry_name(b) for b in branches]
+    issue = api_request("GET", f"{base}/issues/{issue_number}", auth) or {}
+    comments = api_request(
+        "GET", f"{base}/issues/{issue_number}/comments", auth
+    ) or []
+    claim_entry = issue_claim_heartbeat.classify_issue_claim(
+        issue=issue,
+        comments=comments,
+        open_prs=open_prs,
+        branch_names=branch_names,
+    )
+    return open_prs, branch_names, claim_entry
+
+
+# Injectable duplicate-work context fetcher (#400). Production uses the live
+# Gitea API path above; unit tests patch this symbol instead of hitting the
+# network.
+issue_duplicate_context_fetcher = _live_fetch_issue_duplicate_context
+
+
+def _collect_issue_duplicate_context(
+    h: str,
+    o: str,
+    r: str,
+    auth: str,
+    issue_number: int,
+) -> tuple[list[dict], list[str], dict]:
+    return issue_duplicate_context_fetcher(h, o, r, auth, issue_number)
+
+
+def _assess_issue_duplicate_gate(
+    issue_number: int,
+    *,
+    h: str,
+    o: str,
+    r: str,
+    auth: str,
+    locked_branch: str | None = None,
+    phase: str,
+) -> dict:
+    open_prs, branch_names, claim_entry = _collect_issue_duplicate_context(
+        h, o, r, auth, issue_number
+    )
+    return issue_work_duplicate_gate.assess_work_issue_duplicate_gate(
+        issue_number,
+        open_prs=open_prs,
+        branch_names=branch_names,
+        claim_entry=claim_entry,
+        locked_branch=locked_branch,
+        phase=phase,
+    )
+
+
+def _duplicate_gate_block_response(gate: dict, **extra) -> dict:
+    out = {
+        "success": False,
+        "performed": False,
+        "reasons": list(gate.get("reasons") or []),
+        "duplicate_gate": gate,
+        "safe_next_action": gate.get("safe_next_action"),
+    }
+    out.update(extra)
+    return out
+
+
+def _enforce_locked_issue_duplicate_recheck(
+    remote: str,
+    phase: str,
+    *,
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict | None:
+    """Re-check duplicate-work gates for the locked issue (#400)."""
+    lock_data = _load_existing_issue_lock()
+    if not lock_data:
+        return None
+    issue_number = int(lock_data.get("issue_number") or 0)
+    locked_branch = lock_data.get("branch_name")
+    if not issue_number:
+        return None
+    h, o, r = _resolve(
+        remote or lock_data.get("remote") or "dadeschools",
+        host or lock_data.get("host"),
+        org or lock_data.get("org"),
+        repo or lock_data.get("repo"),
+    )
+    auth = _auth(h)
+    gate = _assess_issue_duplicate_gate(
+        issue_number,
+        h=h,
+        o=o,
+        r=r,
+        auth=auth,
+        locked_branch=locked_branch,
+        phase=phase,
+    )
+    if gate.get("block"):
+        return gate
+    return None
+
+
+def _branch_entry_commit_sha(branch: dict | str) -> str | None:
+    """Best-effort head SHA for a Gitea branch entry (None when absent)."""
+    if not isinstance(branch, dict):
+        return None
+    commit = branch.get("commit")
+    if isinstance(commit, dict):
+        sha = commit.get("id") or commit.get("sha")
+        if sha:
+            return str(sha)
+    sha = branch.get("commit_sha")
+    return str(sha) if sha else None
+
+
 def _reveal_endpoints() -> bool:
     """Admin/debug opt-in (#120): include endpoint URLs and token source
     names in tool output. Off by default so normal LLM-facing responses
@@ -973,6 +1184,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.
 
@@ -985,6 +1197,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).
@@ -1011,7 +1224,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(
@@ -1050,76 +1263,6 @@ def gitea_create_issue(
     return _with_optional_url({"number": data["number"]}, data.get("html_url"))
 
 
-def _list_repo_branch_names(h: str, o: str, r: str, auth: str, *, limit: int = 200) -> list[str]:
-    branches = api_get_all(f"{repo_api_url(h, o, r)}/branches", auth, limit=limit)
-    return [_branch_entry_name(branch) for branch in branches]
-
-
-def _gather_author_duplicate_work_context(
-    issue_number: int,
-    *,
-    h: str,
-    o: str,
-    r: str,
-    auth: str,
-    exclude_branch_name: str | None = None,
-) -> dict:
-    base = repo_api_url(h, o, r)
-    issue = api_request("GET", f"{base}/issues/{issue_number}", auth)
-    comments = api_request("GET", f"{base}/issues/{issue_number}/comments", auth) or []
-    open_prs = api_get_all(f"{base}/pulls?state=open", auth)
-    branch_names = _list_repo_branch_names(h, o, r, auth)
-    if exclude_branch_name:
-        branch_names = [
-            name for name in branch_names
-            if name != exclude_branch_name
-        ]
-    return {
-        "issue": issue,
-        "comments": comments,
-        "open_prs": open_prs,
-        "branch_names": branch_names,
-    }
-
-
-def _enforce_author_duplicate_work_gate(
-    issue_number: int,
-    stage: str,
-    *,
-    h: str,
-    o: str,
-    r: str,
-    auth: str,
-    allow_stale_takeover: bool = False,
-    exclude_branch_name: str | None = None,
-) -> dict:
-    """Fail closed when duplicate work is detected (#400)."""
-    ctx = _gather_author_duplicate_work_context(
-        issue_number,
-        h=h,
-        o=o,
-        r=r,
-        auth=auth,
-        exclude_branch_name=exclude_branch_name,
-    )
-    result = author_duplicate_work_gate.classify_and_assess(
-        ctx["issue"],
-        stage=stage,
-        comments=ctx["comments"],
-        open_prs=ctx["open_prs"],
-        branch_names=ctx["branch_names"],
-        allow_stale_takeover=allow_stale_takeover,
-    )
-    assessment = result.get("duplicate_work") or {}
-    if assessment.get("block"):
-        reasons = "; ".join(assessment.get("reasons") or ["duplicate work detected"])
-        raise RuntimeError(
-            f"Author duplicate-work gate (#400) blocked at stage '{stage}': "
-            f"{reasons} (fail closed)"
-        )
-    return result
-
-
 @mcp.tool()
 def gitea_lock_issue(
     issue_number: int,
@@ -1157,8 +1300,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,
@@ -1182,17 +1326,44 @@ def gitea_lock_issue(
             issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment)
         )
 
-    h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
-    _enforce_author_duplicate_work_gate(
+    duplicate_gate = _assess_issue_duplicate_gate(
         issue_number,
-        "lock",
         h=h,
         o=o,
         r=r,
         auth=auth,
-        exclude_branch_name=branch_name,
+        locked_branch=branch_name,
+        phase=issue_work_duplicate_gate.PHASE_LOCK,
     )
+    if duplicate_gate.get("block"):
+        raise ValueError("; ".join(duplicate_gate.get("reasons") or [
+            f"duplicate work gate blocked issue #{issue_number} (fail closed)"
+        ]))
+
+    branch_url = f"{repo_api_url(h, o, r)}/branches"
+    try:
+        branches = api_get_all(branch_url, auth)
+    except Exception as e:
+        raise RuntimeError(f"Could not list branches to verify issue lock: {e}")
+    existing_branch_entries = [
+        {
+            "name": _branch_entry_name(branch),
+            "commit_sha": _branch_entry_commit_sha(branch),
+        }
+        for branch in branches
+    ]
+    adoption = issue_lock_adoption.assess_own_branch_adoption(
+        issue_number=issue_number,
+        requested_branch=branch_name,
+        existing_branches=existing_branch_entries,
+    )
+    if adoption["block"]:
+        competing = ", ".join(adoption["competing_branches"])
+        raise ValueError(
+            f"Issue #{issue_number} already has matching branch '{competing}' "
+            "that is not the requested branch (fail closed)"
+        )
 
     work_lease = _build_author_issue_work_lease(
         issue_number=issue_number,
@@ -1210,11 +1381,7 @@ def gitea_lock_issue(
         "work_lease": work_lease,
     }
 
-    try:
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
-            json.dump(data, f)
-    except Exception as e:
-        raise RuntimeError(f"Could not write issue lock file: {e}")
+    lock_file_path = _save_issue_lock(data)
 
     agent_artifacts = agent_temp_artifacts.find_agent_temp_artifacts_from_porcelain(
         git_state.get("porcelain_status") or ""
@@ -1229,7 +1396,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): "
@@ -1238,6 +1420,39 @@ def gitea_lock_issue(
     return result
 
 
+@mcp.tool()
+def gitea_assess_work_issue_duplicate(
+    issue_number: int,
+    branch_name: str | None = None,
+    phase: str = issue_work_duplicate_gate.PHASE_LOCK,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Read-only duplicate-work gate for author sessions before mutations (#400)."""
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "performed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    gate = _assess_issue_duplicate_gate(
+        issue_number,
+        h=h,
+        o=o,
+        r=r,
+        auth=auth,
+        locked_branch=branch_name,
+        phase=phase,
+    )
+    return {"success": not gate.get("block"), **gate}
+
+
 @mcp.tool()
 def gitea_create_pr(
     title: str,
@@ -1288,15 +1503,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")
@@ -1313,17 +1521,6 @@ def gitea_create_pr(
             f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)"
         )
 
-    auth = _auth(h)
-    _enforce_author_duplicate_work_gate(
-        int(locked_issue),
-        "create_pr",
-        h=h,
-        o=o,
-        r=r,
-        auth=auth,
-        exclude_branch_name=locked_branch,
-    )
-
     # Check for forbidden terms anywhere in title/body
     forbidden_terms = ["equivalent", "related", "same as"]
     text_to_check = f"{title} {body}".lower()
@@ -1340,6 +1537,22 @@ def gitea_create_pr(
             f"PR title or body must contain 'Closes #{locked_issue}' or 'Fixes #{locked_issue}' exactly to ensure durable tracking (fail closed)"
         )
 
+    duplicate_block = _enforce_locked_issue_duplicate_recheck(
+        remote,
+        issue_work_duplicate_gate.PHASE_CREATE_PR,
+        host=host,
+        org=org,
+        repo=repo,
+    )
+    if duplicate_block:
+        return _duplicate_gate_block_response(
+            duplicate_block,
+            number=None,
+            issue_number=locked_issue,
+            branch_name=locked_branch,
+        )
+
+    auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/pulls"
     payload = {"title": title, "body": body, "head": head, "base": base}
     meta = {"title": title, "head": head, "base": base}
@@ -2802,13 +3015,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:
@@ -2943,6 +3150,20 @@ def gitea_commit_files(
     if blocked:
         return blocked
 
+    duplicate_block = _enforce_locked_issue_duplicate_recheck(
+        remote,
+        issue_work_duplicate_gate.PHASE_COMMIT,
+        host=host,
+        org=org,
+        repo=repo,
+    )
+    if duplicate_block:
+        return _duplicate_gate_block_response(
+            duplicate_block,
+            commit="",
+            branch="",
+        )
+
     verify_preflight_purity(remote)
     processed_files, source_proofs = _prepare_commit_payload_files(files)
 
@@ -6053,14 +6274,6 @@ def gitea_mark_issue(
         )
 
     if action == "start":
-        _enforce_author_duplicate_work_gate(
-            issue_number,
-            "claim",
-            h=h,
-            o=o,
-            r=r,
-            auth=auth,
-        )
         with _audited("label_issue", host=h, remote=remote, org=o, repo=r,
                       issue_number=issue_number,
                       request_metadata={"op": "add", "label": "status:in-progress"}):
@@ -6142,65 +6355,6 @@ def gitea_post_heartbeat(
     )
 
 
-@mcp.tool()
-def gitea_assess_author_duplicate_work(
-    issue_number: int,
-    stage: str,
-    remote: str = "dadeschools",
-    host: str | None = None,
-    org: str | None = None,
-    repo: str | None = None,
-    allow_stale_takeover: bool = False,
-    exclude_branch_name: str | None = None,
-) -> dict:
-    """Read-only: assess duplicate-work risk before author mutations (#400).
-
-    Call at claim, lock, worktree, edit, commit, push, and create_pr stages.
-    """
-    read_block = _profile_operation_gate("gitea.read")
-    if read_block:
-        return {
-            "success": False,
-            "reasons": read_block,
-            "permission_report": _permission_block_report("gitea.read"),
-        }
-    h, o, r = _resolve(remote, host, org, repo)
-    auth = _auth(h)
-    ctx = _gather_author_duplicate_work_context(
-        issue_number,
-        h=h,
-        o=o,
-        r=r,
-        auth=auth,
-        exclude_branch_name=exclude_branch_name,
-    )
-    result = author_duplicate_work_gate.classify_and_assess(
-        ctx["issue"],
-        stage=stage,
-        comments=ctx["comments"],
-        open_prs=ctx["open_prs"],
-        branch_names=ctx["branch_names"],
-        allow_stale_takeover=allow_stale_takeover,
-    )
-    assessment = result.get("duplicate_work") or {}
-    return {
-        "success": True,
-        "performed": False,
-        "issue_number": issue_number,
-        "stage": stage,
-        "allowed": assessment.get("allowed"),
-        "block": assessment.get("block"),
-        "eligibility_class": assessment.get("eligibility_class"),
-        "outcome": assessment.get("outcome"),
-        "linked_open_pr": assessment.get("linked_open_pr"),
-        "matching_branches": assessment.get("matching_branches"),
-        "claim_status": assessment.get("claim_status"),
-        "reasons": assessment.get("reasons"),
-        "safe_next_action": assessment.get("safe_next_action"),
-        "claim": result.get("claim"),
-    }
-
-
 @mcp.tool()
 def gitea_reconcile_issue_claims(
     state: str = "open",

issue_lock_adoption.py

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

issue_lock_store.py

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

issue_work_duplicate_gate.py

@@ -0,0 +1,180 @@
+"""Early duplicate-work detection for author work-issue sessions (#400)."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import issue_claim_heartbeat as claim_hb
+
+PHASE_LOCK = "lock_issue"
+PHASE_COMMIT = "commit"
+PHASE_PUSH = "push"
+PHASE_CREATE_PR = "create_pr"
+
+OUTCOME_DUPLICATE_PR_PREVENTED = "duplicate_pr_prevented"
+OUTCOME_DUPLICATE_BRANCH_PREVENTED = "duplicate_branch_prevented"
+OUTCOME_DUPLICATE_COMMIT_PREVENTED = "duplicate_commit_prevented"
+OUTCOME_DUPLICATE_WORK_NOT_PREVENTED = "duplicate_work_not_prevented"
+
+_ACTIVE_CLAIM_STATUSES = frozenset({"active", "awaiting_review"})
+
+
+def _issue_pattern(issue_number: int) -> str:
+    return f"issue-{int(issue_number)}"
+
+
+def _linked_open_pr(issue_number: int, open_prs: list[dict]) -> dict | None:
+    return claim_hb._linked_open_pr(issue_number, open_prs)
+
+
+def _matching_branches(
+    issue_number: int,
+    branch_names: list[str],
+    *,
+    locked_branch: str | None = None,
+) -> list[str]:
+    pattern = _issue_pattern(issue_number)
+    matches = [
+        name for name in (branch_names or [])
+        if pattern in (name or "").lower()
+    ]
+    if locked_branch:
+        locked = locked_branch.strip()
+        matches = [name for name in matches if name != locked]
+    return matches
+
+
+def assess_work_issue_duplicate_gate(
+    issue_number: int,
+    *,
+    open_prs: list[dict] | None = None,
+    branch_names: list[str] | None = None,
+    claim_entry: dict | None = None,
+    locked_branch: str | None = None,
+    phase: str = PHASE_LOCK,
+) -> dict[str, Any]:
+    """Fail closed when duplicate work is already in flight for an issue."""
+    reasons: list[str] = []
+    outcome = OUTCOME_DUPLICATE_WORK_NOT_PREVENTED
+    prs = list(open_prs or [])
+    branches = list(branch_names or [])
+    pattern = _issue_pattern(issue_number)
+
+    linked = _linked_open_pr(issue_number, prs)
+    if linked:
+        reasons.append(
+            f"open PR #{linked.get('number')} already covers issue "
+            f"#{issue_number} (fail closed)"
+        )
+        outcome = OUTCOME_DUPLICATE_PR_PREVENTED
+
+    conflicting_branches = _matching_branches(
+        issue_number, branches, locked_branch=locked_branch
+    )
+    if conflicting_branches:
+        names = ", ".join(conflicting_branches[:5])
+        reasons.append(
+            f"remote branch(es) already match issue pattern '{pattern}': "
+            f"{names} (fail closed)"
+        )
+        if outcome == OUTCOME_DUPLICATE_WORK_NOT_PREVENTED:
+            outcome = OUTCOME_DUPLICATE_BRANCH_PREVENTED
+
+    entry = claim_entry or {}
+    if entry.get("linked_open_pr") and not linked:
+        reasons.append(
+            f"claim inventory reports open PR #{entry['linked_open_pr']} "
+            f"for issue #{issue_number} (fail closed)"
+        )
+        outcome = OUTCOME_DUPLICATE_PR_PREVENTED
+
+    status = (entry.get("status") or "").strip().lower()
+    if status in _ACTIVE_CLAIM_STATUSES and not linked:
+        heartbeat = entry.get("latest_heartbeat") or {}
+        claim_branch = (heartbeat.get("branch") or "").strip()
+        if locked_branch and claim_branch and claim_branch != locked_branch:
+            reasons.append(
+                f"active claim lease on branch '{claim_branch}' blocks "
+                f"work on '{locked_branch}' for issue #{issue_number} "
+                "(fail closed)"
+            )
+            if outcome == OUTCOME_DUPLICATE_WORK_NOT_PREVENTED:
+                outcome = OUTCOME_DUPLICATE_BRANCH_PREVENTED
+        elif not locked_branch and status == "active":
+            reasons.append(
+                f"issue #{issue_number} has an active claim lease "
+                "(fail closed)"
+            )
+            if outcome == OUTCOME_DUPLICATE_WORK_NOT_PREVENTED:
+                outcome = OUTCOME_DUPLICATE_BRANCH_PREVENTED
+
+    if phase in {PHASE_COMMIT, PHASE_PUSH} and reasons:
+        if outcome == OUTCOME_DUPLICATE_PR_PREVENTED:
+            outcome = OUTCOME_DUPLICATE_COMMIT_PREVENTED
+        elif outcome == OUTCOME_DUPLICATE_BRANCH_PREVENTED:
+            outcome = OUTCOME_DUPLICATE_COMMIT_PREVENTED
+
+    block = bool(reasons)
+    return {
+        "block": block,
+        "performed": not block,
+        "issue_number": issue_number,
+        "phase": phase,
+        "outcome": outcome,
+        "linked_open_pr": linked.get("number") if linked else entry.get("linked_open_pr"),
+        "conflicting_branches": conflicting_branches,
+        "claim_status": status or None,
+        "reasons": reasons,
+        "safe_next_action": (
+            "stop before mutating; preserve local work and produce a "
+            "reconciliation handoff if a concurrent PR appeared after push"
+            if block and phase == PHASE_CREATE_PR
+            else "stop before mutating; do not commit or push duplicate work"
+            if block
+            else "proceed"
+        ),
+    }
+
+
+def assess_work_issue_duplicate_report(report_text: str) -> dict[str, Any]:
+    """Require explicit duplicate-work outcome wording in work-issue reports."""
+    text = (report_text or "").lower()
+    markers = {
+        OUTCOME_DUPLICATE_PR_PREVENTED: (
+            "duplicate pr prevented",
+            "duplicate_pr_prevented",
+        ),
+        OUTCOME_DUPLICATE_BRANCH_PREVENTED: (
+            "duplicate branch prevented",
+            "duplicate_branch_prevented",
+        ),
+        OUTCOME_DUPLICATE_COMMIT_PREVENTED: (
+            "duplicate commit prevented",
+            "duplicate_commit_prevented",
+        ),
+        OUTCOME_DUPLICATE_WORK_NOT_PREVENTED: (
+            "duplicate work not prevented",
+            "duplicate_work_not_prevented",
+            "no duplicate work",
+        ),
+    }
+    matched = [
+        key for key, phrases in markers.items()
+        if any(phrase in text for phrase in phrases)
+    ]
+    if len(matched) != 1:
+        return {
+            "complete": False,
+            "downgraded": True,
+            "reasons": [
+                "work-issue report must state exactly one duplicate-work "
+                "outcome (duplicate PR/branch/commit prevented, or "
+                "duplicate work not prevented)"
+            ],
+        }
+    return {
+        "complete": True,
+        "downgraded": False,
+        "outcome": matched[0],
+        "reasons": [],
+    }

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]:

review_proofs.py

@@ -3622,33 +3622,21 @@ def assess_work_issue_mode_isolation(report_text: str) -> dict:
     }
 
 
-def assess_work_issue_duplicate_prevention_report(report_text, **kwargs):
-    """#400: work-issue reports must classify duplicate-work prevention."""
-    from author_duplicate_work_gate import (
-        assess_work_issue_duplicate_prevention_report as _assess,
-    )
-
-    return _assess(report_text, **kwargs)
-
-
 def assess_work_issue_final_report(report_text: str) -> dict:
     """#139: composite verifier for work-issue final reports."""
+    from issue_work_duplicate_gate import assess_work_issue_duplicate_report
+
     checks = {
         "workflow_source": assess_work_issue_workflow_source(report_text),
         "mode_isolation": assess_work_issue_mode_isolation(report_text),
-        "duplicate_prevention": assess_work_issue_duplicate_prevention_report(
-            report_text
-        ),
+        "duplicate_work_outcome": assess_work_issue_duplicate_report(report_text),
     }
 
     reasons = []
     downgraded = False
     for name, result in checks.items():
         verdict = result.get("verdict")
-        if result.get("block"):
-            downgraded = True
-            reasons.extend(result.get("reasons") or [])
-        elif verdict in ("missing", "incomplete"):
+        if verdict in ("missing", "incomplete"):
             downgraded = True
             reasons.extend(result.get("reasons") or [])
         elif result.get("downgraded") or not result.get("complete", True):
@@ -3656,9 +3644,6 @@ def assess_work_issue_final_report(report_text: str) -> dict:
             reasons.extend(
                 f"{name}: {r}" for r in (result.get("reasons") or [])
             )
-        elif result.get("proven") is False:
-            downgraded = True
-            reasons.extend(result.get("reasons") or [])
 
     grade = "A" if not downgraded else "downgraded"
     return {

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

@@ -277,24 +277,6 @@ Do not select an issue based only on memory from a previous session.
 
 Before claiming or working on an issue, check whether there is already an open PR, branch, or active claim for that issue.
 
-Run `gitea_assess_author_duplicate_work` at these stages and stop when `block` is true:
-
-* `claim` — before `gitea_mark_issue`
-* `lock` — before `gitea_lock_issue`
-* `worktree` / `edit` — before creating a worktree or editing files
-* `commit` — immediately before `git commit`
-* `push` — immediately before `git push`
-* `create_pr` — immediately before `gitea_create_pr` (also enforced server-side)
-
-`gitea_mark_issue`, `gitea_lock_issue`, and `gitea_create_pr` enforce the same gate server-side and fail closed.
-
-If a concurrent open PR appears after work begins:
-
-* before commit or push — stop and preserve local work without pushing
-* after push but before PR creation — produce a reconciliation handoff instead of opening a PR
-
-Final reports must name the duplicate-work outcome (`duplicate PR prevented`, `duplicate branch prevented`, `duplicate commit prevented`, `duplicate push prevented`, `duplicate work not prevented`, or `reconciliation handoff`).
-
 If an open PR already exists for the issue, do not implement duplicate work.
 
 Classify the issue as:
@@ -315,6 +297,36 @@ Report:
 
 Do not create another branch/PR for the same issue unless the project explicitly allows taking over or updating existing work and exact capability is proven.
 
+### 10A. Duplicate-work gate phases (#400)
+
+Before any file edits, prove duplicate-work clearance with
+`gitea_assess_work_issue_duplicate` or `gitea_lock_issue` (which runs the same
+gate). The gate checks live:
+
+* open PRs linked to the issue (head branch or Closes/Fixes reference),
+* remote branches matching `issue-<number>`,
+* active claim leases from structured heartbeats.
+
+Re-check immediately before:
+
+* `gitea_commit_files` (commit),
+* branch push,
+* `gitea_create_pr` (PR creation).
+
+If a concurrent open PR appears after work begins:
+
+* before commit/push → stop and preserve local work without pushing,
+* after commit but before push → stop without pushing,
+* after push but before PR creation → stop and produce a reconciliation
+  handoff instead of opening a PR.
+
+Final reports must state exactly one duplicate-work outcome:
+
+* `duplicate PR prevented`
+* `duplicate branch prevented`
+* `duplicate commit prevented`
+* `duplicate work not prevented`
+
 ## 11. Claim or lock the issue before implementation
 
 Claim/lock the issue before implementation if the project provides a claim/lock mechanism.

tests/test_agent_temp_artifacts.py

@@ -74,18 +74,23 @@ 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.issue_duplicate_context_fetcher",
+        return_value=([], [], {"status": "not_claimed"}),
+    )
     @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_author_duplicate_work_gate.py

@@ -1,102 +0,0 @@
-"""Tests for early author duplicate-work gate (#400)."""
-import sys
-import unittest
-from pathlib import Path
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-from author_duplicate_work_gate import (  # noqa: E402
-    ELIGIBILITY_OPEN_PR_EXISTS,
-    assess_author_duplicate_work,
-    assess_work_issue_duplicate_prevention_report,
-)
-
-
-def _open_pr(number: int = 397, issue: int = 395) -> dict:
-    return {
-        "number": number,
-        "head": {"ref": f"feat/issue-{issue}-example"},
-        "title": f"feat: example (Closes #{issue})",
-        "body": f"Closes #{issue}",
-    }
-
-
-class TestAuthorDuplicateWorkGate(unittest.TestCase):
-    def test_clear_when_no_duplicates(self):
-        result = assess_author_duplicate_work(
-            400,
-            stage="claim",
-            open_prs=[],
-            branch_names=["master", "feat/issue-399-other"],
-        )
-        self.assertTrue(result["allowed"])
-        self.assertFalse(result["block"])
-
-    def test_open_pr_blocks_claim(self):
-        result = assess_author_duplicate_work(
-            395,
-            stage="claim",
-            open_prs=[_open_pr()],
-            branch_names=[],
-        )
-        self.assertFalse(result["allowed"])
-        self.assertEqual(result["eligibility_class"], ELIGIBILITY_OPEN_PR_EXISTS)
-
-    def test_matching_branch_blocks_lock_not_create_pr(self):
-        branches = ["feat/issue-400-early-duplicate-work-gate"]
-        lock = assess_author_duplicate_work(
-            400,
-            stage="lock",
-            open_prs=[],
-            branch_names=branches,
-        )
-        self.assertFalse(lock["allowed"])
-
-        create_pr = assess_author_duplicate_work(
-            400,
-            stage="create_pr",
-            open_prs=[],
-            branch_names=branches,
-            matching_branches=[],
-        )
-        self.assertTrue(create_pr["allowed"])
-
-    def test_open_pr_blocks_create_pr_stage(self):
-        result = assess_author_duplicate_work(
-            395,
-            stage="create_pr",
-            open_prs=[_open_pr()],
-            branch_names=["feat/issue-395-proof-backed-review-handoff"],
-        )
-        self.assertFalse(result["allowed"])
-        self.assertEqual(result["outcome"], "duplicate_pr_prevented")
-
-    def test_push_stage_blocks_on_concurrent_pr(self):
-        result = assess_author_duplicate_work(
-            395,
-            stage="push",
-            open_prs=[_open_pr()],
-            branch_names=[],
-        )
-        self.assertFalse(result["allowed"])
-        self.assertEqual(result["outcome"], "duplicate_push_prevented")
-
-    def test_duplicate_prevention_report_requires_outcome(self):
-        bad = assess_work_issue_duplicate_prevention_report(
-            "Duplicate work detected for issue #395."
-        )
-        self.assertFalse(bad["proven"])
-
-        good = assess_work_issue_duplicate_prevention_report(
-            "Duplicate PR prevented; reconciliation handoff produced."
-        )
-        self.assertTrue(good["proven"])
-
-    def test_exported_from_review_proofs(self):
-        from review_proofs import assess_work_issue_duplicate_prevention_report as exported
-
-        self.assertTrue(callable(exported))
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_commit_files_capability.py

@@ -76,7 +76,14 @@ class CommitFilesCapabilityBase(unittest.TestCase):
         with open(self.config_path, "w", encoding="utf-8") as fh:
             fh.write(json.dumps(CONFIG))
 
+        self._dup_fetcher_patcher = patch(
+            "mcp_server.issue_duplicate_context_fetcher",
+            return_value=([], [], {"status": "not_claimed"}),
+        )
+        self._dup_fetcher_patcher.start()
+
     def tearDown(self):
+        self._dup_fetcher_patcher.stop()
         self._remotes.stop()
         mcp_server._IDENTITY_CACHE.clear()
         mcp_server._preflight_whoami_called, mcp_server._preflight_capability_called = (

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
@@ -84,7 +90,14 @@ class TestCommitPayloads(unittest.TestCase):
         mcp_server._preflight_whoami_called = True
         mcp_server._preflight_capability_called = True
 
+        self._dup_fetcher_patcher = patch(
+            "mcp_server.issue_duplicate_context_fetcher",
+            return_value=([], [], {"status": "not_claimed"}),
+        )
+        self._dup_fetcher_patcher.start()
+
     def tearDown(self):
+        self._dup_fetcher_patcher.stop()
         self._remotes.stop()
         mcp_server._IDENTITY_CACHE.clear()
 
@@ -93,8 +106,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 +115,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_lock_adoption.py

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

tests/test_issue_lock_store.py

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

tests/test_issue_work_duplicate_gate.py

@@ -0,0 +1,251 @@
+"""Tests for early duplicate-work detection (#400)."""
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import issue_work_duplicate_gate as dup_gate
+import mcp_server
+from issue_work_duplicate_gate import (
+    OUTCOME_DUPLICATE_BRANCH_PREVENTED,
+    OUTCOME_DUPLICATE_COMMIT_PREVENTED,
+    OUTCOME_DUPLICATE_PR_PREVENTED,
+    OUTCOME_DUPLICATE_WORK_NOT_PREVENTED,
+    PHASE_COMMIT,
+    PHASE_CREATE_PR,
+    PHASE_LOCK,
+    assess_work_issue_duplicate_gate,
+    assess_work_issue_duplicate_report,
+)
+
+
+class TestDuplicateGateAssessment(unittest.TestCase):
+    def test_clear_issue_passes(self):
+        result = assess_work_issue_duplicate_gate(
+            400,
+            open_prs=[],
+            branch_names=["feat/other-issue-99"],
+            claim_entry={"status": "not_claimed"},
+            locked_branch="feat/issue-400-duplicate-work-preflight",
+            phase=PHASE_LOCK,
+        )
+        self.assertFalse(result["block"])
+        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
+
+    def test_open_pr_blocks(self):
+        prs = [{
+            "number": 397,
+            "title": "feat: handoff",
+            "body": "Closes #395",
+            "head": {"ref": "feat/issue-395-proof-backed-review-handoff"},
+        }]
+        result = assess_work_issue_duplicate_gate(
+            395,
+            open_prs=prs,
+            branch_names=[],
+            phase=PHASE_LOCK,
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_PR_PREVENTED)
+
+    def test_conflicting_remote_branch_blocks(self):
+        result = assess_work_issue_duplicate_gate(
+            395,
+            open_prs=[],
+            branch_names=["feat/issue-395-proof-backed-handoff-claims"],
+            locked_branch="feat/issue-395-new-attempt",
+            phase=PHASE_LOCK,
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_BRANCH_PREVENTED)
+
+    def test_active_claim_on_other_branch_blocks(self):
+        result = assess_work_issue_duplicate_gate(
+            398,
+            open_prs=[],
+            branch_names=[],
+            claim_entry={
+                "status": "active",
+                "latest_heartbeat": {
+                    "branch": "feat/issue-398-validation-cwd-proof",
+                },
+            },
+            locked_branch="feat/issue-398-other-branch",
+            phase=PHASE_LOCK,
+        )
+        self.assertTrue(result["block"])
+
+    def test_commit_phase_maps_to_commit_outcome(self):
+        prs = [{
+            "number": 411,
+            "title": "x",
+            "body": "Closes #398",
+            "head": {"ref": "feat/issue-398-validation-cwd-proof"},
+        }]
+        result = assess_work_issue_duplicate_gate(
+            398,
+            open_prs=prs,
+            branch_names=[],
+            locked_branch="feat/issue-398-alt",
+            phase=PHASE_COMMIT,
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_COMMIT_PREVENTED)
+
+    def test_stale_claim_does_not_block_by_status_alone(self):
+        result = assess_work_issue_duplicate_gate(
+            400,
+            open_prs=[],
+            branch_names=[],
+            claim_entry={"status": "reclaimable", "reasons": ["stale"]},
+            locked_branch="feat/issue-400-duplicate-work-preflight",
+            phase=PHASE_LOCK,
+        )
+        self.assertFalse(result["block"])
+
+
+class TestDuplicateReportOutcome(unittest.TestCase):
+    def test_requires_exactly_one_outcome(self):
+        bad = assess_work_issue_duplicate_report("work finished")
+        self.assertFalse(bad["complete"])
+
+        good = assess_work_issue_duplicate_report(
+            "Duplicate work not prevented for issue #400."
+        )
+        self.assertTrue(good["complete"])
+        self.assertEqual(good["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
+
+
+class TestInjectableDuplicateFetcher(unittest.TestCase):
+    @patch("mcp_server.get_auth_header", return_value="token x")
+    def test_lock_issue_uses_injected_fetcher(self, _auth):
+        seen = {}
+
+        def fetcher(h, o, r, auth, issue_number):
+            seen["issue_number"] = issue_number
+            return [], [], {"status": "not_claimed"}
+
+        with patch(
+            "mcp_server.issue_duplicate_context_fetcher",
+            side_effect=fetcher,
+        ), patch(
+            "mcp_server.issue_lock_worktree.read_worktree_git_state",
+            return_value={
+                "current_branch": "master",
+                "porcelain_status": "",
+                "base_equivalent": True,
+            },
+        ), patch.dict(os.environ, {
+            "GITEA_ALLOWED_OPERATIONS": "gitea.issue.comment",
+        }, clear=True):
+            with patch.object(mcp_server, "ISSUE_LOCK_FILE", tempfile.mktemp()):
+                mcp_server.gitea_lock_issue(
+                    issue_number=400,
+                    branch_name="feat/issue-400-duplicate-work-preflight",
+                    remote="prgs",
+                )
+        self.assertEqual(seen["issue_number"], 400)
+
+
+class TestMcpDuplicateRecheck(unittest.TestCase):
+    def setUp(self):
+        self._dir = tempfile.TemporaryDirectory()
+        self.lock_path = os.path.join(self._dir.name, "gitea_issue_lock.json")
+        self._lock_patch = patch.object(
+            mcp_server, "ISSUE_LOCK_FILE", self.lock_path
+        )
+        self._lock_patch.start()
+        self._remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {"host": "gitea.example.com", "org": "Example-Org",
+                     "repo": "Example-Repo"},
+        })
+        self._remotes.start()
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def tearDown(self):
+        patch.stopall()
+        self._dir.cleanup()
+
+    def _write_lock(self, issue_number=400, branch="feat/issue-400-x"):
+        with open(self.lock_path, "w", encoding="utf-8") as fh:
+            json.dump({
+                "issue_number": issue_number,
+                "branch_name": branch,
+                "remote": "prgs",
+            }, fh)
+
+    @patch("mcp_server._assess_issue_duplicate_gate")
+    @patch("mcp_server.get_profile", return_value={
+        "profile_name": "test-author",
+        "allowed_operations": ["gitea.read", "gitea.repo.commit"],
+        "forbidden_operations": [],
+        "audit_label": "test-author",
+    })
+    @patch("mcp_server.get_auth_header", return_value="token x")
+    def test_commit_files_blocked_on_recheck(self, _auth, _profile, mock_gate):
+        self._write_lock()
+        mock_gate.return_value = {
+            "block": True,
+            "reasons": ["open PR #412 already covers issue #400"],
+            "outcome": OUTCOME_DUPLICATE_COMMIT_PREVENTED,
+            "safe_next_action": "stop",
+        }
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check("capability", resolved_role="author")
+        with patch(
+            "mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+            return_value=(True, []),
+        ):
+            result = mcp_server.gitea_commit_files(
+                files=[{
+                    "operation": "create",
+                    "path": "a.txt",
+                    "content_plain": "hi",
+                }],
+                message="test",
+                remote="prgs",
+            )
+        self.assertFalse(result["success"])
+        self.assertIn("duplicate_gate", result)
+
+    @patch("mcp_server._assess_issue_duplicate_gate")
+    @patch("mcp_server.get_profile", return_value={
+        "profile_name": "test-author",
+        "allowed_operations": ["gitea.read", "gitea.pr.create"],
+        "forbidden_operations": [],
+        "audit_label": "test-author",
+    })
+    @patch("mcp_server.get_auth_header", return_value="token x")
+    def test_create_pr_returns_handoff_on_duplicate(self, _auth, _profile, mock_gate):
+        self._write_lock()
+        mock_gate.return_value = {
+            "block": True,
+            "reasons": ["open PR #412 already covers issue #400"],
+            "outcome": OUTCOME_DUPLICATE_PR_PREVENTED,
+            "safe_next_action": "reconciliation handoff",
+        }
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check("capability", resolved_role="author")
+        with patch(
+            "mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+            return_value=(True, []),
+        ):
+            result = mcp_server.gitea_create_pr(
+                title="feat: x (Closes #400)",
+                head="feat/issue-400-x",
+                base="master",
+                body="Closes #400",
+                remote="prgs",
+            )
+        self.assertFalse(result["success"])
+        self.assertIsNone(result.get("number"))
+        self.assertIn("duplicate_gate", result)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_llm_workflow_split.py

@@ -95,12 +95,6 @@ def test_create_issue_workflow_contract():
     assert "## 9. Duplicate search before mutation" in text
 
 
-def test_author_duplicate_work_gate_exported():
-    from review_proofs import assess_work_issue_duplicate_prevention_report
-
-    assert callable(assess_work_issue_duplicate_prevention_report)
-
-
 def test_work_issue_workflow_contract():
     text = (SKILL_DIR / "workflows" / "work-issue.md").read_text(encoding="utf-8")
     assert "canonical: true" in text

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,32 @@ 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 _clear_duplicate_context_fetcher(*_args, **_kwargs):
+    """Default injectable duplicate-work context for lock/create_pr tests."""
+    return [], [], {"status": "not_claimed"}
+
+
+def _bind_test_lock(**overrides) -> str:
+    remote = overrides.get("remote", "dadeschools")
+    record = _sample_issue_lock(**overrides)
+    if remote in mcp_server.REMOTES:
+        profile = mcp_server.REMOTES[remote]
+        record.setdefault("org", profile["org"])
+        record.setdefault("repo", profile["repo"])
+    record["remote"] = remote
+    return issue_lock_store.bind_session_lock(record)
+
+
 # ---------------------------------------------------------------------------
 # Create Issue
 # ---------------------------------------------------------------------------
@@ -166,55 +186,78 @@ class TestCreateIssue(unittest.TestCase):
 # ---------------------------------------------------------------------------
 class TestCreatePR(unittest.TestCase):
 
+    @patch(
+        "mcp_server.issue_duplicate_context_fetcher",
+        return_value=([], [], {"status": "not_claimed"}),
+    )
     @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
            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, _dup_fetcher):
+        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")
         self.assertIn("Closes #123", payload["title"])
 
+    @patch(
+        "mcp_server.issue_duplicate_context_fetcher",
+        return_value=([], [], {"status": "not_claimed"}),
+    )
     @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
            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, _dup_fetcher):
+        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,22 +3085,37 @@ class TestIssueLocking(unittest.TestCase):
     """Test issue locking and PR gating constraints."""
 
     def setUp(self):
-        self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
+        self._lock_dir = tempfile.TemporaryDirectory()
+        env = {
+            **ISSUE_WRITE_ENV,
+            "GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
+        }
+        self._env_patcher = patch.dict(os.environ, env, clear=True)
         self._env_patcher.start()
+        self._dup_fetcher_patcher = patch(
+            "mcp_server.issue_duplicate_context_fetcher",
+            return_value=([], [], {"status": "not_claimed"}),
+        )
+        self.mock_dup_fetcher = self._dup_fetcher_patcher.start()
 
     def tearDown(self):
+        self._dup_fetcher_patcher.stop()
         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",
         return_value=_clean_master_git_state_for_lock(),
     )
-    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_lock_issue_success(self, _auth, mock_api, _git_state):
-        mock_api.return_value = [] # no open PRs
+    def test_lock_issue_success(self, _auth, _api, _git_state):
         res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertTrue(res["success"])
         self.assertEqual(res["work_lease"]["operation_type"], "author_issue_work")
@@ -3067,9 +3125,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)
 
@@ -3082,35 +3139,48 @@ class TestIssueLocking(unittest.TestCase):
         "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_reused_by_open_pr_branch(self, _auth, mock_api, _git_state):
-        mock_api.return_value = [{
+    def test_lock_issue_reused_by_open_pr_branch(self, _auth, _git_state):
+        self.mock_dup_fetcher.return_value = ([{
             "number": 200,
             "head": {"ref": "feat/issue-196-boundary"},
             "title": "Some PR",
-            "body": "No closes ref"
-        }]
+            "body": "No closes ref",
+        }], [], {"status": "not_claimed"})
         with self.assertRaises(ValueError) as ctx:
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
-        self.assertIn("already tied to an open PR", str(ctx.exception))
+        self.assertIn("open PR #200 already covers issue", str(ctx.exception))
 
     @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_reused_by_open_pr_closes_ref(self, _auth, mock_api, _git_state):
-        mock_api.return_value = [{
+    def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, _git_state):
+        self.mock_dup_fetcher.return_value = ([{
             "number": 200,
             "head": {"ref": "feat/other-branch"},
             "title": "Some PR",
-            "body": "fixes #196"
-        }]
+            "body": "fixes #196",
+        }], [], {"status": "not_claimed"})
         with self.assertRaises(ValueError) as ctx:
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
-        self.assertIn("already tied to an open PR", str(ctx.exception))
+        self.assertIn("open PR #200 already covers issue", str(ctx.exception))
+
+    @patch(
+        "mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value=_clean_master_git_state_for_lock(),
+    )
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_reused_by_remote_branch(self, _auth, _git_state):
+        self.mock_dup_fetcher.return_value = (
+            [],
+            ["feat/issue-196-existing-work"],
+            {"status": "not_claimed"},
+        )
+        with self.assertRaises(ValueError) as ctx:
+            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("remote branch(es) already match issue pattern", str(ctx.exception))
 
     @patch(
         "mcp_server.issue_lock_worktree.read_worktree_git_state",
@@ -3118,41 +3188,75 @@ class TestIssueLocking(unittest.TestCase):
     )
     @patch("mcp_server.api_get_all")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_lock_issue_reused_by_remote_branch(self, _auth, mock_api, _git_state):
-        mock_api.side_effect = [
-            [],
-            [{"name": "feat/issue-196-existing-work"}],
-        ]
-        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))
+    def test_lock_issue_adopts_exact_own_branch(self, _auth, mock_api, _git_state):
+        branch = "feat/issue-196-mutations"
+        self.mock_dup_fetcher.return_value = ([], [branch], {"status": "not_claimed"})
+        mock_api.return_value = [{"name": branch, "commit": {"id": "abc123"}}]
+        res = gitea_lock_issue(issue_number=196, branch_name=branch, remote="prgs")
+        self.assertTrue(res["success"])
+        self.assertIn("adoption", res)
+        self.assertEqual(res["adoption"]["branch_head_commit"], "abc123")
 
-    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", 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 +3323,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 +3332,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 +3397,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 +3420,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_review_proofs.py

@@ -2346,6 +2346,7 @@ class TestWorkIssueFinalReport(unittest.TestCase):
             "- Safe next action: open PR",
             "- Next: open PR",
             "- Safety statement: no review/merge",
+            "- Duplicate work outcome: duplicate work not prevented",
         ])
 
     def test_complete_work_issue_report_earns_a(self):

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>"
+    )