Merge branch 'master' into feat/issue-628-autonomous-handoffs-orchestration

Address REQUEST_CHANGES on PR #795: stop claiming all 21 umbrella ACs
from a single unit-test module. Scope docs and cases to child issue #878
(CTH/format, ownership classify_skip, dependency edges). Does not close
umbrella #628.

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
+`/policy`, `/insights`) resolve to graceful read-only stub pages instead of
-backing views land in later child issues of #631 (the inventory surfaces are
+404s; their backing views land in later child issues of #631 (the inventory
-backed by #636). `/policy` is a live read-only surface (#646), not a stub.
+surfaces are backed by #636). Mutating methods on stub routes still fail closed
-Mutating methods on stub routes still fail closed with `read-only-mvp`.
+with `read-only-mvp`.
 
 ## System-health dashboard (#639)
 

tests/test_issue_628_orchestration.py

@@ -0,0 +1,199 @@
+"""Regression tests for #628 building blocks (child scope only).
+
+Honest scope: unit coverage of pre-existing APIs used by umbrella #628.
+This module does **not** implement or verify all 21 umbrella acceptance
+criteria, automatic handoff store/retrieve, multi-worker product wiring,
+or end-to-end orchestration.
+
+Covered building blocks:
+- CTH format / parse / assess (`format_cth_body`, `parse_cth_comment`,
+  `assess_cth_comment`)
+- Exclusive-ownership skip classification (`classify_skip` with
+  OWNERSHIP_FOREIGN vs OWNERSHIP_OWN)
+- Durable dependency edges (`upsert_dependency_edge` / list) and skip
+  when dependency_unmet
+- Edge state transition UNMET -> MET
+
+Parent umbrella remains #628; this slice is a scoped child issue only.
+"""
+
+import unittest
+from unittest.mock import MagicMock, patch
+import os
+import json
+import tempfile
+
+from canonical_thread_handoff import (
+    format_cth_body,
+    parse_cth_comment,
+    assess_cth_comment,
+)
+import dependency_graph
+from control_plane_db import ControlPlaneDB
+from allocator_service import (
+    WorkCandidate,
+    classify_skip,
+    ROLE_AUTHOR,
+    ROLE_REVIEWER,
+    ROLE_MERGER,
+    ROLE_RECONCILER,
+    OWNERSHIP_OWN,
+    OWNERSHIP_FOREIGN,
+)
+
+
+class TestIssue628Orchestration(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
+        self.db = ControlPlaneDB(self.db_path)
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_canonical_handoff_serialization_and_retrieval(self):
+        """Building block: format/parse/assess a CTH body (not full AC1/AC2 product path)."""
+        handoff = format_cth_body(
+            cth_type="Author Handoff",
+            status="completed",
+            next_owner="reviewer",
+            current_blocker="none",
+            decision="Implementation complete, tests passing",
+            proof="pytest tests/test_issue_628_orchestration.py passed",
+            next_action="Review PR and run reviewer pre-flight",
+            ready_to_paste_prompt="Review PR for child issue #878 (parent #628)",
+        )
+        self.assertIn("CTH: Author Handoff", handoff)
+
+        parsed = parse_cth_comment(handoff)
+        self.assertIsNotNone(parsed)
+        self.assertEqual(parsed["cth_type"], "Author Handoff")
+
+        assessment = assess_cth_comment(handoff)
+        self.assertFalse(assessment["block"])
+
+    def test_exclusive_task_unit_single_owner(self):
+        """Building block: classify_skip foreign vs own ownership."""
+        candidate = WorkCandidate(
+            kind="issue",
+            number=878,
+            title="Child #878 ownership classify candidate",
+            state="open",
+            labels=["status:in-progress"],
+            blocked=False,
+            dependency_unmet=False,
+        )
+        # Foreign ownership MUST be skipped
+        skip_foreign = classify_skip(
+            c=candidate,
+            role=ROLE_AUTHOR,
+            terminal_pr=None,
+            claim_ownership=OWNERSHIP_FOREIGN,
+        )
+        self.assertIsNotNone(skip_foreign)
+        self.assertIn("active lease", skip_foreign)
+
+        # Own/Self claim remains selectable for session resumption
+        skip_self = classify_skip(
+            c=candidate,
+            role=ROLE_AUTHOR,
+            terminal_pr=None,
+            claim_ownership=OWNERSHIP_OWN,
+        )
+        self.assertIsNone(skip_self)
+
+    def test_durable_dependency_graph_blocking(self):
+        """Building block: unmet dependency edge + classify_skip on dependency_unmet."""
+        # Upsert a blocking dependency edge between issue 878 and blocker 601
+        self.db.upsert_dependency_edge(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            source_kind="issue",
+            source_number=878,
+            target_kind="issue",
+            target_number=601,
+            edge_type=dependency_graph.EDGE_ISSUE_BLOCKED_BY_ISSUE,
+            state=dependency_graph.STATE_UNMET,
+            blocking_condition="Target issue #601 is not closed",
+            completion_condition="Target issue #601 is closed",
+            evidence={"source": "unit_test"},
+        )
+
+        edges = self.db.list_dependency_edges(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            source_kind="issue",
+            source_number=878,
+        )
+        self.assertEqual(len(edges), 1)
+        self.assertEqual(edges[0]["state"], "unmet")
+        self.assertEqual(edges[0]["target_number"], 601)
+
+        # When dependency is unmet, candidate is blocked from selection
+        candidate = WorkCandidate(
+            kind="issue",
+            number=878,
+            title="Blocked candidate",
+            state="open",
+            labels=[],
+            blocked=False,
+            dependency_unmet=True,
+            dependency_reason="issue#878 is blocked by unmet dependency issue#601",
+        )
+        skip_reason = classify_skip(
+            c=candidate,
+            role=ROLE_AUTHOR,
+            terminal_pr=None,
+            claim_ownership=OWNERSHIP_OWN,
+        )
+        self.assertIsNotNone(skip_reason)
+        self.assertIn("issue#601", skip_reason)
+
+    def test_dependency_completion_reevaluation(self):
+        """Building block: dependency edge state can transition UNMET -> MET."""
+        self.db.upsert_dependency_edge(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            source_kind="issue",
+            source_number=878,
+            target_kind="issue",
+            target_number=601,
+            edge_type=dependency_graph.EDGE_ISSUE_BLOCKED_BY_ISSUE,
+            state=dependency_graph.STATE_UNMET,
+            blocking_condition="Target issue #601 is open",
+            completion_condition="Target issue #601 is closed",
+            evidence={"source": "unit_test"},
+        )
+
+        # Mark edge as met upon target issue closure
+        self.db.upsert_dependency_edge(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            source_kind="issue",
+            source_number=878,
+            target_kind="issue",
+            target_number=601,
+            edge_type=dependency_graph.EDGE_ISSUE_BLOCKED_BY_ISSUE,
+            state=dependency_graph.STATE_MET,
+            blocking_condition="Target issue #601 is open",
+            completion_condition="Target issue #601 is closed",
+            evidence={"source": "target_closed_event"},
+        )
+
+        edges = self.db.list_dependency_edges(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            source_kind="issue",
+            source_number=878,
+        )
+        self.assertEqual(len(edges), 1)
+        self.assertEqual(edges[0]["state"], "met")
+
+
+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>"
-    )