Compare commits

..
1 Commits
Author SHA1 Message Date
jcwalker3andClaude Opus 4.8 08061b7b8a feat(webui): evolve application shell for Phase 1 console IA (Closes #638)
Introduce a single nav-config module (webui/nav.py) driving grouped
navigation across the epic #631 Phase 1 information architecture:
Health, Traffic, Runtime/Sessions, Projects, Inventory, Timeline,
Policy (placeholder), and Insights (placeholder). The shell header now
carries a read-only environment badge (local/remote from WEBUI_HOST), a
mode: read-only badge, and a Docs link. Home summarizes the console
purpose, lists the Phase 1 surfaces by group, and links the MVP legacy
pages.

Not-yet-implemented surfaces (/sessions, /inventory, /timeline,
/policy, /insights) resolve to graceful read-only stub pages rather than
404s; their backing views land in later child issues of #631 (inventory
surfaces backed by #636). Mutating methods on stub routes still fail
closed with read-only-mvp. No privileged action controls are added.

Adds tests/test_webui_shell.py covering nav groups, badge presence,
environment classification, docs link, stub routes (200 + read-only),
and home content. Updates docs/webui-local-dev.md with the new routes
and a Phase 1 shell section.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-22 17:56:28 -05:00
12 changed files with 515 additions and 1879 deletions
+27 -8
View File
@@ -43,10 +43,6 @@ for the console architecture: layer and authority boundaries, the redaction
boundary, `/api/v1/...` versioning, the target page map, and the phase gates
that govern when a write path may open (#632, epic #631).
See [webui-project-registry-api.md](webui-project-registry-api.md) for the
versioned project registry contract: registry schema versions 1 and 2, project
status, onboarding checklist state, and the fail-closed error payloads (#635).
## Routes (MVP)
| Path | Description |
@@ -55,11 +51,9 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
| `/health` | JSON liveness (`status`, `service`, `mode`, `timestamp`) |
| `/queue` | Live PR and issue queue dashboard (#429) |
| `/api/queue` | JSON queue export with pagination metadata |
| `/projects` | Project registry list with status and onboarding progress (#427, #635) |
| `/projects` | Project registry list (#427) |
| `/projects/{id}` | Project detail + onboarding checklist |
| `/api/v1/projects` | Versioned JSON registry export (#635) |
| `/api/v1/projects/{id}` | Versioned JSON project detail (#635) |
| `/api/projects` | JSON registry export — unversioned Phase 1 alias of `/api/v1/projects` |
| `/api/projects` | JSON registry export |
| `/prompts` | Prompt library with per-prompt copy buttons (#428) |
| `/api/prompts` | JSON prompt export with workflow hashes |
| `/runtime` | MCP runtime health and stale detection (#430) |
@@ -73,6 +67,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
@@ -153,6 +152,26 @@ 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.
## Application shell — Phase 1 (#638)
The console shell (`webui/layout.py`) renders a grouped navigation driven by a
single nav-config module, `webui/nav.py`. Nav groups follow the epic #631
Phase 1 information architecture: **Health, Traffic, Runtime/Sessions,
Projects, Inventory, Timeline, Policy** (placeholder), and **Insights**
(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)
MVP serves on loopback by default. Binding `0.0.0.0` or `::` is **refused**
-213
View File
@@ -1,213 +0,0 @@
# Project registry API (#635)
Phase 1 of the [console architecture ADR](architecture/webui-control-plane-console-architecture-adr.md)
gives the project registry a versioned, read-only API. This document is the
field-by-field contract for that API and for the registry file behind it.
Everything here is **read-only**. The console never writes the registry; an
operator edits the JSON file, and an invalid file fails closed rather than
rendering a partial inventory.
## Routes
| Route | Method | Description |
|-------|--------|-------------|
| `/api/v1/projects` | GET | Versioned registry export: all projects, with provenance |
| `/api/v1/projects/{project_id}` | GET | Single project; `404` with `project_not_found` when unknown |
| `/api/projects` | GET | Unversioned MVP alias (#427), retained for all of Phase 1 |
| `/projects` | GET | HTML list — status and onboarding progress per project |
| `/projects/{project_id}` | GET | HTML detail — identity, profiles, paths, checklist |
Per ADR section 6 the unversioned alias may be retired no earlier than Phase 2,
and only after this document and `webui-local-dev.md` record the swap. The alias
returns the same payload as `/api/v1/projects`, including the legacy `version`
and `source_path` keys #427 consumers already read.
The HTML views render from the same DTO the JSON routes serialize
(`project_to_dict`), so the console and the API cannot disagree about a
project's status or onboarding progress.
## Registry file
Default location: `webui/data/projects.registry.json`. Override with the
`WEBUI_PROJECT_REGISTRY` environment variable.
Schema versions: **1** and **2** are accepted; **2** is current. A version 1
file loads unchanged and is normalized with the documented defaults, so an
existing operator registry keeps working without edits.
### Root
| Field | Type | Required | Notes |
|-------|------|----------|-------|
| `version` | int | yes | `1` or `2`. Anything else fails closed |
| `projects` | array | yes | Must be non-empty |
### Project
| Field | Type | Required | Default | Notes |
|-------|------|----------|---------|-------|
| `id` | string | yes | — | Stable registry id used in URLs |
| `repo_name` | string | yes | — | Gitea repository name |
| `gitea_owner` | string | yes | — | Owning org or user |
| `remote_host` | string | yes | — | Instance base URL, no credentials |
| `remote_name` | string | no | `null` | Logical remote label, e.g. `prgs` (v2) |
| `default_branch` | string | yes | — | Stable branch name |
| `local_checkout_path` | string | yes | — | Control checkout path |
| `status` | string | no | `active` | `active`, `onboarding`, `paused`, `archived` (v2) |
| `profiles` | object | yes | — | Must map `author`, `reviewer`, `reconciler` |
| `workflow_paths` | object | yes | — | Non-empty; label to repo-relative path |
| `schema_paths` | object | no | `{}` | Label to repo-relative path |
| `onboarding_checklist` | array | no | `[]` | See below |
| `last_seen_health` | object | no | `null` | Redacted health only (v2) |
### Onboarding step
| Field | Type | Required | Default | Notes |
|-------|------|----------|---------|-------|
| `id` | string | yes | — | Stable step id |
| `title` | string | yes | — | Short operator-facing label |
| `description` | string | yes | — | Self-contained; assumes no chat history |
| `state` | string | no | `pending` | `complete`, `pending`, `blocked`, `not_applicable` (v2) |
| `required` | bool | no | `true` | Optional steps never block readiness (v2) |
### Last-seen health
| Field | Type | Required | Notes |
|-------|------|----------|-------|
| `status` | string | no (default `unknown`) | `healthy`, `degraded`, `unreachable`, `unknown` |
| `checked_at` | string | no | ISO-8601 UTC timestamp, e.g. `2026-01-01T00:00:00Z` |
| `detail` | string | no | Short redacted note |
Health is recorded metadata, not a live probe: Phase 1 performs no outbound
health checks. Endpoints, tokens, and keychain identifiers must never appear
here.
## Response shape
`GET /api/v1/projects`:
```json
{
"api_version": "v1",
"schema_version": 2,
"version": 2,
"source_path": "/path/to/webui/data/projects.registry.json",
"source": {
"kind": "file",
"path": "/path/to/webui/data/projects.registry.json",
"inventory_complete": true
},
"project_count": 1,
"projects": [
{
"id": "example",
"repo_name": "Example",
"gitea_owner": "Org",
"repo_full_name": "Org/Example",
"remote_host": "https://gitea.example.invalid",
"remote_name": "example-remote",
"default_branch": "main",
"local_checkout_path": ".",
"status": "active",
"profiles": {"author": "...", "reviewer": "...", "reconciler": "..."},
"workflow_paths": {"skill": "skills/..."},
"schema_paths": {},
"onboarding_checklist": [
{
"id": "profiles",
"title": "Configure execution profiles",
"description": "...",
"state": "complete",
"required": true
}
],
"onboarding_summary": {
"total": 1,
"complete": 1,
"pending": 0,
"blocked": 0,
"not_applicable": 0,
"required_outstanding": 0,
"onboarding_complete": true
},
"last_seen_health": null
}
]
}
```
`GET /api/v1/projects/{project_id}` returns `api_version`, `schema_version`,
`source`, and a single `project` object with the same fields.
The `source` block satisfies the ADR section 6 provenance rule: every payload
states where the data came from and whether the inventory is complete. A
file-backed registry is always complete — there is no pagination to truncate it.
`onboarding_summary` is derived, never stored. `required_outstanding` counts
steps that are `required` **and** in state `pending` or `blocked`;
`onboarding_complete` is true when that count is zero.
## Fail-closed errors
Validation failures raise `RegistryError`, which routes render instead of a
traceback.
`404` — unknown project id on `/api/v1/projects/{project_id}`:
```json
{
"error": "project_not_found",
"project_id": "not-registered",
"known_project_ids": ["example"],
"remediation": "Request one of the known project ids, or add the project ...",
"source": {"kind": "file", "path": "...", "inventory_complete": true}
}
```
`500` — invalid registry, on both the versioned route and the alias:
```json
{
"error": "registry_invalid",
"detail": "unsupported registry version: 42",
"remediation": "Set 'version' to one of 1, 2 (current schema is 2) ...",
"field_path": "version",
"source_path": "/path/to/registry.json"
}
```
`field_path` points at the offending location (`projects[0].profiles.reconciler`,
`projects[0].onboarding_checklist[2].state`, and so on). The HTML routes render
the same detail, field, source, and remediation on a "Project registry
unavailable" page.
Conditions that fail closed:
* file missing or unreadable;
* invalid JSON (the remediation names line and column);
* root not an object, or `projects` missing/empty;
* unsupported `version`;
* a credential-shaped key anywhere in the file (`token`, `*_secret`, `auth_*`, and similar);
* a project missing a required field, or missing an `author`/`reviewer`/`reconciler` profile;
* an unknown `status`, onboarding `state`, or health `status`.
## Credential rule
The registry stores redacted metadata only. Credential-shaped keys are
rejected at load time, before any DTO is built, consistent with
[safety-model.md](safety-model.md) and
[credential-isolation.md](credential-isolation.md). Tokens live in the keychain
and are resolved server-side by `gitea_auth`.
## Migrating a version 1 registry
1. Set `"version": 2`.
2. Optionally add `"status"` per project (omitted means `active`).
3. Optionally add `"remote_name"` per project.
4. Optionally add `"state"` and `"required"` to each onboarding step (omitted
means `pending` and `true`).
5. Optionally add `"last_seen_health"`.
No step is mandatory: a version 1 file keeps loading. Bumping the version only
declares that the file may use the v2 fields.
+2 -18
View File
@@ -9173,27 +9173,11 @@ def gitea_publish_unpublished_issue_branch(
if blocked:
return blocked
# #815: resolve the caller's worktree *before* preflight and forward it, so
# every workspace-resolution layer behind verify_preflight_purity — including
# the #618 branches-only guard — judges the registered issue worktree this
# publication actually operates on. Resolving it afterwards let preflight
# fall back to the MCP process root, so a daemon rooted at the stable control
# checkout refused a valid explicit worktree before the assessor ever ran.
# A caller supplying nothing usable forwards None and keeps the ordinary
# fail-closed fallback.
explicit_worktree = (worktree_path or "").strip()
workspace = os.path.realpath(os.path.abspath(explicit_worktree or "."))
verify_preflight_purity(
remote,
worktree_path=workspace if explicit_worktree else None,
task=task,
org=org,
repo=repo,
)
verify_preflight_purity(remote, task=task, org=org, repo=repo)
h, o, r = _resolve(remote, host, org, repo)
git_remote = (git_remote_name or remote or "").strip()
workspace = os.path.realpath(os.path.abspath((worktree_path or "").strip() or "."))
existing_lock = issue_lock_store.load_issue_lock(
remote=remote, org=o, repo=r, issue_number=int(issue_number)
@@ -1,622 +0,0 @@
"""Publication preflight must receive the caller's worktree (#815).
``gitea_publish_unpublished_issue_branch`` takes a **required** ``worktree_path``
but resolved it only *after* ``verify_preflight_purity`` had already run. Every
workspace-resolution layer behind that preflight — canonical root, root checkout,
create-issue bootstrap, the #618 branches-only guard, issue scope, and anti-stomp
— therefore received ``None`` and fell back to the MCP process root. A daemon
rooted at the stable control checkout refused a valid registered issue worktree
that the caller had explicitly supplied, before the publication assessor ever ran.
The #812 suite could not see this. Its fixture sets ``self.worktree =
os.path.realpath(self.repo)`` and patches ``PROJECT_ROOT`` to that same path, so
the fallback resolved to the very worktree the argument named. The production
topology — control checkout on a stable branch, issue worktree somewhere else —
was never constructed, and preflight additionally no-ops under pytest unless
production guards are forced on.
These tests build that topology honestly:
* ``PROJECT_ROOT`` is a control checkout sitting on ``master``;
* the registered issue worktree is a genuinely separate path under ``branches/``;
* ``GITEA_TEST_FORCE_PRODUCTION_GUARDS`` is set so the #618 guard really runs;
* no patch makes the issue worktree appear to be ``PROJECT_ROOT``.
Nothing here reads, writes, or references the protected worktree named in #812
AC17 and #815 AC9. Every fixture is built from scratch against a local bare
remote, so publication and read-after-write verification genuinely execute.
"""
from __future__ import annotations
import os
import subprocess
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import issue_lock_provenance # noqa: E402
import issue_lock_store # noqa: E402
import mcp_server # noqa: E402
from mutation_profile_fixture import shared_mutation_env # noqa: E402
ISSUE = 9815
BRANCH = f"feat/issue-{ISSUE}-forwarding-fixture"
WORKTREE_DIRNAME = BRANCH.replace("/", "-")
IDENTITY = "example-user"
PROFILE = "test-author-prgs"
ORG = "Scaled-Tech-Consulting"
REPO = "Gitea-Tools"
GIT_REMOTE = "prgs"
AUTHOR_PROFILE = {
"profile_name": "prgs-author",
"role": "author",
"allowed_operations": [
"gitea.read", "gitea.issue.create", "gitea.issue.comment",
"gitea.pr.create", "gitea.repo.commit", "gitea.branch.push",
],
"forbidden_operations": [],
"audit_label": "prgs-author",
}
def _ts(hours: int) -> str:
return (
(datetime.now(timezone.utc) + timedelta(hours=hours))
.isoformat()
.replace("+00:00", "Z")
)
class TestPreflightReceivesTheWorktree(unittest.TestCase):
"""AC1 — the supplied path reaches ``verify_preflight_purity`` itself.
Follows the #735 capture pattern: replace preflight with a recorder that
raises, so the argument can be proven forwarded without performing the
mutation. This is the direct unit-level statement of the defect.
"""
def _capture_preflight(self, **kwargs):
captured: dict = {}
def _capture(*a, **kw):
captured.update(kw)
captured["_args"] = a
raise RuntimeError("capture-only")
with patch.object(
mcp_server, "verify_preflight_purity", side_effect=_capture
), patch.object(
mcp_server, "get_profile", return_value=AUTHOR_PROFILE
), patch.object(
mcp_server, "_resolve",
return_value=("gitea.prgs.cc", ORG, REPO),
), patch.object(
mcp_server, "_auth", return_value="token fake",
), patch.object(
mcp_server.role_session_router,
"check_author_mutation_after_reviewer_stop",
return_value=(True, []),
), patch.object(
mcp_server, "_namespace_mutation_block", return_value=None
), patch.object(
mcp_server, "_profile_permission_block", return_value=None
):
try:
mcp_server.gitea_publish_unpublished_issue_branch(**kwargs)
except RuntimeError as exc:
if "capture-only" not in str(exc) and not captured:
raise
self.assertTrue(
captured,
"gitea_publish_unpublished_issue_branch never called "
"verify_preflight_purity",
)
return captured
def _base_kwargs(self, **overrides):
kwargs = {
"issue_number": ISSUE,
"branch_name": BRANCH,
"worktree_path": "/tmp/issue-815-explicit-worktree",
"expected_head": "a" * 40,
"remote": "prgs",
"org": ORG,
"repo": REPO,
"git_remote_name": GIT_REMOTE,
}
kwargs.update(overrides)
return kwargs
def test_explicit_worktree_path_reaches_preflight(self):
captured = self._capture_preflight(**self._base_kwargs())
self.assertEqual(
captured.get("worktree_path"),
os.path.realpath(os.path.abspath("/tmp/issue-815-explicit-worktree")),
"the authoritative worktree_path must be forwarded into preflight",
)
def test_forwarded_path_is_the_one_publication_uses(self):
"""AC4 — preflight and publication must judge the same resolved path."""
raw = "/tmp/issue-815-explicit-worktree/./"
captured = self._capture_preflight(**self._base_kwargs(worktree_path=raw))
expected = os.path.realpath(os.path.abspath(raw.strip()))
self.assertEqual(captured.get("worktree_path"), expected)
def test_blank_worktree_path_forwards_none(self):
"""AC5/AC8 — nothing usable supplied keeps the fail-closed fallback."""
for blank in ("", " "):
with self.subTest(blank=repr(blank)):
captured = self._capture_preflight(
**self._base_kwargs(worktree_path=blank)
)
self.assertIsNone(
captured.get("worktree_path"),
"a blank worktree must not resolve to the process cwd",
)
def test_org_repo_and_task_forwarding_are_not_regressed(self):
"""AC6 — #735's org/repo forwarding and the task name still hold."""
captured = self._capture_preflight(**self._base_kwargs())
self.assertEqual(captured.get("org"), ORG)
self.assertEqual(captured.get("repo"), REPO)
self.assertEqual(captured.get("task"), "publish_unpublished_branch")
class _ProductionTopologyBase(unittest.TestCase):
"""Control checkout on master + a distinct registered issue worktree.
This is the shape the production daemon runs in and the shape the #812
fixture never built. ``PROJECT_ROOT`` is the control checkout; the issue
worktree is a real registered worktree at a different path; production
guards are forced on so the #618 branches-only guard genuinely evaluates.
"""
def setUp(self):
self.lock_dir = tempfile.TemporaryDirectory()
self.addCleanup(self.lock_dir.cleanup)
self.origin = tempfile.mkdtemp(prefix="issue815-origin-")
self.control = tempfile.mkdtemp(prefix="issue815-control-")
for path in (self.origin, self.control):
self.addCleanup(
lambda p=path: subprocess.run(["rm", "-rf", p], check=False)
)
self._init_repos()
self.remotes = patch.dict(
mcp_server.REMOTES,
{"prgs": {"host": "gitea.prgs.cc", "org": ORG, "repo": REPO}},
)
self.remotes.start()
self.addCleanup(patch.stopall)
mcp_server._IDENTITY_CACHE.clear()
def _git(self, *args, cwd=None):
return subprocess.run(
["git", "-C", cwd or self.control, *args],
capture_output=True, text=True, check=True,
)
def _init_repos(self):
subprocess.run(
["git", "init", "-q", "--bare", "-b", "master", self.origin], check=True
)
self._git("init", "-q", "-b", "master")
self._git("config", "user.email", "[email protected]")
self._git("config", "user.name", "Test")
self._git("remote", "add", GIT_REMOTE, self.origin)
with open(os.path.join(self.control, "seed.txt"), "w") as fh:
fh.write("seed\n")
# The real repository gitignores branches/, so a registered worktree
# living there does not dirty the stable control checkout. Mirror that,
# or the #615 dirty-runtime block fires on the worktree we just created.
with open(os.path.join(self.control, ".gitignore"), "w") as fh:
fh.write("branches/\n")
self._git("add", "seed.txt", ".gitignore")
self._git("commit", "-q", "-m", "seed")
self.base_sha = self._git("rev-parse", "HEAD").stdout.strip()
self._git("push", "-q", GIT_REMOTE, "master")
# The control checkout STAYS on master. This is the whole point: the
# daemon's process root is the stable control checkout, never the
# worktree the publication targets.
self.worktree = os.path.realpath(
os.path.join(self.control, "branches", WORKTREE_DIRNAME)
)
self._git("worktree", "add", "-q", "-b", BRANCH, self.worktree, "master")
with open(os.path.join(self.worktree, "work.txt"), "w") as fh:
fh.write("unpublished implementation\n")
self._git("add", "work.txt", cwd=self.worktree)
self._git("commit", "-q", "-m", "unpublished implementation", cwd=self.worktree)
self.head_sha = self._git("rev-parse", "HEAD", cwd=self.worktree).stdout.strip()
self.control_branch = self._git(
"rev-parse", "--abbrev-ref", "HEAD"
).stdout.strip()
# ── durable lock naming the caller and the issue worktree ────────────
def lock_path(self):
return issue_lock_store.lock_file_path(
remote="prgs", org=ORG, repo=REPO, issue_number=ISSUE,
lock_dir=self.lock_dir.name,
)
def write_lock(self, *, bind_session=True, **overrides):
path = self.lock_path()
claimant = overrides.pop(
"claimant", {"username": IDENTITY, "profile": PROFILE}
)
pid = overrides.pop("session_pid", os.getpid())
lease = {
"operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
"issue_number": ISSUE,
"pr_number": None,
"branch": overrides.get("branch_name", BRANCH),
"worktree_path": overrides.get("worktree_path", self.worktree),
"claimant": claimant,
"created_at": _ts(-2),
"last_heartbeat_at": _ts(-2),
"expires_at": _ts(-1),
}
lease.update(overrides.pop("work_lease", {}))
data = {
"issue_number": ISSUE,
"branch_name": BRANCH,
"remote": "prgs",
"org": ORG,
"repo": REPO,
"worktree_path": self.worktree,
"session_pid": pid,
"pid": pid,
"lock_generation": 1,
"work_lease": lease,
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
tool="gitea_lock_issue", claimant=claimant
),
}
data.update(overrides)
data["lock_file_path"] = path
issue_lock_store.save_lock_file(path, data)
# Bind the session pointer so the #683 issue-scope guard resolves an
# owning issue for this author session. In real production the publish
# task does not require a session lock — require_author_lock is keyed on
# the test-only production_guards_forced() flag, which this suite must
# set to make preflight run at all — so this pointer is fixture
# scaffolding to clear a guard production would not apply here, never a
# softening of the worktree-forwarding behaviour under test. The
# preflight-negative cases below leave it unbound precisely so the #618
# guard is reached with no session fallback to rescue a bad worktree.
if bind_session:
pointer = {
"pid": os.getpid(),
"lock_file_path": path,
"issue_number": ISSUE,
"branch_name": data["branch_name"],
"remote": "prgs",
"org": ORG,
"repo": REPO,
}
issue_lock_store.save_lock_file(
issue_lock_store.session_pointer_path(self.lock_dir.name), pointer
)
return path
def _tool_env(self):
env = shared_mutation_env(
PROFILE, include_example_repo=True,
GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
)
env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
# The defect only exists where preflight actually runs. Under pytest the
# production root/branches/scope guards are skipped unless forced on, so
# force them: this test exists to exercise the #618 guard, not to bypass
# it. Parity is pinned to the server's own startup head so repointing
# PROJECT_ROOT does not read as a stale daemon.
env["GITEA_TEST_FORCE_PRODUCTION_GUARDS"] = "1"
# Production is a promoted stable-control runtime. The pytest process
# itself runs from a branches/ worktree, which the #615 runtime-mode
# gate correctly classifies as dev-test; declaring the sanctioned mode
# models the production daemon rather than defeating the gate. Without
# this, forcing production guards on would trip the *runtime-mode* block
# for a reason unrelated to the #815 worktree-forwarding defect.
env["GITEA_MCP_RUNTIME_MODE"] = "stable-control"
startup_head = mcp_server._STARTUP_PARITY.get("startup_head") or ""
env["GITEA_TEST_CURRENT_HEAD"] = startup_head
env["GITEA_TEST_LIVE_REMOTE_HEAD"] = startup_head
return env
def run_publish(self, *, open_prs=None, expected_head=None, **kwargs):
"""Drive the public tool with PROJECT_ROOT pinned to the CONTROL checkout."""
env = self._tool_env()
with patch(
"mcp_server._list_open_pulls", return_value=list(open_prs or [])
), patch(
"mcp_server._auth", return_value="token x"
), patch(
"mcp_server.get_auth_header", return_value="token x"
), patch(
"mcp_server._work_lease_claimant",
return_value={"username": IDENTITY, "profile": PROFILE},
), patch.object(
# NOTE: the control checkout — deliberately NOT self.worktree.
mcp_server, "PROJECT_ROOT", self.control
), patch.dict(os.environ, env, clear=True):
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
return mcp_server.gitea_publish_unpublished_issue_branch(
issue_number=kwargs.pop("issue_number", ISSUE),
branch_name=kwargs.pop("branch_name", BRANCH),
worktree_path=kwargs.pop("worktree_path", self.worktree),
expected_head=expected_head or self.head_sha,
remote="prgs",
git_remote_name=kwargs.pop("git_remote_name", GIT_REMOTE),
**kwargs,
)
def remote_head(self, branch=BRANCH):
res = subprocess.run(
["git", "-C", self.origin, "rev-parse", "--verify", "--quiet", branch],
capture_output=True, text=True, check=False,
)
return (res.stdout or "").strip() or None
class TestForwardingClearsThe618Guard(_ProductionTopologyBase):
"""AC2 — the faithful production reproduction, and the sharpest fix proof.
The production recovery worker had **no** session issue lock — acquiring one
was the very thing the deadlock prevented — so preflight had nothing but the
explicit ``worktree_path`` argument to resolve the workspace from. This class
reproduces exactly that: no session pointer is bound, so there is no
author-lock fallback to rescue a dropped argument.
With the argument forwarded (fixed source) the #618 branches-only guard
accepts the registered issue worktree and the call advances to the next
guard. With the argument dropped (the buggy source this issue reports)
preflight falls back to ``PROJECT_ROOT`` — the stable control checkout — and
the #618 guard traps the call there. The two outcomes are told apart by the
guard that fired, on its own error text.
This test therefore *fails* against the unpatched source (the call is trapped
at #618 instead of clearing it), which is what makes it a regression rather
than a smoke test.
"""
_CONTROL_CHECKOUT_MARKERS = ("stable control checkout", "#618")
def test_explicit_worktree_clears_618_without_a_session_lock(self):
# No write_lock(): the session is deliberately unbound, as in production.
with self.assertRaises(RuntimeError) as ctx:
self.run_publish()
message = str(ctx.exception)
# The workspace guard is satisfied — the failure is the *later* scope
# guard (no owning issue), never the control-checkout refusal. If the
# argument were dropped, this call would be trapped at #618 instead.
for marker in self._CONTROL_CHECKOUT_MARKERS:
self.assertNotIn(
marker, message,
f"the explicit worktree must clear #618; got a control-checkout "
f"refusal instead: {message}",
)
self.assertIn(
"owning issue", message,
f"expected the downstream scope guard to fire, got: {message}",
)
self.assertIsNone(self.remote_head())
def test_dropped_argument_would_be_trapped_at_618(self):
# Simulate the buggy call shape directly: no session lock, and preflight
# given no worktree, exactly as the unpatched source left it. This pins
# the control-checkout refusal that the fix eliminates, so the pair of
# tests brackets the defect from both sides regardless of which source
# version is loaded.
env = self._tool_env()
with patch(
"mcp_server._list_open_pulls", return_value=[]
), patch(
"mcp_server._auth", return_value="token x"
), patch(
"mcp_server.get_auth_header", return_value="token x"
), patch(
"mcp_server._work_lease_claimant",
return_value={"username": IDENTITY, "profile": PROFILE},
), patch.object(
mcp_server, "PROJECT_ROOT", self.control
), patch.dict(os.environ, env, clear=True):
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
with self.assertRaises(RuntimeError) as ctx:
# Drive verify_preflight_purity the way the buggy body did:
# no worktree_path forwarded at all.
mcp_server.verify_preflight_purity(
"prgs",
task="publish_unpublished_branch",
org=ORG,
repo=REPO,
)
message = str(ctx.exception)
self.assertTrue(
any(m in message for m in self._CONTROL_CHECKOUT_MARKERS),
f"a dropped worktree must trap at the control checkout: {message}",
)
self.assertIsNone(self.remote_head())
class TestProductionTopologyPublishes(_ProductionTopologyBase):
"""AC2/AC4/AC7 — the explicit registered worktree is what preflight validates."""
def test_fixture_is_genuinely_the_production_topology(self):
"""Guard the guard: if this drifts, the regression stops meaning anything."""
self.assertNotEqual(
os.path.realpath(self.control), self.worktree,
"the issue worktree must not be PROJECT_ROOT",
)
self.assertEqual(
self.control_branch, "master",
"the control checkout must sit on a stable branch",
)
self.assertTrue(
os.path.realpath(self.worktree).startswith(
os.path.realpath(os.path.join(self.control, "branches")) + os.sep
),
"the issue worktree must live under branches/",
)
listed = subprocess.run(
["git", "-C", self.control, "worktree", "list"],
capture_output=True, text=True, check=True,
).stdout
self.assertIn(
self.worktree, listed, "the issue worktree must be genuinely registered"
)
def test_publishes_from_a_control_rooted_daemon(self):
"""The exact production failure: this refused with #618 before the fix."""
self.write_lock()
self.assertIsNone(self.remote_head(), "fixture must start unpublished")
res = self.run_publish()
self.assertTrue(res.get("success"), res)
self.assertTrue(res.get("performed"), res)
self.assertEqual(self.remote_head(), self.head_sha)
def test_dry_run_uses_the_explicit_worktree(self):
"""AC4 — dry-run reaches the same decision without publishing."""
self.write_lock()
res = self.run_publish(dry_run=True)
self.assertTrue(res.get("success"), res)
self.assertFalse(res.get("performed"), res)
self.assertTrue(res.get("would_publish"), res)
self.assertIsNone(self.remote_head(), "dry-run must not publish")
def test_dry_run_and_apply_agree_on_the_same_worktree(self):
"""AC4 — both paths resolve the same workspace, so both succeed."""
self.write_lock()
dry = self.run_publish(dry_run=True)
self.assertTrue(dry.get("would_publish"), dry)
applied = self.run_publish()
self.assertTrue(applied.get("performed"), applied)
self.assertEqual(self.remote_head(), self.head_sha)
def test_read_after_write_verification_still_runs(self):
"""AC6 — PR #814's post-publication verification is unchanged."""
self.write_lock()
res = self.run_publish()
self.assertTrue(res.get("verified"), res)
self.assertEqual(res.get("remote_head_sha"), self.head_sha)
class TestProductionTopologyFailsClosed(_ProductionTopologyBase):
"""AC3/AC5/AC8 — the fix does not weaken any refusal.
A refusal reaches the caller by one of two mechanisms, and this class holds
them apart deliberately. A bad *workspace* is caught by the #618 preflight
guard, which raises before the assessor is built. A bad *content/ownership*
fact passes preflight (the worktree itself is fine) and is then refused by
the publication assessor, which returns ``success: False``. Both are
fail-closed; asserting the wrong mechanism would hide a regression.
"""
# ── #618 preflight refusals: no session lock, so nothing rescues a bad
# workspace and the guard fires exactly as it does in production ──────
def _assert_preflight_raises(self, **kwargs):
with self.assertRaises(RuntimeError) as ctx:
self.run_publish(**kwargs)
self.assertIsNone(
self.remote_head(), "a blocked publication must not reach the remote"
)
return str(ctx.exception)
def test_blank_worktree_path_fails_closed_via_618(self):
"""AC5 — a blank path forwards None, so preflight sees the control root."""
for blank in ("", " "):
with self.subTest(blank=repr(blank)):
message = self._assert_preflight_raises(worktree_path=blank)
self.assertIn("618", message)
def test_control_checkout_as_worktree_fails_closed_via_618(self):
"""AC5 — naming the stable control checkout explicitly is still refused."""
message = self._assert_preflight_raises(worktree_path=self.control)
self.assertIn("618", message)
def test_unregistered_directory_fails_closed(self):
"""AC3 — a plain directory under branches/ is not a registered worktree."""
bogus = os.path.join(self.control, "branches", "not-a-worktree")
os.makedirs(bogus, exist_ok=True)
self._assert_preflight_raises(worktree_path=bogus)
def test_missing_worktree_path_fails_closed(self):
"""AC3 — a path that does not exist is refused, not silently replaced."""
missing = os.path.join(self.control, "branches", "absent-worktree")
self._assert_preflight_raises(worktree_path=missing)
# ── assessor refusals: preflight passes on a valid worktree, then the
# publication assessor refuses on content/ownership evidence ──────────
def _assert_assessor_refuses(self, **kwargs):
res = self.run_publish(**kwargs)
self.assertFalse(res.get("success"), res)
self.assertFalse(res.get("performed"), res)
self.assertIsNone(self.remote_head())
return res
def test_changed_local_head_still_refuses(self):
"""AC6 — the declared expected_head remains authoritative."""
self.write_lock()
self._assert_assessor_refuses(expected_head="b" * 40)
def test_foreign_claimant_still_refuses(self):
"""AC6 — ownership still comes from the durable lock record."""
self.write_lock(claimant={"username": "someone-else", "profile": PROFILE})
self._assert_assessor_refuses()
def test_dirty_worktree_still_refuses(self):
"""AC6 — cleanliness enforcement survives the forwarding change."""
self.write_lock()
with open(os.path.join(self.worktree, "work.txt"), "a") as fh:
fh.write("uncommitted drift\n")
self._assert_assessor_refuses()
def test_competing_open_pr_still_refuses(self):
"""AC6 — a rival claim on another branch still blocks."""
self.write_lock()
self._assert_assessor_refuses(
open_prs=[{"number": 4242, "head": {"ref": f"fix/issue-{ISSUE}-rival"}}]
)
def test_issue_lock_record_is_not_mutated_by_a_refusal(self):
"""AC6 — record separation (#812 AC23) is unaffected by this change."""
path = self.write_lock()
with open(path, "rb") as fh:
before = fh.read()
self._assert_assessor_refuses(expected_head="c" * 40)
with open(path, "rb") as fh:
self.assertEqual(before, fh.read())
class TestProtectedFixtureNotReferenced(unittest.TestCase):
"""AC9 — this regression never names the protected #635 fixture.
The forbidden tokens are reconstructed from fragments so this assertion
file does not itself contain them and produce a false positive.
"""
def test_no_reference_to_the_protected_worktree(self):
forbidden = [
"issue-635-" + "project-registry-api",
"b2f6e9a6dc40e9651ef8" + "76f322dd0a68bddebfd8",
]
here = os.path.abspath(__file__)
with open(here, "r", encoding="utf-8") as fh:
text = fh.read()
for token in forbidden:
self.assertNotIn(
token, text,
f"the protected #635 fixture must not be referenced: {token}",
)
if __name__ == "__main__":
unittest.main()
+30 -335
View File
@@ -1,4 +1,4 @@
"""Tests for web UI project registry (#427) and its API evolution (#635)."""
"""Tests for web UI project registry (#427)."""
import json
import sys
import tempfile
@@ -11,246 +11,57 @@ from starlette.testclient import TestClient
from webui.app import create_app
from webui.project_registry import (
CURRENT_SCHEMA_VERSION,
REGISTRY_API_VERSION,
SUPPORTED_SCHEMA_VERSIONS,
RegistryError,
default_registry_path,
load_registry,
onboarding_summary,
project_to_dict,
)
from webui.registry_safety import is_forbidden_key
_REPO_ROOT = Path(__file__).resolve().parent.parent
_API_DOC = _REPO_ROOT / "docs" / "webui-project-registry-api.md"
def _valid_project(**overrides):
project = {
"id": "example",
"repo_name": "Example",
"gitea_owner": "Org",
"remote_host": "https://gitea.example.invalid",
"default_branch": "main",
"local_checkout_path": ".",
"profiles": {"author": "a", "reviewer": "r", "reconciler": "c"},
"workflow_paths": {"skill": "skills/x.md"},
}
project.update(overrides)
return project
def _write_registry(payload) -> Path:
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
json.dump(payload, handle)
return Path(handle.name)
class RegistryFileCase(unittest.TestCase):
"""Base class that cleans up temporary registry files."""
def setUp(self):
self._temp_paths: list[Path] = []
def tearDown(self):
for path in self._temp_paths:
path.unlink(missing_ok=True)
def write_registry(self, payload) -> Path:
path = _write_registry(payload)
self._temp_paths.append(path)
return path
class TestProjectRegistryLoader(RegistryFileCase):
class TestProjectRegistryLoader(unittest.TestCase):
def test_default_registry_loads_gitea_tools(self):
registry = load_registry()
self.assertEqual(registry.version, CURRENT_SCHEMA_VERSION)
self.assertEqual(registry.schema_version, CURRENT_SCHEMA_VERSION)
self.assertEqual(registry.api_version, REGISTRY_API_VERSION)
self.assertEqual(registry.version, 1)
self.assertEqual(len(registry.projects), 1)
project = registry.projects[0]
self.assertEqual(project.id, "gitea-tools")
self.assertEqual(project.repo_name, "Gitea-Tools")
self.assertEqual(project.gitea_owner, "Scaled-Tech-Consulting")
self.assertEqual(project.repo_full_name, "Scaled-Tech-Consulting/Gitea-Tools")
self.assertEqual(project.remote_host, "https://gitea.prgs.cc")
self.assertEqual(project.remote_name, "prgs")
self.assertEqual(project.status, "active")
self.assertEqual(project.profiles["author"], "prgs-author")
self.assertEqual(project.profiles["reviewer"], "prgs-reviewer")
self.assertEqual(project.profiles["reconciler"], "prgs-reconciler")
self.assertIn("skill", project.workflow_paths)
self.assertGreaterEqual(len(project.onboarding_checklist), 4)
def test_default_registry_onboarding_summary_is_complete(self):
summary = onboarding_summary(load_registry().projects[0])
self.assertEqual(summary.total, summary.complete)
self.assertEqual(summary.required_outstanding, 0)
self.assertTrue(summary.onboarding_complete)
def test_version_1_registry_still_loads_with_defaults(self):
path = self.write_registry({
"version": 1,
"projects": [
_valid_project(
onboarding_checklist=[
{"id": "step", "title": "Step", "description": "Do it"}
]
)
],
})
registry = load_registry(path)
self.assertEqual(registry.schema_version, 1)
self.assertIn(1, SUPPORTED_SCHEMA_VERSIONS)
project = registry.projects[0]
self.assertEqual(project.status, "active")
self.assertIsNone(project.remote_name)
self.assertIsNone(project.last_seen_health)
step = project.onboarding_checklist[0]
self.assertEqual(step.state, "pending")
self.assertTrue(step.required)
self.assertFalse(onboarding_summary(project).onboarding_complete)
def test_onboarding_summary_counts_states(self):
path = self.write_registry({
"version": 2,
"projects": [
_valid_project(
onboarding_checklist=[
{"id": "a", "title": "A", "description": "d", "state": "complete"},
{"id": "b", "title": "B", "description": "d", "state": "blocked"},
{
"id": "c",
"title": "C",
"description": "d",
"state": "pending",
"required": False,
},
{
"id": "d",
"title": "D",
"description": "d",
"state": "not_applicable",
},
]
)
],
})
summary = onboarding_summary(load_registry(path).projects[0])
self.assertEqual(summary.total, 4)
self.assertEqual(summary.complete, 1)
self.assertEqual(summary.blocked, 1)
self.assertEqual(summary.pending, 1)
self.assertEqual(summary.not_applicable, 1)
# Only the blocked step is both required and outstanding.
self.assertEqual(summary.required_outstanding, 1)
self.assertFalse(summary.onboarding_complete)
def test_last_seen_health_is_parsed_when_present(self):
path = self.write_registry({
"version": 2,
"projects": [
_valid_project(
last_seen_health={
"status": "degraded",
"checked_at": "2026-01-01T00:00:00Z",
"detail": "daemon restart pending",
}
)
],
})
health = load_registry(path).projects[0].last_seen_health
self.assertIsNotNone(health)
self.assertEqual(health.status, "degraded")
self.assertEqual(health.checked_at, "2026-01-01T00:00:00Z")
def test_registry_rejects_credential_keys(self):
path = self.write_registry({
payload = {
"version": 1,
"projects": [_valid_project(id="bad", api_token="redacted-placeholder")],
})
with self.assertRaises(RegistryError) as ctx:
load_registry(path)
self.assertIn("credential", ctx.exception.remediation.lower())
self.assertEqual(ctx.exception.field_path, "projects[0].api_token")
def test_unsupported_version_fails_closed_with_remediation(self):
path = self.write_registry({"version": 99, "projects": [_valid_project()]})
with self.assertRaises(RegistryError) as ctx:
load_registry(path)
self.assertIn("unsupported registry version", ctx.exception.message)
self.assertIn(str(CURRENT_SCHEMA_VERSION), ctx.exception.remediation)
self.assertEqual(ctx.exception.field_path, "version")
def test_missing_required_field_fails_closed(self):
broken = _valid_project()
del broken["default_branch"]
path = self.write_registry({"version": 2, "projects": [broken]})
with self.assertRaises(RegistryError) as ctx:
load_registry(path)
self.assertIn("default_branch", ctx.exception.message)
self.assertEqual(ctx.exception.field_path, "projects[0]")
def test_unknown_status_fails_closed(self):
path = self.write_registry({
"version": 2,
"projects": [_valid_project(status="mystery")],
})
with self.assertRaises(RegistryError) as ctx:
load_registry(path)
self.assertEqual(ctx.exception.field_path, "projects[0].status")
self.assertIn("active", ctx.exception.remediation)
def test_unknown_onboarding_state_fails_closed(self):
path = self.write_registry({
"version": 2,
"projects": [
_valid_project(
onboarding_checklist=[
{"id": "a", "title": "A", "description": "d", "state": "almost"}
]
)
{
"id": "bad",
"repo_name": "Bad",
"gitea_owner": "Org",
"remote_host": "https://gitea.example.invalid",
"default_branch": "main",
"local_checkout_path": ".",
"profiles": {
"author": "a",
"reviewer": "r",
"reconciler": "c",
},
"workflow_paths": {"skill": "skills/x.md"},
"api_token": "secret",
}
],
})
with self.assertRaises(RegistryError) as ctx:
load_registry(path)
self.assertEqual(
ctx.exception.field_path,
"projects[0].onboarding_checklist[0].state",
)
def test_missing_profile_role_fails_closed(self):
path = self.write_registry({
"version": 2,
"projects": [_valid_project(profiles={"author": "a", "reviewer": "r"})],
})
with self.assertRaises(RegistryError) as ctx:
load_registry(path)
self.assertEqual(ctx.exception.field_path, "projects[0].profiles.reconciler")
def test_empty_projects_fails_closed(self):
path = self.write_registry({"version": 2, "projects": []})
with self.assertRaises(RegistryError) as ctx:
load_registry(path)
self.assertEqual(ctx.exception.field_path, "projects")
def test_invalid_json_fails_closed_with_location(self):
}
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
handle.write("{not json")
json.dump(payload, handle)
path = Path(handle.name)
self._temp_paths.append(path)
with self.assertRaises(RegistryError) as ctx:
load_registry(path)
self.assertIn("not valid JSON", ctx.exception.message)
self.assertIn("line", ctx.exception.remediation)
def test_missing_file_fails_closed(self):
missing = Path(tempfile.gettempdir()) / "webui-registry-does-not-exist.json"
with self.assertRaises(RegistryError) as ctx:
load_registry(missing)
self.assertIn("could not be read", ctx.exception.message)
try:
with self.assertRaises(ValueError):
load_registry(path)
finally:
path.unlink(missing_ok=True)
def test_default_registry_path_points_at_packaged_data(self):
path = default_registry_path()
@@ -270,147 +81,31 @@ class TestProjectRegistryRoutes(unittest.TestCase):
self.assertIn("prgs-author", response.text)
self.assertNotIn("child issue", response.text.lower())
def test_projects_page_shows_status_and_progress(self):
response = self.client.get("/projects")
self.assertIn("Status", response.text)
self.assertIn("Onboarding", response.text)
self.assertIn("4/4 complete", response.text)
def test_project_detail_renders_checklist(self):
response = self.client.get("/projects/gitea-tools")
self.assertEqual(response.status_code, 200)
self.assertIn("Onboarding checklist", response.text)
self.assertIn("Configure execution profiles", response.text)
self.assertIn("branches/", response.text)
self.assertIn("Complete", response.text)
self.assertIn("required outstanding 0", response.text)
def test_project_detail_404(self):
response = self.client.get("/projects/unknown-repo")
self.assertEqual(response.status_code, 404)
def test_api_projects_alias_stays_compatible(self):
def test_api_projects_json(self):
response = self.client.get("/api/projects")
self.assertEqual(response.status_code, 200)
data = response.json()
# #427 consumers keep these keys.
self.assertEqual(data["version"], CURRENT_SCHEMA_VERSION)
self.assertIn("source_path", data)
self.assertEqual(data["version"], 1)
self.assertEqual(len(data["projects"]), 1)
self.assertEqual(data["projects"][0]["id"], "gitea-tools")
self.assertIn("onboarding_checklist", data["projects"][0])
def test_api_v1_projects_payload(self):
response = self.client.get("/api/v1/projects")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(data["api_version"], REGISTRY_API_VERSION)
self.assertEqual(data["schema_version"], CURRENT_SCHEMA_VERSION)
self.assertEqual(data["project_count"], 1)
self.assertEqual(data["source"]["kind"], "file")
self.assertTrue(data["source"]["inventory_complete"])
project = data["projects"][0]
self.assertEqual(project["status"], "active")
self.assertEqual(project["remote_name"], "prgs")
self.assertEqual(
project["repo_full_name"], "Scaled-Tech-Consulting/Gitea-Tools"
)
self.assertTrue(project["onboarding_summary"]["onboarding_complete"])
self.assertEqual(project["onboarding_checklist"][0]["state"], "complete")
self.assertIsNone(project["last_seen_health"])
def test_api_v1_project_detail(self):
response = self.client.get("/api/v1/projects/gitea-tools")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(data["api_version"], REGISTRY_API_VERSION)
self.assertEqual(data["project"]["id"], "gitea-tools")
self.assertEqual(data["source"]["kind"], "file")
def test_api_v1_project_detail_missing_fails_closed(self):
response = self.client.get("/api/v1/projects/not-registered")
self.assertEqual(response.status_code, 404)
data = response.json()
self.assertEqual(data["error"], "project_not_found")
self.assertEqual(data["project_id"], "not-registered")
self.assertIn("gitea-tools", data["known_project_ids"])
self.assertIn("remediation", data)
def test_api_v1_projects_is_read_only(self):
response = self.client.post("/api/v1/projects", json={})
self.assertEqual(response.status_code, 405)
self.assertEqual(response.json()["error"], "read-only-mvp")
def test_project_to_dict_is_json_safe(self):
registry = load_registry()
dto = project_to_dict(registry.projects[0])
encoded = json.dumps(dto)
encoded = json.dumps(project_to_dict(registry.projects[0]))
self.assertIn("gitea-tools", encoded)
# Prose may mention tokens; no serialized *key* may look like a secret.
for key in dto:
with self.subTest(key=key):
self.assertFalse(is_forbidden_key(key))
class TestInvalidRegistryFailsClosedOverHttp(RegistryFileCase):
def setUp(self):
super().setUp()
self.path = self.write_registry({"version": 42, "projects": []})
self.client = TestClient(create_app())
def _with_bad_registry(self, url: str):
import os
from unittest import mock
with mock.patch.dict(
os.environ, {"WEBUI_PROJECT_REGISTRY": str(self.path)}, clear=False
):
return self.client.get(url)
def test_api_v1_reports_actionable_error(self):
response = self._with_bad_registry("/api/v1/projects")
self.assertEqual(response.status_code, 500)
data = response.json()
self.assertEqual(data["error"], "registry_invalid")
self.assertIn("unsupported registry version", data["detail"])
self.assertTrue(data["remediation"])
self.assertEqual(data["field_path"], "version")
def test_unversioned_alias_reports_actionable_error(self):
response = self._with_bad_registry("/api/projects")
self.assertEqual(response.status_code, 500)
self.assertEqual(response.json()["error"], "registry_invalid")
def test_html_page_reports_actionable_error(self):
response = self._with_bad_registry("/projects")
self.assertEqual(response.status_code, 500)
self.assertIn("Project registry unavailable", response.text)
self.assertIn("Remediation", response.text)
class TestProjectRegistryApiDocs(unittest.TestCase):
def test_api_contract_is_documented(self):
self.assertTrue(_API_DOC.is_file(), f"missing {_API_DOC}")
text = _API_DOC.read_text(encoding="utf-8")
for token in (
"/api/v1/projects",
"/api/v1/projects/{project_id}",
"/api/projects",
"onboarding_summary",
"last_seen_health",
"registry_invalid",
"#635",
):
with self.subTest(token=token):
self.assertIn(token, text)
def test_route_table_lists_versioned_routes(self):
local_dev = (_REPO_ROOT / "docs" / "webui-local-dev.md").read_text(
encoding="utf-8"
)
self.assertIn("/api/v1/projects", local_dev)
self.assertIn("webui-project-registry-api.md", local_dev)
if __name__ == "__main__":
unittest.main()
unittest.main()
+135
View File
@@ -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()
+59 -83
View File
@@ -11,20 +11,9 @@ from starlette.routing import Route
from webui.deployment_boundary import deployment_snapshot
from webui.layout import render_page
from webui.project_registry import (
ProjectRegistry,
RegistryError,
find_project,
known_project_ids,
load_registry,
project_detail_to_dict,
registry_to_dict,
)
from webui.project_views import (
render_project_detail,
render_projects_list,
render_registry_error,
)
from webui.nav import NAV_GROUPS, STUB_PAGES
from webui.project_registry import find_project, load_registry, registry_to_dict
from webui.project_views import render_project_detail, render_projects_list
from webui.prompt_library import find_prompt, library_to_dict
from webui.prompt_views import render_prompt_detail, render_prompts_page
from final_report_validator import FINAL_REPORT_TASK_KINDS
@@ -55,24 +44,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>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({
@@ -93,26 +120,14 @@ async def api_queue(_request: Request) -> JSONResponse:
return JSONResponse(queue_snapshot_to_dict(load_queue_snapshot()))
def _load_project_registry() -> tuple[ProjectRegistry | None, RegistryError | None]:
"""Load the registry, converting validation failure into a fail-closed pair."""
try:
return load_registry(), None
except RegistryError as exc:
return None, exc
async def projects(_request: Request) -> HTMLResponse:
registry, error = _load_project_registry()
if error is not None:
return HTMLResponse(render_registry_error(error), status_code=500)
registry = load_registry()
return HTMLResponse(render_projects_list(registry))
async def project_detail(request: Request) -> HTMLResponse:
project_id = request.path_params["project_id"]
registry, error = _load_project_registry()
if error is not None:
return HTMLResponse(render_registry_error(error), status_code=500)
registry = load_registry()
project = find_project(registry, project_id)
if project is None:
return HTMLResponse(
@@ -130,47 +145,10 @@ async def project_detail(request: Request) -> HTMLResponse:
async def api_projects(_request: Request) -> JSONResponse:
"""Unversioned MVP alias, retained through Phase 1 (#632 section 6)."""
registry, error = _load_project_registry()
if error is not None:
return JSONResponse(error.to_dict(), status_code=500)
registry = load_registry()
return JSONResponse(registry_to_dict(registry))
async def api_v1_projects(_request: Request) -> JSONResponse:
registry, error = _load_project_registry()
if error is not None:
return JSONResponse(error.to_dict(), status_code=500)
return JSONResponse(registry_to_dict(registry))
async def api_v1_project_detail(request: Request) -> JSONResponse:
project_id = request.path_params["project_id"]
registry, error = _load_project_registry()
if error is not None:
return JSONResponse(error.to_dict(), status_code=500)
project = find_project(registry, project_id)
if project is None:
return JSONResponse(
{
"error": "project_not_found",
"project_id": project_id,
"known_project_ids": known_project_ids(registry),
"remediation": (
"Request one of the known project ids, or add the project to the "
"registry file named in 'source'."
),
"source": {
"kind": "file",
"path": str(registry.source_path),
"inventory_complete": True,
},
},
status_code=404,
)
return JSONResponse(project_detail_to_dict(registry, project))
async def prompts(_request: Request) -> HTMLResponse:
return HTMLResponse(render_prompts_page())
@@ -329,12 +307,6 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
Route("/projects", projects, methods=["GET"]),
Route("/projects/{project_id}", project_detail, methods=["GET"]),
Route("/api/projects", api_projects, methods=["GET"]),
Route("/api/v1/projects", api_v1_projects, methods=["GET"]),
Route(
"/api/v1/projects/{project_id}",
api_v1_project_detail,
methods=["GET"],
),
Route("/prompts", prompts, methods=["GET"]),
Route("/prompts/{prompt_id}", prompt_detail, methods=["GET"]),
Route("/api/prompts", api_prompts, methods=["GET"]),
@@ -358,6 +330,10 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
methods=["POST"],
),
Route("/api/leases", api_leases, methods=["GET"]),
*[
Route(path, phase_stub, methods=["GET"])
for path in STUB_PAGES
],
],
exception_handlers={405: method_not_allowed},
)
+6 -16
View File
@@ -1,15 +1,13 @@
{
"version": 2,
"version": 1,
"projects": [
{
"id": "gitea-tools",
"repo_name": "Gitea-Tools",
"gitea_owner": "Scaled-Tech-Consulting",
"remote_name": "prgs",
"remote_host": "https://gitea.prgs.cc",
"default_branch": "master",
"local_checkout_path": ".",
"status": "active",
"profiles": {
"author": "prgs-author",
"reviewer": "prgs-reviewer",
@@ -28,32 +26,24 @@
{
"id": "profiles",
"title": "Configure execution profiles",
"description": "Install author, reviewer, and reconciler MCP profiles (prgs-author, prgs-reviewer, prgs-reconciler) in separate namespaces. Tokens stay in keychain — never in this registry.",
"state": "complete",
"required": true
"description": "Install author, reviewer, and reconciler MCP profiles (prgs-author, prgs-reviewer, prgs-reconciler) in separate namespaces. Tokens stay in keychain — never in this registry."
},
{
"id": "mcp_config",
"title": "Wire MCP v2 contexts",
"description": "Copy and customize gitea-mcp.v2-contexts.example.json for your machine. Map this repo path under projects with default_owner Scaled-Tech-Consulting and default_repo Gitea-Tools.",
"state": "complete",
"required": true
"description": "Copy and customize gitea-mcp.v2-contexts.example.json for your machine. Map this repo path under projects with default_owner Scaled-Tech-Consulting and default_repo Gitea-Tools."
},
{
"id": "wiki_gate",
"title": "Wiki publication readiness",
"description": "For wiki-tracked work, satisfy the live Gitea Wiki proof gate (#224) before closing issues. See docs/wiki/Safety-and-Gates.md.",
"state": "complete",
"required": true
"description": "For wiki-tracked work, satisfy the live Gitea Wiki proof gate (#224) before closing issues. See docs/wiki/Safety-and-Gates.md."
},
{
"id": "branches_layout",
"title": "Isolate work under branches/",
"description": "All LLM task edits happen in worktrees under branches/. Main checkout stays clean; use skills/llm-project-workflow templates for start-issue and review flows.",
"state": "complete",
"required": true
"description": "All LLM task edits happen in worktrees under branches/. Main checkout stays clean; use skills/llm-project-workflow templates for start-issue and review flows."
}
]
}
]
}
}
+95 -17
View File
@@ -2,28 +2,66 @@
from __future__ import annotations
NAV_ITEMS = (
("/", "Home"),
("/queue", "Queue"),
("/projects", "Projects"),
("/prompts", "Prompts"),
("/runtime", "Runtime"),
("/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>
@@ -53,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;
@@ -166,7 +241,10 @@ def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
</head>
<body>
<header>
<h1>MCP Control Plane</h1>
<div class="header-top">
<h1>MCP Control Plane</h1>
{header_badges}
</div>
<nav>{nav_links}</nav>
</header>
<main>
+111
View File
@@ -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())
+25 -460
View File
@@ -1,16 +1,4 @@
"""Load and validate the web UI project registry (#427, evolved for #635).
Phase 1 of the console architecture ADR keeps this loader read-only. It owns
the versioned project registry contract served at ``/api/v1/projects``:
* the on-disk file carries a ``version`` (schema version 1 or 2);
* version 1 files stay loadable and are normalized with explicit defaults, so
an operator registry written for #427 keeps working;
* every validation failure raises :class:`RegistryError`, which carries an
actionable ``remediation`` string instead of leaking a traceback;
* serialization never emits credentials — credential-shaped keys are rejected
at load time, before any DTO is built.
"""
"""Load and validate the web UI project registry (#427)."""
from __future__ import annotations
@@ -20,59 +8,7 @@ from dataclasses import dataclass
from pathlib import Path
from typing import Any
from webui.registry_safety import is_forbidden_key
#: Version of the JSON contract served under ``/api/v1/...``.
REGISTRY_API_VERSION = "v1"
#: Schema version written by this repository's packaged registry.
CURRENT_SCHEMA_VERSION = 2
#: Schema versions this loader accepts. Version 1 is normalized on load.
SUPPORTED_SCHEMA_VERSIONS = (1, 2)
#: Lifecycle state of a registered project.
PROJECT_STATUSES = ("active", "onboarding", "paused", "archived")
_DEFAULT_PROJECT_STATUS = "active"
#: Completion state of a single onboarding step.
ONBOARDING_STATES = ("complete", "pending", "blocked", "not_applicable")
_DEFAULT_ONBOARDING_STATE = "pending"
#: Redacted, last-seen health of a project's control plane.
HEALTH_STATUSES = ("healthy", "degraded", "unreachable", "unknown")
class RegistryError(ValueError):
"""A registry file could not be loaded or failed validation.
Carries an operator-facing ``remediation`` so routes can fail closed with
an actionable message rather than a stack trace.
"""
def __init__(
self,
message: str,
*,
remediation: str,
source_path: Path | None = None,
field_path: str | None = None,
) -> None:
super().__init__(message)
self.message = message
self.remediation = remediation
self.source_path = source_path
self.field_path = field_path
def to_dict(self) -> dict[str, Any]:
"""Serialize for a fail-closed JSON error response."""
return {
"error": "registry_invalid",
"detail": self.message,
"remediation": self.remediation,
"field_path": self.field_path,
"source_path": str(self.source_path) if self.source_path else None,
}
from webui.registry_safety import reject_credential_keys as _reject_credential_keys
_REQUIRED_PROJECT_FIELDS = (
"id",
@@ -93,30 +29,6 @@ class OnboardingStep:
id: str
title: str
description: str
state: str = _DEFAULT_ONBOARDING_STATE
required: bool = True
@dataclass(frozen=True)
class OnboardingSummary:
"""Aggregate onboarding progress for a single project."""
total: int
complete: int
pending: int
blocked: int
not_applicable: int
required_outstanding: int
onboarding_complete: bool
@dataclass(frozen=True)
class ProjectHealth:
"""Redacted last-seen health. Never carries endpoints or credentials."""
status: str
checked_at: str | None
detail: str | None
@dataclass(frozen=True)
@@ -131,13 +43,6 @@ class ProjectRecord:
workflow_paths: dict[str, str]
schema_paths: dict[str, str]
onboarding_checklist: tuple[OnboardingStep, ...]
status: str = _DEFAULT_PROJECT_STATUS
remote_name: str | None = None
last_seen_health: ProjectHealth | None = None
@property
def repo_full_name(self) -> str:
return f"{self.gitea_owner}/{self.repo_name}"
@dataclass(frozen=True)
@@ -146,15 +51,6 @@ class ProjectRegistry:
projects: tuple[ProjectRecord, ...]
source_path: Path
@property
def schema_version(self) -> int:
"""Alias of :attr:`version` — the schema version read from disk."""
return self.version
@property
def api_version(self) -> str:
return REGISTRY_API_VERSION
def default_registry_path() -> Path:
override = os.environ.get("WEBUI_PROJECT_REGISTRY", "").strip()
@@ -163,221 +59,40 @@ def default_registry_path() -> Path:
return (Path(__file__).resolve().parent / "data" / "projects.registry.json").resolve()
def _reject_credential_keys(obj: Any, *, path: str = "", source: Path | None = None) -> None:
"""Recursive credential-key guard that reports an actionable ``field_path``.
Key *shape* is decided by :func:`webui.registry_safety.is_forbidden_key`, the
single source of truth shared with the worker registry (#798).
"""
if isinstance(obj, dict):
for key, value in obj.items():
key_path = f"{path}.{key}" if path else key
if is_forbidden_key(key):
raise RegistryError(
f"registry must not store credentials ({key_path})",
remediation=(
f"Remove the credential-shaped key '{key_path}' from the registry. "
"Tokens live in the keychain and are resolved server-side by "
"gitea_auth; the registry is redacted metadata only."
),
source_path=source,
field_path=key_path,
)
_reject_credential_keys(value, path=key_path, source=source)
elif isinstance(obj, list):
for index, item in enumerate(obj):
_reject_credential_keys(item, path=f"{path}[{index}]", source=source)
def _require_enum(
value: Any,
*,
allowed: tuple[str, ...],
field_path: str,
source: Path | None,
) -> str:
text = str(value)
if text not in allowed:
raise RegistryError(
f"{field_path} must be one of {', '.join(allowed)} (got {text!r})",
remediation=(
f"Set {field_path} to one of: {', '.join(allowed)}. "
"Unknown values fail closed so the console never renders an "
"unverified state."
),
source_path=source,
field_path=field_path,
)
return text
def _parse_onboarding(
raw: Any,
*,
project_path: str,
source: Path | None,
) -> tuple[OnboardingStep, ...]:
if raw is None:
def _parse_onboarding(raw: list[dict[str, Any]] | None) -> tuple[OnboardingStep, ...]:
if not raw:
return ()
if not isinstance(raw, list):
raise RegistryError(
f"{project_path}.onboarding_checklist must be an array",
remediation=(
f"Rewrite {project_path}.onboarding_checklist as a JSON array of "
"steps with id, title, description, and optional state."
),
source_path=source,
field_path=f"{project_path}.onboarding_checklist",
)
steps: list[OnboardingStep] = []
for index, item in enumerate(raw):
step_path = f"{project_path}.onboarding_checklist[{index}]"
if not isinstance(item, dict):
raise RegistryError(
f"{step_path} must be an object",
remediation=f"Rewrite {step_path} as an object with id, title, description.",
source_path=source,
field_path=step_path,
)
missing = [field for field in ("id", "title", "description") if field not in item]
if missing:
raise RegistryError(
f"{step_path} missing required fields: {', '.join(missing)}",
remediation=(
f"Add {', '.join(missing)} to {step_path}. Every onboarding step "
"must be self-describing for an operator who has no chat history."
),
source_path=source,
field_path=step_path,
)
state = _require_enum(
item.get("state", _DEFAULT_ONBOARDING_STATE),
allowed=ONBOARDING_STATES,
field_path=f"{step_path}.state",
source=source,
)
for item in raw:
steps.append(
OnboardingStep(
id=str(item["id"]),
title=str(item["title"]),
description=str(item["description"]),
state=state,
required=bool(item.get("required", True)),
)
)
return tuple(steps)
def _parse_health(
raw: Any,
*,
project_path: str,
source: Path | None,
) -> ProjectHealth | None:
if raw is None:
return None
if not isinstance(raw, dict):
raise RegistryError(
f"{project_path}.last_seen_health must be an object when present",
remediation=(
f"Rewrite {project_path}.last_seen_health as an object with status "
f"(one of {', '.join(HEALTH_STATUSES)}), optional checked_at and detail, "
"or remove it. Never store endpoints or credentials here."
),
source_path=source,
field_path=f"{project_path}.last_seen_health",
)
status = _require_enum(
raw.get("status", "unknown"),
allowed=HEALTH_STATUSES,
field_path=f"{project_path}.last_seen_health.status",
source=source,
)
checked_at = raw.get("checked_at")
detail = raw.get("detail")
return ProjectHealth(
status=status,
checked_at=str(checked_at) if checked_at is not None else None,
detail=str(detail) if detail is not None else None,
)
def _parse_project(raw: Any, *, index: int, source: Path | None) -> ProjectRecord:
project_path = f"projects[{index}]"
if not isinstance(raw, dict):
raise RegistryError(
f"{project_path} must be an object",
remediation=f"Rewrite {project_path} as a JSON object describing one project.",
source_path=source,
field_path=project_path,
)
def _parse_project(raw: dict[str, Any]) -> ProjectRecord:
missing = [field for field in _REQUIRED_PROJECT_FIELDS if field not in raw]
if missing:
raise RegistryError(
f"{project_path} missing required fields: {', '.join(missing)}",
remediation=(
f"Add {', '.join(missing)} to {project_path}. See "
"docs/webui-project-registry-api.md for the field-by-field contract."
),
source_path=source,
field_path=project_path,
)
raise ValueError(f"project missing required fields: {', '.join(missing)}")
profiles = raw["profiles"]
if not isinstance(profiles, dict):
raise RegistryError(
f"{project_path}.profiles must be an object",
remediation=(
f"Rewrite {project_path}.profiles as an object mapping "
f"{', '.join(_REQUIRED_PROFILE_ROLES)} to MCP profile names."
),
source_path=source,
field_path=f"{project_path}.profiles",
)
raise ValueError("profiles must be an object")
for role in _REQUIRED_PROFILE_ROLES:
if role not in profiles or not profiles[role]:
raise RegistryError(
f"{project_path}.profiles.{role} is required",
remediation=(
f"Set {project_path}.profiles.{role} to the configured MCP profile "
"name for that role. Role separation is a workflow-safety invariant."
),
source_path=source,
field_path=f"{project_path}.profiles.{role}",
)
raise ValueError(f"profiles.{role} is required")
workflow_paths = raw["workflow_paths"]
if not isinstance(workflow_paths, dict) or not workflow_paths:
raise RegistryError(
f"{project_path}.workflow_paths must be a non-empty object",
remediation=(
f"Add at least a 'skill' entry to {project_path}.workflow_paths pointing "
"at the project's canonical workflow skill."
),
source_path=source,
field_path=f"{project_path}.workflow_paths",
)
raise ValueError("workflow_paths must be a non-empty object")
schema_paths = raw.get("schema_paths") or {}
if not isinstance(schema_paths, dict):
raise RegistryError(
f"{project_path}.schema_paths must be an object when present",
remediation=(
f"Rewrite {project_path}.schema_paths as an object of label to repo path, "
"or remove it."
),
source_path=source,
field_path=f"{project_path}.schema_paths",
)
status = _require_enum(
raw.get("status", _DEFAULT_PROJECT_STATUS),
allowed=PROJECT_STATUSES,
field_path=f"{project_path}.status",
source=source,
)
remote_name = raw.get("remote_name")
raise ValueError("schema_paths must be an object when present")
return ProjectRecord(
id=str(raw["id"]),
@@ -389,211 +104,61 @@ def _parse_project(raw: Any, *, index: int, source: Path | None) -> ProjectRecor
profiles={role: str(profiles[role]) for role in _REQUIRED_PROFILE_ROLES},
workflow_paths={key: str(value) for key, value in workflow_paths.items()},
schema_paths={key: str(value) for key, value in schema_paths.items()},
onboarding_checklist=_parse_onboarding(
raw.get("onboarding_checklist"),
project_path=project_path,
source=source,
),
status=status,
remote_name=str(remote_name) if remote_name else None,
last_seen_health=_parse_health(
raw.get("last_seen_health"),
project_path=project_path,
source=source,
),
onboarding_checklist=_parse_onboarding(raw.get("onboarding_checklist")),
)
def load_registry(path: Path | None = None) -> ProjectRegistry:
"""Load the versioned project registry from disk.
Raises:
RegistryError: whenever the file is unreadable, is not valid JSON, or
fails schema validation. The error carries an operator remediation.
"""
"""Load the versioned project registry from disk."""
source = (path or default_registry_path()).resolve()
try:
raw_text = source.read_text(encoding="utf-8")
except OSError as exc:
raise RegistryError(
f"registry file could not be read: {exc.strerror or exc}",
remediation=(
f"Create a readable registry at {source}, or point "
"WEBUI_PROJECT_REGISTRY at an existing file."
),
source_path=source,
) from exc
try:
payload = json.loads(raw_text)
except json.JSONDecodeError as exc:
raise RegistryError(
f"registry is not valid JSON: {exc.msg} (line {exc.lineno}, column {exc.colno})",
remediation=(
f"Fix the JSON syntax in {source} at line {exc.lineno}, column {exc.colno}."
),
source_path=source,
) from exc
raw_text = source.read_text(encoding="utf-8")
payload = json.loads(raw_text)
if not isinstance(payload, dict):
raise RegistryError(
"registry root must be an object",
remediation=(
"Wrap the registry in a JSON object with 'version' and 'projects' keys."
),
source_path=source,
)
raise ValueError("registry root must be an object")
version = payload.get("version")
if version not in SUPPORTED_SCHEMA_VERSIONS:
supported = ", ".join(str(item) for item in SUPPORTED_SCHEMA_VERSIONS)
raise RegistryError(
f"unsupported registry version: {version!r}",
remediation=(
f"Set 'version' to one of {supported} (current schema is "
f"{CURRENT_SCHEMA_VERSION}). Migration notes live in "
"docs/webui-project-registry-api.md."
),
source_path=source,
field_path="version",
)
if version != 1:
raise ValueError(f"unsupported registry version: {version!r}")
_reject_credential_keys(payload, source=source)
_reject_credential_keys(payload)
projects_raw = payload.get("projects")
if not isinstance(projects_raw, list) or not projects_raw:
raise RegistryError(
"projects must be a non-empty array",
remediation=(
"Add at least one project object to 'projects'. An empty console "
"registry fails closed rather than rendering a blank inventory."
),
source_path=source,
field_path="projects",
)
raise ValueError("projects must be a non-empty array")
projects = tuple(
_parse_project(item, index=index, source=source)
for index, item in enumerate(projects_raw)
)
return ProjectRegistry(version=int(version), projects=projects, source_path=source)
def onboarding_summary(project: ProjectRecord) -> OnboardingSummary:
"""Aggregate a project's onboarding checklist state."""
steps = project.onboarding_checklist
counts = {state: 0 for state in ONBOARDING_STATES}
for step in steps:
counts[step.state] += 1
required_outstanding = sum(
1
for step in steps
if step.required and step.state in ("pending", "blocked")
)
return OnboardingSummary(
total=len(steps),
complete=counts["complete"],
pending=counts["pending"],
blocked=counts["blocked"],
not_applicable=counts["not_applicable"],
required_outstanding=required_outstanding,
onboarding_complete=required_outstanding == 0,
)
projects = tuple(_parse_project(item) for item in projects_raw)
return ProjectRegistry(version=version, projects=projects, source_path=source)
def project_to_dict(project: ProjectRecord) -> dict[str, Any]:
"""Serialize a project for JSON API responses and HTML views.
The HTML views render from this same DTO, so the console and the API can
never disagree about a project's status or onboarding progress.
"""
summary = onboarding_summary(project)
health = project.last_seen_health
"""Serialize a project for JSON API responses."""
return {
"id": project.id,
"repo_name": project.repo_name,
"gitea_owner": project.gitea_owner,
"repo_full_name": project.repo_full_name,
"remote_host": project.remote_host,
"remote_name": project.remote_name,
"default_branch": project.default_branch,
"local_checkout_path": project.local_checkout_path,
"status": project.status,
"profiles": dict(project.profiles),
"workflow_paths": dict(project.workflow_paths),
"schema_paths": dict(project.schema_paths),
"onboarding_checklist": [
{
"id": step.id,
"title": step.title,
"description": step.description,
"state": step.state,
"required": step.required,
}
{"id": step.id, "title": step.title, "description": step.description}
for step in project.onboarding_checklist
],
"onboarding_summary": {
"total": summary.total,
"complete": summary.complete,
"pending": summary.pending,
"blocked": summary.blocked,
"not_applicable": summary.not_applicable,
"required_outstanding": summary.required_outstanding,
"onboarding_complete": summary.onboarding_complete,
},
"last_seen_health": (
None
if health is None
else {
"status": health.status,
"checked_at": health.checked_at,
"detail": health.detail,
}
),
}
def registry_to_dict(registry: ProjectRegistry) -> dict[str, Any]:
"""Serialize the whole registry, including API provenance (#632 section 6)."""
return {
"api_version": registry.api_version,
"schema_version": registry.schema_version,
# Retained for the unversioned MVP alias consumers (#427).
"version": registry.version,
"source_path": str(registry.source_path),
"source": {
"kind": "file",
"path": str(registry.source_path),
"inventory_complete": True,
},
"project_count": len(registry.projects),
"projects": [project_to_dict(project) for project in registry.projects],
}
def project_detail_to_dict(
registry: ProjectRegistry,
project: ProjectRecord,
) -> dict[str, Any]:
"""Serialize a single project for ``/api/v1/projects/{project_id}``."""
return {
"api_version": registry.api_version,
"schema_version": registry.schema_version,
"source": {
"kind": "file",
"path": str(registry.source_path),
"inventory_complete": True,
},
"project": project_to_dict(project),
}
def find_project(registry: ProjectRegistry, project_id: str) -> ProjectRecord | None:
for project in registry.projects:
if project.id == project_id:
return project
return None
def known_project_ids(registry: ProjectRegistry) -> list[str]:
return [project.id for project in registry.projects]
return None
+25 -107
View File
@@ -1,65 +1,34 @@
"""HTML views for project registry pages (#427, evolved for #635).
Every view renders from :func:`webui.project_registry.project_to_dict`, the
same DTO the ``/api/v1/projects`` JSON responses use, so the HTML console and
the API can never disagree about status or onboarding progress.
"""
"""HTML views for project registry pages (#427)."""
from __future__ import annotations
import html
from typing import Any
from webui.layout import render_page
from webui.project_registry import (
ProjectRecord,
ProjectRegistry,
RegistryError,
project_to_dict,
)
_STATE_LABELS = {
"complete": "Complete",
"pending": "Pending",
"blocked": "Blocked",
"not_applicable": "Not applicable",
}
from webui.project_registry import ProjectRecord, ProjectRegistry
def _escape(text: str) -> str:
return html.escape(text, quote=True)
def _progress_label(summary: dict[str, Any]) -> str:
total = summary["total"]
if not total:
return "no steps"
label = f"{summary['complete']}/{total} complete"
if summary["blocked"]:
label += f", {summary['blocked']} blocked"
return label
def render_projects_list(registry: ProjectRegistry) -> str:
rows = []
for project in registry.projects:
dto = project_to_dict(project)
rows.append(
"<tr>"
f"<td><a href=\"/projects/{_escape(dto['id'])}\">{_escape(dto['repo_name'])}</a></td>"
f"<td>{_escape(dto['gitea_owner'])}</td>"
f"<td>{_escape(dto['remote_host'])}</td>"
f"<td>{_escape(dto['default_branch'])}</td>"
f"<td><code>{_escape(dto['status'])}</code></td>"
f"<td>{_escape(_progress_label(dto['onboarding_summary']))}</td>"
f"<td><code>{_escape(dto['profiles']['author'])}</code></td>"
f"<td><a href=\"/projects/{_escape(project.id)}\">{_escape(project.repo_name)}</a></td>"
f"<td>{_escape(project.gitea_owner)}</td>"
f"<td>{_escape(project.remote_host)}</td>"
f"<td>{_escape(project.default_branch)}</td>"
f"<td><code>{_escape(project.profiles['author'])}</code></td>"
"</tr>"
)
table = (
"<table class=\"registry\">"
"<thead><tr>"
"<th>Repository</th><th>Owner</th><th>Remote</th>"
"<th>Branch</th><th>Status</th><th>Onboarding</th><th>Author profile</th>"
"<th>Branch</th><th>Author profile</th>"
"</tr></thead>"
f"<tbody>{''.join(rows)}</tbody></table>"
)
@@ -67,39 +36,32 @@ def render_projects_list(registry: ProjectRegistry) -> str:
"<h2>Projects</h2>"
"<p>Configured repositories managed by the MCP Control Plane.</p>"
f"<p class=\"meta\">Registry: <code>{_escape(str(registry.source_path))}</code> "
f"(schema version {registry.schema_version}, "
f"API {_escape(registry.api_version)})</p>"
f"(version {registry.version})</p>"
f"{table}"
"<p><a href=\"/api/v1/projects\">JSON API</a> "
"(<a href=\"/api/projects\">unversioned alias</a>)</p>"
"<p><a href=\"/api/projects\">JSON API</a></p>"
)
return render_page(title="Projects", body_html=body)
def render_project_detail(project: ProjectRecord) -> str:
dto = project_to_dict(project)
profile_rows = "".join(
f"<tr><th>{_escape(role)}</th><td><code>{_escape(name)}</code></td></tr>"
for role, name in dto["profiles"].items()
for role, name in project.profiles.items()
)
workflow_rows = "".join(
f"<tr><th>{_escape(key)}</th><td><code>{_escape(path)}</code></td></tr>"
for key, path in dto["workflow_paths"].items()
for key, path in project.workflow_paths.items()
)
schema_rows = "".join(
f"<tr><th>{_escape(key)}</th><td><code>{_escape(path)}</code></td></tr>"
for key, path in dto["schema_paths"].items()
for key, path in project.schema_paths.items()
)
checklist_items = []
for index, step in enumerate(dto["onboarding_checklist"], start=1):
state_label = _STATE_LABELS.get(step["state"], step["state"])
requirement = "required" if step["required"] else "optional"
for index, step in enumerate(project.onboarding_checklist, start=1):
checklist_items.append(
f"<li class=\"step-{_escape(step['state'])}\">"
f"<strong>{index}. {_escape(step['title'])}</strong>"
f" <span class=\"badge\">{_escape(state_label)}</span>"
f" <span class=\"meta\">({_escape(requirement)})</span>"
f"<p>{_escape(step['description'])}</p>"
"<li>"
f"<strong>{index}. {_escape(step.title)}</strong>"
f"<p>{_escape(step.description)}</p>"
"</li>"
)
checklist_html = (
@@ -107,38 +69,16 @@ def render_project_detail(project: ProjectRecord) -> str:
if checklist_items
else "<p>No onboarding steps defined.</p>"
)
summary = dto["onboarding_summary"]
summary_html = (
"<p class=\"meta\">Onboarding: "
f"{_escape(_progress_label(summary))}; required outstanding "
f"{summary['required_outstanding']}.</p>"
)
health = dto["last_seen_health"]
health_html = (
"<p class=\"meta\">No health probe recorded (Phase 1 is read-only).</p>"
if health is None
else (
"<table class=\"detail\">"
f"<tr><th>Status</th><td><code>{_escape(health['status'])}</code></td></tr>"
f"<tr><th>Checked at</th><td>{_escape(str(health['checked_at'] or 'unknown'))}</td></tr>"
f"<tr><th>Detail</th><td>{_escape(str(health['detail'] or ''))}</td></tr>"
"</table>"
)
)
remote_name = dto["remote_name"] or "unset"
body = (
f"<h2>{_escape(dto['repo_name'])}</h2>"
f"<h2>{_escape(project.repo_name)}</h2>"
"<p><a href=\"/projects\">← All projects</a></p>"
"<h3>Identity</h3>"
"<table class=\"detail\">"
f"<tr><th>Registry id</th><td><code>{_escape(dto['id'])}</code></td></tr>"
f"<tr><th>Status</th><td><code>{_escape(dto['status'])}</code></td></tr>"
f"<tr><th>Gitea owner</th><td>{_escape(dto['gitea_owner'])}</td></tr>"
f"<tr><th>Repository</th><td><code>{_escape(dto['repo_full_name'])}</code></td></tr>"
f"<tr><th>Remote name</th><td><code>{_escape(remote_name)}</code></td></tr>"
f"<tr><th>Remote host</th><td>{_escape(dto['remote_host'])}</td></tr>"
f"<tr><th>Default branch</th><td><code>{_escape(dto['default_branch'])}</code></td></tr>"
f"<tr><th>Local checkout</th><td><code>{_escape(dto['local_checkout_path'])}</code></td></tr>"
f"<tr><th>Registry id</th><td><code>{_escape(project.id)}</code></td></tr>"
f"<tr><th>Gitea owner</th><td>{_escape(project.gitea_owner)}</td></tr>"
f"<tr><th>Remote host</th><td>{_escape(project.remote_host)}</td></tr>"
f"<tr><th>Default branch</th><td><code>{_escape(project.default_branch)}</code></td></tr>"
f"<tr><th>Local checkout</th><td><code>{_escape(project.local_checkout_path)}</code></td></tr>"
"</table>"
"<h3>Profiles</h3>"
f"<table class=\"detail\">{profile_rows}</table>"
@@ -146,30 +86,8 @@ def render_project_detail(project: ProjectRecord) -> str:
f"<table class=\"detail\">{workflow_rows}</table>"
"<h3>Schema paths</h3>"
f"<table class=\"detail\">{schema_rows}</table>"
"<h3>Last seen health</h3>"
f"{health_html}"
"<h3>Onboarding checklist</h3>"
"<p class=\"meta\">Read-only — complete these steps outside the UI.</p>"
f"{summary_html}"
"<p class=\"meta\">Read-only MVP — complete these steps outside the UI.</p>"
f"{checklist_html}"
f"<p><a href=\"/api/v1/projects/{_escape(dto['id'])}\">JSON detail</a></p>"
)
return render_page(title=dto["repo_name"], body_html=body)
def render_registry_error(error: RegistryError) -> str:
"""Render a fail-closed page for an invalid registry."""
source = str(error.source_path) if error.source_path else "unknown"
field = error.field_path or "n/a"
body = (
"<h2>Project registry unavailable</h2>"
"<p>The registry failed validation, so the console refuses to render a "
"partial inventory.</p>"
"<table class=\"detail\">"
f"<tr><th>Detail</th><td>{_escape(error.message)}</td></tr>"
f"<tr><th>Field</th><td><code>{_escape(field)}</code></td></tr>"
f"<tr><th>Source</th><td><code>{_escape(source)}</code></td></tr>"
f"<tr><th>Remediation</th><td>{_escape(error.remediation)}</td></tr>"
"</table>"
)
return render_page(title="Project registry unavailable", body_html=body)
return render_page(title=project.repo_name, body_html=body)