Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
64b6eb5d54 | ||
|
|
f21f81f9b5 | ||
|
|
08061b7b8a |
+23
-13
@@ -64,8 +64,6 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
|
||||
| `/api/prompts` | JSON prompt export with workflow hashes |
|
||||
| `/runtime` | MCP runtime health and stale detection (#430) |
|
||||
| `/api/runtime` | JSON runtime health export |
|
||||
| `/policy` | Workflow policy and guardrail configuration visibility (#646) |
|
||||
| `/api/v1/policy` | Versioned JSON guardrail inventory (redacted, read-only) |
|
||||
| `/audit` | Report audit paste + validator preview (#431) |
|
||||
| `/api/audit` | JSON validator preview (POST `report_text`, optional `task_kind`) |
|
||||
| `/worktrees` | Worktree hygiene dashboard (#432) |
|
||||
@@ -75,6 +73,11 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
|
||||
| `/api/actions/{id}/preview` | Mutation ledger preview (GET, read-only) |
|
||||
| `/leases` | Lease and collision visibility (#433) |
|
||||
| `/api/leases` | JSON lease/collision export |
|
||||
| `/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
|
||||
`read-only-mvp`, except `/audit` and `/api/audit` which accept POST for
|
||||
@@ -155,18 +158,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)
|
||||
## Application shell — Phase 1 (#638)
|
||||
|
||||
`/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.
|
||||
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**
|
||||
(placeholder). Live views and Phase 1 placeholders (`stub`) are declared in one
|
||||
place so the layout and the route table cannot drift.
|
||||
|
||||
The header carries two read-only status badges — an **environment** badge
|
||||
(`local` for loopback binds, `remote` otherwise, derived from `WEBUI_HOST`) and
|
||||
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
|
||||
404s; their backing views land in later child issues of #631 (the inventory
|
||||
surfaces are backed by #636). Mutating methods on stub routes still fail closed
|
||||
with `read-only-mvp`.
|
||||
|
||||
## Deployment boundary (#435)
|
||||
|
||||
|
||||
@@ -1,222 +0,0 @@
|
||||
"""Tests for the read-only workflow policy/guardrail visibility view (#646).
|
||||
|
||||
Covers issue #646 acceptance criteria:
|
||||
|
||||
1. Console lists major guardrails with source pointers.
|
||||
2. Secrets redacted.
|
||||
3. Tests ensure sample secrets never appear.
|
||||
4. Docs explain read-only nature (asserted here for the page copy; the doc
|
||||
itself is covered by inspection).
|
||||
"""
|
||||
|
||||
import json
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from webui import console_redaction
|
||||
from webui import policy_inventory
|
||||
from webui.app import create_app
|
||||
from webui.policy_inventory import (
|
||||
PolicyEntry,
|
||||
PolicyInventorySnapshot,
|
||||
SourcePointer,
|
||||
load_policy_inventory,
|
||||
snapshot_to_dict,
|
||||
)
|
||||
from webui.policy_views import render_policy_page
|
||||
|
||||
|
||||
def _entry(key, category, *, active=None, error=None):
|
||||
return PolicyEntry(
|
||||
key=key,
|
||||
title=key.replace("_", " ").title(),
|
||||
category=category,
|
||||
summary=f"summary for {key}",
|
||||
sources=(SourcePointer("src", f"{key}.py", "module"),),
|
||||
active=active,
|
||||
documented_default=None,
|
||||
diff=None,
|
||||
error=error,
|
||||
)
|
||||
|
||||
|
||||
def _snapshot(entries):
|
||||
return PolicyInventorySnapshot(
|
||||
schema_version=1,
|
||||
read_only=True,
|
||||
note="read-only projection",
|
||||
entries=tuple(entries),
|
||||
categories=tuple(dict.fromkeys(e.category for e in entries)),
|
||||
build_errors=(),
|
||||
)
|
||||
|
||||
# The guardrail categories issue #646 names as in-scope.
|
||||
_EXPECTED_CATEGORIES = {
|
||||
"role_separation",
|
||||
"lease_rules",
|
||||
"worktree_rules",
|
||||
"merge_confirmation",
|
||||
"redaction",
|
||||
"contamination",
|
||||
"allocator_policy",
|
||||
"audit_logging",
|
||||
"mutation_gating",
|
||||
}
|
||||
|
||||
|
||||
class TestPolicyInventoryModel(unittest.TestCase):
|
||||
def test_major_guardrails_present(self):
|
||||
snapshot = load_policy_inventory()
|
||||
categories = {e.category for e in snapshot.entries}
|
||||
self.assertEqual(_EXPECTED_CATEGORIES, categories)
|
||||
self.assertGreaterEqual(len(snapshot.entries), len(_EXPECTED_CATEGORIES))
|
||||
|
||||
def test_every_guardrail_has_source_pointers(self):
|
||||
# AC1: source attribution (file/module/doc) for every guardrail.
|
||||
snapshot = load_policy_inventory()
|
||||
for entry in snapshot.entries:
|
||||
with self.subTest(entry=entry.key):
|
||||
self.assertTrue(entry.sources, "guardrail must carry source pointers")
|
||||
for source in entry.sources:
|
||||
self.assertTrue(source.path)
|
||||
self.assertIn(source.kind, {"module", "doc", "script", "config"})
|
||||
|
||||
def test_diff_reported_where_documented_default_declared(self):
|
||||
snapshot = load_policy_inventory()
|
||||
checked_any = False
|
||||
for entry in snapshot.entries:
|
||||
if entry.documented_default is None:
|
||||
self.assertIsNone(entry.diff)
|
||||
continue
|
||||
checked_any = True
|
||||
self.assertIsNotNone(entry.diff)
|
||||
self.assertEqual(
|
||||
entry.diff["status"],
|
||||
"matches_documented_default",
|
||||
f"{entry.key} drifted from its documented default: {entry.diff}",
|
||||
)
|
||||
self.assertTrue(checked_any, "at least one guardrail should declare a default")
|
||||
|
||||
def test_live_projections_populate_active(self):
|
||||
snapshot = load_policy_inventory()
|
||||
by_key = {e.key: e for e in snapshot.entries}
|
||||
for key in ("role_separation", "redaction", "audit_logging"):
|
||||
self.assertIsNone(by_key[key].error, f"{key} projection failed")
|
||||
self.assertIsInstance(by_key[key].active, dict)
|
||||
|
||||
def test_build_entry_is_fail_soft_on_projection_error(self):
|
||||
def _boom():
|
||||
raise RuntimeError("projection exploded")
|
||||
|
||||
row = (
|
||||
"redaction",
|
||||
"Secret redaction",
|
||||
"redaction",
|
||||
"summary",
|
||||
(SourcePointer("x", "webui/console_redaction.py", "module"),),
|
||||
_boom,
|
||||
{"redact_before_persist": True},
|
||||
)
|
||||
entry = policy_inventory._build_entry(row)
|
||||
self.assertIsNone(entry.active)
|
||||
self.assertIsNotNone(entry.error)
|
||||
self.assertEqual(entry.diff["status"], "active_unavailable")
|
||||
|
||||
|
||||
class TestPolicyRedaction(unittest.TestCase):
|
||||
def test_real_snapshot_has_no_secret_shapes(self):
|
||||
# AC3: the real emitted payload never carries a known secret shape.
|
||||
payload = snapshot_to_dict(load_policy_inventory())
|
||||
self.assertEqual(console_redaction.scan_for_secrets(payload), [])
|
||||
|
||||
def test_planted_keychain_secret_is_redacted(self):
|
||||
# AC2/AC3: a secret planted in an active projection is masked before emit.
|
||||
snapshot = _snapshot([
|
||||
_entry(
|
||||
"redaction",
|
||||
"redaction",
|
||||
active={"leaked": "keychain:prgs-author-super-secret", "roles": ["author"]},
|
||||
)
|
||||
])
|
||||
payload = snapshot_to_dict(snapshot)
|
||||
blob = json.dumps(payload)
|
||||
self.assertNotIn("keychain:prgs-author-super-secret", blob)
|
||||
self.assertEqual(console_redaction.scan_for_secrets(payload), [])
|
||||
|
||||
def test_planted_credential_assignment_is_redacted(self):
|
||||
snapshot = _snapshot([
|
||||
_entry(
|
||||
"audit_logging",
|
||||
"audit_logging",
|
||||
active={"leaked": "token=abcd1234efgh5678", "append_only": True},
|
||||
)
|
||||
])
|
||||
payload = snapshot_to_dict(snapshot)
|
||||
blob = json.dumps(payload)
|
||||
self.assertNotIn("abcd1234efgh5678", blob)
|
||||
self.assertEqual(console_redaction.scan_for_secrets(payload), [])
|
||||
|
||||
|
||||
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()
|
||||
@@ -0,0 +1,135 @@
|
||||
"""Tests for the Phase 1 operator console application shell (#638)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from starlette.routing import Route
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from webui import layout
|
||||
from webui.app import create_app
|
||||
from webui.nav import NAV_GROUPS, STUB_PAGES, nav_hrefs
|
||||
|
||||
|
||||
class TestShellNav(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_nav_group_labels_present(self):
|
||||
text = self.client.get("/").text
|
||||
for group in NAV_GROUPS:
|
||||
with self.subTest(group=group.label):
|
||||
self.assertIn(f">{group.label}<", text)
|
||||
|
||||
def test_phase1_group_labels_cover_expected_ia(self):
|
||||
labels = {group.label for group in NAV_GROUPS}
|
||||
for expected in (
|
||||
"Health",
|
||||
"Traffic",
|
||||
"Runtime/Sessions",
|
||||
"Projects",
|
||||
"Inventory",
|
||||
"Timeline",
|
||||
"Policy",
|
||||
"Insights",
|
||||
):
|
||||
with self.subTest(label=expected):
|
||||
self.assertIn(expected, labels)
|
||||
|
||||
def test_every_nav_href_resolves_to_a_get_route(self):
|
||||
app = create_app()
|
||||
get_paths = {
|
||||
route.path
|
||||
for route in app.routes
|
||||
if isinstance(route, Route) and "GET" in route.methods
|
||||
}
|
||||
for href in nav_hrefs():
|
||||
with self.subTest(href=href):
|
||||
self.assertIn(href, get_paths, f"nav href {href} has no GET route")
|
||||
|
||||
def test_legacy_hrefs_still_navigable(self):
|
||||
text = self.client.get("/").text
|
||||
for href in ("/queue", "/projects", "/prompts", "/runtime",
|
||||
"/audit", "/worktrees", "/leases", "/actions"):
|
||||
with self.subTest(href=href):
|
||||
self.assertIn(f'href="{href}"', text)
|
||||
|
||||
|
||||
class TestShellBadges(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_mode_badge_present(self):
|
||||
self.assertIn("mode: read-only", self.client.get("/").text)
|
||||
|
||||
def test_environment_badge_present(self):
|
||||
self.assertIn("env:", self.client.get("/").text)
|
||||
|
||||
def test_default_environment_is_local(self):
|
||||
self.assertEqual(layout.environment_label(), "local")
|
||||
|
||||
def test_remote_bind_reports_remote_environment(self):
|
||||
import os
|
||||
|
||||
prior = os.environ.get("WEBUI_HOST")
|
||||
os.environ["WEBUI_HOST"] = "10.0.0.5"
|
||||
try:
|
||||
self.assertEqual(layout.environment_label(), "remote")
|
||||
finally:
|
||||
if prior is None:
|
||||
os.environ.pop("WEBUI_HOST", None)
|
||||
else:
|
||||
os.environ["WEBUI_HOST"] = prior
|
||||
|
||||
def test_docs_link_present(self):
|
||||
text = self.client.get("/").text
|
||||
self.assertIn(layout.DOCS_URL, text)
|
||||
self.assertIn(">Docs<", text)
|
||||
|
||||
|
||||
class TestShellStubs(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_stub_routes_render_200(self):
|
||||
for path, (title, _desc) in STUB_PAGES.items():
|
||||
with self.subTest(path=path):
|
||||
response = self.client.get(path)
|
||||
self.assertEqual(response.status_code, 200, path)
|
||||
self.assertIn(title, response.text)
|
||||
self.assertIn("placeholder", response.text)
|
||||
|
||||
def test_stub_routes_are_read_only(self):
|
||||
for path in STUB_PAGES:
|
||||
with self.subTest(path=path):
|
||||
response = self.client.post(path)
|
||||
self.assertEqual(response.status_code, 405)
|
||||
self.assertEqual(response.json()["error"], "read-only-mvp")
|
||||
|
||||
def test_stub_pages_carry_nav_and_badges(self):
|
||||
response = self.client.get("/inventory")
|
||||
self.assertIn("mode: read-only", response.text)
|
||||
self.assertIn('href="/queue"', response.text)
|
||||
|
||||
|
||||
class TestShellHome(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_home_summarizes_console(self):
|
||||
text = self.client.get("/").text
|
||||
self.assertIn("Operator console", text)
|
||||
self.assertIn("Phase 1", text)
|
||||
|
||||
def test_home_links_legacy_pages(self):
|
||||
text = self.client.get("/").text
|
||||
self.assertIn("MVP legacy pages", text)
|
||||
for href in ("/queue", "/audit", "/leases"):
|
||||
with self.subTest(href=href):
|
||||
self.assertIn(f'href="{href}"', text)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+54
-27
@@ -12,6 +12,7 @@ from starlette.routing import Route
|
||||
|
||||
from webui.deployment_boundary import deployment_snapshot
|
||||
from webui.layout import render_page
|
||||
from webui.nav import NAV_GROUPS, STUB_PAGES
|
||||
from webui.project_registry import (
|
||||
ProjectRegistry,
|
||||
RegistryError,
|
||||
@@ -45,8 +46,6 @@ from webui.worktree_scanner import load_hygiene_snapshot, snapshot_to_dict as wo
|
||||
from webui.worktree_views import render_worktrees_page
|
||||
from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
|
||||
from webui.runtime_views import render_runtime_page
|
||||
from webui.policy_inventory import load_policy_inventory, snapshot_to_dict as policy_snapshot_to_dict
|
||||
from webui.policy_views import render_policy_page
|
||||
|
||||
_READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
|
||||
_AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
|
||||
@@ -61,25 +60,62 @@ def _stub_page(title: str, description: str) -> HTMLResponse:
|
||||
return HTMLResponse(render_page(title=title, body_html=body))
|
||||
|
||||
|
||||
_LEGACY_PAGES = (
|
||||
("/queue", "Queue", "live PR and issue dashboard (#429)"),
|
||||
("/projects", "Projects", "registry and onboarding (#427)"),
|
||||
("/prompts", "Prompts", "canonical workflow prompt library (#428)"),
|
||||
("/runtime", "Runtime", "MCP health and stale-runtime detection (#430)"),
|
||||
("/audit", "Audit", "final-report paste and validator preview (#431)"),
|
||||
("/worktrees", "Worktrees", "branch hygiene dashboard (#432)"),
|
||||
("/leases", "Leases", "collision and lease visibility (#433)"),
|
||||
("/actions", "Actions", "gated write-action framework (#434)"),
|
||||
)
|
||||
|
||||
|
||||
def _render_home_nav_groups() -> str:
|
||||
groups = []
|
||||
for group in NAV_GROUPS:
|
||||
items = "".join(
|
||||
f'<li><a href="{item.href}">{item.label}</a>'
|
||||
+ ("" if item.status == "live" else " <span class=\"muted\">(stub)</span>")
|
||||
+ "</li>"
|
||||
for item in group.items
|
||||
)
|
||||
groups.append(f"<h3>{group.label}</h3><ul>{items}</ul>")
|
||||
return "".join(groups)
|
||||
|
||||
|
||||
async def home(_request: Request) -> HTMLResponse:
|
||||
legacy = "".join(
|
||||
f"<li><strong>{label}</strong> — {desc} "
|
||||
f'(<a href="{href}">{href}</a>)</li>'
|
||||
for href, label, desc in _LEGACY_PAGES
|
||||
)
|
||||
body = (
|
||||
"<h2>Operator console</h2>"
|
||||
"<p>Local entry point for MCP Control Plane operational views.</p>"
|
||||
"<ul>"
|
||||
"<li><strong>Queue</strong> — live PR and issue dashboard (#429)</li>"
|
||||
"<li><strong>Projects</strong> — registry and onboarding (#427)</li>"
|
||||
"<li><strong>Prompts</strong> — canonical workflow prompt library (#428)</li>"
|
||||
"<li><strong>Runtime</strong> — MCP health and stale-runtime detection (#430)</li>"
|
||||
"<li><strong>Policy</strong> — workflow guardrail configuration visibility (#646)</li>"
|
||||
"<li><strong>Audit</strong> — final-report paste and validator preview (#431)</li>"
|
||||
"<li><strong>Worktrees</strong> — branch hygiene dashboard (#432)</li>"
|
||||
"<li><strong>Leases</strong> — collision and lease visibility (#433)</li>"
|
||||
"<li><strong>Actions</strong> — gated write-action framework (#434)</li>"
|
||||
"</ul>"
|
||||
"<p>Read-only home for the MCP Control Plane Phase 1 operator console. "
|
||||
"Gitea, MCP capability gates, and canonical workflows remain the source "
|
||||
"of truth; this console never mutates them.</p>"
|
||||
"<h2>Phase 1 surfaces</h2>"
|
||||
+ _render_home_nav_groups()
|
||||
+ "<h2>MVP legacy pages</h2>"
|
||||
"<ul>" + legacy + "</ul>"
|
||||
)
|
||||
return HTMLResponse(render_page(title="Home", body_html=body))
|
||||
|
||||
|
||||
async def phase_stub(request: Request) -> HTMLResponse:
|
||||
"""Graceful read-only placeholder for a not-yet-implemented Phase 1 surface."""
|
||||
title, description = STUB_PAGES[request.url.path]
|
||||
body = (
|
||||
f"<h2>{title}</h2>"
|
||||
f'<div class="stub"><p>{description}</p>'
|
||||
"<p>Phase 1 shell placeholder — no write actions. Tracked under "
|
||||
"epic #631.</p></div>"
|
||||
)
|
||||
return HTMLResponse(render_page(title=title, body_html=body))
|
||||
|
||||
|
||||
async def health(_request: Request) -> JSONResponse:
|
||||
bind_host = _request.app.state.webui_bind_host
|
||||
return JSONResponse({
|
||||
@@ -213,17 +249,6 @@ async def api_runtime(_request: Request) -> JSONResponse:
|
||||
return JSONResponse(runtime_snapshot_to_dict(load_runtime_snapshot()))
|
||||
|
||||
|
||||
async def policy(_request: Request) -> HTMLResponse:
|
||||
snapshot = load_policy_inventory()
|
||||
return HTMLResponse(
|
||||
render_page(title="Policy", body_html=render_policy_page(snapshot))
|
||||
)
|
||||
|
||||
|
||||
async def api_v1_policy(_request: Request) -> JSONResponse:
|
||||
return JSONResponse(policy_snapshot_to_dict(load_policy_inventory()))
|
||||
|
||||
|
||||
async def _parse_audit_form(request: Request) -> tuple[str, str | None]:
|
||||
if request.method == "GET":
|
||||
return "", None
|
||||
@@ -429,8 +454,6 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
|
||||
Route("/api/prompts", api_prompts, methods=["GET"]),
|
||||
Route("/runtime", runtime, methods=["GET"]),
|
||||
Route("/api/runtime", api_runtime, methods=["GET"]),
|
||||
Route("/policy", policy, methods=["GET"]),
|
||||
Route("/api/v1/policy", api_v1_policy, methods=["GET"]),
|
||||
Route("/audit", audit, methods=["GET", "POST"]),
|
||||
Route("/api/audit", api_audit, methods=["GET", "POST"]),
|
||||
Route("/worktrees", worktrees, methods=["GET"]),
|
||||
@@ -454,6 +477,10 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
|
||||
api_console_security_model,
|
||||
methods=["GET"],
|
||||
),
|
||||
*[
|
||||
Route(path, phase_stub, methods=["GET"])
|
||||
for path in STUB_PAGES
|
||||
],
|
||||
],
|
||||
exception_handlers={405: method_not_allowed},
|
||||
)
|
||||
|
||||
+94
-17
@@ -2,29 +2,66 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
NAV_ITEMS = (
|
||||
("/", "Home"),
|
||||
("/queue", "Queue"),
|
||||
("/projects", "Projects"),
|
||||
("/prompts", "Prompts"),
|
||||
("/runtime", "Runtime"),
|
||||
("/policy", "Policy"),
|
||||
("/audit", "Audit"),
|
||||
("/worktrees", "Worktrees"),
|
||||
("/leases", "Leases"),
|
||||
("/actions", "Actions"),
|
||||
)
|
||||
import os
|
||||
|
||||
from webui.nav import NAV_GROUPS
|
||||
|
||||
MVP_NOTICE = (
|
||||
"Read-only MVP — Gitea, MCP tools, and canonical workflows remain the "
|
||||
"source of truth. No mutation endpoints."
|
||||
)
|
||||
|
||||
# Canonical docs entry point surfaced from the shell header (#638).
|
||||
DOCS_URL = (
|
||||
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/src/branch/"
|
||||
"master/docs/webui-local-dev.md"
|
||||
)
|
||||
|
||||
_LOCAL_HOSTS = frozenset({"", "127.0.0.1", "localhost", "::1"})
|
||||
|
||||
|
||||
def environment_label() -> str:
|
||||
"""Classify the serving environment as ``local`` or ``remote`` (#638).
|
||||
|
||||
Derived from the same ``WEBUI_HOST`` default the app binds to; loopback
|
||||
hosts are ``local``, anything else is ``remote``. Read-only signal only.
|
||||
"""
|
||||
host = (os.environ.get("WEBUI_HOST", "127.0.0.1") or "").strip().lower()
|
||||
return "local" if host in _LOCAL_HOSTS else "remote"
|
||||
|
||||
|
||||
def _render_nav() -> str:
|
||||
groups_html = []
|
||||
for group in NAV_GROUPS:
|
||||
links = "".join(
|
||||
f'<a href="{item.href}"'
|
||||
+ (' class="nav-stub"' if item.status == "stub" else "")
|
||||
+ f'>{item.label}</a>'
|
||||
for item in group.items
|
||||
)
|
||||
groups_html.append(
|
||||
'<div class="nav-group">'
|
||||
f'<span class="nav-group-label">{group.label}</span>'
|
||||
f'<span class="nav-group-links">{links}</span>'
|
||||
"</div>"
|
||||
)
|
||||
return "".join(groups_html)
|
||||
|
||||
|
||||
def _render_badges() -> str:
|
||||
env = environment_label()
|
||||
return (
|
||||
'<div class="header-badges">'
|
||||
f'<span class="badge env-badge env-{env}">env: {env}</span>'
|
||||
'<span class="badge mode-badge">mode: read-only</span>'
|
||||
f'<a class="badge docs-link" href="{DOCS_URL}">Docs</a>'
|
||||
"</div>"
|
||||
)
|
||||
|
||||
|
||||
def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
|
||||
nav_links = "".join(
|
||||
f'<a href="{href}">{label}</a>' for href, label in NAV_ITEMS
|
||||
)
|
||||
nav_links = _render_nav()
|
||||
header_badges = _render_badges()
|
||||
return f"""<!DOCTYPE html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
@@ -54,21 +91,58 @@ def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
|
||||
padding: 0.75rem 1.25rem;
|
||||
}}
|
||||
header h1 {{
|
||||
margin: 0 0 0.5rem;
|
||||
margin: 0;
|
||||
font-size: 1.1rem;
|
||||
font-weight: 600;
|
||||
}}
|
||||
.header-top {{
|
||||
display: flex;
|
||||
flex-wrap: wrap;
|
||||
align-items: center;
|
||||
justify-content: space-between;
|
||||
gap: 0.5rem 1rem;
|
||||
margin-bottom: 0.6rem;
|
||||
}}
|
||||
.header-badges {{ display: inline-flex; flex-wrap: wrap; gap: 0.4rem; }}
|
||||
.env-badge.env-local {{ color: #8fd19e; border-color: #3d6b4a; }}
|
||||
.env-badge.env-remote {{ color: #e0c27a; border-color: #6b5730; }}
|
||||
.mode-badge {{ color: #9ec8f0; border-color: #3d5f7a; }}
|
||||
a.docs-link {{
|
||||
color: var(--accent);
|
||||
border-color: var(--accent);
|
||||
text-decoration: none;
|
||||
text-transform: none;
|
||||
}}
|
||||
a.docs-link:hover {{ filter: brightness(1.12); }}
|
||||
nav {{
|
||||
display: flex;
|
||||
flex-wrap: wrap;
|
||||
gap: 0.75rem 1rem;
|
||||
gap: 0.5rem 1.25rem;
|
||||
}}
|
||||
.nav-group {{
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
gap: 0.15rem;
|
||||
}}
|
||||
.nav-group-label {{
|
||||
font-size: 0.68rem;
|
||||
text-transform: uppercase;
|
||||
letter-spacing: 0.04em;
|
||||
color: var(--muted);
|
||||
}}
|
||||
.nav-group-links {{ display: inline-flex; flex-wrap: wrap; gap: 0.6rem; }}
|
||||
nav a {{
|
||||
color: var(--accent);
|
||||
text-decoration: none;
|
||||
font-size: 0.9rem;
|
||||
}}
|
||||
nav a:hover {{ text-decoration: underline; }}
|
||||
nav a.nav-stub {{ color: var(--muted); }}
|
||||
nav a.nav-stub::after {{
|
||||
content: " ·stub";
|
||||
font-size: 0.7rem;
|
||||
color: var(--muted);
|
||||
}}
|
||||
main {{
|
||||
max-width: 52rem;
|
||||
margin: 0 auto;
|
||||
@@ -167,7 +241,10 @@ def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
|
||||
</head>
|
||||
<body>
|
||||
<header>
|
||||
<div class="header-top">
|
||||
<h1>MCP Control Plane</h1>
|
||||
{header_badges}
|
||||
</div>
|
||||
<nav>{nav_links}</nav>
|
||||
</header>
|
||||
<main>
|
||||
|
||||
+111
@@ -0,0 +1,111 @@
|
||||
"""Navigation IA for the Phase 1 operator console shell (#638).
|
||||
|
||||
Single source of truth for the console navigation so ``webui/layout.py`` and
|
||||
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
|
||||
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.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class NavItem:
|
||||
"""A single navigation destination.
|
||||
|
||||
``status`` is ``"live"`` for implemented views and ``"stub"`` for Phase 1
|
||||
placeholders whose backing view lands in a later child issue.
|
||||
"""
|
||||
|
||||
href: str
|
||||
label: str
|
||||
status: str = "live"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class NavGroup:
|
||||
label: str
|
||||
items: tuple[NavItem, ...]
|
||||
|
||||
|
||||
NAV_GROUPS: tuple[NavGroup, ...] = (
|
||||
NavGroup("Health", (
|
||||
NavItem("/health", "Liveness"),
|
||||
)),
|
||||
NavGroup("Traffic", (
|
||||
NavItem("/queue", "Queue"),
|
||||
NavItem("/leases", "Leases"),
|
||||
NavItem("/actions", "Actions"),
|
||||
)),
|
||||
NavGroup("Runtime/Sessions", (
|
||||
NavItem("/runtime", "Runtime health"),
|
||||
NavItem("/sessions", "Sessions", "stub"),
|
||||
)),
|
||||
NavGroup("Projects", (
|
||||
NavItem("/projects", "Projects"),
|
||||
)),
|
||||
NavGroup("Inventory", (
|
||||
NavItem("/inventory", "Inventory", "stub"),
|
||||
NavItem("/worktrees", "Worktrees"),
|
||||
)),
|
||||
NavGroup("Timeline", (
|
||||
NavItem("/timeline", "Timeline", "stub"),
|
||||
)),
|
||||
NavGroup("Policy", (
|
||||
NavItem("/policy", "Policy", "stub"),
|
||||
NavItem("/prompts", "Prompts"),
|
||||
)),
|
||||
NavGroup("Insights", (
|
||||
NavItem("/insights", "Insights", "stub"),
|
||||
NavItem("/audit", "Audit"),
|
||||
)),
|
||||
)
|
||||
|
||||
|
||||
# Phase 1 placeholder destinations whose backing views land in later child
|
||||
# issues of epic #631. Each maps a path to (title, description). Routes are
|
||||
# registered so nav links resolve to a graceful, read-only stub page.
|
||||
STUB_PAGES: dict[str, tuple[str, str]] = {
|
||||
"/sessions": (
|
||||
"Sessions",
|
||||
"Active session, capability, and role inventory. Backed by the unified "
|
||||
"inventory API (#636) once it lands.",
|
||||
),
|
||||
"/inventory": (
|
||||
"Inventory",
|
||||
"Unified sessions, leases, locks, namespaces, and worktree inventory. "
|
||||
"Backed by the Phase 1 inventory API (#636).",
|
||||
),
|
||||
"/timeline": (
|
||||
"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 "
|
||||
"phase.",
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def iter_nav_items():
|
||||
"""Yield every ``NavItem`` across all groups in declared order."""
|
||||
for group in NAV_GROUPS:
|
||||
for item in group.items:
|
||||
yield item
|
||||
|
||||
|
||||
def nav_hrefs() -> tuple[str, ...]:
|
||||
"""Return every navigation href in declared order."""
|
||||
return tuple(item.href for item in iter_nav_items())
|
||||
@@ -1,387 +0,0 @@
|
||||
"""Read-only workflow policy and guardrail inventory for the web UI (#646).
|
||||
|
||||
Policy and guardrails live in code, profiles, docs, and skills. An operator
|
||||
cannot *see* the active workflow policy configuration from the console without
|
||||
reading the repository tree. This module projects the major guardrails into a
|
||||
redacted, machine-readable inventory with source attribution (file / module /
|
||||
doc), so the console can render them as HTML tables with source pointers.
|
||||
|
||||
Design constraints (Phase 3, #646):
|
||||
|
||||
- **Read-only projection.** Nothing here edits policy or exposes a toggle that
|
||||
could weaken a gate. It reports what is already enforced elsewhere.
|
||||
- **Source attribution without secrets.** Every guardrail carries pointers to
|
||||
the file/module/doc that owns it. Live values are compact summaries derived
|
||||
from the safe policy accessors that already exist (``rbac_matrix``,
|
||||
``redaction_policy``, ``audit_policy``); raw regex, tokens, and endpoints are
|
||||
never embedded.
|
||||
- **Redact before emit.** ``snapshot_to_dict`` runs the whole payload through
|
||||
``console_redaction.redact_payload`` so a planted or accidental secret in any
|
||||
projected value degrades to the placeholder rather than reaching a client.
|
||||
- **Fail soft.** A projection that raises is recorded as a per-entry error and
|
||||
never takes the page down; a guardrail is still listed with its sources.
|
||||
- **Diff vs documented defaults where feasible.** When a guardrail declares a
|
||||
documented invariant, the active projection is compared against it and the
|
||||
result is reported; otherwise the diff is explicitly ``None`` with a reason.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Callable
|
||||
|
||||
from webui import console_audit
|
||||
from webui import console_authz
|
||||
from webui import console_redaction
|
||||
|
||||
SCHEMA_VERSION = 1
|
||||
|
||||
READ_ONLY_NOTE = (
|
||||
"Read-only projection of guardrails enforced in code, profiles, docs, and "
|
||||
"skills. This view never edits policy and exposes no gate-weakening toggle."
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SourcePointer:
|
||||
"""Where a guardrail is defined. Attribution only — never a secret."""
|
||||
|
||||
label: str
|
||||
path: str
|
||||
kind: str # "module" | "doc" | "script" | "config"
|
||||
anchor: str | None = None
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"label": self.label,
|
||||
"path": self.path,
|
||||
"kind": self.kind,
|
||||
"anchor": self.anchor,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PolicyEntry:
|
||||
key: str
|
||||
title: str
|
||||
category: str
|
||||
summary: str
|
||||
sources: tuple[SourcePointer, ...]
|
||||
active: dict[str, Any] | None
|
||||
documented_default: dict[str, Any] | None
|
||||
diff: dict[str, Any] | None
|
||||
error: str | None = None
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"key": self.key,
|
||||
"title": self.title,
|
||||
"category": self.category,
|
||||
"summary": self.summary,
|
||||
"sources": [s.to_dict() for s in self.sources],
|
||||
"active": self.active,
|
||||
"documented_default": self.documented_default,
|
||||
"diff": self.diff,
|
||||
"error": self.error,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PolicyInventorySnapshot:
|
||||
schema_version: int
|
||||
read_only: bool
|
||||
note: str
|
||||
entries: tuple[PolicyEntry, ...]
|
||||
categories: tuple[str, ...]
|
||||
build_errors: tuple[str, ...]
|
||||
|
||||
|
||||
def _diff_active_vs_default(
|
||||
active: dict[str, Any] | None,
|
||||
documented_default: dict[str, Any] | None,
|
||||
) -> dict[str, Any] | None:
|
||||
"""Compare only the keys the documented default declares.
|
||||
|
||||
Returns ``None`` when no documented default is declared (diff not feasible)
|
||||
or when the active projection is unavailable. Otherwise reports, per
|
||||
declared key, whether the active value matches the documented invariant.
|
||||
"""
|
||||
if not documented_default:
|
||||
return None
|
||||
if not active:
|
||||
return {"status": "active_unavailable", "checked": {}}
|
||||
checked: dict[str, Any] = {}
|
||||
matches = True
|
||||
for key, expected in documented_default.items():
|
||||
observed = active.get(key)
|
||||
ok = observed == expected
|
||||
matches = matches and ok
|
||||
checked[key] = {"expected": expected, "observed": observed, "matches": ok}
|
||||
return {
|
||||
"status": "matches_documented_default" if matches else "drift_detected",
|
||||
"checked": checked,
|
||||
}
|
||||
|
||||
|
||||
# ── Live projections (compact, safe, fail-soft) ──────────────────────────────
|
||||
# Each returns a small dict of already-safe machine values. They are module
|
||||
# level so tests can substitute one to prove the redaction pass runs.
|
||||
|
||||
|
||||
def _project_role_separation() -> dict[str, Any]:
|
||||
matrix = console_authz.rbac_matrix()
|
||||
return {
|
||||
"model_version": matrix.get("model_version"),
|
||||
"active_phase": matrix.get("active_phase"),
|
||||
"roles": [r.get("role") for r in matrix.get("roles", [])],
|
||||
"privileged_action_count": len(matrix.get("privileged_actions", [])),
|
||||
"default_decision": matrix.get("default_decision"),
|
||||
"execution_enabled": matrix.get("execution_enabled"),
|
||||
}
|
||||
|
||||
|
||||
def _project_redaction() -> dict[str, Any]:
|
||||
policy = console_redaction.redaction_policy()
|
||||
return {
|
||||
"policy_version": policy.get("policy_version"),
|
||||
"placeholder": policy.get("placeholder"),
|
||||
"applies_to": policy.get("applies_to"),
|
||||
"console_detector_count": len(policy.get("console_rules", [])),
|
||||
"redact_before_persist": policy.get("redact_before_persist"),
|
||||
"failure_mode": policy.get("failure_mode"),
|
||||
}
|
||||
|
||||
|
||||
def _project_audit() -> dict[str, Any]:
|
||||
policy = console_audit.audit_policy()
|
||||
return {
|
||||
"schema_version": policy.get("schema_version"),
|
||||
"required_field_count": len(policy.get("required_fields", [])),
|
||||
"results": policy.get("results"),
|
||||
"retention_defaults_days": policy.get("retention_defaults_days"),
|
||||
"append_only": policy.get("append_only"),
|
||||
"redact_before_persist": policy.get("redact_before_persist"),
|
||||
"enabled": policy.get("enabled"),
|
||||
}
|
||||
|
||||
|
||||
def _static(value: dict[str, Any]) -> Callable[[], dict[str, Any]]:
|
||||
return lambda: dict(value)
|
||||
|
||||
|
||||
# ── Guardrail catalog ────────────────────────────────────────────────────────
|
||||
# One row per major guardrail. ``project`` yields the active value (may raise;
|
||||
# caught per entry). ``documented_default`` drives the feasible diff.
|
||||
|
||||
_CatalogRow = tuple[
|
||||
str,
|
||||
str,
|
||||
str,
|
||||
str,
|
||||
tuple[SourcePointer, ...],
|
||||
Callable[[], dict[str, Any]] | None,
|
||||
dict[str, Any] | None,
|
||||
]
|
||||
|
||||
_CATALOG: tuple[_CatalogRow, ...] = (
|
||||
(
|
||||
"role_separation",
|
||||
"Role separation and RBAC",
|
||||
"role_separation",
|
||||
"Author, reviewer, merger, and reconciler capabilities are disjoint and "
|
||||
"role-exclusive; self-review and self-merge are always blocked. The "
|
||||
"console RBAC model defaults to deny.",
|
||||
(
|
||||
SourcePointer("task capability map", "task_capability_map.py", "module"),
|
||||
SourcePointer("role/namespace gate", "role_namespace_gate.py", "module"),
|
||||
SourcePointer("console RBAC", "webui/console_authz.py", "module"),
|
||||
),
|
||||
_project_role_separation,
|
||||
{"default_decision": "deny", "execution_enabled": False},
|
||||
),
|
||||
(
|
||||
"lease_rules",
|
||||
"Issue and PR lease lifecycle",
|
||||
"lease_rules",
|
||||
"Durable work is claimed through issue locks and control-plane leases "
|
||||
"with freshness, expiry, and dead-session recovery; abandoned or stale "
|
||||
"claims are reclaimed only through the sanctioned recovery path.",
|
||||
(
|
||||
SourcePointer("issue lock store", "issue_lock_store.py", "module"),
|
||||
SourcePointer("branch cleanup guard", "branch_cleanup_guard.py", "module"),
|
||||
SourcePointer("safety model §5", "docs/safety-model.md", "doc", "5-mutation-gating"),
|
||||
),
|
||||
None,
|
||||
None,
|
||||
),
|
||||
(
|
||||
"worktree_rules",
|
||||
"Author worktree binding",
|
||||
"worktree_rules",
|
||||
"Author mutations require a validated worktree under branches/ derived "
|
||||
"from the active issue lock; silent fallback to the stable control "
|
||||
"checkout or master is forbidden (#618).",
|
||||
(
|
||||
SourcePointer("author worktree gate", "author_mutation_worktree.py", "module"),
|
||||
SourcePointer("worktree bootstrap", "scripts/worktree-start", "script"),
|
||||
SourcePointer("workflow scope guard", "workflow_scope_guard.py", "module"),
|
||||
),
|
||||
None,
|
||||
None,
|
||||
),
|
||||
(
|
||||
"merge_confirmation",
|
||||
"Explicit merge confirmation",
|
||||
"merge_confirmation",
|
||||
"A merge fails closed unless the caller passes the exact confirmation "
|
||||
"phrase for that PR; reviewing never implies merging.",
|
||||
(
|
||||
SourcePointer("merge path", "merge_pr.py", "module"),
|
||||
SourcePointer("merge tool gate", "gitea_mcp_server.py", "module"),
|
||||
),
|
||||
_static({"required_confirmation_format": "MERGE PR <n>", "auto_merge": False}),
|
||||
{"auto_merge": False},
|
||||
),
|
||||
(
|
||||
"redaction",
|
||||
"Secret redaction",
|
||||
"redaction",
|
||||
"Every console surface runs the shared gitea_audit pass then console "
|
||||
"patterns before any payload, HTML, log line, or audit record leaves "
|
||||
"the server; unredactable values fail closed to the placeholder.",
|
||||
(
|
||||
SourcePointer("console redaction", "webui/console_redaction.py", "module"),
|
||||
SourcePointer("shared redaction", "gitea_audit.py", "module"),
|
||||
SourcePointer("safety model §3", "docs/safety-model.md", "doc", "3-secret-redaction"),
|
||||
),
|
||||
_project_redaction,
|
||||
{"redact_before_persist": True},
|
||||
),
|
||||
(
|
||||
"contamination",
|
||||
"Contamination containment",
|
||||
"contamination",
|
||||
"A session contaminated by a direct stable-branch push or a manual MCP "
|
||||
"daemon kill is blocked from review, merge, close, and completion "
|
||||
"mutations until cleared (reconciler-exempt).",
|
||||
(
|
||||
SourcePointer("contamination gates", "gitea_mcp_server.py", "module"),
|
||||
SourcePointer("stable-branch audit", "workflow_scope_guard.py", "module"),
|
||||
),
|
||||
None,
|
||||
None,
|
||||
),
|
||||
(
|
||||
"allocator_policy",
|
||||
"Work allocation policy",
|
||||
"allocator_policy",
|
||||
"Workers do not self-select exclusive work; the controller-owned "
|
||||
"allocator ranks the complete queue by priority then PRs-before-issues "
|
||||
"then ascending number, honoring dependency edges and foreign claims.",
|
||||
(
|
||||
SourcePointer("allocator", "gitea_mcp_server.py", "module"),
|
||||
SourcePointer("safety model §5", "docs/safety-model.md", "doc", "5-mutation-gating"),
|
||||
),
|
||||
_static(
|
||||
{
|
||||
"self_select_exclusive_work": False,
|
||||
"ranking": "priority desc, PRs before issues, number asc",
|
||||
"respects_dependency_edges": True,
|
||||
"respects_foreign_claims": True,
|
||||
}
|
||||
),
|
||||
{"self_select_exclusive_work": False},
|
||||
),
|
||||
(
|
||||
"audit_logging",
|
||||
"Audit logging",
|
||||
"audit_logging",
|
||||
"Console intent and authorization outcomes are recorded to an "
|
||||
"append-only, redact-before-persist audit log; MCP mutations are "
|
||||
"recorded by gitea_audit and correlated by request id.",
|
||||
(
|
||||
SourcePointer("console audit", "webui/console_audit.py", "module"),
|
||||
SourcePointer("MCP audit", "gitea_audit.py", "module"),
|
||||
SourcePointer("safety model §1", "docs/safety-model.md", "doc", "1-audit-logging-and-confirmation"),
|
||||
),
|
||||
_project_audit,
|
||||
{"append_only": True, "redact_before_persist": True},
|
||||
),
|
||||
(
|
||||
"mutation_gating",
|
||||
"Mutation gating and master parity",
|
||||
"mutation_gating",
|
||||
"Mutations fail closed while the running server is stale relative to "
|
||||
"master, and every mutation is preceded by identity and capability "
|
||||
"resolution in a fixed pre-flight order.",
|
||||
(
|
||||
SourcePointer("mutation gate", "gitea_mcp_server.py", "module"),
|
||||
SourcePointer("safety model §5", "docs/safety-model.md", "doc", "5-mutation-gating"),
|
||||
),
|
||||
_static(
|
||||
{
|
||||
"stale_runtime_blocks_mutations": True,
|
||||
"preflight_order": "whoami -> resolve_task_capability -> mutation",
|
||||
}
|
||||
),
|
||||
{"stale_runtime_blocks_mutations": True},
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _build_entry(row: _CatalogRow) -> PolicyEntry:
|
||||
key, title, category, summary, sources, project, documented_default = row
|
||||
active: dict[str, Any] | None = None
|
||||
error: str | None = None
|
||||
if project is not None:
|
||||
try:
|
||||
active = project()
|
||||
except Exception as exc: # noqa: BLE001 — fail soft; never take the page down
|
||||
active = None
|
||||
error = f"active projection unavailable: {exc}"
|
||||
diff = _diff_active_vs_default(active, documented_default)
|
||||
return PolicyEntry(
|
||||
key=key,
|
||||
title=title,
|
||||
category=category,
|
||||
summary=summary,
|
||||
sources=sources,
|
||||
active=active,
|
||||
documented_default=documented_default,
|
||||
diff=diff,
|
||||
error=error,
|
||||
)
|
||||
|
||||
|
||||
def load_policy_inventory() -> PolicyInventorySnapshot:
|
||||
"""Build the read-only guardrail inventory. Never raises for one bad entry."""
|
||||
entries: list[PolicyEntry] = []
|
||||
build_errors: list[str] = []
|
||||
for row in _CATALOG:
|
||||
try:
|
||||
entries.append(_build_entry(row))
|
||||
except Exception as exc: # noqa: BLE001 — one row must not break the rest
|
||||
build_errors.append(f"{row[0]}: {exc}")
|
||||
categories = tuple(dict.fromkeys(e.category for e in entries))
|
||||
return PolicyInventorySnapshot(
|
||||
schema_version=SCHEMA_VERSION,
|
||||
read_only=True,
|
||||
note=READ_ONLY_NOTE,
|
||||
entries=tuple(entries),
|
||||
categories=categories,
|
||||
build_errors=tuple(build_errors),
|
||||
)
|
||||
|
||||
|
||||
def snapshot_to_dict(snapshot: PolicyInventorySnapshot) -> dict[str, Any]:
|
||||
"""Serialize the snapshot, redacting the entire payload before it is emitted."""
|
||||
payload = {
|
||||
"schema_version": snapshot.schema_version,
|
||||
"read_only": snapshot.read_only,
|
||||
"note": snapshot.note,
|
||||
"categories": list(snapshot.categories),
|
||||
"entry_count": len(snapshot.entries),
|
||||
"entries": [entry.to_dict() for entry in snapshot.entries],
|
||||
"build_errors": list(snapshot.build_errors),
|
||||
}
|
||||
return console_redaction.redact_payload(payload)
|
||||
@@ -1,104 +0,0 @@
|
||||
"""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 & 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>"
|
||||
)
|
||||
Reference in New Issue
Block a user