Merge branch 'master' into feat/issue-646-policy-guardrail-visibility

Add pure post_restart_reconcile.reconcile_after_restart classifier with a
machine-readable completion proof covering service health, sessions, leases,
capabilities, worktrees, interrupted mutations (never auto-resumed),
duplicates, and queue state. Soft-depends on #660 checkpoints (skipped with
reason when the schema module is absent).

Wire read-only MCP tool gitea_reconcile_after_restart, boot-once hook via
gitea_assess_master_parity, log_only/enforce modes (mutation_hold), and docs.

Closes #662
Related: #655 #652 #653 #660 #661

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

docs/post-restart-reconcile.md

@@ -0,0 +1,56 @@
+# Post-restart MCP reconciliation (#662)
+
+After an MCP process restart, sessions, leases, capabilities, worktrees, and
+interrupted mutations must be reconciled before operators claim a clean runtime.
+This document describes the #662 completion-proof path.
+
+## Components
+
+| Piece | Where | Responsibility |
+|-------|-------|----------------|
+| `post_restart_reconcile.reconcile_after_restart` | `post_restart_reconcile.py` | Pure classification: inventory → completion proof DTO. No I/O. |
+| `RestartCompletionProof` | `post_restart_reconcile.py` | Machine-readable proof (`.as_dict()` is JSON-serializable). |
+| `gitea_reconcile_after_restart` | `gitea_mcp_server.py` | MCP tool: gathers inventory from the #613 control-plane DB + master-parity, classifies, returns the proof. Read-only. |
+| Boot hook | `gitea_assess_master_parity` | First post-restart parity probe also runs reconcile once (log-only by default). |
+
+## Dimensions
+
+The assessor classifies:
+
+- **service_health** — process healthy / parity mutation-safe
+- **clients** — connected client descriptors (optional inventory)
+- **sessions** — active session rows with dead owner pids are unresolved
+- **checkpoints** — soft-depends on #660; skipped with reason when schema absent
+- **leases** — live control-plane leases after restart
+- **capabilities** — master-parity / stale-runtime (#610)
+- **worktrees** — lease-bound paths missing on disk
+- **interrupted_mutations** — mutating lease phases or explicit pending inventory; **never auto-resumed**
+- **duplicates** — multiple live claims on the same work item
+- **queue** — allocator resume safety
+
+## Modes
+
+| Mode | Env / arg | Behavior |
+|------|-----------|----------|
+| `log_only` (default) | unset or `GITEA_POST_RESTART_RECONCILE_MODE=log_only` | Proof only; `mutation_hold=false` |
+| `enforce` | `GITEA_POST_RESTART_RECONCILE_MODE=enforce` or `mode=enforce` | Sets `mutation_hold=true` when overall status is degraded/failed or interrupted mutations remain |
+
+## Follow-up issues
+
+Unresolved dimensions produce `proposed_follow_ups` entries suitable for durable
+Gitea issues. The MCP tool **does not create** those issues in v1 (rollout is
+log-only first). Controllers may file them from the proof payload.
+
+## Links
+
+- Umbrella: #655
+- Vision: #652 · Roadmap: #653
+- Checkpoint schema: #660 (soft dependency)
+- Drain proof: #661 (soft)
+- This issue: #662
+
+## Non-goals
+
+- HA multi-instance failover
+- Automatic silent mutation replay
+- Implementing the #660 checkpoint schema itself

docs/webui-local-dev.md

@@ -66,6 +66,8 @@ 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) |
@@ -78,7 +80,6 @@ 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
@@ -239,12 +240,25 @@ 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** (placeholder), and **Insights**
+Projects, Inventory, Timeline, Policy** (live via #646), and **Insights**
 (placeholder). Live views and Phase 1 placeholders (`stub`) are declared in one
 place so the layout and the route table cannot drift.
 
@@ -254,10 +268,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`,
-`/policy`, `/insights`) resolve to graceful read-only stub pages instead of
+`/insights`) resolve to graceful read-only stub pages instead of 404s; their
-404s; their backing views land in later child issues of #631 (the inventory
+backing views land in later child issues of #631 (the inventory surfaces are
-surfaces are backed by #636). Mutating methods on stub routes still fail closed
+backed by #636). `/policy` is a live read-only surface (#646), not a stub.
-with `read-only-mvp`.
+Mutating methods on stub routes still fail closed with `read-only-mvp`.
 
 ## System-health dashboard (#639)
 

gitea_mcp_server.py

@@ -18022,6 +18022,17 @@ def gitea_assess_master_parity(
     }
     if parity["restart_required"] and enforced:
         out["report"] = master_parity_gate.parity_report(parity)
+    # #662 AC1: first post-restart master-parity probe also runs the boot
+    # reconcile once (log-only by default; never raises).
+    boot_proof = _ensure_boot_post_restart_reconcile()
+    if boot_proof is not None:
+        out["post_restart_reconcile"] = {
+            "reconcile_id": boot_proof.get("reconcile_id"),
+            "overall_status": boot_proof.get("overall_status"),
+            "mutation_hold": boot_proof.get("mutation_hold"),
+            "unresolved_count": boot_proof.get("unresolved_count"),
+            "mode": boot_proof.get("mode"),
+        }
     return out
 
 
@@ -22470,6 +22481,249 @@ def gitea_request_mcp_restart(
     return payload
 
 
+# --- #662 post-restart reconciliation ---------------------------------------
+
+_POST_RESTART_LAST_PROOF: dict | None = None
+_POST_RESTART_BOOT_RAN = False
+
+
+def _post_restart_reconcile_mode() -> str:
+    """Return log_only (default) or enforce from process environment."""
+    raw = (os.environ.get("GITEA_POST_RESTART_RECONCILE_MODE") or "").strip().lower()
+    if raw in {"enforce", "enforced", "hold"}:
+        return "enforce"
+    return "log_only"
+
+
+def _gather_post_restart_inventory(
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    limit: int = 200,
+) -> dict:
+    """Gather live control-plane facts for post-restart reconcile (#662).
+
+    Best-effort and fail-closed: any inventory section that cannot be read is
+    recorded on ``incomplete_reasons`` and ``inventory_complete`` is cleared.
+    Never mutates Gitea or the control-plane DB.
+    """
+    import master_parity_gate
+    import post_restart_reconcile as prr
+
+    inventory_complete = True
+    incomplete_reasons: list[str] = []
+    sessions: list[dict] = []
+    leases: list[dict] = []
+    worktree_bindings: list[dict] = []
+
+    db, db_errs = _control_plane_db_or_error()
+    if db is None:
+        inventory_complete = False
+        incomplete_reasons.extend(
+            db_errs or ["control-plane DB unavailable; cannot reconcile after restart"]
+        )
+    else:
+        try:
+            sessions = db.list_sessions(statuses=("active",), limit=max(1, int(limit)))
+        except Exception as exc:  # noqa: BLE001
+            inventory_complete = False
+            incomplete_reasons.append(
+                f"session inventory failed: {_redact(str(exc))}"
+            )
+        try:
+            lease_result = lease_lifecycle.list_active_leases(
+                db,
+                remote=remote if remote in REMOTES else remote,
+                org=org,
+                repo=repo,
+                role=None,
+                include_non_active=True,
+                limit=max(1, int(limit)),
+            )
+            leases = list(lease_result.get("leases") or [])
+        except Exception as exc:  # noqa: BLE001
+            inventory_complete = False
+            incomplete_reasons.append(
+                f"lease inventory failed: {_redact(str(exc))}"
+            )
+            leases = []
+
+        # Derive worktree bindings from live leases that carry a path.
+        for row in leases:
+            wt = (row.get("worktree_path") or "").strip()
+            if not wt:
+                continue
+            exists = bool(lease_lifecycle.worktree_exists(wt))
+            worktree_bindings.append(
+                {
+                    "path": wt,
+                    "exists": exists,
+                    "missing": not exists,
+                    "lease_id": row.get("lease_id"),
+                    "session_id": row.get("session_id"),
+                    "work_kind": row.get("work_kind"),
+                    "work_number": row.get("work_number"),
+                }
+            )
+
+    # Capability / master-parity dimension (code parity after restart).
+    try:
+        parity = _current_master_parity()
+        capabilities = {
+            "stale": bool(parity.get("stale")),
+            "startup_head": parity.get("startup_head"),
+            "current_head": parity.get("current_head"),
+            "in_parity": parity.get("in_parity"),
+            "mutation_safe": parity.get("mutation_safe"),
+        }
+        service_health = {
+            "healthy": bool(parity.get("mutation_safe", not parity.get("stale"))),
+            "parity_summary": master_parity_gate.format_parity(parity),
+        }
+        boot_head = parity.get("startup_head") or _process_boot_head_sha
+        current_head = parity.get("current_head")
+    except Exception as exc:  # noqa: BLE001
+        inventory_complete = False
+        incomplete_reasons.append(
+            f"master-parity inventory failed: {_redact(str(exc))}"
+        )
+        capabilities = {"stale": None}
+        service_health = {"healthy": None, "error": "parity assessment failed"}
+        boot_head = _process_boot_head_sha
+        current_head = None
+
+    # #660 soft dependency: checkpoint schema not landed → skip dimension.
+    checkpoints_available = False
+    try:
+        import importlib
+
+        importlib.import_module("session_checkpoint_schema")
+        checkpoints_available = True
+    except Exception:  # noqa: BLE001
+        checkpoints_available = False
+
+    return {
+        "inventory_complete": inventory_complete,
+        "incomplete_reasons": incomplete_reasons,
+        "service_health": service_health,
+        "clients": [],  # client transport inventory is host/IDE-owned
+        "sessions": sessions,
+        "leases": leases,
+        "checkpoints_available": checkpoints_available,
+        "checkpoints": [] if checkpoints_available else None,
+        "worktree_bindings": worktree_bindings,
+        "pending_mutations": [],
+        "capabilities": capabilities,
+        "boot_head_sha": boot_head,
+        "current_head_sha": current_head,
+        "queue_state": {"safe_to_resume": inventory_complete},
+        "reconcile_version": prr.RECONCILE_VERSION,
+    }
+
+
+def _run_post_restart_reconcile(
+    *,
+    remote: str = "prgs",
+    org: str | None = None,
+    repo: str | None = None,
+    mode: str | None = None,
+    limit: int = 200,
+) -> dict:
+    """Gather + classify post-restart state; cache the latest proof (#662)."""
+    global _POST_RESTART_LAST_PROOF, _POST_RESTART_BOOT_RAN
+    import post_restart_reconcile as prr
+
+    try:
+        _h, o, r = _resolve(remote, None, org, repo)
+    except ValueError as exc:
+        return {
+            "success": False,
+            "read_only": True,
+            "reasons": [str(exc)],
+        }
+
+    inventory = _gather_post_restart_inventory(
+        remote=remote, org=o, repo=r, limit=limit
+    )
+    proof = prr.reconcile_after_restart(
+        inventory,
+        mode=mode or _post_restart_reconcile_mode(),
+    )
+    payload = proof.as_dict()
+    payload["success"] = True
+    payload["read_only"] = True
+    payload["remote"] = remote
+    payload["org"] = o
+    payload["repo"] = r
+    payload["follow_up_create_supported"] = False
+    payload["follow_up_create_note"] = (
+        "proposed_follow_ups lists durable issues the apply path may create; "
+        "this tool never creates them (log-only by default, #662 rollout)"
+    )
+    _POST_RESTART_LAST_PROOF = payload
+    _POST_RESTART_BOOT_RAN = True
+    return payload
+
+
+def _ensure_boot_post_restart_reconcile() -> dict | None:
+    """Run post-restart reconcile once per process (boot hook, #662 AC1)."""
+    global _POST_RESTART_BOOT_RAN
+    if _POST_RESTART_BOOT_RAN:
+        return _POST_RESTART_LAST_PROOF
+    # Best-effort: never raise from the boot hook.
+    try:
+        return _run_post_restart_reconcile()
+    except Exception:  # noqa: BLE001
+        _POST_RESTART_BOOT_RAN = True
+        return None
+
+
+@mcp.tool()
+def gitea_reconcile_after_restart(
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    mode: str | None = None,
+    limit: int = 200,
+) -> dict:
+    """Run post-restart MCP reconciliation and return a completion proof (#662).
+
+    Gathers live control-plane sessions, leases, worktree bindings, and
+    master-parity evidence, then classifies them with the pure
+    ``post_restart_reconcile.reconcile_after_restart`` assessor. Returns a
+    machine-readable completion proof listing resolved / unresolved dimensions
+    and proposed durable follow-up issues.
+
+    Read-only by design: never restarts MCP, never auto-resumes write
+    mutations, and never creates Gitea issues (those are a separate apply
+    path). Default mode is ``log_only``; set
+    ``GITEA_POST_RESTART_RECONCILE_MODE=enforce`` (or pass ``mode='enforce'``)
+    to set ``mutation_hold`` when anything remains unresolved.
+
+    Soft-depends on #660 for session checkpoints: when the checkpoint schema
+    module is absent the checkpoints dimension is ``skipped`` with an explicit
+    reason rather than inventing a schema.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "read_only": True,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    return _run_post_restart_reconcile(
+        remote=remote,
+        org=org,
+        repo=repo,
+        mode=mode,
+        limit=limit,
+    )
+
+
 @mcp.tool()
 def gitea_inspect_workflow_lease(
     lease_id: str,

post_restart_reconcile.py

@@ -0,0 +1,791 @@
+"""Post-restart MCP reconciliation and completion proof (#662).
+
+After an MCP process restart, sessions, leases, capabilities, worktrees, and
+interrupted mutations are not systematically reconciled; operators rebuild
+context from chat. This module is the pure classification core of the
+post-restart reconcile path.
+
+Design rules (mirrors ``restart_coordinator`` / ``workflow_dashboard``):
+
+* **Pure classification.** :func:`reconcile_after_restart` takes an already
+  gathered inventory and returns a structured *completion proof*. It never
+  touches the network, the filesystem, or a live process, so multi-session
+  fixtures can drive every branch in unit tests.
+* **Fail closed.** Incomplete inventory never reports overall ``complete``.
+  Ambiguous interrupted mutations are ``unresolved`` (never silently resumed).
+* **No blind write resume.** The proof never authorizes replaying a mutation;
+  it only classifies evidence and names follow-up work.
+* **#660 soft dependency.** When durable session checkpoints are not present
+  in the inventory, the checkpoint dimension is ``skipped`` with an explicit
+  reason rather than inventing a schema (#660 lands separately).
+* **Log-only then enforce.** Default mode is ``log_only``. ``enforce`` sets
+  ``mutation_hold`` when anything remains unresolved so callers can block
+  write ops until reconcile is complete or degraded mode is documented.
+
+The single sanctioned gather+classify entry point is the MCP tool
+``gitea_reconcile_after_restart`` (read-only inventory gather + pure classify).
+Creating durable follow-up Gitea issues from unresolved items is an explicit
+apply step outside this pure module.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any, Mapping, Sequence
+from uuid import uuid4
+
+import lease_lifecycle
+
+RECONCILE_VERSION = "1.0.0-issue-662"
+SCHEMA_VERSION = 1
+
+# Overall proof statuses.
+STATUS_COMPLETE = "complete"
+STATUS_DEGRADED = "degraded"
+STATUS_FAILED = "failed"
+
+# Per-dimension item statuses.
+ITEM_RESOLVED = "resolved"
+ITEM_UNRESOLVED = "unresolved"
+ITEM_DEGRADED = "degraded"
+ITEM_SKIPPED = "skipped"
+
+# Modes.
+MODE_LOG_ONLY = "log_only"
+MODE_ENFORCE = "enforce"
+
+# Lease / session phases that imply a write critical section was in flight.
+MUTATING_PHASES = frozenset(
+    {
+        "implementing",
+        "publishing",
+        "merging",
+        "reviewing",
+        "committing",
+        "pushing",
+        "closing",
+        "mutating",
+        "critical_section",
+    }
+)
+
+# Dimensions the acceptance criteria require.
+DIM_SERVICE_HEALTH = "service_health"
+DIM_CLIENTS = "clients"
+DIM_SESSIONS = "sessions"
+DIM_CHECKPOINTS = "checkpoints"
+DIM_LEASES = "leases"
+DIM_CAPABILITIES = "capabilities"
+DIM_WORKTREES = "worktrees"
+DIM_MUTATIONS = "interrupted_mutations"
+DIM_DUPLICATES = "duplicates"
+DIM_QUEUE = "queue"
+
+
+def _utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _ts(dt: datetime) -> str:
+    return dt.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+@dataclass(frozen=True)
+class ReconcileItem:
+    """One dimension of the post-restart reconcile report."""
+
+    dimension: str
+    status: str
+    summary: str
+    details: dict[str, Any] = field(default_factory=dict)
+    follow_up_required: bool = False
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "dimension": self.dimension,
+            "status": self.status,
+            "summary": self.summary,
+            "details": dict(self.details),
+            "follow_up_required": self.follow_up_required,
+        }
+
+
+@dataclass(frozen=True)
+class FollowUpIssue:
+    """A durable follow-up issue the apply path may create for unresolved work."""
+
+    title: str
+    body: str
+    dimension: str
+    severity: str = "high"
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "title": self.title,
+            "body": self.body,
+            "dimension": self.dimension,
+            "severity": self.severity,
+        }
+
+
+@dataclass(frozen=True)
+class RestartCompletionProof:
+    """Machine-readable post-restart completion proof (#662 AC2)."""
+
+    schema_version: int
+    reconcile_version: str
+    reconcile_id: str
+    started_at: str
+    finished_at: str
+    boot_head_sha: str | None
+    current_head_sha: str | None
+    inventory_complete: bool
+    incomplete_reasons: tuple[str, ...]
+    mode: str
+    mutation_hold: bool
+    overall_status: str
+    items: tuple[ReconcileItem, ...]
+    proposed_follow_ups: tuple[FollowUpIssue, ...]
+    resolved_count: int
+    unresolved_count: int
+    skipped_count: int
+    note: str
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "schema_version": self.schema_version,
+            "reconcile_version": self.reconcile_version,
+            "reconcile_id": self.reconcile_id,
+            "started_at": self.started_at,
+            "finished_at": self.finished_at,
+            "boot_head_sha": self.boot_head_sha,
+            "current_head_sha": self.current_head_sha,
+            "inventory_complete": self.inventory_complete,
+            "incomplete_reasons": list(self.incomplete_reasons),
+            "mode": self.mode,
+            "mutation_hold": self.mutation_hold,
+            "overall_status": self.overall_status,
+            "items": [i.as_dict() for i in self.items],
+            "proposed_follow_ups": [f.as_dict() for f in self.proposed_follow_ups],
+            "resolved_count": self.resolved_count,
+            "unresolved_count": self.unresolved_count,
+            "skipped_count": self.skipped_count,
+            "note": self.note,
+            "links": {
+                "umbrella": 655,
+                "vision": 652,
+                "roadmap": 653,
+                "issue": 662,
+                "checkpoint_schema": 660,
+                "drain_proof": 661,
+            },
+        }
+
+
+def _item(
+    dimension: str,
+    status: str,
+    summary: str,
+    *,
+    details: dict[str, Any] | None = None,
+    follow_up: bool = False,
+) -> ReconcileItem:
+    return ReconcileItem(
+        dimension=dimension,
+        status=status,
+        summary=summary,
+        details=dict(details or {}),
+        follow_up_required=follow_up,
+    )
+
+
+def _lease_freshness(lease: Mapping[str, Any]) -> str:
+    fr = lease.get("freshness")
+    if isinstance(fr, Mapping):
+        return str(fr.get("freshness") or fr.get("status") or "unknown")
+    if isinstance(fr, str):
+        return fr
+    # Fall back to pure classifier when raw lease rows are supplied.
+    try:
+        return str(lease_lifecycle.classify_lease_freshness(dict(lease)).get("freshness") or "unknown")
+    except Exception:  # noqa: BLE001 - pure path must not raise on bad rows
+        return "unknown"
+
+
+def _is_live_freshness(freshness: str) -> bool:
+    return freshness in {"active", "live", "fresh"}
+
+
+def _is_mutating_phase(phase: str | None) -> bool:
+    p = (phase or "").strip().lower()
+    if not p:
+        return False
+    if p in MUTATING_PHASES:
+        return True
+    # Soft match for compound phases like "author_implementing".
+    return any(token in p for token in MUTATING_PHASES)
+
+
+def _detect_interrupted_mutations(
+    leases: Sequence[Mapping[str, Any]],
+    pending_mutations: Sequence[Mapping[str, Any]],
+) -> list[dict[str, Any]]:
+    """Return interrupted-mutation evidence (never auto-resumes writes)."""
+    found: list[dict[str, Any]] = []
+
+    for raw in pending_mutations or ():
+        if not isinstance(raw, Mapping):
+            continue
+        found.append(
+            {
+                "source": "pending_mutation_inventory",
+                "status": "unresolved",
+                "phase": raw.get("phase"),
+                "session_id": raw.get("session_id"),
+                "work_kind": raw.get("work_kind") or raw.get("kind"),
+                "work_number": raw.get("work_number") or raw.get("number"),
+                "reason": raw.get("reason")
+                or "pending mutation recorded across process restart",
+                "resume_allowed": False,
+            }
+        )
+
+    for lease in leases or ():
+        if not isinstance(lease, Mapping):
+            continue
+        phase = lease.get("phase")
+        freshness = _lease_freshness(lease)
+        if not _is_mutating_phase(str(phase) if phase is not None else None):
+            continue
+        # A mutating phase whose owner is not live is interrupted.
+        if _is_live_freshness(freshness):
+            # Still live after restart is itself surprising — flag for review.
+            found.append(
+                {
+                    "source": "lease_mutating_phase",
+                    "status": "unresolved",
+                    "phase": phase,
+                    "freshness": freshness,
+                    "lease_id": lease.get("lease_id"),
+                    "session_id": lease.get("session_id"),
+                    "work_kind": lease.get("work_kind"),
+                    "work_number": lease.get("work_number"),
+                    "worktree_path": lease.get("worktree_path"),
+                    "reason": (
+                        "mutating lease phase still classified live after restart; "
+                        "do not auto-resume writes"
+                    ),
+                    "resume_allowed": False,
+                }
+            )
+        else:
+            found.append(
+                {
+                    "source": "lease_mutating_phase",
+                    "status": "unresolved",
+                    "phase": phase,
+                    "freshness": freshness,
+                    "lease_id": lease.get("lease_id"),
+                    "session_id": lease.get("session_id"),
+                    "work_kind": lease.get("work_kind"),
+                    "work_number": lease.get("work_number"),
+                    "worktree_path": lease.get("worktree_path"),
+                    "reason": (
+                        f"mutating lease phase '{phase}' with non-live freshness "
+                        f"'{freshness}' — interrupted by restart"
+                    ),
+                    "resume_allowed": False,
+                }
+            )
+    return found
+
+
+def _detect_duplicate_work(
+    leases: Sequence[Mapping[str, Any]],
+) -> list[dict[str, Any]]:
+    """Surface duplicate live claims on the same work item."""
+    by_work: dict[tuple[Any, Any], list[Mapping[str, Any]]] = {}
+    for lease in leases or ():
+        if not isinstance(lease, Mapping):
+            continue
+        if not _is_live_freshness(_lease_freshness(lease)):
+            continue
+        key = (lease.get("work_kind"), lease.get("work_number"))
+        if key[0] is None or key[1] is None:
+            continue
+        by_work.setdefault(key, []).append(lease)
+
+    dups: list[dict[str, Any]] = []
+    for (kind, number), rows in sorted(by_work.items(), key=lambda kv: str(kv[0])):
+        if len(rows) < 2:
+            continue
+        dups.append(
+            {
+                "work_kind": kind,
+                "work_number": number,
+                "claim_count": len(rows),
+                "session_ids": [r.get("session_id") for r in rows],
+                "lease_ids": [r.get("lease_id") for r in rows],
+            }
+        )
+    return dups
+
+
+def _follow_up_for_item(item: ReconcileItem) -> FollowUpIssue | None:
+    if not item.follow_up_required:
+        return None
+    title = f"[post-restart] unresolved {item.dimension} after MCP restart"
+    body = (
+        f"## Post-restart reconcile follow-up (#662)\n\n"
+        f"**Dimension:** `{item.dimension}`\n"
+        f"**Status:** `{item.status}`\n"
+        f"**Summary:** {item.summary}\n\n"
+        f"```json\n{item.details!r}\n```\n\n"
+        f"Parent umbrella: #655 · Vision: #652 · Roadmap: #653 · Reconcile: #662\n"
+        f"Do **not** auto-resume write mutations; reconcile evidence first.\n"
+    )
+    return FollowUpIssue(
+        title=title,
+        body=body,
+        dimension=item.dimension,
+        severity="high" if item.dimension == DIM_MUTATIONS else "medium",
+    )
+
+
+def reconcile_after_restart(
+    inventory: Mapping[str, Any],
+    *,
+    now: datetime | None = None,
+    mode: str = MODE_LOG_ONLY,
+    reconcile_id: str | None = None,
+) -> RestartCompletionProof:
+    """Classify a post-restart inventory into a completion proof (#662).
+
+    Parameters
+    ----------
+    inventory:
+        Gathered facts. Expected keys (all optional except completeness):
+
+        * ``inventory_complete`` (bool) — fail closed when false
+        * ``incomplete_reasons`` (list[str])
+        * ``service_health`` (dict with ``healthy`` bool)
+        * ``clients`` (list) — connected client descriptors
+        * ``sessions`` (list)
+        * ``leases`` (list, optionally with ``freshness``)
+        * ``checkpoints`` (list | None) — durable session checkpoints (#660)
+        * ``checkpoints_available`` (bool) — False when #660 schema absent
+        * ``worktree_bindings`` (list)
+        * ``pending_mutations`` (list) — explicit interrupted-mutation evidence
+        * ``capabilities`` (dict with optional ``stale`` / heads)
+        * ``boot_head_sha`` / ``current_head_sha``
+        * ``queue_state`` (dict)
+    mode:
+        ``log_only`` (default) or ``enforce`` (sets mutation_hold on unresolved).
+    """
+    started = now or _utc_now()
+    mode_norm = (mode or MODE_LOG_ONLY).strip().lower()
+    if mode_norm not in {MODE_LOG_ONLY, MODE_ENFORCE}:
+        mode_norm = MODE_LOG_ONLY
+
+    inventory_complete = bool(inventory.get("inventory_complete", False))
+    incomplete_reasons = tuple(
+        str(r) for r in (inventory.get("incomplete_reasons") or []) if str(r).strip()
+    )
+
+    items: list[ReconcileItem] = []
+
+    # --- service health -------------------------------------------------
+    health = inventory.get("service_health") or {}
+    if not isinstance(health, Mapping):
+        health = {}
+    if not inventory_complete and "service_health" not in inventory:
+        items.append(
+            _item(
+                DIM_SERVICE_HEALTH,
+                ITEM_UNRESOLVED,
+                "service health unknown because inventory is incomplete",
+                details={"inventory_complete": False},
+                follow_up=True,
+            )
+        )
+    elif health.get("healthy") is True:
+        items.append(
+            _item(
+                DIM_SERVICE_HEALTH,
+                ITEM_RESOLVED,
+                "service health verified",
+                details=dict(health),
+            )
+        )
+    elif health.get("healthy") is False:
+        items.append(
+            _item(
+                DIM_SERVICE_HEALTH,
+                ITEM_UNRESOLVED,
+                "service health check failed",
+                details=dict(health),
+                follow_up=True,
+            )
+        )
+    else:
+        items.append(
+            _item(
+                DIM_SERVICE_HEALTH,
+                ITEM_DEGRADED,
+                "service health not reported; treating as degraded",
+                details=dict(health),
+                follow_up=True,
+            )
+        )
+
+    # --- clients --------------------------------------------------------
+    clients = list(inventory.get("clients") or [])
+    disconnected = [
+        c
+        for c in clients
+        if isinstance(c, Mapping) and c.get("connected") is False
+    ]
+    if "clients" not in inventory:
+        items.append(
+            _item(
+                DIM_CLIENTS,
+                ITEM_SKIPPED,
+                "client inventory not supplied",
+                details={},
+            )
+        )
+    elif disconnected:
+        items.append(
+            _item(
+                DIM_CLIENTS,
+                ITEM_UNRESOLVED,
+                f"{len(disconnected)} disconnected client(s) need reconnect",
+                details={"disconnected": disconnected, "total": len(clients)},
+                follow_up=True,
+            )
+        )
+    else:
+        items.append(
+            _item(
+                DIM_CLIENTS,
+                ITEM_RESOLVED,
+                f"{len(clients)} client(s) accounted for",
+                details={"total": len(clients)},
+            )
+        )
+
+    # --- sessions -------------------------------------------------------
+    sessions = [s for s in (inventory.get("sessions") or []) if isinstance(s, Mapping)]
+    orphan_sessions = [
+        s
+        for s in sessions
+        if str(s.get("status") or "").lower() == "active"
+        and s.get("pid") is not None
+        and not lease_lifecycle.is_process_alive(s.get("pid"))
+    ]
+    if orphan_sessions:
+        items.append(
+            _item(
+                DIM_SESSIONS,
+                ITEM_UNRESOLVED,
+                f"{len(orphan_sessions)} active session row(s) with dead owner pid",
+                details={
+                    "orphan_session_ids": [s.get("session_id") for s in orphan_sessions],
+                    "total_sessions": len(sessions),
+                },
+                follow_up=True,
+            )
+        )
+    else:
+        items.append(
+            _item(
+                DIM_SESSIONS,
+                ITEM_RESOLVED,
+                f"{len(sessions)} session row(s) reconciled (no dead-pid orphans)",
+                details={"total_sessions": len(sessions)},
+            )
+        )
+
+    # --- checkpoints (#660 soft) ----------------------------------------
+    checkpoints_available = inventory.get("checkpoints_available")
+    checkpoints = inventory.get("checkpoints")
+    if checkpoints_available is False or (
+        checkpoints is None and "checkpoints" not in inventory
+    ):
+        items.append(
+            _item(
+                DIM_CHECKPOINTS,
+                ITEM_SKIPPED,
+                "durable session checkpoint schema not available yet (#660)",
+                details={"depends_on": 660},
+            )
+        )
+    else:
+        cp_list = [c for c in (checkpoints or []) if isinstance(c, Mapping)]
+        stale_cp = [c for c in cp_list if c.get("stale") or c.get("invalid")]
+        if stale_cp:
+            items.append(
+                _item(
+                    DIM_CHECKPOINTS,
+                    ITEM_UNRESOLVED,
+                    f"{len(stale_cp)} checkpoint(s) invalid or stale vs live state",
+                    details={"stale_count": len(stale_cp), "total": len(cp_list)},
+                    follow_up=True,
+                )
+            )
+        else:
+            items.append(
+                _item(
+                    DIM_CHECKPOINTS,
+                    ITEM_RESOLVED,
+                    f"{len(cp_list)} checkpoint(s) consistent with live state",
+                    details={"total": len(cp_list)},
+                )
+            )
+
+    # --- leases / locks -------------------------------------------------
+    leases = [L for L in (inventory.get("leases") or []) if isinstance(L, Mapping)]
+    live_leases = [L for L in leases if _is_live_freshness(_lease_freshness(L))]
+    items.append(
+        _item(
+            DIM_LEASES,
+            ITEM_RESOLVED if inventory_complete else ITEM_DEGRADED,
+            f"{len(live_leases)} live lease(s) of {len(leases)} inventoried",
+            details={
+                "live_count": len(live_leases),
+                "total": len(leases),
+                "live_lease_ids": [L.get("lease_id") for L in live_leases],
+            },
+            follow_up=not inventory_complete,
+        )
+    )
+
+    # --- capabilities / stale runtime -----------------------------------
+    caps = inventory.get("capabilities") or {}
+    if not isinstance(caps, Mapping):
+        caps = {}
+    if caps.get("stale") is True:
+        items.append(
+            _item(
+                DIM_CAPABILITIES,
+                ITEM_UNRESOLVED,
+                "runtime code is stale vs on-disk master; restart did not reach parity",
+                details=dict(caps),
+                follow_up=True,
+            )
+        )
+    else:
+        items.append(
+            _item(
+                DIM_CAPABILITIES,
+                ITEM_RESOLVED,
+                "capability/runtime parity acceptable",
+                details=dict(caps) if caps else {"stale": False},
+            )
+        )
+
+    # --- worktrees ------------------------------------------------------
+    bindings = [
+        b for b in (inventory.get("worktree_bindings") or []) if isinstance(b, Mapping)
+    ]
+    missing_wt = [
+        b
+        for b in bindings
+        if b.get("missing") is True or b.get("exists") is False
+    ]
+    if "worktree_bindings" not in inventory:
+        items.append(
+            _item(
+                DIM_WORKTREES,
+                ITEM_SKIPPED,
+                "worktree binding inventory not supplied",
+            )
+        )
+    elif missing_wt:
+        items.append(
+            _item(
+                DIM_WORKTREES,
+                ITEM_UNRESOLVED,
+                f"{len(missing_wt)} worktree binding(s) missing on disk",
+                details={"missing": missing_wt, "total": len(bindings)},
+                follow_up=True,
+            )
+        )
+    else:
+        items.append(
+            _item(
+                DIM_WORKTREES,
+                ITEM_RESOLVED,
+                f"{len(bindings)} worktree binding(s) present",
+                details={"total": len(bindings)},
+            )
+        )
+
+    # --- interrupted mutations (AC4) ------------------------------------
+    pending = [
+        m
+        for m in (inventory.get("pending_mutations") or [])
+        if isinstance(m, Mapping)
+    ]
+    interrupted = _detect_interrupted_mutations(leases, pending)
+    if interrupted:
+        items.append(
+            _item(
+                DIM_MUTATIONS,
+                ITEM_UNRESOLVED,
+                f"{len(interrupted)} interrupted mutation(s); write resume forbidden",
+                details={"interrupted": interrupted},
+                follow_up=True,
+            )
+        )
+    else:
+        items.append(
+            _item(
+                DIM_MUTATIONS,
+                ITEM_RESOLVED,
+                "no interrupted mutations detected",
+                details={"interrupted": []},
+            )
+        )
+
+    # --- duplicates -----------------------------------------------------
+    dups = _detect_duplicate_work(leases)
+    if dups:
+        items.append(
+            _item(
+                DIM_DUPLICATES,
+                ITEM_UNRESOLVED,
+                f"{len(dups)} work item(s) have multiple live claims",
+                details={"duplicates": dups},
+                follow_up=True,
+            )
+        )
+    else:
+        items.append(
+            _item(
+                DIM_DUPLICATES,
+                ITEM_RESOLVED,
+                "no duplicate live claims detected",
+                details={"duplicates": []},
+            )
+        )
+
+    # --- queue ----------------------------------------------------------
+    queue = inventory.get("queue_state")
+    if queue is None:
+        items.append(
+            _item(
+                DIM_QUEUE,
+                ITEM_SKIPPED,
+                "allocator queue state not supplied",
+            )
+        )
+    elif isinstance(queue, Mapping) and queue.get("safe_to_resume") is False:
+        items.append(
+            _item(
+                DIM_QUEUE,
+                ITEM_UNRESOLVED,
+                "allocator queue not safe to resume",
+                details=dict(queue),
+                follow_up=True,
+            )
+        )
+    else:
+        items.append(
+            _item(
+                DIM_QUEUE,
+                ITEM_RESOLVED,
+                "allocator queue state acceptable",
+                details=dict(queue) if isinstance(queue, Mapping) else {},
+            )
+        )
+
+    # Incomplete inventory always degrades the whole proof.
+    if not inventory_complete:
+        # Ensure at least one follow-up names the incomplete inventory.
+        items.append(
+            _item(
+                "inventory",
+                ITEM_UNRESOLVED,
+                "control-plane inventory incomplete; reconcile cannot claim success",
+                details={"reasons": list(incomplete_reasons)},
+                follow_up=True,
+            )
+        )
+
+    resolved = sum(1 for i in items if i.status == ITEM_RESOLVED)
+    unresolved = sum(1 for i in items if i.status in {ITEM_UNRESOLVED, ITEM_DEGRADED})
+    skipped = sum(1 for i in items if i.status == ITEM_SKIPPED)
+
+    if not inventory_complete or any(i.status == ITEM_UNRESOLVED for i in items):
+        if any(i.status == ITEM_UNRESOLVED for i in items) and inventory_complete:
+            overall = STATUS_DEGRADED
+        elif not inventory_complete:
+            overall = STATUS_FAILED
+        else:
+            overall = STATUS_DEGRADED
+    elif any(i.status == ITEM_DEGRADED for i in items):
+        overall = STATUS_DEGRADED
+    else:
+        overall = STATUS_COMPLETE
+
+    # Enforce mode holds mutations whenever anything is unresolved/failed.
+    mutation_hold = False
+    if mode_norm == MODE_ENFORCE and overall in {STATUS_DEGRADED, STATUS_FAILED}:
+        mutation_hold = True
+    if mode_norm == MODE_ENFORCE and any(
+        i.dimension == DIM_MUTATIONS and i.status == ITEM_UNRESOLVED for i in items
+    ):
+        mutation_hold = True
+
+    follow_ups = tuple(
+        fu for i in items if (fu := _follow_up_for_item(i)) is not None
+    )
+
+    finished = _utc_now() if now is None else now
+    note = (
+        "Read-only completion proof. Never auto-resumes write mutations. "
+        "Unresolved items require durable follow-up before claiming clean restart. "
+        f"Mode={mode_norm}."
+    )
+
+    return RestartCompletionProof(
+        schema_version=SCHEMA_VERSION,
+        reconcile_version=RECONCILE_VERSION,
+        reconcile_id=(reconcile_id or f"reconcile-{uuid4().hex[:12]}"),
+        started_at=_ts(started),
+        finished_at=_ts(finished),
+        boot_head_sha=(
+            str(inventory.get("boot_head_sha")).strip()
+            if inventory.get("boot_head_sha")
+            else None
+        ),
+        current_head_sha=(
+            str(inventory.get("current_head_sha")).strip()
+            if inventory.get("current_head_sha")
+            else None
+        ),
+        inventory_complete=inventory_complete,
+        incomplete_reasons=incomplete_reasons,
+        mode=mode_norm,
+        mutation_hold=mutation_hold,
+        overall_status=overall,
+        items=tuple(items),
+        proposed_follow_ups=follow_ups,
+        resolved_count=resolved,
+        unresolved_count=unresolved,
+        skipped_count=skipped,
+        note=note,
+    )
+
+
+def mutations_allowed(proof: RestartCompletionProof | Mapping[str, Any] | None) -> bool:
+    """Return whether write mutations may proceed under the given proof."""
+    if proof is None:
+        return True  # no proof yet → caller decides; enforce path sets hold
+    if isinstance(proof, RestartCompletionProof):
+        return not proof.mutation_hold
+    if isinstance(proof, Mapping):
+        return not bool(proof.get("mutation_hold"))
+    return True

task_capability_map.py

@@ -132,6 +132,16 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.branch.push",
         "role": "author",
     },
+    # #662: post-restart reconcile is read-only inventory + pure classification.
+    # Durable follow-up issue creation is a separate apply path (not this task).
+    "reconcile_after_restart": {
+        "permission": "gitea.read",
+        "role": "author",
+    },
+    "gitea_reconcile_after_restart": {
+        "permission": "gitea.read",
+        "role": "author",
+    },
     # PR synchronization lifecycle: assess is read-only (any role with gitea.read);
     # update-by-merge is author-only and mutates the PR head via Gitea API.
     "assess_pr_sync_status": {

tests/test_issue_662_post_restart_reconcile.py

@@ -0,0 +1,249 @@
+"""Tests for post-restart MCP reconciliation and completion proof (#662)."""
+
+from __future__ import annotations
+
+import json
+import os
+import unittest
+from datetime import datetime, timezone
+
+import post_restart_reconcile as prr
+
+NOW = datetime(2026, 7, 24, 12, 0, 0, tzinfo=timezone.utc)
+
+
+def _base_inventory(**overrides):
+    inv = {
+        "inventory_complete": True,
+        "incomplete_reasons": [],
+        "service_health": {"healthy": True},
+        "clients": [{"session_id": "c1", "connected": True}],
+        "sessions": [
+            {
+                "session_id": "s-live",
+                "status": "active",
+                "pid": os.getpid(),
+                "role": "author",
+            }
+        ],
+        "leases": [],
+        "checkpoints_available": False,
+        "worktree_bindings": [{"path": "/tmp/wt", "exists": True}],
+        "pending_mutations": [],
+        "capabilities": {"stale": False},
+        "boot_head_sha": "a" * 40,
+        "current_head_sha": "a" * 40,
+        "queue_state": {"safe_to_resume": True},
+    }
+    inv.update(overrides)
+    return inv
+
+
+class IncompleteInventoryTest(unittest.TestCase):
+    def test_incomplete_inventory_fails_closed(self) -> None:
+        proof = prr.reconcile_after_restart(
+            {
+                "inventory_complete": False,
+                "incomplete_reasons": ["control-plane DB unavailable"],
+            },
+            now=NOW,
+            mode=prr.MODE_ENFORCE,
+            reconcile_id="test-incomplete",
+        )
+        self.assertEqual(proof.overall_status, prr.STATUS_FAILED)
+        self.assertTrue(proof.mutation_hold)
+        self.assertFalse(proof.inventory_complete)
+        self.assertTrue(proof.proposed_follow_ups)
+        self.assertIn("control-plane DB unavailable", proof.incomplete_reasons)
+
+
+class HappyPathTest(unittest.TestCase):
+    def test_clean_restart_is_complete_without_mutation_hold(self) -> None:
+        proof = prr.reconcile_after_restart(
+            _base_inventory(),
+            now=NOW,
+            mode=prr.MODE_ENFORCE,
+            reconcile_id="test-clean",
+        )
+        self.assertEqual(proof.overall_status, prr.STATUS_COMPLETE)
+        self.assertFalse(proof.mutation_hold)
+        self.assertEqual(proof.unresolved_count, 0)
+        cp = next(i for i in proof.items if i.dimension == prr.DIM_CHECKPOINTS)
+        self.assertEqual(cp.status, prr.ITEM_SKIPPED)
+        links = proof.as_dict()["links"]
+        self.assertEqual(links["umbrella"], 655)
+        self.assertEqual(links["issue"], 662)
+        self.assertEqual(links["vision"], 652)
+        self.assertEqual(links["roadmap"], 653)
+
+
+class InterruptedMutationTest(unittest.TestCase):
+    def test_mutating_lease_with_dead_owner_is_unresolved(self) -> None:
+        proof = prr.reconcile_after_restart(
+            _base_inventory(
+                leases=[
+                    {
+                        "lease_id": "lease-mut",
+                        "session_id": "s-dead",
+                        "phase": "implementing",
+                        "work_kind": "issue",
+                        "work_number": 662,
+                        "worktree_path": "/tmp/wt-662",
+                        "freshness": {"freshness": "stale_dead_process"},
+                    }
+                ]
+            ),
+            now=NOW,
+            mode=prr.MODE_ENFORCE,
+        )
+        mut = next(i for i in proof.items if i.dimension == prr.DIM_MUTATIONS)
+        self.assertEqual(mut.status, prr.ITEM_UNRESOLVED)
+        self.assertTrue(mut.follow_up_required)
+        interrupted = mut.details["interrupted"]
+        self.assertEqual(len(interrupted), 1)
+        self.assertFalse(interrupted[0]["resume_allowed"])
+        self.assertTrue(proof.mutation_hold)
+        self.assertTrue(
+            any(f.dimension == prr.DIM_MUTATIONS for f in proof.proposed_follow_ups)
+        )
+
+    def test_explicit_pending_mutation_inventory(self) -> None:
+        proof = prr.reconcile_after_restart(
+            _base_inventory(
+                pending_mutations=[
+                    {
+                        "session_id": "s1",
+                        "phase": "publishing",
+                        "work_kind": "pr",
+                        "work_number": 856,
+                        "reason": "push interrupted mid-flight",
+                    }
+                ]
+            ),
+            now=NOW,
+            mode=prr.MODE_LOG_ONLY,
+        )
+        mut = next(i for i in proof.items if i.dimension == prr.DIM_MUTATIONS)
+        self.assertEqual(mut.status, prr.ITEM_UNRESOLVED)
+        # log_only never holds mutations even when unresolved
+        self.assertFalse(proof.mutation_hold)
+        self.assertEqual(proof.overall_status, prr.STATUS_DEGRADED)
+
+
+class DuplicateClaimsTest(unittest.TestCase):
+    def test_duplicate_live_claims_flagged(self) -> None:
+        proof = prr.reconcile_after_restart(
+            _base_inventory(
+                leases=[
+                    {
+                        "lease_id": "l1",
+                        "session_id": "s1",
+                        "phase": "allocated",
+                        "work_kind": "issue",
+                        "work_number": 100,
+                        "freshness": {"freshness": "active"},
+                    },
+                    {
+                        "lease_id": "l2",
+                        "session_id": "s2",
+                        "phase": "allocated",
+                        "work_kind": "issue",
+                        "work_number": 100,
+                        "freshness": {"freshness": "active"},
+                    },
+                ]
+            ),
+            now=NOW,
+            mode=prr.MODE_ENFORCE,
+        )
+        dups = next(i for i in proof.items if i.dimension == prr.DIM_DUPLICATES)
+        self.assertEqual(dups.status, prr.ITEM_UNRESOLVED)
+        self.assertEqual(dups.details["duplicates"][0]["claim_count"], 2)
+        self.assertTrue(proof.mutation_hold)
+
+
+class OrphanSessionTest(unittest.TestCase):
+    def test_active_session_dead_pid_is_unresolved(self) -> None:
+        proof = prr.reconcile_after_restart(
+            _base_inventory(
+                sessions=[
+                    {
+                        "session_id": "ghost",
+                        "status": "active",
+                        "pid": 2_000_000_000,
+                        "role": "author",
+                    }
+                ]
+            ),
+            now=NOW,
+            mode=prr.MODE_ENFORCE,
+        )
+        sess = next(i for i in proof.items if i.dimension == prr.DIM_SESSIONS)
+        self.assertEqual(sess.status, prr.ITEM_UNRESOLVED)
+        self.assertIn("ghost", sess.details["orphan_session_ids"])
+
+
+class CapabilityStaleTest(unittest.TestCase):
+    def test_stale_runtime_unresolved(self) -> None:
+        proof = prr.reconcile_after_restart(
+            _base_inventory(capabilities={"stale": True, "startup_head": "aaa"}),
+            now=NOW,
+            mode=prr.MODE_ENFORCE,
+        )
+        caps = next(i for i in proof.items if i.dimension == prr.DIM_CAPABILITIES)
+        self.assertEqual(caps.status, prr.ITEM_UNRESOLVED)
+        self.assertTrue(proof.mutation_hold)
+
+
+class MutationsAllowedHelperTest(unittest.TestCase):
+    def test_mutations_allowed_respects_hold(self) -> None:
+        held = prr.reconcile_after_restart(
+            _base_inventory(
+                pending_mutations=[{"phase": "merging", "session_id": "x"}]
+            ),
+            now=NOW,
+            mode=prr.MODE_ENFORCE,
+        )
+        self.assertFalse(prr.mutations_allowed(held))
+        self.assertFalse(prr.mutations_allowed(held.as_dict()))
+        clean = prr.reconcile_after_restart(
+            _base_inventory(), now=NOW, mode=prr.MODE_ENFORCE
+        )
+        self.assertTrue(prr.mutations_allowed(clean))
+
+
+class CheckpointSoftDependencyTest(unittest.TestCase):
+    def test_checkpoints_when_schema_present(self) -> None:
+        proof = prr.reconcile_after_restart(
+            _base_inventory(
+                checkpoints_available=True,
+                checkpoints=[{"session_id": "s1", "stale": False}],
+            ),
+            now=NOW,
+        )
+        cp = next(i for i in proof.items if i.dimension == prr.DIM_CHECKPOINTS)
+        self.assertEqual(cp.status, prr.ITEM_RESOLVED)
+
+    def test_stale_checkpoints_unresolved(self) -> None:
+        proof = prr.reconcile_after_restart(
+            _base_inventory(
+                checkpoints_available=True,
+                checkpoints=[{"session_id": "s1", "stale": True}],
+            ),
+            now=NOW,
+            mode=prr.MODE_ENFORCE,
+        )
+        cp = next(i for i in proof.items if i.dimension == prr.DIM_CHECKPOINTS)
+        self.assertEqual(cp.status, prr.ITEM_UNRESOLVED)
+
+
+class ProofSerializationTest(unittest.TestCase):
+    def test_as_dict_is_json_friendly(self) -> None:
+        proof = prr.reconcile_after_restart(_base_inventory(), now=NOW)
+        blob = json.dumps(proof.as_dict())
+        self.assertIn("reconcile_id", blob)
+        self.assertIn("proposed_follow_ups", blob)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_webui_policy_visibility.py

@@ -0,0 +1,243 @@
+"""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,6 +46,8 @@ 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.analytics_loader import (
     load_analytics,
@@ -308,6 +310,17 @@ 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
@@ -721,6 +734,8 @@ 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("/analytics", analytics, methods=["GET"]),
             Route("/api/analytics", api_v1_analytics, methods=["GET"]),

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 (placeholder), and
+Runtime/Sessions, Projects, Inventory, Timeline, Policy (live via #646), 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", "stub"),
+        NavItem("/policy", "Policy"),
         NavItem("/prompts", "Prompts"),
     )),
     NavGroup("Insights", (
@@ -89,10 +89,6 @@ 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

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

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