Compare commits

..
Author SHA1 Message Date
jcwalker3andClaude Opus 4.8 ab33337a94 feat(webui): read-only workflow policy & guardrail visibility (Closes #646)
Phase 3 child of the Web Console epic #631. Operators can now see the active
workflow policy/guardrail configuration from the console instead of reading the
repo tree.

- webui/policy_inventory.py (new): redacted, machine-readable guardrail
  inventory. One row per major guardrail (role separation/RBAC, lease rules,
  worktree binding, merge confirmation, redaction, contamination, allocator
  policy, audit logging, mutation gating) with source pointers (file/module/doc)
  and a compact active projection from the existing safe policy accessors.
  Fail-soft per entry; whole payload run through console_redaction before emit;
  diff vs documented default where feasible.
- webui/policy_views.py (new): HTML cards with source pointers, active config,
  and the documented-default diff; read-only page copy, no forms.
- webui/app.py: register GET /policy and GET /api/v1/policy (additive).
- webui/layout.py: add Policy nav item.
- tests/test_webui_policy_visibility.py (new): guardrail presence + source
  pointers (AC1), redaction incl. planted-secret masking and scan_for_secrets
  (AC2/AC3), read-only page + no-mutation routes (AC4), fail-soft rendering.
- docs/webui-local-dev.md: route table + read-only policy-visibility section.

Read-only throughout; no policy editing, no gate-weakening toggle, secrets
redacted.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-23 03:20:56 -05:00
9 changed files with 746 additions and 347 deletions
+1 -103
View File
@@ -53,8 +53,6 @@ OUTCOME_CANDIDATE_SET_DRIFT = "candidate_set_drift"
SKIP_CLAIMED_BY_OTHER_SESSION = "claimed_by_other_session" SKIP_CLAIMED_BY_OTHER_SESSION = "claimed_by_other_session"
# #776: controller-supplied pre-rank exclusion. # #776: controller-supplied pre-rank exclusion.
SKIP_EXCLUDED_BY_CONTROLLER = "excluded_by_controller" SKIP_EXCLUDED_BY_CONTROLLER = "excluded_by_controller"
# #844: epic / child-only implementation container (pre-rank).
SKIP_EPIC_OR_CHILD_ONLY_CONTAINER = "epic_or_child_only_container"
# Ownership verdicts for a live claim on a candidate (#765). # Ownership verdicts for a live claim on a candidate (#765).
OWNERSHIP_OWN = "own" OWNERSHIP_OWN = "own"
@@ -132,39 +130,6 @@ ROLE_ACTIONS: dict[str, tuple[tuple[str, ...], tuple[str, ...]]] = {
} }
# Body phrases that prove an issue is an implementation container, not a
# unit of direct author work (#844). Matched case-insensitively against the
# issue body. Title alone is never sufficient (ordinary issues may mention
# "epic" incidentally).
_CHILD_ONLY_BODY_MARKERS: tuple[str, ...] = (
"implementation is delivered via child issues only",
"implementation is delivered through child issues only",
"implementation is delivered via child issues",
"implementation is delivered through child issues",
"do not implement product features in this epic",
"do not implement product features in this epic issue itself",
"no product feature implementation is claimed complete solely on this epic",
"implementable child issues remain independently eligible",
"owns the product roadmap and linkage",
"this epic owns the product roadmap",
"coordination container",
"child-only container",
"implementation is delegated to child",
)
# Explicit epic / umbrella labels (structured evidence preferred over title).
_EPIC_LABELS: frozenset[str] = frozenset(
{
"type:epic",
"epic",
"kind:epic",
"scope:epic",
"type:umbrella",
"umbrella",
}
)
@dataclass @dataclass
class WorkCandidate: class WorkCandidate:
"""One assignable Gitea issue or PR presented to the allocator.""" """One assignable Gitea issue or PR presented to the allocator."""
@@ -174,7 +139,6 @@ class WorkCandidate:
state: str = "open" state: str = "open"
labels: tuple[str, ...] = () labels: tuple[str, ...] = ()
title: str = "" title: str = ""
body: str = ""
priority: int = 0 priority: int = 0
head_sha: str | None = None head_sha: str | None = None
# Routing signals (callers derive from Gitea / review feedback). # Routing signals (callers derive from Gitea / review feedback).
@@ -194,7 +158,6 @@ class WorkCandidate:
self.labels = tuple( self.labels = tuple(
str(x).strip().lower() for x in (self.labels or ()) if str(x).strip() str(x).strip().lower() for x in (self.labels or ()) if str(x).strip()
) )
self.body = str(self.body or "")
if self.kind not in WORK_KINDS: if self.kind not in WORK_KINDS:
raise InvalidWorkKindError( raise InvalidWorkKindError(
f"candidate kind '{self.kind}' is not assignable; only " f"candidate kind '{self.kind}' is not assignable; only "
@@ -208,7 +171,6 @@ class WorkCandidate:
"state": self.state, "state": self.state,
"labels": list(self.labels), "labels": list(self.labels),
"title": self.title, "title": self.title,
"body": self.body,
"priority": self.priority, "priority": self.priority,
"head_sha": self.head_sha, "head_sha": self.head_sha,
"request_changes_current_head": self.request_changes_current_head, "request_changes_current_head": self.request_changes_current_head,
@@ -222,51 +184,6 @@ class WorkCandidate:
} }
def classify_epic_or_child_only_container(
c: WorkCandidate,
) -> tuple[bool, str | None]:
"""Return whether *c* is an epic / child-only implementation container (#844).
Exclusion uses structured evidence first (labels, body scope language).
A bare title containing the word "epic" is **not** enough — ordinary
implementable issues may mention epics incidentally. A title that is
explicitly prefixed ``Epic:`` only counts when the body also proves
child-only / no-direct-implementation scope (or an epic label is present).
PRs are never classified as containers here (they already have a head).
"""
if c.kind != "issue":
return False, None
labels = set(c.labels)
epic_label = sorted(labels & _EPIC_LABELS)
body_l = (c.body or "").lower()
title = (c.title or "").strip()
title_l = title.lower()
body_hits = [m for m in _CHILD_ONLY_BODY_MARKERS if m in body_l]
title_epic_prefix = title_l.startswith("epic:") or title_l.startswith("epic ")
if epic_label:
detail = f"label={epic_label[0]}"
if body_hits:
detail = f"{detail}; body_marker={body_hits[0]!r}"
return True, detail
if body_hits:
# Body proves child-only / umbrella scope. Title "Epic:" is corroborating
# but not required — containers without the word still exclude.
detail = f"body_marker={body_hits[0]!r}"
if title_epic_prefix:
detail = f"title_epic_prefix; {detail}"
return True, detail
# Title-only "Epic:" without body scope evidence is insufficient (#844 AC:
# eligibility does not rely solely on the word "Epic" in a title).
# Similarly, incidental "epic" mid-title without markers stays eligible.
return False, None
@dataclass @dataclass
class SkipRecord: class SkipRecord:
kind: str kind: str
@@ -933,8 +850,7 @@ def allocate_next_work(
ownership_defects: list[dict[str, Any]] = [] ownership_defects: list[dict[str, Any]] = []
controller_excluded: list[dict[str, Any]] = [] controller_excluded: list[dict[str, Any]] = []
# #776 AC2 + #844: remove excluded numbers *and* epic/child-only containers # #776 AC2: remove excluded numbers *before* ranking / selection / lease.
# *before* ranking / selection / lease so they never receive assignments.
rankable: list[WorkCandidate] = [] rankable: list[WorkCandidate] = []
for c in candidates: for c in candidates:
if int(c.number) in exclude_set: if int(c.number) in exclude_set:
@@ -1013,23 +929,6 @@ def allocate_next_work(
}, },
} }
continue continue
# #844: epics / child-only containers are never direct implement targets.
is_container, container_detail = classify_epic_or_child_only_container(c)
if is_container:
detail = container_detail or "epic or child-only container"
reason = (
f"{c.kind}#{c.number} {SKIP_EPIC_OR_CHILD_ONLY_CONTAINER}: "
f"{detail}; implementation is delegated to child issues"
)
skipped.append(
SkipRecord(
c.kind,
c.number,
reason,
SKIP_EPIC_OR_CHILD_ONLY_CONTAINER,
)
)
continue
rankable.append(c) rankable.append(c)
ordered = sort_candidates(rankable) ordered = sort_candidates(rankable)
@@ -1442,7 +1341,6 @@ def candidate_from_dict(data: dict[str, Any]) -> WorkCandidate:
state=str(data.get("state") or "open"), state=str(data.get("state") or "open"),
labels=tuple(data.get("labels") or ()), labels=tuple(data.get("labels") or ()),
title=str(data.get("title") or ""), title=str(data.get("title") or ""),
body=str(data.get("body") or ""),
priority=priority, priority=priority,
head_sha=data.get("head_sha"), head_sha=data.get("head_sha"),
request_changes_current_head=bool(data.get("request_changes_current_head")), request_changes_current_head=bool(data.get("request_changes_current_head")),
+15
View File
@@ -64,6 +64,8 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
| `/api/prompts` | JSON prompt export with workflow hashes | | `/api/prompts` | JSON prompt export with workflow hashes |
| `/runtime` | MCP runtime health and stale detection (#430) | | `/runtime` | MCP runtime health and stale detection (#430) |
| `/api/runtime` | JSON runtime health export | | `/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) | | `/audit` | Report audit paste + validator preview (#431) |
| `/api/audit` | JSON validator preview (POST `report_text`, optional `task_kind`) | | `/api/audit` | JSON validator preview (POST `report_text`, optional `task_kind`) |
| `/worktrees` | Worktree hygiene dashboard (#432) | | `/worktrees` | Worktree hygiene dashboard (#432) |
@@ -153,6 +155,19 @@ health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
checkout is behind merged safety-gate changes. Restart guidance links to #420; checkout is behind merged safety-gate changes. Restart guidance links to #420;
no tokens or MCP restart actions are exposed. 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.
## Deployment boundary (#435) ## Deployment boundary (#435)
MVP serves on loopback by default. Binding `0.0.0.0` or `::` is **refused** MVP serves on loopback by default. Binding `0.0.0.0` or `::` is **refused**
-1
View File
@@ -19912,7 +19912,6 @@ def _allocator_candidates_from_gitea(
state="open", state="open",
labels=tuple(labels), labels=tuple(labels),
title=title, title=title,
body=body,
priority=20 if "status:ready" in labels else 1, priority=20 if "status:ready" in labels else 1,
blocked=blocked, blocked=blocked,
dependency_unmet=dep_unmet, dependency_unmet=dep_unmet,
@@ -1,243 +0,0 @@
"""Allocator epic / child-only container pre-rank exclusion (#844).
Covers:
* Issue #631-shaped child-only epic is excluded before ranking.
* Implementable child issues remain eligible and can be selected.
* Ordinary issues that merely mention "epic" in title/body are not excluded.
* Excluded containers never receive assignments or workflow leases.
* Structured skip reason ``epic_or_child_only_container`` is reported.
"""
from __future__ import annotations
import os
import tempfile
import unittest
from allocator_service import (
OUTCOME_ASSIGNED,
OUTCOME_PREVIEW,
SKIP_EPIC_OR_CHILD_ONLY_CONTAINER,
WorkCandidate,
allocate_next_work,
classify_epic_or_child_only_container,
)
from control_plane_db import ControlPlaneDB
REMOTE = "prgs"
ORG = "Scaled-Tech-Consulting"
REPO = "Gitea-Tools"
# Minimal body mirroring issue #631 authoritative scope language.
_EPIC_631_BODY = """
## Scope (umbrella)
This epic owns the **product roadmap and linkage** for the Web Console.
Implementation is delivered via child issues only.
## Explicit non-goals
* Do not implement product features in this epic issue itself.
* No product feature implementation is claimed complete solely on this epic.
"""
_CHILD_BODY = """
## Problem
Operators need a workflow-event timeline model for Phase 1.
## Acceptance criteria
- [ ] Timeline model API exists
"""
def _issue(
number: int,
*,
title: str = "",
body: str = "",
labels: tuple[str, ...] = ("status:ready", "type:feature"),
priority: int = 20,
) -> WorkCandidate:
return WorkCandidate(
kind="issue",
number=number,
state="open",
labels=labels,
title=title or f"issue {number}",
body=body,
priority=priority,
)
class ClassifyEpicContainerTest(unittest.TestCase):
def test_631_shaped_body_and_title_is_container(self) -> None:
c = _issue(
631,
title="Epic: MCP Control Plane Web Console",
body=_EPIC_631_BODY,
)
is_c, detail = classify_epic_or_child_only_container(c)
self.assertTrue(is_c)
self.assertIsNotNone(detail)
self.assertIn("body_marker", detail or "")
def test_body_markers_without_epic_title(self) -> None:
c = _issue(
900,
title="Control plane roadmap tracker",
body="Implementation is delivered via child issues only.",
)
is_c, _ = classify_epic_or_child_only_container(c)
self.assertTrue(is_c)
def test_epic_label_alone_is_container(self) -> None:
c = _issue(
901,
title="Roadmap linkage",
body="Track children.",
labels=("status:ready", "type:epic"),
)
is_c, detail = classify_epic_or_child_only_container(c)
self.assertTrue(is_c)
self.assertIn("type:epic", detail or "")
def test_title_epic_prefix_alone_not_container(self) -> None:
"""Title-only 'Epic:' without body scope evidence stays eligible (#844)."""
c = _issue(
902,
title="Epic: something mentioned only in title",
body="Implement a concrete fix for the allocator skip list.",
)
is_c, detail = classify_epic_or_child_only_container(c)
self.assertFalse(is_c)
self.assertIsNone(detail)
def test_incidental_epic_word_not_container(self) -> None:
c = _issue(
903,
title="Document epic handoff conventions",
body=(
"Update the docs so implementable issues that mention an epic "
"remain independently executable."
),
)
is_c, _ = classify_epic_or_child_only_container(c)
self.assertFalse(is_c)
def test_prs_never_classified(self) -> None:
pr = WorkCandidate(
kind="pr",
number=10,
state="open",
title="Epic: fake",
body="Implementation is delivered via child issues only.",
head_sha="a" * 40,
priority=5,
)
is_c, _ = classify_epic_or_child_only_container(pr)
self.assertFalse(is_c)
class AllocateEpicContainerExclusionTest(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
def _alloc(self, candidates, **kwargs):
defaults = dict(
session_id="sess-844",
role="author",
remote=REMOTE,
org=ORG,
repo=REPO,
profile_name="prgs-author",
username="jcwalker3",
claims={},
apply=False,
)
defaults.update(kwargs)
return allocate_next_work(self.db, candidates=candidates, **defaults)
def test_631_shaped_epic_excluded_child_selected(self) -> None:
epic = _issue(
631,
title="Epic: MCP Control Plane Web Console",
body=_EPIC_631_BODY,
)
child = _issue(
637,
title="Web Console: Workflow-event timeline model (Phase 1)",
body=_CHILD_BODY,
)
res = self._alloc([epic, child], apply=False)
self.assertTrue(res["success"], res)
self.assertEqual(res["outcome"], OUTCOME_PREVIEW)
self.assertEqual(res["selected"]["number"], 637)
skipped = {s["number"]: s for s in res["skipped"]}
self.assertIn(631, skipped)
self.assertEqual(
skipped[631]["reason_code"], SKIP_EPIC_OR_CHILD_ONLY_CONTAINER
)
self.assertIn(SKIP_EPIC_OR_CHILD_ONLY_CONTAINER, skipped[631]["reason"])
def test_container_cannot_receive_assignment_or_lease(self) -> None:
epic = _issue(
631,
title="Epic: MCP Control Plane Web Console",
body=_EPIC_631_BODY,
)
res = self._alloc([epic], apply=True)
self.assertTrue(res["success"], res)
# Only container present → no safe work; never assigned_work.
self.assertNotEqual(res["outcome"], OUTCOME_ASSIGNED)
self.assertIsNone(res.get("assignment"))
self.assertIsNone(res.get("selected"))
skipped = {s["number"]: s for s in res["skipped"]}
self.assertEqual(
skipped[631]["reason_code"], SKIP_EPIC_OR_CHILD_ONLY_CONTAINER
)
# No lease row for the epic.
leases = self.db.list_active_leases(
remote=REMOTE, org=ORG, repo=REPO
) if hasattr(self.db, "list_active_leases") else []
# Prefer generic inventory if available.
if not leases and hasattr(self.db, "list_leases"):
leases = self.db.list_leases(remote=REMOTE, org=ORG, repo=REPO)
for lease in leases or []:
work_number = lease.get("work_number") if isinstance(lease, dict) else None
self.assertNotEqual(work_number, 631)
def test_incidental_epic_title_remains_eligible(self) -> None:
ordinary = _issue(
700,
title="Document epic handoff conventions",
body="Write runbook text about epic vs child issues.",
)
res = self._alloc([ordinary], apply=False)
self.assertTrue(res["success"], res)
self.assertEqual(res["selected"]["number"], 700)
self.assertEqual(res["skipped"], [])
def test_apply_selects_child_not_epic(self) -> None:
epic = _issue(
631,
title="Epic: MCP Control Plane Web Console",
body=_EPIC_631_BODY,
)
child = _issue(
637,
title="Web Console: Workflow-event timeline model (Phase 1)",
body=_CHILD_BODY,
)
res = self._alloc([epic, child], apply=True)
self.assertTrue(res["success"], res)
self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
self.assertEqual(res["selected"]["number"], 637)
self.assertEqual(res["assignment"]["work_number"], 637)
if __name__ == "__main__":
unittest.main()
+222
View File
@@ -0,0 +1,222 @@
"""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), [])
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):
text = self.client.get("/").text
self.assertIn('href="/policy"', 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()
+16
View File
@@ -45,6 +45,8 @@ from webui.worktree_scanner import load_hygiene_snapshot, snapshot_to_dict as wo
from webui.worktree_views import render_worktrees_page 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_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
from webui.runtime_views import render_runtime_page 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
_READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"}) _READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
_AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"}) _AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
@@ -68,6 +70,7 @@ async def home(_request: Request) -> HTMLResponse:
"<li><strong>Projects</strong> — registry and onboarding (#427)</li>" "<li><strong>Projects</strong> — registry and onboarding (#427)</li>"
"<li><strong>Prompts</strong> — canonical workflow prompt library (#428)</li>" "<li><strong>Prompts</strong> — canonical workflow prompt library (#428)</li>"
"<li><strong>Runtime</strong> — MCP health and stale-runtime detection (#430)</li>" "<li><strong>Runtime</strong> — MCP health and stale-runtime detection (#430)</li>"
"<li><strong>Policy</strong> — workflow guardrail configuration visibility (#646)</li>"
"<li><strong>Audit</strong> — final-report paste and validator preview (#431)</li>" "<li><strong>Audit</strong> — final-report paste and validator preview (#431)</li>"
"<li><strong>Worktrees</strong> — branch hygiene dashboard (#432)</li>" "<li><strong>Worktrees</strong> — branch hygiene dashboard (#432)</li>"
"<li><strong>Leases</strong> — collision and lease visibility (#433)</li>" "<li><strong>Leases</strong> — collision and lease visibility (#433)</li>"
@@ -210,6 +213,17 @@ async def api_runtime(_request: Request) -> JSONResponse:
return JSONResponse(runtime_snapshot_to_dict(load_runtime_snapshot())) 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]: async def _parse_audit_form(request: Request) -> tuple[str, str | None]:
if request.method == "GET": if request.method == "GET":
return "", None return "", None
@@ -415,6 +429,8 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
Route("/api/prompts", api_prompts, methods=["GET"]), Route("/api/prompts", api_prompts, methods=["GET"]),
Route("/runtime", runtime, methods=["GET"]), Route("/runtime", runtime, methods=["GET"]),
Route("/api/runtime", api_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("/audit", audit, methods=["GET", "POST"]), Route("/audit", audit, methods=["GET", "POST"]),
Route("/api/audit", api_audit, methods=["GET", "POST"]), Route("/api/audit", api_audit, methods=["GET", "POST"]),
Route("/worktrees", worktrees, methods=["GET"]), Route("/worktrees", worktrees, methods=["GET"]),
+1
View File
@@ -8,6 +8,7 @@ NAV_ITEMS = (
("/projects", "Projects"), ("/projects", "Projects"),
("/prompts", "Prompts"), ("/prompts", "Prompts"),
("/runtime", "Runtime"), ("/runtime", "Runtime"),
("/policy", "Policy"),
("/audit", "Audit"), ("/audit", "Audit"),
("/worktrees", "Worktrees"), ("/worktrees", "Worktrees"),
("/leases", "Leases"), ("/leases", "Leases"),
+387
View File
@@ -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)
+104
View File
@@ -0,0 +1,104 @@
"""HTML views for the workflow policy and guardrail inventory (#646)."""
from __future__ import annotations
import html
import json
from webui.policy_inventory import PolicyEntry, PolicyInventorySnapshot
def _source_pointer(source) -> str:
path = source.path
if source.anchor:
path = f"{path}#{source.anchor}"
return (
f"<li>{html.escape(source.label)}"
f"<code>{html.escape(path)}</code> "
f"<span class='muted'>({html.escape(source.kind)})</span></li>"
)
def _active_block(entry: PolicyEntry) -> str:
if entry.error:
return (
"<p class='muted'><strong>Active value unavailable:</strong> "
f"{html.escape(entry.error)}</p>"
)
if not entry.active:
return "<p class='muted'>No live projection for this guardrail.</p>"
pretty = json.dumps(entry.active, indent=2, sort_keys=True, default=str)
return f"<pre class='prompt-text'>{html.escape(pretty)}</pre>"
def _diff_block(entry: PolicyEntry) -> str:
if entry.diff is None:
if entry.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 = entry.diff.get("status", "unknown")
badge = "badge-claimed" if status == "matches_documented_default" else "badge-blocked"
rows = []
for key, cell in (entry.diff.get("checked") or {}).items():
marker = "" if cell.get("matches") else ""
rows.append(
"<tr>"
f"<td><code>{html.escape(str(key))}</code></td>"
f"<td><code>{html.escape(str(cell.get('expected')))}</code></td>"
f"<td><code>{html.escape(str(cell.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: PolicyEntry) -> str:
sources = "".join(_source_pointer(s) for s in entry.sources)
return (
"<div class='prompt-card'>"
f"<h3>{html.escape(entry.title)} "
f"<span class='badge'>{html.escape(entry.category)}</span></h3>"
f"<p>{html.escape(entry.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) -> str:
categories = ", ".join(html.escape(c) for c in snapshot.categories) or "none"
cards = "".join(_entry_card(e) for e in snapshot.entries)
build_errors = ""
if snapshot.build_errors:
items = "".join(
f"<li>{html.escape(err)}</li>" for err in snapshot.build_errors
)
build_errors = (
"<div class='stub'><p><strong>Some guardrails could not be built:"
f"</strong></p><ul>{items}</ul></div>"
)
return (
"<h2>Workflow policy &amp; guardrails</h2>"
f"<p class='muted'>{html.escape(snapshot.note)}</p>"
f"<p class='meta'>Schema v{snapshot.schema_version} · "
f"{len(snapshot.entries)} 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>"
)