test: fix reviewer lease gate regressions in MCP integration tests (#407)

Align merge/review/audit mocks with the per-PR reviewer lease gate introduced
in #407: install owned session leases after init_review_decision_lock, stub
_fetch_pr_comments and _authenticated_username to preserve mock ordering, and
update head-SHA mismatch expectations for lease-first fail-closed paths.

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

docs/webui-local-dev.md

@@ -1,89 +0,0 @@
-# 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,48 +440,10 @@ 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)"
         )
 
-    workspace = author_mutation_worktree.resolve_mutation_workspace(
+    if worktree_path:
-        worktree_path,
+        dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(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(workspace, dirty_files)
+            details = _preflight_workspace_details(worktree_path, dirty_files)
             raise RuntimeError(
                 "Pre-flight order violation: Active task workspace has tracked "
                 "file edits before mutation (fail closed). "
@@ -541,6 +503,7 @@ import issue_lock_worktree  # noqa: E402
 import already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
+import reviewer_pr_lease  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
 import reconciler_profile  # noqa: E402
 import reconciliation_workflow  # noqa: E402
@@ -1010,7 +973,6 @@ 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.
 
@@ -1023,7 +985,6 @@ 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).
@@ -1050,7 +1011,7 @@ def gitea_create_issue(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(remote)
     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(
@@ -1868,6 +1829,7 @@ def init_review_decision_lock(remote: str | None, task: str | None):
         (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
         or profile_name
     )
+    reviewer_pr_lease.clear_session_lease()
     _save_review_decision_lock({
         "task": task,
         "remote": remote,
@@ -2310,6 +2272,20 @@ def _evaluate_pr_review_submission(
             result["permission_report"] = elig["permission_report"]
         return result
 
+    if live:
+        reasons.extend(_reviewer_pr_lease_gate(
+            pr_number=pr_number,
+            remote=remote,
+            host=host,
+            org=org,
+            repo=repo,
+            mutation=action,
+            live_head_sha=result.get("head_sha"),
+            pinned_head_sha=expected_head_sha,
+        ))
+        if reasons:
+            return result
+
     auth_user = result["authenticated_user"]
     pr_author = result["pr_author"]
     if action == "approve" and auth_user and pr_author and auth_user == pr_author:
@@ -3095,6 +3071,19 @@ def gitea_merge_pr(
             result["permission_report"] = elig["permission_report"]
         return result
 
+    reasons.extend(_reviewer_pr_lease_gate(
+        pr_number=pr_number,
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+        mutation="merge",
+        live_head_sha=result.get("head_sha"),
+        pinned_head_sha=expected_head_sha,
+    ))
+    if reasons:
+        return result
+
     # Gate 4 — head SHA must match if the caller pinned a reviewed SHA.
     actual_sha = result["head_sha"]
     if expected_head_sha and actual_sha and expected_head_sha != actual_sha:
@@ -4394,6 +4383,261 @@ def _namespace_mutation_block(mutation_task: str, **extra_fields) -> dict | None
     return blocked
 
 
+def _fetch_pr_comments(
+    pr_number: int,
+    *,
+    remote: str,
+    host: str | None,
+    org: str | None,
+    repo: str | None,
+) -> list[dict]:
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
+    return api_request("GET", api, auth) or []
+
+
+def _reviewer_pr_lease_gate(
+    *,
+    pr_number: int,
+    remote: str,
+    host: str | None,
+    org: str | None,
+    repo: str | None,
+    mutation: str,
+    live_head_sha: str | None,
+    pinned_head_sha: str | None,
+) -> list[str]:
+    """Return block reasons when the session lacks an owned PR reviewer lease."""
+    session = reviewer_pr_lease.get_session_lease()
+    session_id = (session or {}).get("session_id")
+    identity = _authenticated_username(remote) or ""
+    try:
+        comments = _fetch_pr_comments(
+            pr_number, remote=remote, host=host, org=org, repo=repo)
+    except Exception as exc:
+        return [f"cannot fetch PR comments for lease gate: {_redact(str(exc))}"]
+    assessment = reviewer_pr_lease.assess_mutation_lease_gate(
+        pr_number=pr_number,
+        comments=comments,
+        reviewer_identity=identity,
+        session_id=session_id,
+        mutation=mutation,
+        live_head_sha=live_head_sha,
+        pinned_head_sha=pinned_head_sha,
+    )
+    return list(assessment.get("reasons") or []) if assessment.get("block") else []
+
+
+@mcp.tool()
+def gitea_acquire_reviewer_pr_lease(
+    pr_number: int,
+    worktree: str,
+    candidate_head: str | None = None,
+    target_branch: str = "master",
+    target_branch_sha: str | None = None,
+    issue_number: int | None = None,
+    session_id: str | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Acquire a per-PR reviewer lease before review/merge mutations (#407)."""
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "acquired": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+    comment_block = _profile_operation_gate("gitea.pr.comment")
+    if comment_block:
+        return {
+            "success": False,
+            "acquired": False,
+            "reasons": comment_block,
+            "permission_report": _permission_block_report("gitea.pr.comment"),
+        }
+
+    verify_preflight_purity(remote)
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    profile = get_profile()
+    identity = _authenticated_username(remote) or profile.get("username") or ""
+    sid = (session_id or "").strip() or reviewer_pr_lease.new_session_id()
+    repo_label = f"{o}/{r}"
+
+    comments = _fetch_pr_comments(
+        pr_number, remote=remote, host=host, org=org, repo=repo)
+    assessment = reviewer_pr_lease.assess_acquire_lease(
+        comments,
+        pr_number=pr_number,
+        reviewer_identity=identity,
+        profile=profile.get("profile_name") or "unknown",
+        session_id=sid,
+        repo=repo_label,
+        issue_number=issue_number,
+        worktree=worktree,
+        candidate_head=candidate_head,
+        target_branch=target_branch,
+        target_branch_sha=target_branch_sha,
+    )
+    if not assessment.get("acquire_allowed"):
+        return {
+            "success": False,
+            "acquired": False,
+            "reasons": assessment.get("reasons") or [],
+            "existing_lease": assessment.get("existing_lease"),
+        }
+
+    body = assessment["lease_body"]
+    comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
+    with _audited(
+        "comment_pr",
+        host=h,
+        remote=remote,
+        org=o,
+        repo=r,
+        pr_number=pr_number,
+        request_metadata={"source": "acquire_reviewer_pr_lease"},
+    ):
+        posted = api_request("POST", comment_url, auth, {"body": body})
+
+    session_lease = reviewer_pr_lease.record_session_lease({
+        "pr_number": pr_number,
+        "issue_number": issue_number,
+        "session_id": sid,
+        "reviewer_identity": identity,
+        "profile": profile.get("profile_name"),
+        "worktree": worktree,
+        "phase": "claimed",
+        "candidate_head": candidate_head,
+        "target_branch": target_branch,
+        "target_branch_sha": target_branch_sha,
+        "repo": repo_label,
+        "comment_id": posted.get("id"),
+    })
+    return {
+        "success": True,
+        "acquired": True,
+        "pr_number": pr_number,
+        "session_id": sid,
+        "comment_id": posted.get("id"),
+        "session_lease": session_lease,
+        "reasons": [],
+    }
+
+
+@mcp.tool()
+def gitea_heartbeat_reviewer_pr_lease(
+    pr_number: int,
+    phase: str,
+    worktree: str | None = None,
+    candidate_head: str | None = None,
+    target_branch_sha: str | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Post a reviewer lease heartbeat / phase update on the PR thread (#407)."""
+    comment_block = _profile_operation_gate("gitea.pr.comment")
+    if comment_block:
+        return {
+            "success": False,
+            "posted": False,
+            "reasons": comment_block,
+            "permission_report": _permission_block_report("gitea.pr.comment"),
+        }
+    session = reviewer_pr_lease.get_session_lease()
+    if not session or session.get("pr_number") != pr_number:
+        return {
+            "success": False,
+            "posted": False,
+            "reasons": [
+                f"no in-session lease for PR #{pr_number}; acquire first "
+                "(fail closed)"
+            ],
+        }
+
+    verify_preflight_purity(remote)
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    body = reviewer_pr_lease.format_lease_body(
+        repo=f"{o}/{r}",
+        pr_number=pr_number,
+        issue_number=session.get("issue_number"),
+        reviewer_identity=session.get("reviewer_identity") or "",
+        profile=session.get("profile") or "unknown",
+        session_id=session.get("session_id") or reviewer_pr_lease.new_session_id(),
+        worktree=worktree or session.get("worktree") or "",
+        phase=phase,
+        candidate_head=candidate_head or session.get("candidate_head"),
+        target_branch=session.get("target_branch") or "master",
+        target_branch_sha=target_branch_sha or session.get("target_branch_sha"),
+    )
+    comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
+    with _audited(
+        "comment_pr",
+        host=h,
+        remote=remote,
+        org=o,
+        repo=r,
+        pr_number=pr_number,
+        request_metadata={"source": "heartbeat_reviewer_pr_lease", "phase": phase},
+    ):
+        posted = api_request("POST", comment_url, auth, {"body": body})
+
+    updated = reviewer_pr_lease.record_session_lease({
+        **session,
+        "phase": phase,
+        "worktree": worktree or session.get("worktree"),
+        "candidate_head": candidate_head or session.get("candidate_head"),
+        "target_branch_sha": target_branch_sha or session.get("target_branch_sha"),
+        "last_comment_id": posted.get("id"),
+    })
+    return {
+        "success": True,
+        "posted": True,
+        "pr_number": pr_number,
+        "phase": phase,
+        "comment_id": posted.get("id"),
+        "session_lease": updated,
+        "reasons": [],
+    }
+
+
+@mcp.tool()
+def gitea_assess_reviewer_pr_lease(
+    pr_number: int,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Read-only: assess active reviewer lease state for a PR (#407)."""
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+    comments = _fetch_pr_comments(
+        pr_number, remote=remote, host=host, org=org, repo=repo)
+    active = reviewer_pr_lease.find_active_reviewer_lease(
+        comments, pr_number=pr_number)
+    return {
+        "success": True,
+        "pr_number": pr_number,
+        "active_lease": active,
+        "session_lease": reviewer_pr_lease.get_session_lease(),
+        "reasons": [],
+    }
+
+
 @mcp.tool()
 def gitea_list_issue_comments(
     issue_number: int,

reviewer_pr_lease.py

@@ -0,0 +1,382 @@
+"""Per-PR reviewer leases for safe parallel review sessions (#407)."""
+
+from __future__ import annotations
+
+import os
+import re
+import uuid
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+MARKER = "<!-- mcp-review-lease:v1 -->"
+
+_FIELD_RE = re.compile(
+    r"^\s*([a-z_]+)\s*:\s*(.+?)\s*$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
+
+_TERMINAL_PHASES = frozenset({"done", "released", "blocked"})
+_ACTIVE_PHASES = frozenset({
+    "claimed",
+    "validating",
+    "approved",
+    "request-changes",
+    "merging",
+})
+
+DEFAULT_LEASE_TTL_MINUTES = 120
+STALE_WARNING_MINUTES = 30
+RECLAIMABLE_MINUTES = 60
+
+_SESSION_LEASE: dict[str, Any] | None = None
+
+
+def _parse_timestamp(value: str | None) -> datetime | None:
+    if not value:
+        return None
+    text = value.strip()
+    if text.endswith("Z"):
+        text = text[:-1] + "+00:00"
+    try:
+        parsed = datetime.fromisoformat(text)
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        return parsed.replace(tzinfo=timezone.utc)
+    return parsed.astimezone(timezone.utc)
+
+
+def _normalize_sha(value: str | None) -> str | None:
+    text = (value or "").strip().lower()
+    return text if text and _FULL_SHA.match(text) else None
+
+
+def _parse_pr_ref(value: str | None) -> int | None:
+    digits = re.sub(r"[^\d]", "", value or "")
+    return int(digits) if digits.isdigit() else None
+
+
+def new_session_id() -> str:
+    return f"{os.getpid()}-{uuid.uuid4().hex[:12]}"
+
+
+def format_lease_body(
+    *,
+    repo: str,
+    pr_number: int,
+    issue_number: int | None,
+    reviewer_identity: str,
+    profile: str,
+    session_id: str,
+    worktree: str,
+    phase: str,
+    candidate_head: str | None,
+    target_branch: str,
+    target_branch_sha: str | None,
+    last_activity: datetime | None = None,
+    expires_at: datetime | None = None,
+    blocker: str = "none",
+) -> str:
+    now = last_activity or datetime.now(timezone.utc)
+    expires = expires_at or (now + timedelta(minutes=DEFAULT_LEASE_TTL_MINUTES))
+    last_text = now.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace(
+        "+00:00", "Z"
+    )
+    expires_text = expires.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace(
+        "+00:00", "Z"
+    )
+    issue_text = f"#{issue_number}" if issue_number else "none"
+    lines = [
+        MARKER,
+        f"repo: {repo}",
+        f"pr: #{pr_number}",
+        f"issue: {issue_text}",
+        f"reviewer_identity: {reviewer_identity}",
+        f"profile: {profile}",
+        f"session_id: {session_id}",
+        f"worktree: {worktree}",
+        f"phase: {phase}",
+        f"candidate_head: {candidate_head or 'none'}",
+        f"target_branch: {target_branch}",
+        f"target_branch_sha: {target_branch_sha or 'none'}",
+        f"last_activity: {last_text}",
+        f"expires_at: {expires_text}",
+        f"blocker: {blocker}",
+    ]
+    return "\n".join(lines)
+
+
+def parse_lease_comment(body: str) -> dict[str, Any] | None:
+    text = body or ""
+    if MARKER not in text:
+        return None
+    fields: dict[str, str] = {}
+    for match in _FIELD_RE.finditer(text):
+        fields[match.group(1).strip().lower()] = match.group(2).strip()
+    if not fields:
+        return None
+    return {
+        "repo": fields.get("repo"),
+        "pr_number": _parse_pr_ref(fields.get("pr")),
+        "issue_number": _parse_pr_ref(fields.get("issue")),
+        "reviewer_identity": fields.get("reviewer_identity"),
+        "profile": fields.get("profile"),
+        "session_id": fields.get("session_id"),
+        "worktree": fields.get("worktree"),
+        "phase": (fields.get("phase") or "").strip().lower() or None,
+        "candidate_head": _normalize_sha(fields.get("candidate_head")),
+        "target_branch": fields.get("target_branch"),
+        "target_branch_sha": _normalize_sha(fields.get("target_branch_sha")),
+        "last_activity": fields.get("last_activity"),
+        "expires_at": fields.get("expires_at"),
+        "blocker": fields.get("blocker"),
+        "raw_fields": fields,
+    }
+
+
+def _lease_entries(comments: list[dict], *, pr_number: int) -> list[dict]:
+    entries: list[dict] = []
+    for comment in comments or []:
+        parsed = parse_lease_comment(comment.get("body") or "")
+        if not parsed:
+            continue
+        if parsed.get("pr_number") not in (None, pr_number):
+            continue
+        entries.append({
+            **parsed,
+            "comment_id": comment.get("id"),
+            "author": (comment.get("user") or {}).get("login") or comment.get("author"),
+            "created_at": comment.get("created_at"),
+            "updated_at": comment.get("updated_at"),
+        })
+    return entries
+
+
+def _lease_expired(lease: dict, *, now: datetime) -> bool:
+    expires_at = _parse_timestamp(lease.get("expires_at"))
+    return bool(expires_at and expires_at <= now)
+
+
+def _minutes_since_activity(lease: dict, *, now: datetime) -> float | None:
+    last = _parse_timestamp(lease.get("last_activity"))
+    if not last:
+        return None
+    return (now - last).total_seconds() / 60.0
+
+
+def classify_lease_freshness(lease: dict, *, now: datetime | None = None) -> str:
+    """Return active, stale_warning, reclaimable, expired, or terminal."""
+    now = now or datetime.now(timezone.utc)
+    phase = (lease.get("phase") or "").strip().lower()
+    if phase in _TERMINAL_PHASES:
+        return "terminal"
+    if _lease_expired(lease, now=now):
+        return "expired"
+    minutes = _minutes_since_activity(lease, now=now)
+    if minutes is None:
+        return "active"
+    if minutes >= RECLAIMABLE_MINUTES:
+        return "reclaimable"
+    if minutes >= STALE_WARNING_MINUTES:
+        return "stale_warning"
+    return "active"
+
+
+def find_active_reviewer_lease(
+    comments: list[dict],
+    *,
+    pr_number: int,
+    now: datetime | None = None,
+) -> dict[str, Any] | None:
+    """Newest non-terminal, unexpired lease for *pr_number*."""
+    now = now or datetime.now(timezone.utc)
+    for lease in reversed(_lease_entries(comments, pr_number=pr_number)):
+        phase = (lease.get("phase") or "").strip().lower()
+        if phase in _TERMINAL_PHASES:
+            continue
+        if _lease_expired(lease, now=now):
+            continue
+        if phase in _ACTIVE_PHASES or phase:
+            lease = dict(lease)
+            lease["freshness"] = classify_lease_freshness(lease, now=now)
+            return lease
+    return None
+
+
+def assess_acquire_lease(
+    comments: list[dict],
+    *,
+    pr_number: int,
+    reviewer_identity: str,
+    profile: str,
+    session_id: str,
+    repo: str,
+    issue_number: int | None,
+    worktree: str,
+    candidate_head: str | None,
+    target_branch: str,
+    target_branch_sha: str | None,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Fail closed when another session holds an active lease."""
+    now = now or datetime.now(timezone.utc)
+    reasons: list[str] = []
+    existing = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
+    if existing:
+        owner_session = (existing.get("session_id") or "").strip()
+        freshness = existing.get("freshness") or classify_lease_freshness(existing, now=now)
+        if owner_session and owner_session != session_id and freshness in {
+            "active", "stale_warning"
+        }:
+            reasons.append(
+                f"PR #{pr_number} already has active reviewer lease "
+                f"(session_id={owner_session}, phase={existing.get('phase')})"
+            )
+        elif owner_session and owner_session != session_id and freshness == "reclaimable":
+            reasons.append(
+                f"PR #{pr_number} lease is reclaimable but still held by "
+                f"session_id={owner_session}; explicit reclaim not implemented "
+                "(fail closed)"
+            )
+
+    if not (reviewer_identity or "").strip():
+        reasons.append("reviewer identity required for lease acquisition")
+    if not (session_id or "").strip():
+        reasons.append("session_id required for lease acquisition")
+    if not (worktree or "").strip():
+        reasons.append("worktree path required for lease acquisition")
+
+    allowed = not reasons
+    body = None
+    if allowed:
+        body = format_lease_body(
+            repo=repo,
+            pr_number=pr_number,
+            issue_number=issue_number,
+            reviewer_identity=reviewer_identity,
+            profile=profile,
+            session_id=session_id,
+            worktree=worktree,
+            phase="claimed",
+            candidate_head=candidate_head,
+            target_branch=target_branch,
+            target_branch_sha=target_branch_sha,
+            last_activity=now,
+        )
+    return {
+        "acquire_allowed": allowed,
+        "reasons": reasons,
+        "existing_lease": existing,
+        "lease_body": body,
+        "session_id": session_id,
+    }
+
+
+def record_session_lease(lease: dict[str, Any]) -> dict[str, Any]:
+    global _SESSION_LEASE
+    _SESSION_LEASE = dict(lease)
+    return dict(_SESSION_LEASE)
+
+
+def clear_session_lease() -> None:
+    global _SESSION_LEASE
+    _SESSION_LEASE = None
+
+
+def get_session_lease() -> dict[str, Any] | None:
+    return dict(_SESSION_LEASE) if _SESSION_LEASE else None
+
+
+def assess_mutation_lease_gate(
+    *,
+    pr_number: int,
+    comments: list[dict],
+    reviewer_identity: str,
+    session_id: str | None,
+    mutation: str,
+    live_head_sha: str | None,
+    pinned_head_sha: str | None,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Reviewer mutations require an owned, current PR lease."""
+    now = now or datetime.now(timezone.utc)
+    reasons: list[str] = []
+    session = get_session_lease()
+    active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
+
+    if not session:
+        reasons.append(
+            f"no in-session reviewer lease recorded; acquire via "
+            f"gitea_acquire_reviewer_pr_lease before {mutation}"
+        )
+    elif session.get("pr_number") != pr_number:
+        reasons.append(
+            f"session lease is for PR #{session.get('pr_number')}, not #{pr_number}"
+        )
+    elif (session.get("session_id") or "") != (session_id or session.get("session_id")):
+        reasons.append("session lease session_id mismatch (fail closed)")
+
+    if active:
+        owner = (active.get("session_id") or "").strip()
+        if owner and session_id and owner != session_id:
+            reasons.append(
+                f"active PR lease owned by session_id={owner}; current session "
+                f"cannot {mutation}"
+            )
+        pinned = _normalize_sha(pinned_head_sha)
+        live = _normalize_sha(live_head_sha)
+        lease_head = active.get("candidate_head")
+        if pinned and live and pinned != live:
+            reasons.append(
+                "PR head changed during lease; stop and re-validate before "
+                f"reviewer {mutation}"
+            )
+        if lease_head and live and lease_head != live:
+            reasons.append(
+                "live PR head differs from lease candidate_head; refresh lease "
+                f"before {mutation}"
+            )
+        freshness = active.get("freshness") or classify_lease_freshness(active, now=now)
+        if freshness in {"expired", "reclaimable"}:
+            reasons.append(f"reviewer lease freshness is '{freshness}' (fail closed)")
+    else:
+        reasons.append(f"no active reviewer lease found on PR #{pr_number}")
+
+    allowed = not reasons
+    return {
+        "mutation_allowed": allowed,
+        "block": not allowed,
+        "reasons": reasons,
+        "active_lease": active,
+        "session_lease": session,
+    }
+
+
+def assess_lease_inventory(
+    comments_by_pr: dict[int, list[dict]],
+    *,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Summarize lease states across PR comment threads."""
+    now = now or datetime.now(timezone.utc)
+    active: list[dict] = []
+    stale: list[dict] = []
+    reclaimable: list[dict] = []
+    for pr_number, comments in (comments_by_pr or {}).items():
+        lease = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
+        if not lease:
+            continue
+        freshness = lease.get("freshness") or classify_lease_freshness(lease, now=now)
+        entry = {"pr_number": pr_number, "session_id": lease.get("session_id"), "freshness": freshness}
+        if freshness == "stale_warning":
+            stale.append(entry)
+        elif freshness == "reclaimable":
+            reclaimable.append(entry)
+        else:
+            active.append(entry)
+    return {
+        "active_review_leases": active,
+        "stale_review_leases": stale,
+        "reclaimable_review_leases": reclaimable,
+    }

scripts/run-webui

@@ -1,6 +0,0 @@
-#!/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 "$@"

skills/llm-project-workflow/workflows/review-merge-pr.md

@@ -732,6 +732,26 @@ The final report must identify:
 * whether same-PR merge continuation was allowed
 * whether the run stopped as required
 
+## 26B. Per-PR reviewer lease (#407)
+
+Parallel reviewer sessions are allowed only when each session holds a distinct,
+live PR lease.
+
+Before validation or review mutation on a selected PR:
+
+1. Call `gitea_acquire_reviewer_pr_lease` with worktree path, candidate head SHA,
+   and target branch SHA.
+2. Post heartbeats via `gitea_heartbeat_reviewer_pr_lease` before validation,
+   after validation, before review mutation, and before merge.
+3. Do not approve, request changes, or merge unless the in-session lease
+   matches the selected PR.
+
+If PR head or target branch advances during the lease, stop and refresh
+inventory before continuing.
+
+Final reports must include lease session id, acquisition proof, heartbeat
+status, and release/blocked status.
+
 ## 27. Merge rules
 
 Before merge, rerun fresh live checks:

tests/test_audit.py

@@ -286,6 +286,21 @@ class TestSimpleToolAudit(_AuditWiringBase):
 
 class TestGatedToolAudit(_AuditWiringBase):
 
+    def setUp(self):
+        super().setUp()
+        from tests.test_mcp_server import _install_owned_reviewer_lease
+        import reviewer_pr_lease
+
+        self._lease_patch = _install_owned_reviewer_lease(8)
+        self._lease_patch.start()
+        self._auth_identity_patch = patch(
+            "mcp_server._authenticated_username", return_value="reviewer-bot"
+        )
+        self._auth_identity_patch.start()
+        self.addCleanup(self._auth_identity_patch.stop)
+        self.addCleanup(self._lease_patch.stop)
+        self.addCleanup(reviewer_pr_lease.clear_session_lease)
+
     def _pr(self, author, state="open", sha="abc123", mergeable=True):
         return {"user": {"login": author}, "state": state,
                 "head": {"sha": sha}, "mergeable": mergeable}
@@ -344,11 +359,22 @@ class TestGatedToolAudit(_AuditWiringBase):
                         GITEA_ALLOWED_OPERATIONS="read,review,approve")
         with patch.dict(os.environ, env, clear=True):
             from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
+            from tests.test_mcp_server import _install_owned_reviewer_lease
+            import reviewer_pr_lease
+
             init_review_decision_lock("prgs", "review_pr")
             gitea_mark_final_review_decision(8, "approve", remote="prgs")
-            r = gitea_submit_pr_review(pr_number=8, action="approve",
+            lease_patch = _install_owned_reviewer_lease(8)
-                                       body="LGTM", remote="prgs",
+            lease_patch.start()
-                                       final_review_decision_ready=True)
+            try:
+                r = gitea_submit_pr_review(
+                    pr_number=8, action="approve",
+                    body="LGTM", remote="prgs",
+                    final_review_decision_ready=True,
+                )
+            finally:
+                lease_patch.stop()
+                reviewer_pr_lease.clear_session_lease()
         self.assertTrue(r["performed"])
         recs = self._records()
         self.assertEqual(len(recs), 1)

tests/test_create_issue_workspace_guard.py

@@ -1,145 +0,0 @@
-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_mcp_server.py

@@ -79,6 +79,64 @@ def _visible_approval_reviews(reviewer="reviewer-bot", sha="abc123"):
     return [_formal_review(reviewer, "APPROVED", sha=sha)]
 
 
+_DEFAULT_LEASE_SESSION = "mcp-test-reviewer-lease"
+
+
+def _reviewer_lease_comment(
+    pr_number,
+    *,
+    session_id=_DEFAULT_LEASE_SESSION,
+    head_sha="abc123",
+    reviewer="reviewer-bot",
+):
+    from datetime import datetime, timezone
+
+    import reviewer_pr_lease
+
+    body = reviewer_pr_lease.format_lease_body(
+        repo="Scaled-Tech-Consulting/Gitea-Tools",
+        pr_number=pr_number,
+        issue_number=407,
+        reviewer_identity=reviewer,
+        profile="gitea-reviewer",
+        session_id=session_id,
+        worktree="branches/review-test",
+        phase="claimed",
+        candidate_head=head_sha,
+        target_branch="master",
+        target_branch_sha="b" * 40,
+        last_activity=datetime.now(timezone.utc),
+    )
+    return {"id": 9001, "body": body, "user": {"login": reviewer}}
+
+
+def _install_owned_reviewer_lease(
+    pr_number,
+    *,
+    session_id=_DEFAULT_LEASE_SESSION,
+    head_sha="abc123",
+):
+    import reviewer_pr_lease
+
+    reviewer_pr_lease.clear_session_lease()
+    reviewer_pr_lease.record_session_lease({
+        "pr_number": pr_number,
+        "session_id": session_id,
+        "candidate_head": head_sha,
+        "target_branch": "master",
+    })
+    return patch(
+        "mcp_server._fetch_pr_comments",
+        return_value=[
+            _reviewer_lease_comment(
+                pr_number,
+                session_id=session_id,
+                head_sha=head_sha,
+            )
+        ],
+    )
+
+
 # Issue-write tools are profile-gated (#69).
 ISSUE_WRITE_ENV = {
     "GITEA_ALLOWED_OPERATIONS": (
@@ -512,6 +570,19 @@ class TestViewPR(unittest.TestCase):
 class TestMergePR(unittest.TestCase):
     """Gated merge workflow (#16). gitea_merge_pr is the only merge path."""
 
+    def setUp(self):
+        import reviewer_pr_lease
+
+        self._lease_patch = _install_owned_reviewer_lease(8)
+        self._lease_patch.start()
+        self._auth_identity_patch = patch(
+            "mcp_server._authenticated_username", return_value="reviewer-bot"
+        )
+        self._auth_identity_patch.start()
+        self.addCleanup(self._auth_identity_patch.stop)
+        self.addCleanup(self._lease_patch.stop)
+        self.addCleanup(reviewer_pr_lease.clear_session_lease)
+
     def _pr(self, author, state="open", sha="abc123", mergeable=True):
         return {
             "user": {"login": author},
@@ -772,9 +843,11 @@ class TestMergePR(unittest.TestCase):
                 pr_number=8, confirmation=self._confirm(8),
                 expected_head_sha="deadbeef", remote="prgs")
         self.assertFalse(r["performed"])
-        self.assertIn(
+        self.assertTrue(any(
-            "expected head SHA does not match current PR head (fail closed)",
+            "expected head SHA does not match current PR head (fail closed)" in reason
-            r["reasons"])
+            or "PR head changed during lease" in reason
+            for reason in r["reasons"]
+        ))
         self._assert_no_merge_call(mock_api)
 
     @patch("mcp_server.api_request")
@@ -1653,7 +1726,20 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
         }
 
     def setUp(self):
+        import reviewer_pr_lease
+
         init_review_decision_lock("prgs", "review_pr")
+        self._lease_patch = _install_owned_reviewer_lease(
+            self.PR, head_sha=self.SHA,
+        )
+        self._lease_patch.start()
+        self._auth_identity_patch = patch(
+            "mcp_server._authenticated_username", return_value="reviewer-bot"
+        )
+        self._auth_identity_patch.start()
+        self.addCleanup(self._auth_identity_patch.stop)
+        self.addCleanup(self._lease_patch.stop)
+        self.addCleanup(reviewer_pr_lease.clear_session_lease)
 
     def _env(self):
         return patch.dict(os.environ, {
@@ -1748,8 +1834,19 @@ class TestSubmitPrReview(unittest.TestCase):
     """Gated review-mutation tool (#15)."""
 
     def setUp(self):
+        import reviewer_pr_lease
+
         init_review_decision_lock("prgs", "review_pr")
         gitea_mark_final_review_decision(8, "approve", remote="prgs")
+        self._lease_patch = _install_owned_reviewer_lease(8)
+        self._lease_patch.start()
+        self._auth_identity_patch = patch(
+            "mcp_server._authenticated_username", return_value="reviewer-bot"
+        )
+        self._auth_identity_patch.start()
+        self.addCleanup(self._auth_identity_patch.stop)
+        self.addCleanup(self._lease_patch.stop)
+        self.addCleanup(reviewer_pr_lease.clear_session_lease)
 
     def _pr(self, author, state="open", sha="abc123", mergeable=True):
         return {
@@ -1987,9 +2084,11 @@ class TestSubmitPrReview(unittest.TestCase):
                     final_review_decision_ready=True,
                 )
             self.assertFalse(r["performed"])
-            self.assertIn(
+            self.assertTrue(any(
-                "expected head SHA does not match current PR head (fail closed)",
+                "expected head SHA does not match current PR head (fail closed)" in reason
-                r["reasons"])
+                or "PR head changed during lease" in reason
+                for reason in r["reasons"]
+            ))
             self._assert_no_mutation(mock_api)
 
     def test_head_sha_match_allows(self):
@@ -2052,9 +2151,9 @@ class TestSubmitPrReview(unittest.TestCase):
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
         with patch.dict(os.environ, env, clear=True):
-            gitea_mark_final_review_decision(5, "approve", remote="prgs")
+            gitea_mark_final_review_decision(8, "approve", remote="prgs")
             r = gitea_submit_pr_review(
-                pr_number=5, action="approve", remote="prgs",
+                pr_number=8, action="approve", remote="prgs",
                 final_review_decision_ready=True,
             )
         self.assertFalse(r["performed"])
@@ -2273,6 +2372,13 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
         self.assertEqual(res["cleanup_status"].get(1), "not present")
 
     def test_merge_pr_with_closes_removes_label(self):
+        import reviewer_pr_lease
+
+        lease_patch = _install_owned_reviewer_lease(1, head_sha="sha123")
+        lease_patch.start()
+        self.addCleanup(lease_patch.stop)
+        self.addCleanup(reviewer_pr_lease.clear_session_lease)
+
         def api_side_effect(method, url, auth, payload=None):
             if method == "GET" and "/user" in url:
                 return {"login": "merger"}
@@ -2307,6 +2413,13 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
         self.assertEqual(res["cleanup_status"].get(123), "released")
 
     def test_merge_pr_with_branch_name_removes_label(self):
+        import reviewer_pr_lease
+
+        lease_patch = _install_owned_reviewer_lease(1, head_sha="sha123")
+        lease_patch.start()
+        self.addCleanup(lease_patch.stop)
+        self.addCleanup(reviewer_pr_lease.clear_session_lease)
+
         def api_side_effect(method, url, auth, payload=None):
             if method == "GET" and "/user" in url:
                 return {"login": "merger"}
@@ -2988,10 +3101,12 @@ class TestVerifyMutationAuthority(unittest.TestCase):
         # profile; the active profile resolves as reviewer — side-channel
         # override rejected even with a matching in-process authority.
         self._authority()
-        with patch.dict(os.environ,
+        with patch("mcp_server.gitea_config.is_runtime_switching_enabled",
-                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}):
+                   return_value=False):
-            with self.assertRaises(RuntimeError) as ctx:
+            with patch.dict(os.environ,
-                mcp_server.verify_mutation_authority("prgs")
+                            {"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}):
+                with self.assertRaises(RuntimeError) as ctx:
+                    mcp_server.verify_mutation_authority("prgs")
         self.assertIn("side-channel override rejected", str(ctx.exception))
 
     def test_foreign_pid_authority_is_not_trusted(self):

tests/test_pr_queue_inventory.py

@@ -156,9 +156,22 @@ class TestPRQueueInventory(unittest.TestCase):
         ]
 
         from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
-        init_review_decision_lock("prgs", "review_pr")
+        from tests.test_mcp_server import _install_owned_reviewer_lease
-        gitea_mark_final_review_decision(1, "approve", remote="prgs")
+        import reviewer_pr_lease
-        result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs", final_review_decision_ready=True)
+
+        with patch("mcp_server._authenticated_username", return_value="reviewer1"):
+            init_review_decision_lock("prgs", "review_pr")
+            gitea_mark_final_review_decision(1, "approve", remote="prgs")
+            lease_patch = _install_owned_reviewer_lease(
+                1, head_sha="abc1", session_id="inventory-review-lease",
+            )
+            lease_patch.start()
+            self.addCleanup(lease_patch.stop)
+            self.addCleanup(reviewer_pr_lease.clear_session_lease)
+            result = gitea_review_pr(
+                pr_number=1, event="APPROVE", remote="prgs",
+                final_review_decision_ready=True,
+            )
         self.assertTrue(result["success"])
         self.assertIn("=== PR Queue Inventory ===", result["message"])
         self.assertIn("Repository:", result["message"])

tests/test_reviewer_pr_lease.py

@@ -0,0 +1,193 @@
+"""Tests for per-PR reviewer leases (#407)."""
+
+import sys
+import unittest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import reviewer_pr_lease as leases
+
+
+def _lease_comment(
+    pr_number: int,
+    session_id: str,
+    *,
+    phase: str = "claimed",
+    minutes_ago: int = 0,
+    candidate_head: str = "a" * 40,
+) -> dict:
+    now = datetime.now(timezone.utc) - timedelta(minutes=minutes_ago)
+    body = leases.format_lease_body(
+        repo="Scaled-Tech-Consulting/Gitea-Tools",
+        pr_number=pr_number,
+        issue_number=295,
+        reviewer_identity="rev1",
+        profile="prgs-reviewer",
+        session_id=session_id,
+        worktree="branches/review-pr382",
+        phase=phase,
+        candidate_head=candidate_head,
+        target_branch="master",
+        target_branch_sha="b" * 40,
+        last_activity=now,
+    )
+    return {"id": 1, "body": body, "user": {"login": "rev1"}}
+
+
+class TestReviewerLeaseAcquire(unittest.TestCase):
+    def setUp(self):
+        leases.clear_session_lease()
+
+    def test_two_reviewers_cannot_lease_same_pr(self):
+        comments = [_lease_comment(382, "session-a")]
+        result = leases.assess_acquire_lease(
+            comments,
+            pr_number=382,
+            reviewer_identity="rev2",
+            profile="prgs-reviewer",
+            session_id="session-b",
+            repo="Scaled-Tech-Consulting/Gitea-Tools",
+            issue_number=295,
+            worktree="branches/review-pr382-b",
+            candidate_head="c" * 40,
+            target_branch="master",
+            target_branch_sha="d" * 40,
+        )
+        self.assertFalse(result["acquire_allowed"])
+        self.assertTrue(any("already has active" in r for r in result["reasons"]))
+
+    def test_two_reviewers_can_lease_different_prs(self):
+        comments = [_lease_comment(382, "session-a")]
+        result = leases.assess_acquire_lease(
+            comments,
+            pr_number=383,
+            reviewer_identity="rev2",
+            profile="prgs-reviewer",
+            session_id="session-b",
+            repo="Scaled-Tech-Consulting/Gitea-Tools",
+            issue_number=296,
+            worktree="branches/review-pr383",
+            candidate_head="c" * 40,
+            target_branch="master",
+            target_branch_sha="d" * 40,
+        )
+        self.assertTrue(result["acquire_allowed"])
+        self.assertIsNotNone(result["lease_body"])
+
+
+class TestReviewerLeaseFreshness(unittest.TestCase):
+    def test_stale_warning_after_30_minutes(self):
+        lease = leases.parse_lease_comment(
+            _lease_comment(382, "session-a", minutes_ago=35)["body"]
+        )
+        self.assertEqual(
+            leases.classify_lease_freshness(lease),
+            "stale_warning",
+        )
+
+    def test_reclaimable_after_60_minutes(self):
+        lease = leases.parse_lease_comment(
+            _lease_comment(382, "session-a", minutes_ago=65)["body"]
+        )
+        self.assertEqual(
+            leases.classify_lease_freshness(lease),
+            "reclaimable",
+        )
+
+
+class TestReviewerLeaseMutationGate(unittest.TestCase):
+    def setUp(self):
+        leases.clear_session_lease()
+
+    def test_reviewer_without_lease_cannot_mutate(self):
+        head = "f" * 40
+        comments = [_lease_comment(382, "other-session", candidate_head=head)]
+        result = leases.assess_mutation_lease_gate(
+            pr_number=382,
+            comments=comments,
+            reviewer_identity="rev1",
+            session_id="my-session",
+            mutation="approve",
+            live_head_sha=head,
+            pinned_head_sha=head,
+        )
+        self.assertTrue(result["block"])
+
+    def test_owned_lease_allows_mutation(self):
+        head = "f" * 40
+        comments = [_lease_comment(382, "my-session", candidate_head=head)]
+        leases.record_session_lease({
+            "pr_number": 382,
+            "session_id": "my-session",
+            "candidate_head": head,
+            "target_branch": "master",
+        })
+        result = leases.assess_mutation_lease_gate(
+            pr_number=382,
+            comments=comments,
+            reviewer_identity="rev1",
+            session_id="my-session",
+            mutation="approve",
+            live_head_sha=head,
+            pinned_head_sha=head,
+        )
+        self.assertFalse(result["block"])
+
+    def test_head_change_invalidates_lease(self):
+        reviewed = "f" * 40
+        live = "e" * 40
+        comments = [_lease_comment(382, "my-session", candidate_head=reviewed)]
+        leases.record_session_lease({
+            "pr_number": 382,
+            "session_id": "my-session",
+            "candidate_head": reviewed,
+        })
+        result = leases.assess_mutation_lease_gate(
+            pr_number=382,
+            comments=comments,
+            reviewer_identity="rev1",
+            session_id="my-session",
+            mutation="merge",
+            live_head_sha=live,
+            pinned_head_sha=reviewed,
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(any("head" in r.lower() for r in result["reasons"]))
+
+
+class TestReviewerLeaseMcpGate(unittest.TestCase):
+    def setUp(self):
+        leases.clear_session_lease()
+        patch("mcp_server.verify_preflight_purity").start()
+        patch("gitea_audit.audit_enabled", return_value=False).start()
+        mcp_server = __import__("mcp_server")
+        mcp_server._IDENTITY_CACHE.clear()
+        mcp_server.init_review_decision_lock("prgs", "review_pr")
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check("capability", "reviewer")
+
+    def tearDown(self):
+        patch.stopall()
+        leases.clear_session_lease()
+
+    def test_reviewer_pr_lease_gate_helper_blocks_without_session(self):
+        import mcp_server
+        head = "a" * 40
+        with patch("mcp_server._fetch_pr_comments", return_value=[]):
+            reasons = mcp_server._reviewer_pr_lease_gate(
+                pr_number=382,
+                remote="prgs",
+                host=None,
+                org=None,
+                repo=None,
+                mutation="approve",
+                live_head_sha=head,
+                pinned_head_sha=head,
+            )
+        self.assertTrue(any("lease" in r.lower() for r in reasons))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_webui_project_registry.py

@@ -1,111 +0,0 @@
-"""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

@@ -1,90 +0,0 @@
-"""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

@@ -1,245 +0,0 @@
-"""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,
-    _classify_issue,
-    _classify_pr,
-    _extract_linked_issue,
-    load_queue_snapshot,
-    snapshot_to_dict,
-)
-
-_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 test_queue_fail_closed_hides_empty_state_copy(self):
-        with mock.patch("webui.queue_loader.get_auth_header", return_value=None):
-            snapshot = load_queue_snapshot()
-        with mock.patch("webui.app.load_queue_snapshot", return_value=snapshot):
-            response = self.client.get("/queue")
-        self.assertEqual(response.status_code, 200)
-        self.assertIn("Queue unavailable", response.text)
-        self.assertIn("not fetched", response.text.lower())
-        self.assertNotIn("No open items.", response.text)
-        self.assertNotIn("Open pull requests", response.text)
-        self.assertNotIn("pages_fetched", response.text)
-
-    def test_queue_successful_empty_inventory_shows_empty_state(self):
-        def _empty_prs(*_args, **_kwargs):
-            return [], _mock_pagination(0)
-
-        def _empty_issues(*_args, **_kwargs):
-            return [], _mock_pagination(0)
-
-        snapshot = load_queue_snapshot(
-            fetch_prs=_empty_prs,
-            fetch_issues=_empty_issues,
-        )
-        with mock.patch("webui.app.load_queue_snapshot", return_value=snapshot):
-            response = self.client.get("/queue")
-        self.assertEqual(response.status_code, 200)
-        self.assertIsNone(snapshot.fetch_error)
-        self.assertEqual(response.text.count("No open items."), 2)
-        self.assertIn("pages_fetched", response.text)
-        self.assertIn("pagination (complete)", response.text)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_webui_skeleton.py

@@ -1,73 +0,0 @@
-"""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()

webui/__init__.py

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

webui/__main__.py

@@ -1,19 +0,0 @@
-"""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

@@ -1,180 +0,0 @@
-"""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

@@ -1,49 +0,0 @@
-{
-  "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

@@ -1,176 +0,0 @@
-"""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

@@ -1,196 +0,0 @@
-"""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

@@ -1,93 +0,0 @@
-"""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

@@ -1,197 +0,0 @@
-"""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

@@ -1,87 +0,0 @@
-"""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

@@ -1,385 +0,0 @@
-"""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

@@ -1,121 +0,0 @@
-"""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], ...],
-) -> str:
-    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>"
-        )
-        inventory_block = (
-            "<p class='muted'><strong>Inventory:</strong> not fetched "
-            "(fail closed — queue tables omitted).</p>"
-        )
-        return (
-            "<h2>Live queue</h2>"
-            f"<p class='meta'>Repository: <code>{html.escape(snapshot.repo_label)}</code> "
-            f"· project <code>{html.escape(snapshot.project_id)}</code></p>"
-            f"{error_block}"
-            f"{inventory_block}"
-            "<p class='muted'>Read-only MVP — claims, reviews, and merges stay in Gitea/MCP tools.</p>"
-        )
-
-    pr_section = _queue_table(
-        title="Open pull requests",
-        items=snapshot.prs,
-        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,
-        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"{_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>"
-    )