fix(author-lock): dead-session recovery cross-session PR base-sync (Closes #872)

docs/webui-local-dev.md

@@ -66,8 +66,6 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
 | `/api/prompts` | JSON prompt export with workflow hashes |
 | `/runtime` | MCP runtime health and stale detection (#430) |
 | `/api/runtime` | JSON runtime health export |
-| `/policy` | Workflow policy and guardrail configuration visibility (#646) |
-| `/api/v1/policy` | Versioned JSON guardrail inventory (redacted, read-only) |
 | `/audit` | Report audit paste + validator preview (#431) |
 | `/api/audit` | JSON validator preview (POST `report_text`, optional `task_kind`) |
 | `/worktrees` | Worktree hygiene dashboard (#432) |
@@ -80,6 +78,7 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
 | `/sessions` | Phase 1 shell stub — session inventory (backed by #636) |
 | `/inventory` | Phase 1 shell stub — unified inventory (backed by #636) |
 | `/timeline` | Phase 1 shell stub — workflow event timeline |
+| `/policy` | Phase 1 shell stub — capability/role policy placeholder |
 | `/insights` | Phase 1 shell stub — operational insights placeholder |
 
 Most routes are GET-only. POST/PUT/PATCH/DELETE return `405` with
@@ -240,25 +239,12 @@ health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
 checkout is behind merged safety-gate changes. Restart guidance links to #420;
 no tokens or MCP restart actions are exposed.
 
-## Policy & guardrail visibility (#646)
-
-`/policy` (HTML) and `/api/v1/policy` (JSON) surface a **read-only** projection
-of the major workflow guardrails — role separation/RBAC, lease lifecycle,
-author worktree binding, merge confirmation, secret redaction, contamination
-containment, allocator policy, audit logging, and mutation gating. Each entry
-carries source pointers to the file/module/doc that owns it, a compact active
-value derived from the existing safe policy accessors, and — where a documented
-default is declared — a diff of active vs documented. The whole payload is run
-through the console redaction pass before it is emitted, so a planted or
-accidental secret degrades to the placeholder rather than reaching a client.
-The view never edits policy and exposes no gate-weakening toggle.
-
 ## Application shell — Phase 1 (#638)
 
 The console shell (`webui/layout.py`) renders a grouped navigation driven by a
 single nav-config module, `webui/nav.py`. Nav groups follow the epic #631
 Phase 1 information architecture: **Health, Traffic, Runtime/Sessions,
-Projects, Inventory, Timeline, Policy** (live via #646), and **Insights**
+Projects, Inventory, Timeline, Policy** (placeholder), and **Insights**
 (placeholder). Live views and Phase 1 placeholders (`stub`) are declared in one
 place so the layout and the route table cannot drift.
 
@@ -268,10 +254,10 @@ a **mode: read-only** badge — plus a **Docs** link to this document. No
 privileged action controls are present in the Phase 1 shell.
 
 Not-yet-implemented surfaces (`/sessions`, `/inventory`, `/timeline`,
-`/insights`) resolve to graceful read-only stub pages instead of 404s; their
-backing views land in later child issues of #631 (the inventory surfaces are
-backed by #636). `/policy` is a live read-only surface (#646), not a stub.
-Mutating methods on stub routes still fail closed with `read-only-mvp`.
+`/policy`, `/insights`) resolve to graceful read-only stub pages instead of
+404s; their backing views land in later child issues of #631 (the inventory
+surfaces are backed by #636). Mutating methods on stub routes still fail closed
+with `read-only-mvp`.
 
 ## System-health dashboard (#639)
 

gitea_mcp_server.py

@@ -2485,6 +2485,7 @@ def _evaluate_issue_lock_recovery(
             worktree_path,
             prior_head_sha=local_head,
             synced_head_sha=remote_head,
+            remote=remote,
         )
 
     # #772: with no remote branch there is no head to measure against, so the

issue_lock_worktree.py

@@ -150,23 +150,14 @@ def read_merge_sync_provenance(
     *,
     prior_head_sha: str | None,
     synced_head_sha: str | None,
+    remote: str | None = None,
 ) -> dict:
-    """Observe whether ``synced_head_sha`` is a sanctioned merge-based branch sync
-    that advanced the PR branch past ``prior_head_sha`` (#871).
+    """Observe whether ``synced_head_sha`` is a sanctioned merge-sync of a base
+    into the branch above ``prior_head_sha`` (#871/#872).
 
-    ``gitea_update_pr_branch_by_merge`` advances a PR branch by merging the base
-    branch *into* the branch (``POST /pulls/{n}/update?style=merge``). The result
-    is a merge commit ``M`` on the branch whose **first** parent is the prior
-    branch head and whose second parent is the base tip. When the owning session
-    then dies without the durable lock's recorded head being refreshed, the local
-    worktree still sits at ``prior_head_sha`` while the live PR head is ``M``.
-
-    Recovering that drift safely requires proving the remote head is *exactly*
-    such a merge-sync — not a rewrite, rebase, force-push, or an unrelated
-    commit. This is that server-side observation. It reports facts only; the
-    disposition lives in ``issue_lock_recovery``. Every field is read from git in
-    the declared worktree — nothing is supplied by, or reachable from, an MCP
-    caller (#871).
+    Reports server-derived git facts only; the recovery disposition lives in
+    ``issue_lock_recovery``. All comparisons are executed locally in the
+    declared worktree -- nothing is taken from caller parameters.
 
     Provenance is proven only when ALL hold:
 
@@ -183,6 +174,7 @@ def read_merge_sync_provenance(
     path = (worktree_path or "").strip()
     prior = (prior_head_sha or "").strip()
     synced = (synced_head_sha or "").strip()
+    target_remote = (remote or "").strip() or None
     result: dict = {
         "prior_head_sha": prior or None,
         "synced_head_sha": synced or None,
@@ -235,6 +227,23 @@ def read_merge_sync_provenance(
     try:
         result["prior_present"] = _present(prior)
         result["synced_present"] = _present(synced)
+        if not result["synced_present"] and path and os.path.isdir(path):
+            if target_remote:
+                subprocess.run(
+                    ["git", "-C", path, "fetch", target_remote, "--quiet"],
+                    capture_output=True,
+                    text=True,
+                    check=False,
+                )
+                result["synced_present"] = _present(synced)
+            if not result["synced_present"]:
+                subprocess.run(
+                    ["git", "-C", path, "fetch", "--quiet"],
+                    capture_output=True,
+                    text=True,
+                    check=False,
+                )
+                result["synced_present"] = _present(synced)
     except OSError as exc:  # git unavailable — fail closed, never assume
         result["reasons"].append(f"merge-sync provenance probe could not run: {exc}")
         return result

tests/test_issue_872_dead_session_two_base_syncs.py

@@ -0,0 +1,332 @@
+"""Dead-session recovery for cross-session PR base-syncs (#872).
+
+Validates that when a sanctioned server-side base-sync (``gitea_update_pr_branch_by_merge``)
+advances a PR's remote head from A to B (a merge commit whose first parent is A),
+a subsequent author session whose local worktree is at A can recover the dead-session
+lock via HEAD_RELATION_REMOTE_MERGE_SYNCED and execute a second base-sync (B to C)
+without deadlock or RuntimeError.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import issue_lock_recovery  # noqa: E402
+import issue_lock_store  # noqa: E402
+import issue_lock_worktree  # noqa: E402
+
+ISSUE = 8720
+PR_NUMBER = 8721
+BRANCH = f"fix/issue-{ISSUE}-dead-session-two-base-syncs"
+IDENTITY = "example-author"
+PROFILE = "prgs-author"
+REMOTE = "prgs"
+ORG = "ExampleOrg"
+REPO = "ExampleRepo"
+
+
+def dead_pid() -> int:
+    proc = subprocess.Popen([sys.executable, "-c", "pass"])
+    proc.wait()
+    return proc.pid
+
+
+def future_ts(hours: int = 4) -> str:
+    return (
+        (datetime.now(timezone.utc) + timedelta(hours=hours))
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+def _git(cwd, *args):
+    return subprocess.run(
+        ["git", "-C", cwd, *args],
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+
+
+def _rev(cwd, ref="HEAD") -> str:
+    return _git(cwd, "rev-parse", ref).stdout.strip()
+
+
+def build_two_base_sync_repo(tmp: str) -> dict:
+    """Build a repo simulating two sequential base-sync merges."""
+    _git(tmp, "init", "-q", "-b", "master")
+    _git(tmp, "config", "user.email", "[email protected]")
+    _git(tmp, "config", "user.name", "Author")
+    Path(tmp, "base.txt").write_text("base 1\n")
+    _git(tmp, "add", "-A")
+    _git(tmp, "commit", "-q", "-m", "initial master")
+
+    # Feature branch cut from initial master -> Head A (prior/recorded head)
+    _git(tmp, "checkout", "-q", "-b", BRANCH)
+    Path(tmp, "feature.txt").write_text("feature work\n")
+    _git(tmp, "add", "-A")
+    _git(tmp, "commit", "-q", "-m", "feature commit A")
+    head_a = _rev(tmp)
+
+    # Master advances -> Master 1
+    _git(tmp, "checkout", "-q", "master")
+    Path(tmp, "base.txt").write_text("base 1\nbase 2\n")
+    _git(tmp, "add", "-A")
+    _git(tmp, "commit", "-q", "-m", "master advance 1")
+    master_1 = _rev(tmp)
+
+    # First sync: merge master into feature -> Head B (merge commit, first parent = A)
+    _git(tmp, "checkout", "-q", BRANCH)
+    _git(tmp, "merge", "-q", "--no-ff", "-m", "First base-sync (merge master)", "master")
+    head_b = _rev(tmp)
+
+    # Master advances again -> Master 2
+    _git(tmp, "checkout", "-q", "master")
+    Path(tmp, "base.txt").write_text("base 1\nbase 2\nbase 3\n")
+    _git(tmp, "add", "-A")
+    _git(tmp, "commit", "-q", "-m", "master advance 2")
+    master_2 = _rev(tmp)
+
+    # Second sync: merge master into feature -> Head C (merge commit, first parent = B)
+    _git(tmp, "checkout", "-q", BRANCH)
+    _git(tmp, "merge", "-q", "--no-ff", "-m", "Second base-sync (merge master)", "master")
+    head_c = _rev(tmp)
+
+    # Non-merge rebase/force-pushed branch shape
+    _git(tmp, "checkout", "-q", "-b", "rebased-branch", head_a)
+    Path(tmp, "rebase.txt").write_text("rebased\n")
+    _git(tmp, "add", "-A")
+    _git(tmp, "commit", "-q", "-m", "rebased commit")
+    rebased_head = _rev(tmp)
+
+    # Reset worktree back to head A, as a dead session leaving local worktree at A
+    _git(tmp, "checkout", "-q", BRANCH)
+    _git(tmp, "reset", "-q", "--hard", head_a)
+
+    return {
+        "head_a": head_a,
+        "master_1": master_1,
+        "head_b": head_b,
+        "master_2": master_2,
+        "head_c": head_c,
+        "rebased_head": rebased_head,
+    }
+
+
+def make_dead_lock(worktree, **over):
+    pid = dead_pid()
+    lock = {
+        "issue_number": ISSUE,
+        "branch_name": BRANCH,
+        "worktree_path": worktree,
+        "remote": REMOTE,
+        "org": ORG,
+        "repo": REPO,
+        "session_pid": pid,
+        "pid": pid,
+        "claimant": {"username": IDENTITY, "profile": PROFILE},
+        "work_lease": {
+            "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+            "issue_number": ISSUE,
+            "branch": BRANCH,
+            "worktree_path": worktree,
+            "claimant": {"username": IDENTITY, "profile": PROFILE},
+            "expires_at": future_ts(),
+        },
+    }
+    lock.update(over)
+    return lock
+
+
+class TestDeadSessionTwoBaseSyncs(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp()
+        self.shas = build_two_base_sync_repo(self.tmp)
+        self.head_a = self.shas["head_a"]
+        self.head_b = self.shas["head_b"]
+        self.head_c = self.shas["head_c"]
+
+    def test_first_base_sync_dead_session_recovery(self):
+        """AC1: dead session after first base sync (remote=B, local=A) recovers via merge-sync."""
+        lock = make_dead_lock(self.tmp, synced_pr_head=self.head_b)
+        obs = issue_lock_worktree.read_merge_sync_provenance(
+            self.tmp,
+            prior_head_sha=self.head_a,
+            synced_head_sha=self.head_b,
+            remote=REMOTE,
+        )
+        self.assertTrue(obs["is_merge_sync"], obs["reasons"])
+        self.assertTrue(obs["prior_is_ancestor"])
+        self.assertTrue(obs["synced_is_merge"])
+        self.assertTrue(obs["first_parent_reaches_prior"])
+
+        res = issue_lock_recovery.assess_dead_session_lock_recovery(
+            lock,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=self.tmp,
+            remote=REMOTE,
+            org=ORG,
+            repo=REPO,
+            identity=IDENTITY,
+            profile=PROFILE,
+            current_branch=BRANCH,
+            porcelain_status="",
+            head_sha=self.head_a,
+            remote_head_sha=self.head_b,
+            pr_head_sha=self.head_b,
+            pr_number=PR_NUMBER,
+            competing_live_locks=[],
+            candidate_branches=[BRANCH],
+            current_pid=os.getpid(),
+            remote_branch_exists=True,
+            sync_provenance=obs,
+        )
+        self.assertEqual(res["outcome"], issue_lock_recovery.RECOVERY_SANCTIONED, res["reasons"])
+        self.assertEqual(
+            res["evidence"]["head_relation"],
+            issue_lock_recovery.HEAD_RELATION_REMOTE_MERGE_SYNCED,
+        )
+        self.assertEqual(res["evidence"]["accepted_head"], self.head_b)
+
+    def test_second_base_sync_head_refresh_after_recovery(self):
+        """AC2: after recovery, second base sync (remote B -> C) updates durable lock head."""
+        lock_dir = tempfile.mkdtemp()
+        lock_data = make_dead_lock(self.tmp, synced_pr_head=self.head_b)
+        # Rebind to current PID as gitea_lock_issue does upon sanctioned recovery
+        lock_data["session_pid"] = os.getpid()
+        lock_data["pid"] = os.getpid()
+        issue_lock_store.save_lock_file(
+            issue_lock_store.lock_file_path(
+                remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE, lock_dir=lock_dir
+            ),
+            lock_data,
+        )
+
+        refresh_res = issue_lock_store.apply_durable_lock_head_refresh(
+            remote=REMOTE,
+            org=ORG,
+            repo=REPO,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=self.tmp,
+            pr_number=PR_NUMBER,
+            identity=IDENTITY,
+            profile=PROFILE,
+            current_pid=os.getpid(),
+            expected_old_head=self.head_b,
+            new_head=self.head_c,
+            synced_at=future_ts(0),
+            base_head=self.shas["master_2"],
+            lock_dir=lock_dir,
+        )
+        self.assertTrue(refresh_res["refreshed"], refresh_res["reasons"])
+        self.assertTrue(refresh_res["read_after_write_ok"])
+        loaded = issue_lock_store.load_issue_lock(
+            remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE, lock_dir=lock_dir
+        )
+        self.assertEqual(loaded.get("synced_pr_head"), self.head_c)
+
+    def test_dirty_worktree_blocks_recovery(self):
+        """AC3: tracked dirty edits block dead-session recovery."""
+        lock = make_dead_lock(self.tmp)
+        obs = issue_lock_worktree.read_merge_sync_provenance(
+            self.tmp, prior_head_sha=self.head_a, synced_head_sha=self.head_b
+        )
+        res = issue_lock_recovery.assess_dead_session_lock_recovery(
+            lock,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=self.tmp,
+            remote=REMOTE,
+            org=ORG,
+            repo=REPO,
+            identity=IDENTITY,
+            profile=PROFILE,
+            current_branch=BRANCH,
+            porcelain_status=" M feature.txt\n",
+            head_sha=self.head_a,
+            remote_head_sha=self.head_b,
+            pr_head_sha=self.head_b,
+            pr_number=PR_NUMBER,
+            competing_live_locks=[],
+            candidate_branches=[BRANCH],
+            current_pid=os.getpid(),
+            remote_branch_exists=True,
+            sync_provenance=obs,
+        )
+        self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
+        self.assertTrue(any("dirty" in r or "tracked" in r for r in res["reasons"]))
+
+    def test_foreign_reclaimer_blocks_recovery(self):
+        """AC4: a foreign claimant cannot recover a dead-session lock."""
+        lock = make_dead_lock(self.tmp)
+        obs = issue_lock_worktree.read_merge_sync_provenance(
+            self.tmp, prior_head_sha=self.head_a, synced_head_sha=self.head_b
+        )
+        res = issue_lock_recovery.assess_dead_session_lock_recovery(
+            lock,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=self.tmp,
+            remote=REMOTE,
+            org=ORG,
+            repo=REPO,
+            identity="intruder-user",
+            profile=PROFILE,
+            current_branch=BRANCH,
+            porcelain_status="",
+            head_sha=self.head_a,
+            remote_head_sha=self.head_b,
+            pr_head_sha=self.head_b,
+            pr_number=PR_NUMBER,
+            competing_live_locks=[],
+            candidate_branches=[BRANCH],
+            current_pid=os.getpid(),
+            remote_branch_exists=True,
+            sync_provenance=obs,
+        )
+        self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
+
+    def test_non_merge_rebased_remote_head_blocks_recovery(self):
+        """AC5: a rebased/force-pushed remote head (not a merge commit) is refused."""
+        lock = make_dead_lock(self.tmp)
+        obs = issue_lock_worktree.read_merge_sync_provenance(
+            self.tmp, prior_head_sha=self.head_a, synced_head_sha=self.shas["rebased_head"]
+        )
+        self.assertFalse(obs["is_merge_sync"])
+        res = issue_lock_recovery.assess_dead_session_lock_recovery(
+            lock,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=self.tmp,
+            remote=REMOTE,
+            org=ORG,
+            repo=REPO,
+            identity=IDENTITY,
+            profile=PROFILE,
+            current_branch=BRANCH,
+            porcelain_status="",
+            head_sha=self.head_a,
+            remote_head_sha=self.shas["rebased_head"],
+            pr_head_sha=self.shas["rebased_head"],
+            pr_number=PR_NUMBER,
+            competing_live_locks=[],
+            candidate_branches=[BRANCH],
+            current_pid=os.getpid(),
+            remote_branch_exists=True,
+            sync_provenance=obs,
+        )
+        self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_webui_policy_visibility.py

@@ -1,243 +0,0 @@
-"""Tests for the read-only workflow policy/guardrail visibility view (#646).
-
-Covers issue #646 acceptance criteria:
-
-1. Console lists major guardrails with source pointers.
-2. Secrets redacted.
-3. Tests ensure sample secrets never appear.
-4. Docs explain read-only nature (asserted here for the page copy; the doc
-   itself is covered by inspection).
-"""
-
-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 import console_redaction
-from webui import policy_inventory
-from webui.app import create_app
-from webui.policy_inventory import (
-    PolicyEntry,
-    PolicyInventorySnapshot,
-    SourcePointer,
-    load_policy_inventory,
-    snapshot_to_dict,
-)
-from webui.policy_views import render_policy_page
-
-
-def _entry(key, category, *, active=None, error=None):
-    return PolicyEntry(
-        key=key,
-        title=key.replace("_", " ").title(),
-        category=category,
-        summary=f"summary for {key}",
-        sources=(SourcePointer("src", f"{key}.py", "module"),),
-        active=active,
-        documented_default=None,
-        diff=None,
-        error=error,
-    )
-
-
-def _snapshot(entries):
-    return PolicyInventorySnapshot(
-        schema_version=1,
-        read_only=True,
-        note="read-only projection",
-        entries=tuple(entries),
-        categories=tuple(dict.fromkeys(e.category for e in entries)),
-        build_errors=(),
-    )
-
-# The guardrail categories issue #646 names as in-scope.
-_EXPECTED_CATEGORIES = {
-    "role_separation",
-    "lease_rules",
-    "worktree_rules",
-    "merge_confirmation",
-    "redaction",
-    "contamination",
-    "allocator_policy",
-    "audit_logging",
-    "mutation_gating",
-}
-
-
-class TestPolicyInventoryModel(unittest.TestCase):
-    def test_major_guardrails_present(self):
-        snapshot = load_policy_inventory()
-        categories = {e.category for e in snapshot.entries}
-        self.assertEqual(_EXPECTED_CATEGORIES, categories)
-        self.assertGreaterEqual(len(snapshot.entries), len(_EXPECTED_CATEGORIES))
-
-    def test_every_guardrail_has_source_pointers(self):
-        # AC1: source attribution (file/module/doc) for every guardrail.
-        snapshot = load_policy_inventory()
-        for entry in snapshot.entries:
-            with self.subTest(entry=entry.key):
-                self.assertTrue(entry.sources, "guardrail must carry source pointers")
-                for source in entry.sources:
-                    self.assertTrue(source.path)
-                    self.assertIn(source.kind, {"module", "doc", "script", "config"})
-
-    def test_diff_reported_where_documented_default_declared(self):
-        snapshot = load_policy_inventory()
-        checked_any = False
-        for entry in snapshot.entries:
-            if entry.documented_default is None:
-                self.assertIsNone(entry.diff)
-                continue
-            checked_any = True
-            self.assertIsNotNone(entry.diff)
-            self.assertEqual(
-                entry.diff["status"],
-                "matches_documented_default",
-                f"{entry.key} drifted from its documented default: {entry.diff}",
-            )
-        self.assertTrue(checked_any, "at least one guardrail should declare a default")
-
-    def test_live_projections_populate_active(self):
-        snapshot = load_policy_inventory()
-        by_key = {e.key: e for e in snapshot.entries}
-        for key in ("role_separation", "redaction", "audit_logging"):
-            self.assertIsNone(by_key[key].error, f"{key} projection failed")
-            self.assertIsInstance(by_key[key].active, dict)
-
-    def test_build_entry_is_fail_soft_on_projection_error(self):
-        def _boom():
-            raise RuntimeError("projection exploded")
-
-        row = (
-            "redaction",
-            "Secret redaction",
-            "redaction",
-            "summary",
-            (SourcePointer("x", "webui/console_redaction.py", "module"),),
-            _boom,
-            {"redact_before_persist": True},
-        )
-        entry = policy_inventory._build_entry(row)
-        self.assertIsNone(entry.active)
-        self.assertIsNotNone(entry.error)
-        self.assertEqual(entry.diff["status"], "active_unavailable")
-
-
-class TestPolicyRedaction(unittest.TestCase):
-    def test_real_snapshot_has_no_secret_shapes(self):
-        # AC3: the real emitted payload never carries a known secret shape.
-        payload = snapshot_to_dict(load_policy_inventory())
-        self.assertEqual(console_redaction.scan_for_secrets(payload), [])
-
-    def test_planted_keychain_secret_is_redacted(self):
-        # AC2/AC3: a secret planted in an active projection is masked before emit.
-        snapshot = _snapshot([
-            _entry(
-                "redaction",
-                "redaction",
-                active={"leaked": "keychain:prgs-author-super-secret", "roles": ["author"]},
-            )
-        ])
-        payload = snapshot_to_dict(snapshot)
-        blob = json.dumps(payload)
-        self.assertNotIn("keychain:prgs-author-super-secret", blob)
-        self.assertEqual(console_redaction.scan_for_secrets(payload), [])
-
-    def test_planted_credential_assignment_is_redacted(self):
-        snapshot = _snapshot([
-            _entry(
-                "audit_logging",
-                "audit_logging",
-                active={"leaked": "token=abcd1234efgh5678", "append_only": True},
-            )
-        ])
-        payload = snapshot_to_dict(snapshot)
-        blob = json.dumps(payload)
-        self.assertNotIn("abcd1234efgh5678", blob)
-        self.assertEqual(console_redaction.scan_for_secrets(payload), [])
-
-    def test_planted_secret_is_redacted_in_html_emit(self):
-        # AC2/AC3: HTML emit path runs redaction before rendering HTML cards.
-        snapshot = _snapshot([
-            _entry(
-                "redaction",
-                "redaction",
-                active={"leaked": "keychain:prgs-author-super-secret", "roles": ["author"]},
-            )
-        ])
-        html_output = render_policy_page(snapshot)
-        self.assertNotIn("keychain:prgs-author-super-secret", html_output)
-        self.assertEqual(console_redaction.scan_for_secrets(html_output), [])
-
-
-class TestPolicyRoutes(unittest.TestCase):
-    def setUp(self):
-        self.client = TestClient(create_app())
-
-    def test_policy_html_lists_guardrails_with_sources(self):
-        response = self.client.get("/policy")
-        self.assertEqual(response.status_code, 200)
-        text = response.text
-        self.assertIn("Workflow policy", text)
-        self.assertIn("Role separation and RBAC", text)
-        self.assertIn("Source pointers", text)
-        self.assertIn("task_capability_map.py", text)
-        self.assertIn("docs/safety-model.md", text)
-
-    def test_policy_html_states_read_only(self):
-        # AC4: the page explains its read-only nature.
-        text = self.client.get("/policy").text
-        self.assertIn("read-only", text.lower())
-        self.assertNotIn("<form", text.lower())
-
-    def test_policy_html_has_no_secret_shapes(self):
-        text = self.client.get("/policy").text
-        self.assertEqual(console_redaction.scan_for_secrets(text), [])
-
-    def test_api_v1_policy_returns_inventory(self):
-        response = self.client.get("/api/v1/policy")
-        self.assertEqual(response.status_code, 200)
-        data = response.json()
-        self.assertEqual(data["schema_version"], policy_inventory.SCHEMA_VERSION)
-        self.assertTrue(data["read_only"])
-        self.assertEqual(data["entry_count"], len(data["entries"]))
-        self.assertEqual(set(data["categories"]), _EXPECTED_CATEGORIES)
-
-    def test_policy_is_read_only_no_post(self):
-        # AC4 / non-goal: no mutation endpoint.
-        response = self.client.post("/policy")
-        self.assertIn(response.status_code, (404, 405))
-
-    def test_nav_links_policy(self):
-        # Policy is graduated live: not in STUB_PAGES and not labeled ·stub in nav.
-        from webui.nav import STUB_PAGES, iter_nav_items
-
-        self.assertNotIn("/policy", STUB_PAGES)
-        policy_items = [i for i in iter_nav_items() if i.href == "/policy"]
-        self.assertEqual(len(policy_items), 1)
-        self.assertEqual(policy_items[0].status, "live")
-        text = self.client.get("/").text
-        self.assertIn('href="/policy">Policy</a>', text)
-        self.assertNotIn('href="/policy" class="nav-stub"', text)
-
-
-class TestPolicyViewFailSoft(unittest.TestCase):
-    def test_page_renders_when_a_projection_errors(self):
-        snapshot = _snapshot([
-            _entry("role_separation", "role_separation", error="active projection unavailable: boom"),
-            _entry("redaction", "redaction", active={"redact_before_persist": True}),
-        ])
-        page = render_policy_page(snapshot)
-        # The errored guardrail surfaces its error; other guardrails still render.
-        self.assertIn("Active value unavailable", page)
-        self.assertIn("Redaction", page)
-        self.assertIn("Workflow policy", page)
-
-
-if __name__ == "__main__":
-    unittest.main()

webui/app.py

@@ -46,8 +46,6 @@ from webui.worktree_scanner import load_hygiene_snapshot, snapshot_to_dict as wo
 from webui.worktree_views import render_worktrees_page
 from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
 from webui.runtime_views import render_runtime_page
-from webui.policy_inventory import load_policy_inventory, snapshot_to_dict as policy_snapshot_to_dict
-from webui.policy_views import render_policy_page
 from webui.timeline import load_timeline, snapshot_to_dict as timeline_snapshot_to_dict
 from webui.system_health import (
     API_PATH as SYSTEM_HEALTH_API_PATH,
@@ -304,17 +302,6 @@ async def api_runtime(_request: Request) -> JSONResponse:
     return JSONResponse(runtime_snapshot_to_dict(load_runtime_snapshot()))
 
 
-async def policy(_request: Request) -> HTMLResponse:
-    snapshot = load_policy_inventory()
-    return HTMLResponse(
-        render_page(title="Policy", body_html=render_policy_page(snapshot))
-    )
-
-
-async def api_v1_policy(_request: Request) -> JSONResponse:
-    return JSONResponse(policy_snapshot_to_dict(load_policy_inventory()))
-
-
 async def _parse_audit_form(request: Request) -> tuple[str, str | None]:
     if request.method == "GET":
         return "", None
@@ -620,8 +607,6 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
             Route("/api/prompts", api_prompts, methods=["GET"]),
             Route("/runtime", runtime, methods=["GET"]),
             Route("/api/runtime", api_runtime, methods=["GET"]),
-            Route("/policy", policy, methods=["GET"]),
-            Route("/api/v1/policy", api_v1_policy, methods=["GET"]),
             Route("/api/v1/timeline", api_v1_timeline, methods=["GET"]),
             Route("/audit", audit, methods=["GET", "POST"]),
             Route("/api/audit", api_audit, methods=["GET", "POST"]),

webui/nav.py

@@ -5,7 +5,7 @@ the ``webui/app.py`` route table stay aligned with epic #631. Read-only: every
 destination is a GET view or a Phase 1 placeholder. No mutation links.
 
 Nav groups follow the #631 Phase 1 information architecture: Health, Traffic,
-Runtime/Sessions, Projects, Inventory, Timeline, Policy (live via #646), and
+Runtime/Sessions, Projects, Inventory, Timeline, Policy (placeholder), and
 Insights (placeholder). Later-phase surfaces are declared as ``stub`` items and
 backed by ``STUB_PAGES`` so their nav links resolve to a graceful placeholder
 instead of a 404.
@@ -60,7 +60,7 @@ NAV_GROUPS: tuple[NavGroup, ...] = (
         NavItem("/timeline", "Timeline", "stub"),
     )),
     NavGroup("Policy", (
-        NavItem("/policy", "Policy"),
+        NavItem("/policy", "Policy", "stub"),
         NavItem("/prompts", "Prompts"),
     )),
     NavGroup("Insights", (
@@ -88,6 +88,10 @@ STUB_PAGES: dict[str, tuple[str, str]] = {
         "Timeline",
         "Workflow event timeline across issues and PRs. A later Phase 1 surface.",
     ),
+    "/policy": (
+        "Policy",
+        "Capability and role policy surface. Placeholder until a later phase.",
+    ),
     "/insights": (
         "Insights",
         "Aggregate operational insights and trends. Placeholder until a later "

webui/policy_inventory.py

@@ -1,387 +0,0 @@
-"""Read-only workflow policy and guardrail inventory for the web UI (#646).
-
-Policy and guardrails live in code, profiles, docs, and skills. An operator
-cannot *see* the active workflow policy configuration from the console without
-reading the repository tree. This module projects the major guardrails into a
-redacted, machine-readable inventory with source attribution (file / module /
-doc), so the console can render them as HTML tables with source pointers.
-
-Design constraints (Phase 3, #646):
-
-- **Read-only projection.** Nothing here edits policy or exposes a toggle that
-  could weaken a gate. It reports what is already enforced elsewhere.
-- **Source attribution without secrets.** Every guardrail carries pointers to
-  the file/module/doc that owns it. Live values are compact summaries derived
-  from the safe policy accessors that already exist (``rbac_matrix``,
-  ``redaction_policy``, ``audit_policy``); raw regex, tokens, and endpoints are
-  never embedded.
-- **Redact before emit.** ``snapshot_to_dict`` runs the whole payload through
-  ``console_redaction.redact_payload`` so a planted or accidental secret in any
-  projected value degrades to the placeholder rather than reaching a client.
-- **Fail soft.** A projection that raises is recorded as a per-entry error and
-  never takes the page down; a guardrail is still listed with its sources.
-- **Diff vs documented defaults where feasible.** When a guardrail declares a
-  documented invariant, the active projection is compared against it and the
-  result is reported; otherwise the diff is explicitly ``None`` with a reason.
-"""
-
-from __future__ import annotations
-
-from dataclasses import dataclass
-from typing import Any, Callable
-
-from webui import console_audit
-from webui import console_authz
-from webui import console_redaction
-
-SCHEMA_VERSION = 1
-
-READ_ONLY_NOTE = (
-    "Read-only projection of guardrails enforced in code, profiles, docs, and "
-    "skills. This view never edits policy and exposes no gate-weakening toggle."
-)
-
-
-@dataclass(frozen=True)
-class SourcePointer:
-    """Where a guardrail is defined. Attribution only — never a secret."""
-
-    label: str
-    path: str
-    kind: str  # "module" | "doc" | "script" | "config"
-    anchor: str | None = None
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "label": self.label,
-            "path": self.path,
-            "kind": self.kind,
-            "anchor": self.anchor,
-        }
-
-
-@dataclass(frozen=True)
-class PolicyEntry:
-    key: str
-    title: str
-    category: str
-    summary: str
-    sources: tuple[SourcePointer, ...]
-    active: dict[str, Any] | None
-    documented_default: dict[str, Any] | None
-    diff: dict[str, Any] | None
-    error: str | None = None
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "key": self.key,
-            "title": self.title,
-            "category": self.category,
-            "summary": self.summary,
-            "sources": [s.to_dict() for s in self.sources],
-            "active": self.active,
-            "documented_default": self.documented_default,
-            "diff": self.diff,
-            "error": self.error,
-        }
-
-
-@dataclass(frozen=True)
-class PolicyInventorySnapshot:
-    schema_version: int
-    read_only: bool
-    note: str
-    entries: tuple[PolicyEntry, ...]
-    categories: tuple[str, ...]
-    build_errors: tuple[str, ...]
-
-
-def _diff_active_vs_default(
-    active: dict[str, Any] | None,
-    documented_default: dict[str, Any] | None,
-) -> dict[str, Any] | None:
-    """Compare only the keys the documented default declares.
-
-    Returns ``None`` when no documented default is declared (diff not feasible)
-    or when the active projection is unavailable. Otherwise reports, per
-    declared key, whether the active value matches the documented invariant.
-    """
-    if not documented_default:
-        return None
-    if not active:
-        return {"status": "active_unavailable", "checked": {}}
-    checked: dict[str, Any] = {}
-    matches = True
-    for key, expected in documented_default.items():
-        observed = active.get(key)
-        ok = observed == expected
-        matches = matches and ok
-        checked[key] = {"expected": expected, "observed": observed, "matches": ok}
-    return {
-        "status": "matches_documented_default" if matches else "drift_detected",
-        "checked": checked,
-    }
-
-
-# ── Live projections (compact, safe, fail-soft) ──────────────────────────────
-# Each returns a small dict of already-safe machine values. They are module
-# level so tests can substitute one to prove the redaction pass runs.
-
-
-def _project_role_separation() -> dict[str, Any]:
-    matrix = console_authz.rbac_matrix()
-    return {
-        "model_version": matrix.get("model_version"),
-        "active_phase": matrix.get("active_phase"),
-        "roles": [r.get("role") for r in matrix.get("roles", [])],
-        "privileged_action_count": len(matrix.get("privileged_actions", [])),
-        "default_decision": matrix.get("default_decision"),
-        "execution_enabled": matrix.get("execution_enabled"),
-    }
-
-
-def _project_redaction() -> dict[str, Any]:
-    policy = console_redaction.redaction_policy()
-    return {
-        "policy_version": policy.get("policy_version"),
-        "placeholder": policy.get("placeholder"),
-        "applies_to": policy.get("applies_to"),
-        "console_detector_count": len(policy.get("console_rules", [])),
-        "redact_before_persist": policy.get("redact_before_persist"),
-        "failure_mode": policy.get("failure_mode"),
-    }
-
-
-def _project_audit() -> dict[str, Any]:
-    policy = console_audit.audit_policy()
-    return {
-        "schema_version": policy.get("schema_version"),
-        "required_field_count": len(policy.get("required_fields", [])),
-        "results": policy.get("results"),
-        "retention_defaults_days": policy.get("retention_defaults_days"),
-        "append_only": policy.get("append_only"),
-        "redact_before_persist": policy.get("redact_before_persist"),
-        "enabled": policy.get("enabled"),
-    }
-
-
-def _static(value: dict[str, Any]) -> Callable[[], dict[str, Any]]:
-    return lambda: dict(value)
-
-
-# ── Guardrail catalog ────────────────────────────────────────────────────────
-# One row per major guardrail. ``project`` yields the active value (may raise;
-# caught per entry). ``documented_default`` drives the feasible diff.
-
-_CatalogRow = tuple[
-    str,
-    str,
-    str,
-    str,
-    tuple[SourcePointer, ...],
-    Callable[[], dict[str, Any]] | None,
-    dict[str, Any] | None,
-]
-
-_CATALOG: tuple[_CatalogRow, ...] = (
-    (
-        "role_separation",
-        "Role separation and RBAC",
-        "role_separation",
-        "Author, reviewer, merger, and reconciler capabilities are disjoint and "
-        "role-exclusive; self-review and self-merge are always blocked. The "
-        "console RBAC model defaults to deny.",
-        (
-            SourcePointer("task capability map", "task_capability_map.py", "module"),
-            SourcePointer("role/namespace gate", "role_namespace_gate.py", "module"),
-            SourcePointer("console RBAC", "webui/console_authz.py", "module"),
-        ),
-        _project_role_separation,
-        {"default_decision": "deny", "execution_enabled": False},
-    ),
-    (
-        "lease_rules",
-        "Issue and PR lease lifecycle",
-        "lease_rules",
-        "Durable work is claimed through issue locks and control-plane leases "
-        "with freshness, expiry, and dead-session recovery; abandoned or stale "
-        "claims are reclaimed only through the sanctioned recovery path.",
-        (
-            SourcePointer("issue lock store", "issue_lock_store.py", "module"),
-            SourcePointer("branch cleanup guard", "branch_cleanup_guard.py", "module"),
-            SourcePointer("safety model §5", "docs/safety-model.md", "doc", "5-mutation-gating"),
-        ),
-        None,
-        None,
-    ),
-    (
-        "worktree_rules",
-        "Author worktree binding",
-        "worktree_rules",
-        "Author mutations require a validated worktree under branches/ derived "
-        "from the active issue lock; silent fallback to the stable control "
-        "checkout or master is forbidden (#618).",
-        (
-            SourcePointer("author worktree gate", "author_mutation_worktree.py", "module"),
-            SourcePointer("worktree bootstrap", "scripts/worktree-start", "script"),
-            SourcePointer("workflow scope guard", "workflow_scope_guard.py", "module"),
-        ),
-        None,
-        None,
-    ),
-    (
-        "merge_confirmation",
-        "Explicit merge confirmation",
-        "merge_confirmation",
-        "A merge fails closed unless the caller passes the exact confirmation "
-        "phrase for that PR; reviewing never implies merging.",
-        (
-            SourcePointer("merge path", "merge_pr.py", "module"),
-            SourcePointer("merge tool gate", "gitea_mcp_server.py", "module"),
-        ),
-        _static({"required_confirmation_format": "MERGE PR <n>", "auto_merge": False}),
-        {"auto_merge": False},
-    ),
-    (
-        "redaction",
-        "Secret redaction",
-        "redaction",
-        "Every console surface runs the shared gitea_audit pass then console "
-        "patterns before any payload, HTML, log line, or audit record leaves "
-        "the server; unredactable values fail closed to the placeholder.",
-        (
-            SourcePointer("console redaction", "webui/console_redaction.py", "module"),
-            SourcePointer("shared redaction", "gitea_audit.py", "module"),
-            SourcePointer("safety model §3", "docs/safety-model.md", "doc", "3-secret-redaction"),
-        ),
-        _project_redaction,
-        {"redact_before_persist": True},
-    ),
-    (
-        "contamination",
-        "Contamination containment",
-        "contamination",
-        "A session contaminated by a direct stable-branch push or a manual MCP "
-        "daemon kill is blocked from review, merge, close, and completion "
-        "mutations until cleared (reconciler-exempt).",
-        (
-            SourcePointer("contamination gates", "gitea_mcp_server.py", "module"),
-            SourcePointer("stable-branch audit", "workflow_scope_guard.py", "module"),
-        ),
-        None,
-        None,
-    ),
-    (
-        "allocator_policy",
-        "Work allocation policy",
-        "allocator_policy",
-        "Workers do not self-select exclusive work; the controller-owned "
-        "allocator ranks the complete queue by priority then PRs-before-issues "
-        "then ascending number, honoring dependency edges and foreign claims.",
-        (
-            SourcePointer("allocator", "gitea_mcp_server.py", "module"),
-            SourcePointer("safety model §5", "docs/safety-model.md", "doc", "5-mutation-gating"),
-        ),
-        _static(
-            {
-                "self_select_exclusive_work": False,
-                "ranking": "priority desc, PRs before issues, number asc",
-                "respects_dependency_edges": True,
-                "respects_foreign_claims": True,
-            }
-        ),
-        {"self_select_exclusive_work": False},
-    ),
-    (
-        "audit_logging",
-        "Audit logging",
-        "audit_logging",
-        "Console intent and authorization outcomes are recorded to an "
-        "append-only, redact-before-persist audit log; MCP mutations are "
-        "recorded by gitea_audit and correlated by request id.",
-        (
-            SourcePointer("console audit", "webui/console_audit.py", "module"),
-            SourcePointer("MCP audit", "gitea_audit.py", "module"),
-            SourcePointer("safety model §1", "docs/safety-model.md", "doc", "1-audit-logging-and-confirmation"),
-        ),
-        _project_audit,
-        {"append_only": True, "redact_before_persist": True},
-    ),
-    (
-        "mutation_gating",
-        "Mutation gating and master parity",
-        "mutation_gating",
-        "Mutations fail closed while the running server is stale relative to "
-        "master, and every mutation is preceded by identity and capability "
-        "resolution in a fixed pre-flight order.",
-        (
-            SourcePointer("mutation gate", "gitea_mcp_server.py", "module"),
-            SourcePointer("safety model §5", "docs/safety-model.md", "doc", "5-mutation-gating"),
-        ),
-        _static(
-            {
-                "stale_runtime_blocks_mutations": True,
-                "preflight_order": "whoami -> resolve_task_capability -> mutation",
-            }
-        ),
-        {"stale_runtime_blocks_mutations": True},
-    ),
-)
-
-
-def _build_entry(row: _CatalogRow) -> PolicyEntry:
-    key, title, category, summary, sources, project, documented_default = row
-    active: dict[str, Any] | None = None
-    error: str | None = None
-    if project is not None:
-        try:
-            active = project()
-        except Exception as exc:  # noqa: BLE001 — fail soft; never take the page down
-            active = None
-            error = f"active projection unavailable: {exc}"
-    diff = _diff_active_vs_default(active, documented_default)
-    return PolicyEntry(
-        key=key,
-        title=title,
-        category=category,
-        summary=summary,
-        sources=sources,
-        active=active,
-        documented_default=documented_default,
-        diff=diff,
-        error=error,
-    )
-
-
-def load_policy_inventory() -> PolicyInventorySnapshot:
-    """Build the read-only guardrail inventory. Never raises for one bad entry."""
-    entries: list[PolicyEntry] = []
-    build_errors: list[str] = []
-    for row in _CATALOG:
-        try:
-            entries.append(_build_entry(row))
-        except Exception as exc:  # noqa: BLE001 — one row must not break the rest
-            build_errors.append(f"{row[0]}: {exc}")
-    categories = tuple(dict.fromkeys(e.category for e in entries))
-    return PolicyInventorySnapshot(
-        schema_version=SCHEMA_VERSION,
-        read_only=True,
-        note=READ_ONLY_NOTE,
-        entries=tuple(entries),
-        categories=categories,
-        build_errors=tuple(build_errors),
-    )
-
-
-def snapshot_to_dict(snapshot: PolicyInventorySnapshot) -> dict[str, Any]:
-    """Serialize the snapshot, redacting the entire payload before it is emitted."""
-    payload = {
-        "schema_version": snapshot.schema_version,
-        "read_only": snapshot.read_only,
-        "note": snapshot.note,
-        "categories": list(snapshot.categories),
-        "entry_count": len(snapshot.entries),
-        "entries": [entry.to_dict() for entry in snapshot.entries],
-        "build_errors": list(snapshot.build_errors),
-    }
-    return console_redaction.redact_payload(payload)

webui/policy_views.py

@@ -1,145 +0,0 @@
-"""HTML views for the workflow policy and guardrail inventory (#646)."""
-
-from __future__ import annotations
-
-import html
-import json
-from typing import Any
-
-from webui import console_redaction
-from webui.policy_inventory import (
-    PolicyEntry,
-    PolicyInventorySnapshot,
-    SourcePointer,
-    snapshot_to_dict,
-)
-
-
-def _source_pointer(source: dict[str, Any] | SourcePointer) -> str:
-    if isinstance(source, dict):
-        label = str(source.get("label") or "")
-        path = str(source.get("path") or "")
-        anchor = source.get("anchor")
-        kind = str(source.get("kind") or "")
-    else:
-        label = source.label
-        path = source.path
-        anchor = source.anchor
-        kind = source.kind
-
-    if anchor:
-        path = f"{path}#{anchor}"
-    return (
-        f"<li>{html.escape(label)} — "
-        f"<code>{html.escape(path)}</code> "
-        f"<span class='muted'>({html.escape(kind)})</span></li>"
-    )
-
-
-def _active_block(entry: dict[str, Any] | PolicyEntry) -> str:
-    error = entry.get("error") if isinstance(entry, dict) else entry.error
-    active = entry.get("active") if isinstance(entry, dict) else entry.active
-    if error:
-        return (
-            "<p class='muted'><strong>Active value unavailable:</strong> "
-            f"{html.escape(error)}</p>"
-        )
-    if not active:
-        return "<p class='muted'>No live projection for this guardrail.</p>"
-    pretty = json.dumps(active, indent=2, sort_keys=True, default=str)
-    return f"<pre class='prompt-text'>{html.escape(pretty)}</pre>"
-
-
-def _diff_block(entry: dict[str, Any] | PolicyEntry) -> str:
-    diff = entry.get("diff") if isinstance(entry, dict) else entry.diff
-    documented_default = entry.get("documented_default") if isinstance(entry, dict) else entry.documented_default
-    if diff is None:
-        if documented_default is None:
-            return "<p class='muted'>Diff vs documented default: not feasible (no declared default).</p>"
-        return "<p class='muted'>Diff vs documented default: unavailable.</p>"
-    status = str(diff.get("status") if isinstance(diff, dict) else "unknown")
-    badge = "badge-claimed" if status == "matches_documented_default" else "badge-blocked"
-    rows = []
-    checked = diff.get("checked") if isinstance(diff, dict) else {}
-    if isinstance(checked, dict):
-        for key, cell in checked.items():
-            cell_dict = cell if isinstance(cell, dict) else {}
-            marker = "✓" if cell_dict.get("matches") else "✗"
-            rows.append(
-                "<tr>"
-                f"<td><code>{html.escape(str(key))}</code></td>"
-                f"<td><code>{html.escape(str(cell_dict.get('expected')))}</code></td>"
-                f"<td><code>{html.escape(str(cell_dict.get('observed')))}</code></td>"
-                f"<td>{marker}</td>"
-                "</tr>"
-            )
-    table = ""
-    if rows:
-        table = (
-            "<table class='detail'><thead><tr>"
-            "<th>Key</th><th>Documented</th><th>Active</th><th>Match</th>"
-            "</tr></thead><tbody>"
-            f"{''.join(rows)}</tbody></table>"
-        )
-    return (
-        f"<p class='meta'>Diff vs documented default: "
-        f"<span class='badge {badge}'>{html.escape(status)}</span></p>"
-        f"{table}"
-    )
-
-
-def _entry_card(entry: dict[str, Any] | PolicyEntry) -> str:
-    title = str(entry.get("title") if isinstance(entry, dict) else entry.title)
-    category = str(entry.get("category") if isinstance(entry, dict) else entry.category)
-    summary = str(entry.get("summary") if isinstance(entry, dict) else entry.summary)
-    sources_data = entry.get("sources", []) if isinstance(entry, dict) else entry.sources
-
-    sources = "".join(_source_pointer(s) for s in sources_data)
-    return (
-        "<div class='prompt-card'>"
-        f"<h3>{html.escape(title)} "
-        f"<span class='badge'>{html.escape(category)}</span></h3>"
-        f"<p>{html.escape(summary)}</p>"
-        "<p class='meta'><strong>Source pointers</strong></p>"
-        f"<ul>{sources}</ul>"
-        "<p class='meta'><strong>Active configuration</strong></p>"
-        f"{_active_block(entry)}"
-        f"{_diff_block(entry)}"
-        "</div>"
-    )
-
-
-def render_policy_page(snapshot: PolicyInventorySnapshot | dict[str, Any]) -> str:
-    if isinstance(snapshot, PolicyInventorySnapshot):
-        payload = snapshot_to_dict(snapshot)
-    elif isinstance(snapshot, dict):
-        payload = console_redaction.redact_payload(snapshot)
-    else:
-        payload = {}
-
-    categories_list = payload.get("categories") or []
-    categories = ", ".join(html.escape(str(c)) for c in categories_list) or "none"
-    entries_list = payload.get("entries") or []
-    cards = "".join(_entry_card(e) for e in entries_list)
-    build_errors = ""
-    errors_list = payload.get("build_errors") or []
-    if errors_list:
-        items = "".join(
-            f"<li>{html.escape(str(err))}</li>" for err in errors_list
-        )
-        build_errors = (
-            "<div class='stub'><p><strong>Some guardrails could not be built:"
-            f"</strong></p><ul>{items}</ul></div>"
-        )
-    note = str(payload.get("note") or "")
-    schema_version = payload.get("schema_version") or 1
-    return (
-        "<h2>Workflow policy &amp; guardrails</h2>"
-        f"<p class='muted'>{html.escape(note)}</p>"
-        f"<p class='meta'>Schema v{schema_version} · "
-        f"{len(entries_list)} guardrails · categories: {categories}</p>"
-        f"{build_errors}"
-        f"{cards}"
-        "<p class='muted'>This page is read-only. It reports enforced policy "
-        "and never edits or weakens a gate. Secret values are redacted.</p>"
-    )