Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
  jcwalker3andClaude Opus 4.8
|
5494696227 | ||
 sysadmin
|
9eb0f29cef | ||
|
sysadmin
|
0b29404031 | ||
|
jcwalker3andClaude Opus 4.8
|
5463f58933 | ||
|
sysadmin
|
5032965e3a | ||
|
jcwalker3andClaude Opus 4.8
|
e33b8d3712 |
@@ -0,0 +1,201 @@
|
||||
# ADR: MCP Control Plane Web Console architecture and information architecture
|
||||
|
||||
- **Status:** Proposed (documentation only; blocks no code, gates every #631 child)
|
||||
- **Date:** 2026-07-22
|
||||
- **Tracking issue:** [#632](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/632) — architecture and information architecture (Phase 1)
|
||||
- **Parent epic:** [#631](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/631) — MCP Control Plane Web Console
|
||||
- **Foundation (closed, extend — do not recreate):** #425 tracker and children #426 skeleton, #427 projects, #428 prompts, #429 queue, #430 runtime, #431 audit paste, #432 worktrees, #433 leases, #434 gated actions, #435 auth/deployment boundary, #436 tests/CI
|
||||
- **Related:** `mcp-allocator-control-plane-observability-adr.md`, `mcp-stable-control-runtime-policy-adr.md`, `control-plane-db-substrate.md`, `../safety-model.md`, `../tool-boundaries.md`, `../credential-isolation.md`, `../webui-local-dev.md`, `../webui-deployment.md`
|
||||
|
||||
## 1. Context
|
||||
|
||||
The MVP web UI shipped under `webui/` as a read-only Starlette application with ten operator routes and a JSON export beside most of them. It is a working foundation, not the console product described by epic #631, and it carries no durable architecture record: no layer contract, no authority boundary, no API versioning rule, no page map, and no statement of which phase may open a write path.
|
||||
|
||||
Twenty children (#632–#651) hang off #631. Without one architecture document each implementer re-derives boundaries, and the most likely failure is not a bad view — it is a privileged action wired into the browser before the authorization and audit model of #633 exists.
|
||||
|
||||
This ADR is the single retrievable design source for the console. It decides structure only. It implements no UI, no API, and no change to deployment topology.
|
||||
|
||||
## 2. Decision summary (core)
|
||||
|
||||
| Layer | Owns | Must not |
|
||||
|-------|------|----------|
|
||||
| **Browser UI** | Rendering, navigation, operator affordances | Hold tokens, call Gitea/providers directly, or execute an action the server did not gate |
|
||||
| **HTTP route layer** (`webui/app.py`) | Versioned routing, authentication, authorization, redaction boundary, audit emission | Contain domain logic or reach past a loader to a raw credential |
|
||||
| **Domain loaders** (`webui/*_loader.py`, `*_scanner.py`, `runtime_health.py`, `project_registry.py`) | Assembling read models from authoritative sources | Mutate anything, or emit unredacted secrets across the boundary |
|
||||
| **Gitea** | Durable work record: issues, PRs, comments, reviews, labels, merges | Be the concurrency lock under multi-session load |
|
||||
| **Control-plane DB** | Sessions, assignment, leases, heartbeats, events | Replace Gitea history |
|
||||
| **MCP tools / capability gates** | Mutation authorization | Be re-implemented, mirrored, or bypassed by console code |
|
||||
| **External providers** (Sentry/GlitchTip, AI providers) | Incident and usage data | Assign work or mutate Gitea outside the #612 bridge |
|
||||
|
||||
**One-liner:** **Gitea records. The DB coordinates. MCP tools authorize. The console projects state and executes only capability-checked, audited actions. Providers observe.**
|
||||
|
||||
## 3. Console surface today versus target
|
||||
|
||||
`webui/app.py` currently registers these routes (see `../webui-local-dev.md` for the operator-facing table): `/`, `/health`, `/queue`, `/projects`, `/projects/{id}`, `/prompts`, `/runtime`, `/audit`, `/worktrees`, `/leases`, `/actions`, and the unversioned exports `/api/queue`, `/api/projects`, `/api/prompts`, `/api/runtime`, `/api/audit`, `/api/worktrees`, `/api/leases`, `/api/actions`, `/api/actions/{id}/preview`, `/api/actions/{id}/attempt`.
|
||||
|
||||
Every one of these is **retained and evolved**. No child issue may recreate a route from scratch; each states in its PR which MVP surface it extends and what it changes.
|
||||
|
||||
## 4. Authority boundaries
|
||||
|
||||
### 4.1 Gitea (durable record)
|
||||
|
||||
Authoritative for issue and PR identity and state, comments, reviews and verdicts, labels, merges, and branch refs. When the console and Gitea disagree about durable state, Gitea wins and the console view is refreshed — never the reverse.
|
||||
|
||||
### 4.2 Control-plane DB (coordination)
|
||||
|
||||
Authoritative for live coordination: which session holds which assignment or lease, heartbeat freshness, expiry, and the allocation event log. The console reads it; only allocator and lease tools write it.
|
||||
|
||||
### 4.3 MCP capability gates (authorization)
|
||||
|
||||
`task_capability_map.py` and `gitea_resolve_task_capability` remain the only authority that decides whether a mutation may run. The console asks; it never answers. A console action that cannot name the MCP tool it delegates to is not an action — it is a defect.
|
||||
|
||||
### 4.4 Filesystem and git (local state)
|
||||
|
||||
Issue lock files, `branches/` worktrees, and registered git worktrees are read through existing scanners. The console never deletes, rebinds, or force-clears local state outside a Phase 2 gated action.
|
||||
|
||||
### 4.5 Providers (observe only)
|
||||
|
||||
Sentry/GlitchTip and AI providers are read surfaces. The #612 incident bridge is the only path that turns an observation into Gitea work.
|
||||
|
||||
## 5. Request flow and the redaction boundary
|
||||
|
||||
```text
|
||||
browser ──HTTP──> route layer ──> domain loader ──> Gitea REST
|
||||
│ ├──> control-plane DB
|
||||
│ ├──> filesystem / git
|
||||
│ └──> providers
|
||||
│
|
||||
[redaction boundary]
|
||||
│
|
||||
audit event
|
||||
```
|
||||
|
||||
| Stage | May hold credentials | Emits |
|
||||
|-------|----------------------|-------|
|
||||
| Loader → route layer | yes (server-side, via `gitea_auth`) | domain objects |
|
||||
| Route layer → browser | **no** | redacted DTOs, HTML |
|
||||
|
||||
Two invariants govern the boundary and are non-negotiable for every child:
|
||||
|
||||
1. **No secrets to the browser.** Tokens, keychain identifiers, Authorization headers, raw provider endpoints, and credential-bearing URLs are redacted by default, consistent with `../safety-model.md` §3 and `../credential-isolation.md`. Serializers redact; templates do not sanitize after the fact.
|
||||
2. **No ungated mutations.** A write reaches an authoritative system only by delegating to an MCP tool that passed its own capability gate. HTML forms and JSON endpoints are transport, never authority.
|
||||
|
||||
## 6. API naming and versioning
|
||||
|
||||
**Decision:** all console APIs added from Phase 1 onward are served under `/api/v1/...`.
|
||||
|
||||
- Nouns are plural and hierarchical: `/api/v1/inventory/leases`, `/api/v1/system/health`.
|
||||
- Read endpoints are `GET` and side-effect free.
|
||||
- Phase 2 action endpoints are `POST /api/v1/actions/{action_id}/preview` and `POST /api/v1/actions/{action_id}/execute`; `preview` stays side-effect free and returns a mutation ledger.
|
||||
- The existing unversioned MVP exports remain as **compatibility aliases** for the whole of Phase 1 so the current operator flow never breaks. They may be retired no earlier than Phase 2, and only after the replacing `v1` route ships and `../webui-local-dev.md` records the swap.
|
||||
- A breaking change to a `v1` payload requires `/api/v2/...`, not an in-place edit.
|
||||
- Every JSON payload carries enough provenance for an auditor to tell where the data came from — at minimum the source system and whether the inventory was complete, matching the pagination-proof habit the MVP queue export already established.
|
||||
|
||||
## 7. Page map
|
||||
|
||||
| Page | Purpose | Owning child | Evolves |
|
||||
|------|---------|--------------|---------|
|
||||
| `/` | Console shell, navigation, next-safe-action summary | #638 | MVP `/` (#426) |
|
||||
| `/system` | System-health dashboard | #639 | new, backed by #634 |
|
||||
| `/traffic` | Workflow traffic control, queues, blockers | #640 | MVP `/queue` (#429) |
|
||||
| `/runtime` | Runtime and session view | #641 | MVP `/runtime` (#430) |
|
||||
| `/projects`, `/projects/{id}` | Project registry and onboarding | #635 | MVP `/projects` (#427) |
|
||||
| `/inventory` | Sessions, leases, locks, worktrees in one surface | #636 | MVP `/leases` (#433) + `/worktrees` (#432) |
|
||||
| `/timeline` | Workflow events and conversation timeline | #637 | new |
|
||||
| `/actions` | Gated action registry, preview, execution | #642, #643, #644 | MVP `/actions` (#434) |
|
||||
| `/gitea` | Issue and PR linkage console | #645 | new |
|
||||
| `/policy` | Guardrail visibility, then versioned editing | #646, #647 | new |
|
||||
| `/notifications` | Human-attention routing | #648 | new |
|
||||
| `/observability` | Sentry/GlitchTip correlation and durable issue creation | #649 | new |
|
||||
| `/providers` | AI-provider connections and insights | #650 | new |
|
||||
| `/analytics` | Usage, token cost, latency, workflow performance | #651 | new |
|
||||
| `/audit` | Final-report validator preview and audit log | #431 foundation, extended by #633 | MVP `/audit` (#431) |
|
||||
| `/prompts`, `/prompts/{id}` | Canonical prompt library | #638 | MVP `/prompts` (#428) |
|
||||
| `/health` | Liveness and deployment metadata | #634 | MVP `/health` (#435) |
|
||||
|
||||
## 8. Component ownership for every epic child
|
||||
|
||||
Each #631 child maps to at least one architectural component defined above.
|
||||
|
||||
| Child | Capability area | Primary component | Phase |
|
||||
|-------|-----------------|-------------------|-------|
|
||||
| #632 | Architecture and information architecture | this ADR | 1 |
|
||||
| #633 | Authorization, RBAC, secret redaction, audit and retention | route layer + redaction boundary (§5) | 1 |
|
||||
| #634 | Read-only system-health API | `/api/v1/system/health` + health loader | 1 |
|
||||
| #635 | Project registry API evolution | `/api/v1/projects` + `project_registry.py` | 1 |
|
||||
| #636 | Session, lease, lock, worktree inventory API | `/api/v1/inventory/*` + `lease_loader.py`, `worktree_scanner.py` | 1 |
|
||||
| #637 | Workflow-event and conversation timeline model | `/api/v1/events` + control-plane DB event log | 1 |
|
||||
| #638 | Application shell evolution | browser UI layer + `layout.py` | 1 |
|
||||
| #639 | System-health dashboard | `/system` page over #634 | 1 |
|
||||
| #640 | Workflow traffic-control view | `/traffic` page over the queue loader | 1 |
|
||||
| #641 | Runtime and session view | `/runtime` page over `runtime_health.py` | 1 |
|
||||
| #642 | Sanctioned restart and graceful reload controls | gated action framework, restart class | 2 |
|
||||
| #643 | Requests, intent preview, authorization, workflow initiation | `/api/v1/actions/*` execute path | 2 |
|
||||
| #644 | Stale-runtime recovery, worktree rebinding, reconciliation controls | gated actions over filesystem/git authority | 2 |
|
||||
| #645 | Gitea issue and PR linkage console | `/gitea` page over Gitea authority | 3 |
|
||||
| #646 | Workflow policy and guardrail visibility | `/policy` read view over the capability map | 3 |
|
||||
| #647 | Versioned policy editing, validation, simulation, approval, rollback | `/policy` write path, gated | 3 |
|
||||
| #648 | Notifications and human-attention routing | notification component over the event model | 3 |
|
||||
| #649 | Sentry/GlitchTip connections, correlation, durable issue creation | provider layer + #612 incident bridge | 4 |
|
||||
| #650 | AI-provider connections and operational insights | provider layer | 4 |
|
||||
| #651 | Model usage, token cost, latency, workflow analytics | analytics component over the event model | 4 |
|
||||
|
||||
Related but **outside** this epic: #667 (restart status, impact preview, and approval controls) belongs to the #655 restart-governance umbrella and must reuse the #642 action class rather than adding a second restart surface.
|
||||
|
||||
## 9. Phase gates
|
||||
|
||||
| Phase | May ship | Entry condition |
|
||||
|-------|----------|-----------------|
|
||||
| **1 — read-only visibility** | `GET` pages and `GET /api/v1/...` | this ADR accepted |
|
||||
| **2 — controlled actions** | gated `POST` action execution | #633 authorization, RBAC, and audit model landed |
|
||||
| **3 — orchestration and policy** | linkage, policy visibility, versioned policy editing | Phase 1 inventory plus the Phase 2 action framework |
|
||||
| **4 — insights** | provider correlation, analytics | evidence-backed sources from Phases 1–3 |
|
||||
|
||||
Phase 1 must not open a mutation endpoint, and the read-only guard that returns `405 read-only-mvp` stays in force until the Phase 2 entry condition is met. A phase is not entered by exception; if a control is urgent, the entry condition is what gets prioritized.
|
||||
|
||||
## 10. Security and workflow safety
|
||||
|
||||
- **Fail closed** on unknown authentication, missing RBAC mapping, or ambiguous lease ownership. An unknown state renders as blocked, never as permitted.
|
||||
- **Redact by default**, per §5.
|
||||
- **Every privileged action** requires a resolved capability, an explicit operator confirmation, and a durable audit event naming actor, action, target, and outcome.
|
||||
- **Contamination surfaces.** Session contamination — including a manually killed MCP daemon (#630) — must be shown and must block clean claims rather than being silently repaired.
|
||||
- **Deployment boundary unchanged.** Loopback by default, with the existing refusal of public binds (#435). This ADR documents that target; it does not widen it.
|
||||
|
||||
## 11. Forbidden paths
|
||||
|
||||
These are rejected designs, not preferences:
|
||||
|
||||
1. **Raw provider incidents as work.** The allocator never receives an unclassified Sentry/GlitchTip incident; only the #612 bridge turns an observation into a Gitea issue.
|
||||
2. **Browser-held tokens.** No credential, keychain identifier, or Authorization header is ever sent to the browser or embedded in a client bundle.
|
||||
3. **Process-kill recovery.** The console must not expose `pkill`, process-identifier termination, or any host process kill as a recovery affordance (#630). Restart is the sanctioned, operator-owned path of #642 and the #655 umbrella.
|
||||
4. **Ungated browser mutations.** No review, approval, merge, close, or comment may originate from the browser without passing an MCP capability gate.
|
||||
5. **Policy invented in the console.** The console projects policy from the capability map and canonical workflows; it never encodes a second copy.
|
||||
6. **Recreating MVP scope.** Re-implementing a #426–#436 surface without an explicit evolve-or-extend statement is out of bounds.
|
||||
|
||||
## 12. Approval checklist (readable without chat history)
|
||||
|
||||
A controller can accept or reject this ADR against these six points alone:
|
||||
|
||||
1. Layers and their owners are defined (§2) and each authority is named (§4).
|
||||
2. The redaction boundary and the two invariants are stated (§5).
|
||||
3. API versioning is decided, including what happens to the existing unversioned routes (§6).
|
||||
4. A page map exists and names an owning child for every page (§7).
|
||||
5. Every #631 child maps to at least one component and one phase (§8).
|
||||
6. Phase gates and forbidden paths are explicit (§9, §11).
|
||||
|
||||
## 13. Open questions and follow-ups
|
||||
|
||||
Unresolved choices are recorded here rather than settled by implication. Each needs its own durable issue before the phase that depends on it:
|
||||
|
||||
- **Authentication mechanism.** Whether the console authenticates via an access proxy (Cloudflare Access or equivalent) or an application-level session is deferred to #633. This ADR requires only that it fail closed.
|
||||
- **Event model substrate.** Whether the #637 timeline reads the control-plane event log directly or through a projection is deferred to #637.
|
||||
- **CI path filter coverage.** `webui/ci_paths.py` triggers the web UI suite on `webui/`, `tests/test_webui_*`, and `docs/webui*`. This ADR lives under `docs/architecture/`, so editing it alone does not trigger that gate; the accompanying `tests/test_webui_architecture_docs.py` does run in the full suite. Widening the filter is a small follow-up, deliberately not bundled into a documentation-only change.
|
||||
- **Retention.** Audit-event retention duration is owned by #633.
|
||||
|
||||
## 14. Acceptance
|
||||
|
||||
Accepting this ADR means:
|
||||
|
||||
- Phase 1 children may proceed against the layers, page map, and API rules above.
|
||||
- Phase 2 children may not open a write path until #633 lands.
|
||||
- Any deviation is recorded as an amendment to this file with its own issue reference, not as an undocumented divergence in code.
|
||||
+87
-1
@@ -37,12 +37,19 @@ Optional environment variables:
|
||||
See [webui-deployment.md](webui-deployment.md) for internal-only serving,
|
||||
Cloudflare Access/WARP/VPN guidance, and unsafe bind overrides (#435).
|
||||
|
||||
See
|
||||
[architecture/webui-control-plane-console-architecture-adr.md](architecture/webui-control-plane-console-architecture-adr.md)
|
||||
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).
|
||||
|
||||
## Routes (MVP)
|
||||
|
||||
| Path | Description |
|
||||
|------|-------------|
|
||||
| `/` | Home / operator overview |
|
||||
| `/health` | JSON liveness (`status`, `service`, `mode`, `timestamp`) |
|
||||
| `/health` | JSON liveness (`status`, `service`, `mode`, `timestamp`, `uptime_seconds`) |
|
||||
| `/api/v1/system/health` | Structured read-only system health (#634) |
|
||||
| `/queue` | Live PR and issue queue dashboard (#429) |
|
||||
| `/api/queue` | JSON queue export with pagination metadata |
|
||||
| `/projects` | Project registry list (#427) |
|
||||
@@ -66,6 +73,85 @@ Most routes are GET-only. POST/PUT/PATCH/DELETE return `405` with
|
||||
`read-only-mvp`, except `/audit` and `/api/audit` which accept POST for
|
||||
local validator preview only (no Gitea mutations, no server-side storage).
|
||||
|
||||
## System health API (#634)
|
||||
|
||||
`GET /api/v1/system/health` is the structured, read-only health surface for
|
||||
automated readiness checks. It is the first console API under the `/api/v1`
|
||||
prefix; the unversioned MVP exports remain as compatibility aliases.
|
||||
|
||||
`/health` is unchanged for existing consumers — every MVP key is still present
|
||||
— and now also carries `started_at`, `uptime_seconds`, and a
|
||||
`system_health_api` pointer. It stays deliberately cheap and runs no dependency
|
||||
probe, because answering readiness costs real work.
|
||||
|
||||
**Status codes.** `200` when ready, `503` when a required dependency failed or
|
||||
was never probed. Automation can branch on the code without parsing the body.
|
||||
|
||||
**Query flags.** The Gitea check is a network call, so it is opt-in:
|
||||
`GET /api/v1/system/health?deep=1` runs it and caches the result for
|
||||
`WEBUI_HEALTH_PROBE_TTL_SECONDS` (default 15s) so dashboard polling does not
|
||||
amplify into remote load. Without the flag that probe reports `skipped`.
|
||||
|
||||
**Dependencies.** `control_plane_db` and `repository` are required and drive
|
||||
readiness. `gitea` is optional: when it fails the overall `status` degrades but
|
||||
`readiness.ready` stays true, because local inventory is still serveable. Each
|
||||
entry carries `status`, `detail`, `required`, and `latency_ms`.
|
||||
|
||||
Two honesty rules are worth knowing before reading the payload:
|
||||
|
||||
* `stale_runtime.mutation_safe` is true only when the runtime, checkout, and
|
||||
remote-tracking commits are all known and equal. An unfetched remote is
|
||||
reported as indeterminate, never as safe.
|
||||
* `mcp_namespaces` entries are always `unproven`. A web process runs outside
|
||||
the IDE-managed MCP client and cannot invoke a namespace tool, so per #543
|
||||
only a `client_namespace` probe can prove that path.
|
||||
|
||||
Sample response (abridged, healthy):
|
||||
|
||||
```json
|
||||
{
|
||||
"status": "ok",
|
||||
"service": "mcp-control-plane-webui",
|
||||
"mode": "read-only",
|
||||
"api": "/api/v1/system/health",
|
||||
"timestamp": "2026-07-22T11:04:18.512034+00:00",
|
||||
"readiness": { "ready": true, "complete": true, "reasons": [] },
|
||||
"version": {
|
||||
"git_sha": "620ed6e9a9550b8da2ceb82d9ab8744e8920490f",
|
||||
"git_describe": "v1.1.0-898-g620ed6e",
|
||||
"control_plane_schema_version": 4,
|
||||
"python_version": "3.14.5",
|
||||
"known": true
|
||||
},
|
||||
"process": { "started_at": "2026-07-22T10:58:02.114+00:00", "uptime_seconds": 376.4 },
|
||||
"deep_probes_requested": false,
|
||||
"dependencies": [
|
||||
{
|
||||
"name": "control_plane_db",
|
||||
"kind": "sqlite",
|
||||
"status": "ok",
|
||||
"detail": "schema v4 readable",
|
||||
"required": true,
|
||||
"healthy": true,
|
||||
"latency_ms": 1.482,
|
||||
"metadata": { "schema_version": 4, "active_leases": 3 }
|
||||
},
|
||||
{ "name": "repository", "kind": "git", "status": "ok", "required": true, "healthy": true },
|
||||
{ "name": "gitea", "kind": "http", "status": "skipped", "required": false, "healthy": false }
|
||||
],
|
||||
"mcp_namespaces": [
|
||||
{ "namespace": "gitea-author", "required_tool": "gitea_whoami", "status": "unproven" }
|
||||
],
|
||||
"stale_runtime": { "stale": false, "determinable": true, "mutation_safe": true, "reasons": [] },
|
||||
"probe_errors": []
|
||||
}
|
||||
```
|
||||
|
||||
No restart, reload, or process-kill control is exposed here: those are Phase 2
|
||||
at the earliest, and #630 forbids process-kill recovery outright. Every probe
|
||||
opens its subject read-only — the control-plane database is opened through a
|
||||
`mode=ro` URI so a health check can never create or migrate a schema.
|
||||
|
||||
## Report audit (#431)
|
||||
|
||||
Paste an LLM final report at `/audit` or POST JSON to `/api/audit`. The UI
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
"""Documentation acceptance for the web console architecture ADR (#632 / epic #631).
|
||||
|
||||
Enforces the acceptance criteria of issue #632:
|
||||
|
||||
* AC1 — the ADR exists and covers layers, authority, phases, API versioning,
|
||||
and a page map.
|
||||
* AC2 — every #631 child (#632–#651) maps to at least one architectural
|
||||
component.
|
||||
* AC3 — the closed MVP (#425–#436) is stated as foundation, not recreated.
|
||||
* AC4 — forbidden paths are explicit: raw provider incidents as work,
|
||||
browser-held tokens, process-kill recovery.
|
||||
* AC5 — a controller can approve the document without reading chat history.
|
||||
|
||||
Plus the linkage requirement: ``docs/webui-local-dev.md`` cross-links the ADR.
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
ADR = (
|
||||
REPO_ROOT
|
||||
/ "docs"
|
||||
/ "architecture"
|
||||
/ "webui-control-plane-console-architecture-adr.md"
|
||||
)
|
||||
ADR_BASENAME = "webui-control-plane-console-architecture-adr.md"
|
||||
LOCAL_DEV = REPO_ROOT / "docs" / "webui-local-dev.md"
|
||||
|
||||
# Epic #631 children, phases 1-4 (twenty capability areas).
|
||||
EPIC_CHILDREN = tuple(f"#{number}" for number in range(632, 652))
|
||||
|
||||
|
||||
def _read(path: Path) -> str:
|
||||
assert path.is_file(), f"missing {path.relative_to(REPO_ROOT)}"
|
||||
return path.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def test_ac1_adr_exists_with_required_sections():
|
||||
text = _read(ADR)
|
||||
lower = text.lower()
|
||||
assert text.lstrip().startswith("#"), "ADR lacks a title"
|
||||
assert "#631" in text and "#632" in text
|
||||
for heading in (
|
||||
"## 2. Decision summary",
|
||||
"## 4. Authority boundaries",
|
||||
"## 5. Request flow and the redaction boundary",
|
||||
"## 6. API naming and versioning",
|
||||
"## 7. Page map",
|
||||
"## 8. Component ownership",
|
||||
"## 9. Phase gates",
|
||||
"## 11. Forbidden paths",
|
||||
):
|
||||
assert heading in text, f"ADR must contain section {heading!r}"
|
||||
assert "browser ui" in lower and "domain loader" in lower
|
||||
assert "control-plane db" in lower and "capability gate" in lower
|
||||
|
||||
|
||||
def test_ac1_api_versioning_is_decided_including_legacy_routes():
|
||||
text = _read(ADR)
|
||||
assert "/api/v1/" in text, "ADR must decide the versioned API prefix"
|
||||
assert "/api/v2/" in text, "ADR must state how breaking changes are handled"
|
||||
lower = text.lower()
|
||||
assert "compatibility alias" in lower, (
|
||||
"ADR must say what happens to the existing unversioned MVP exports"
|
||||
)
|
||||
|
||||
|
||||
def test_ac1_page_map_covers_mvp_routes():
|
||||
text = _read(ADR)
|
||||
for route in ("`/`", "`/health`", "`/projects`", "`/prompts`", "`/runtime`",
|
||||
"`/audit`", "`/actions`"):
|
||||
assert route in text, f"page map must account for MVP route {route}"
|
||||
|
||||
|
||||
def test_ac2_every_epic_child_maps_to_a_component():
|
||||
text = _read(ADR)
|
||||
ownership = text.split("## 8. Component ownership", 1)[-1].split("## 9.", 1)[0]
|
||||
missing = [child for child in EPIC_CHILDREN if child not in ownership]
|
||||
assert not missing, (
|
||||
f"epic #631 children without an architectural component: {missing}"
|
||||
)
|
||||
|
||||
|
||||
def test_ac2_every_child_row_declares_a_phase():
|
||||
text = _read(ADR)
|
||||
ownership = text.split("## 8. Component ownership", 1)[-1].split("## 9.", 1)[0]
|
||||
for child in EPIC_CHILDREN:
|
||||
row = next(
|
||||
(line for line in ownership.splitlines() if line.startswith(f"| {child} ")),
|
||||
None,
|
||||
)
|
||||
assert row is not None, f"no ownership row for {child}"
|
||||
assert row.rstrip().endswith(("| 1 |", "| 2 |", "| 3 |", "| 4 |")), (
|
||||
f"ownership row for {child} must end with its phase: {row!r}"
|
||||
)
|
||||
|
||||
|
||||
def test_ac3_mvp_is_foundation_not_recreated():
|
||||
text = _read(ADR)
|
||||
assert "#425" in text and "#436" in text
|
||||
lower = text.lower()
|
||||
assert "do not recreate" in lower or "recreating mvp scope" in lower
|
||||
assert "retained and evolved" in lower
|
||||
|
||||
|
||||
def test_ac4_forbidden_paths_are_explicit():
|
||||
text = _read(ADR)
|
||||
forbidden = text.split("## 11. Forbidden paths", 1)[-1].split("## 12.", 1)[0]
|
||||
lower = forbidden.lower()
|
||||
assert "raw provider incidents" in lower and "#612" in forbidden
|
||||
assert "browser-held tokens" in lower
|
||||
assert "process-kill recovery" in lower and "#630" in forbidden
|
||||
assert "ungated browser mutations" in lower
|
||||
|
||||
|
||||
def test_ac5_approval_checklist_is_self_contained():
|
||||
text = _read(ADR)
|
||||
assert "## 12. Approval checklist" in text
|
||||
checklist = text.split("## 12. Approval checklist", 1)[-1].split("## 13.", 1)[0]
|
||||
for marker in ("1.", "2.", "3.", "4.", "5.", "6."):
|
||||
assert marker in checklist, f"approval checklist missing item {marker}"
|
||||
|
||||
|
||||
def test_adr_states_the_two_boundary_invariants():
|
||||
text = _read(ADR)
|
||||
lower = text.lower()
|
||||
assert "no secrets to the browser" in lower
|
||||
assert "no ungated mutations" in lower
|
||||
|
||||
|
||||
def test_open_questions_are_recorded_not_implied():
|
||||
text = _read(ADR)
|
||||
assert "## 13. Open questions and follow-ups" in text
|
||||
section = text.split("## 13. Open questions and follow-ups", 1)[-1]
|
||||
assert "#633" in section, "deferred authorization work must name its issue"
|
||||
|
||||
|
||||
def test_local_dev_doc_cross_links_the_adr():
|
||||
text = _read(LOCAL_DEV)
|
||||
assert ADR_BASENAME in text, (
|
||||
"docs/webui-local-dev.md must cross-link the console architecture ADR "
|
||||
"(issue #632 scope)"
|
||||
)
|
||||
|
||||
|
||||
def test_docs_do_not_embed_secrets():
|
||||
for path in (ADR, LOCAL_DEV):
|
||||
text = _read(path)
|
||||
for marker in ("ghp_", "BEGIN PRIVATE KEY", "Authorization: Bearer"):
|
||||
assert marker not in text, f"{path.name} contains {marker!r}"
|
||||
@@ -0,0 +1,499 @@
|
||||
"""Tests for the read-only system-health API (#634).
|
||||
|
||||
Covers the acceptance criteria directly: a structured payload with readiness
|
||||
and a dependency list (AC1), version and uptime when knowable (AC2), stale
|
||||
runtime reported without a false mutation-safe claim (AC3), and the healthy /
|
||||
degraded-dependency / redaction cases (AC4).
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sqlite3
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
import control_plane_db
|
||||
from webui.app import create_app
|
||||
from webui.deployment_boundary import scan_text_for_client_secrets
|
||||
from webui.system_health import (
|
||||
API_PATH,
|
||||
STATUS_DEGRADED,
|
||||
STATUS_DOWN,
|
||||
STATUS_OK,
|
||||
STATUS_SKIPPED,
|
||||
DependencyProbe,
|
||||
StaleRuntime,
|
||||
assess_stale_runtime,
|
||||
clear_probe_cache,
|
||||
load_system_health,
|
||||
namespace_summaries,
|
||||
probe_control_plane_db,
|
||||
probe_gitea,
|
||||
process_uptime,
|
||||
redact,
|
||||
redact_url,
|
||||
snapshot_to_dict,
|
||||
)
|
||||
|
||||
|
||||
def _probe(name, status, *, required=True, detail="detail", kind="test"):
|
||||
return DependencyProbe(
|
||||
name=name,
|
||||
kind=kind,
|
||||
status=status,
|
||||
detail=detail,
|
||||
required=required,
|
||||
latency_ms=1.5,
|
||||
metadata={},
|
||||
)
|
||||
|
||||
|
||||
_ALL_HEALTHY = (
|
||||
_probe("control_plane_db", STATUS_OK, kind="sqlite"),
|
||||
_probe("repository", STATUS_OK, kind="git"),
|
||||
_probe("gitea", STATUS_OK, required=False, kind="http"),
|
||||
)
|
||||
|
||||
_CLEAN_PARITY = StaleRuntime(
|
||||
daemon_head="abc123",
|
||||
checkout_head="abc123",
|
||||
remote_head="abc123",
|
||||
stale=False,
|
||||
determinable=True,
|
||||
mutation_safe=True,
|
||||
reasons=(),
|
||||
)
|
||||
|
||||
|
||||
class CleanParityMixin:
|
||||
"""Pin parity for tests about aggregation rather than staleness.
|
||||
|
||||
Without this the assertions depend on the real checkout: a worktree whose
|
||||
branch is ahead of its upstream is genuinely stale, which would degrade the
|
||||
overall status and make these cases fail for an unrelated reason.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
patcher = mock.patch(
|
||||
"webui.system_health.assess_stale_runtime",
|
||||
return_value=_CLEAN_PARITY,
|
||||
)
|
||||
patcher.start()
|
||||
self.addCleanup(patcher.stop)
|
||||
|
||||
|
||||
class TestDependencyAggregation(CleanParityMixin, unittest.TestCase):
|
||||
"""AC1 — readiness and dependency list derived from probe results."""
|
||||
|
||||
def test_all_healthy_is_ok_and_ready(self):
|
||||
snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="abc123")
|
||||
self.assertEqual(snapshot.status, STATUS_OK)
|
||||
self.assertTrue(snapshot.ready)
|
||||
self.assertTrue(snapshot.readiness_complete)
|
||||
self.assertEqual(snapshot.readiness_reasons, ())
|
||||
self.assertEqual(len(snapshot.dependencies), 3)
|
||||
|
||||
def test_required_dependency_down_blocks_readiness(self):
|
||||
probes = (
|
||||
_probe("control_plane_db", STATUS_DOWN, detail="file missing", kind="sqlite"),
|
||||
_probe("repository", STATUS_OK, kind="git"),
|
||||
_probe("gitea", STATUS_OK, required=False, kind="http"),
|
||||
)
|
||||
snapshot = load_system_health(probes=probes, daemon_head="abc123")
|
||||
self.assertEqual(snapshot.status, STATUS_DOWN)
|
||||
self.assertFalse(snapshot.ready)
|
||||
self.assertTrue(
|
||||
any("control_plane_db" in reason for reason in snapshot.readiness_reasons)
|
||||
)
|
||||
|
||||
def test_optional_dependency_down_degrades_but_stays_ready(self):
|
||||
"""A failing optional probe must not claim the process itself is unready."""
|
||||
probes = (
|
||||
_probe("control_plane_db", STATUS_OK, kind="sqlite"),
|
||||
_probe("repository", STATUS_OK, kind="git"),
|
||||
_probe("gitea", STATUS_DOWN, required=False, detail="timeout", kind="http"),
|
||||
)
|
||||
snapshot = load_system_health(probes=probes, daemon_head="abc123")
|
||||
self.assertEqual(snapshot.status, STATUS_DEGRADED)
|
||||
self.assertTrue(snapshot.ready)
|
||||
self.assertTrue(any("gitea" in reason for reason in snapshot.readiness_reasons))
|
||||
|
||||
def test_unrun_required_probe_leaves_readiness_incomplete(self):
|
||||
"""Not probed is not the same as passing."""
|
||||
probes = (
|
||||
_probe("control_plane_db", STATUS_OK, kind="sqlite"),
|
||||
_probe("repository", STATUS_SKIPPED, detail="offline", kind="git"),
|
||||
)
|
||||
snapshot = load_system_health(probes=probes, daemon_head="abc123")
|
||||
self.assertFalse(snapshot.ready)
|
||||
self.assertFalse(snapshot.readiness_complete)
|
||||
self.assertEqual(snapshot.status, STATUS_DEGRADED)
|
||||
|
||||
def test_skipped_optional_probe_does_not_block_readiness(self):
|
||||
probes = (
|
||||
_probe("control_plane_db", STATUS_OK, kind="sqlite"),
|
||||
_probe("repository", STATUS_OK, kind="git"),
|
||||
_probe("gitea", STATUS_SKIPPED, required=False, kind="http"),
|
||||
)
|
||||
snapshot = load_system_health(probes=probes, daemon_head="abc123")
|
||||
self.assertTrue(snapshot.ready)
|
||||
self.assertTrue(snapshot.readiness_complete)
|
||||
|
||||
|
||||
class TestVersionAndUptime(CleanParityMixin, unittest.TestCase):
|
||||
"""AC2 — version and uptime present when knowable."""
|
||||
|
||||
def test_uptime_and_start_time_present(self):
|
||||
snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="abc123")
|
||||
self.assertGreaterEqual(snapshot.uptime_seconds, 0.0)
|
||||
self.assertIn("T", snapshot.started_at)
|
||||
|
||||
def test_process_uptime_helper_matches_shape(self):
|
||||
started_at, uptime = process_uptime()
|
||||
self.assertIn("T", started_at)
|
||||
self.assertGreaterEqual(uptime, 0.0)
|
||||
|
||||
def test_version_reports_python_and_schema_version(self):
|
||||
probes = (
|
||||
DependencyProbe(
|
||||
name="control_plane_db",
|
||||
kind="sqlite",
|
||||
status=STATUS_OK,
|
||||
detail="ok",
|
||||
required=True,
|
||||
latency_ms=1.0,
|
||||
metadata={"schema_version": control_plane_db.SCHEMA_VERSION},
|
||||
),
|
||||
_probe("repository", STATUS_OK, kind="git"),
|
||||
)
|
||||
snapshot = load_system_health(probes=probes, daemon_head="abc123")
|
||||
self.assertEqual(
|
||||
snapshot.version.control_plane_schema_version,
|
||||
control_plane_db.SCHEMA_VERSION,
|
||||
)
|
||||
self.assertTrue(snapshot.version.python_version)
|
||||
|
||||
def test_version_known_flag_false_when_sha_unavailable(self):
|
||||
with mock.patch("webui.system_health._git", return_value=None):
|
||||
snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="abc")
|
||||
self.assertIsNone(snapshot.version.git_sha)
|
||||
self.assertFalse(snapshot.version.known)
|
||||
|
||||
|
||||
class TestStaleRuntime(unittest.TestCase):
|
||||
"""AC3 — stale runtime reflected without a false mutation-safe claim."""
|
||||
|
||||
def test_matching_commits_are_mutation_safe(self):
|
||||
assessment = assess_stale_runtime(
|
||||
Path("/tmp"),
|
||||
daemon_head="aaa",
|
||||
git_reader=lambda *args: "aaa",
|
||||
)
|
||||
self.assertFalse(assessment.stale)
|
||||
self.assertTrue(assessment.determinable)
|
||||
self.assertTrue(assessment.mutation_safe)
|
||||
|
||||
def test_diverged_commits_are_stale_and_not_mutation_safe(self):
|
||||
reads = {"HEAD": "aaa", "@{upstream}": "bbb"}
|
||||
assessment = assess_stale_runtime(
|
||||
Path("/tmp"),
|
||||
daemon_head="aaa",
|
||||
git_reader=lambda *args: reads.get(args[-1]),
|
||||
)
|
||||
self.assertTrue(assessment.stale)
|
||||
self.assertFalse(assessment.mutation_safe)
|
||||
self.assertTrue(assessment.reasons)
|
||||
|
||||
def test_unknown_remote_is_not_mutation_safe(self):
|
||||
"""Indeterminate must never read as safe."""
|
||||
reads = {"HEAD": "aaa", "@{upstream}": None}
|
||||
assessment = assess_stale_runtime(
|
||||
Path("/tmp"),
|
||||
daemon_head="aaa",
|
||||
git_reader=lambda *args: reads.get(args[-1]),
|
||||
)
|
||||
self.assertFalse(assessment.determinable)
|
||||
self.assertFalse(assessment.mutation_safe)
|
||||
self.assertFalse(assessment.stale)
|
||||
self.assertTrue(
|
||||
any("indeterminate" in reason for reason in assessment.reasons)
|
||||
)
|
||||
|
||||
def test_unobservable_daemon_head_is_disclosed(self):
|
||||
assessment = assess_stale_runtime(
|
||||
Path("/tmp"),
|
||||
git_reader=lambda *args: "aaa",
|
||||
)
|
||||
self.assertTrue(
|
||||
any("not observable" in reason for reason in assessment.reasons)
|
||||
)
|
||||
|
||||
def test_stale_runtime_degrades_overall_status(self):
|
||||
reads = {"HEAD": "aaa", "@{upstream}": "bbb"}
|
||||
# Pinned rather than inherited: this path uses the default git reader,
|
||||
# so the assertion must hold whether or not the suite runs offline.
|
||||
with mock.patch.dict(os.environ, {"WEBUI_TEST_OFFLINE": ""}), mock.patch(
|
||||
"webui.system_health._git",
|
||||
side_effect=lambda repo, *args: reads.get(args[-1]),
|
||||
):
|
||||
snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="aaa")
|
||||
self.assertTrue(snapshot.stale_runtime.stale)
|
||||
self.assertFalse(snapshot.stale_runtime.mutation_safe)
|
||||
self.assertEqual(snapshot.status, STATUS_DEGRADED)
|
||||
|
||||
|
||||
class TestControlPlaneDbProbe(unittest.TestCase):
|
||||
"""The required local dependency, probed read-only."""
|
||||
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.db_path = str(Path(self.tmp.name) / "control-plane.db")
|
||||
|
||||
def _build_db(self, schema_version):
|
||||
conn = sqlite3.connect(self.db_path)
|
||||
conn.execute("CREATE TABLE schema_meta (key TEXT PRIMARY KEY, value TEXT)")
|
||||
conn.execute("CREATE TABLE leases (lease_id TEXT PRIMARY KEY, status TEXT)")
|
||||
conn.execute(
|
||||
"INSERT INTO schema_meta(key, value) VALUES ('schema_version', ?)",
|
||||
(str(schema_version),),
|
||||
)
|
||||
conn.execute("INSERT INTO leases(lease_id, status) VALUES ('l1', 'active')")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
def test_missing_database_is_down(self):
|
||||
probe = probe_control_plane_db(str(Path(self.tmp.name) / "absent.db"))
|
||||
self.assertEqual(probe.status, STATUS_DOWN)
|
||||
self.assertTrue(probe.required)
|
||||
self.assertIsNotNone(probe.latency_ms)
|
||||
|
||||
def test_matching_schema_is_ok(self):
|
||||
self._build_db(control_plane_db.SCHEMA_VERSION)
|
||||
probe = probe_control_plane_db(self.db_path)
|
||||
self.assertEqual(probe.status, STATUS_OK)
|
||||
self.assertEqual(
|
||||
probe.metadata["schema_version"], control_plane_db.SCHEMA_VERSION
|
||||
)
|
||||
self.assertEqual(probe.metadata["active_leases"], 1)
|
||||
|
||||
def test_mismatched_schema_is_degraded(self):
|
||||
self._build_db(control_plane_db.SCHEMA_VERSION + 99)
|
||||
probe = probe_control_plane_db(self.db_path)
|
||||
self.assertEqual(probe.status, STATUS_DEGRADED)
|
||||
|
||||
def test_probe_does_not_create_a_database(self):
|
||||
"""A health check must never initialise the substrate it inspects."""
|
||||
absent = str(Path(self.tmp.name) / "never-created.db")
|
||||
probe_control_plane_db(absent)
|
||||
self.assertFalse(Path(absent).exists())
|
||||
|
||||
def test_unreadable_database_is_down_not_raised(self):
|
||||
Path(self.db_path).write_text("this is not a sqlite database")
|
||||
probe = probe_control_plane_db(self.db_path)
|
||||
self.assertEqual(probe.status, STATUS_DOWN)
|
||||
|
||||
|
||||
class TestRedaction(unittest.TestCase):
|
||||
"""AC4 — redaction. No credential-shaped text crosses the boundary."""
|
||||
|
||||
def test_redacts_token_assignment(self):
|
||||
cleaned = redact("failed with token=ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ012345")
|
||||
self.assertNotIn("ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ012345", cleaned)
|
||||
self.assertIn("[redacted]", cleaned)
|
||||
|
||||
def test_redacts_authorization_header_text(self):
|
||||
cleaned = redact("Authorization: Bearer abcdefghijklmnopqrstuvwxyz123456")
|
||||
self.assertNotIn("abcdefghijklmnopqrstuvwxyz123456", cleaned)
|
||||
|
||||
def test_redacts_long_opaque_strings(self):
|
||||
cleaned = redact("value 0123456789abcdef0123456789abcdef here")
|
||||
self.assertNotIn("0123456789abcdef0123456789abcdef", cleaned)
|
||||
|
||||
def test_url_userinfo_and_query_are_stripped(self):
|
||||
cleaned = redact_url("https://user:[email protected]/api/v1?token=xyz")
|
||||
self.assertNotIn("secretpass", cleaned)
|
||||
self.assertNotIn("token=xyz", cleaned)
|
||||
self.assertEqual(cleaned, "https://gitea.example.com/api/v1")
|
||||
|
||||
def test_url_inside_free_text_is_redacted(self):
|
||||
cleaned = redact("GET https://u:[email protected]/x?token=abc failed")
|
||||
self.assertNotIn("u:p@", cleaned)
|
||||
self.assertNotIn("token=abc", cleaned)
|
||||
|
||||
def test_gitea_probe_failure_detail_is_redacted(self):
|
||||
boom = RuntimeError(
|
||||
"connection refused for https://user:[email protected]/api/v1/version"
|
||||
)
|
||||
with mock.patch("webui.system_health.get_auth_header", return_value="token x"), \
|
||||
mock.patch("webui.system_health.api_request", side_effect=boom):
|
||||
probe = probe_gitea("gitea.example.com")
|
||||
self.assertEqual(probe.status, STATUS_DOWN)
|
||||
self.assertNotIn("hunter2", probe.detail)
|
||||
self.assertEqual(scan_text_for_client_secrets(probe.detail), [])
|
||||
|
||||
def test_credential_guard_refusal_is_a_status_not_a_crash(self):
|
||||
with mock.patch(
|
||||
"webui.system_health.get_auth_header",
|
||||
side_effect=RuntimeError("daemon guard refused"),
|
||||
):
|
||||
probe = probe_gitea("gitea.example.com")
|
||||
self.assertEqual(probe.status, STATUS_DEGRADED)
|
||||
self.assertFalse(probe.required)
|
||||
|
||||
|
||||
class TestNamespaceSummaries(unittest.TestCase):
|
||||
"""A web process cannot prove IDE namespace health, and must not claim to."""
|
||||
|
||||
def test_every_namespace_reports_unproven(self):
|
||||
rows = namespace_summaries()
|
||||
self.assertTrue(rows)
|
||||
for row in rows:
|
||||
with self.subTest(namespace=row["namespace"]):
|
||||
self.assertEqual(row["status"], "unproven")
|
||||
self.assertFalse(row["ide_namespace_proven"])
|
||||
self.assertIn("client_namespace", row["reason"])
|
||||
|
||||
|
||||
class TestSystemHealthRoutes(CleanParityMixin, unittest.TestCase):
|
||||
"""The HTTP surface: versioned path, status codes, read-only guard."""
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
clear_probe_cache()
|
||||
self.addCleanup(clear_probe_cache)
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def _patch_snapshot(self, probes, daemon_head="abc123"):
|
||||
snapshot = load_system_health(probes=probes, daemon_head=daemon_head)
|
||||
patcher = mock.patch(
|
||||
"webui.app.load_system_health",
|
||||
return_value=snapshot,
|
||||
)
|
||||
patcher.start()
|
||||
self.addCleanup(patcher.stop)
|
||||
return snapshot
|
||||
|
||||
def test_versioned_route_is_registered(self):
|
||||
self.assertEqual(API_PATH, "/api/v1/system/health")
|
||||
self._patch_snapshot(_ALL_HEALTHY)
|
||||
response = self.client.get(API_PATH)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_healthy_payload_shape(self):
|
||||
self._patch_snapshot(_ALL_HEALTHY)
|
||||
data = self.client.get(API_PATH).json()
|
||||
self.assertEqual(data["status"], STATUS_OK)
|
||||
self.assertTrue(data["readiness"]["ready"])
|
||||
self.assertTrue(data["readiness"]["complete"])
|
||||
self.assertEqual(data["api"], API_PATH)
|
||||
self.assertEqual(len(data["dependencies"]), 3)
|
||||
for key in ("version", "process", "stale_runtime", "mcp_namespaces"):
|
||||
self.assertIn(key, data)
|
||||
self.assertIn("uptime_seconds", data["process"])
|
||||
self.assertIn("mutation_safe", data["stale_runtime"])
|
||||
|
||||
def test_degraded_dependency_returns_503(self):
|
||||
probes = (
|
||||
_probe("control_plane_db", STATUS_DOWN, detail="missing", kind="sqlite"),
|
||||
_probe("repository", STATUS_OK, kind="git"),
|
||||
)
|
||||
self._patch_snapshot(probes)
|
||||
response = self.client.get(API_PATH)
|
||||
self.assertEqual(response.status_code, 503)
|
||||
data = response.json()
|
||||
self.assertFalse(data["readiness"]["ready"])
|
||||
self.assertTrue(data["readiness"]["reasons"])
|
||||
|
||||
def test_dependency_entries_expose_status_and_latency(self):
|
||||
self._patch_snapshot(_ALL_HEALTHY)
|
||||
data = self.client.get(API_PATH).json()
|
||||
names = {entry["name"] for entry in data["dependencies"]}
|
||||
self.assertEqual(names, {"control_plane_db", "repository", "gitea"})
|
||||
for entry in data["dependencies"]:
|
||||
with self.subTest(dependency=entry["name"]):
|
||||
self.assertIn("status", entry)
|
||||
self.assertIn("required", entry)
|
||||
self.assertIn("latency_ms", entry)
|
||||
|
||||
def test_response_body_carries_no_client_secrets(self):
|
||||
self._patch_snapshot(_ALL_HEALTHY)
|
||||
body = self.client.get(API_PATH).text
|
||||
self.assertEqual(scan_text_for_client_secrets(body), [])
|
||||
|
||||
def test_deep_flag_is_forwarded(self):
|
||||
snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="abc")
|
||||
with mock.patch(
|
||||
"webui.app.load_system_health", return_value=snapshot
|
||||
) as loader:
|
||||
self.client.get(f"{API_PATH}?deep=1")
|
||||
loader.assert_called_once_with(deep=True)
|
||||
|
||||
def test_shallow_is_the_default(self):
|
||||
snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="abc")
|
||||
with mock.patch(
|
||||
"webui.app.load_system_health", return_value=snapshot
|
||||
) as loader:
|
||||
self.client.get(API_PATH)
|
||||
loader.assert_called_once_with(deep=False)
|
||||
|
||||
def test_route_rejects_mutation_methods(self):
|
||||
for method in ("POST", "PUT", "PATCH", "DELETE"):
|
||||
with self.subTest(method=method):
|
||||
response = self.client.request(method, API_PATH)
|
||||
self.assertEqual(response.status_code, 405)
|
||||
self.assertEqual(response.json()["error"], "read-only-mvp")
|
||||
|
||||
def test_default_shallow_call_skips_the_network_probe(self):
|
||||
"""The expensive probe must not run unless it was asked for."""
|
||||
with mock.patch("webui.system_health.probe_gitea") as probe:
|
||||
snapshot = load_system_health(deep=False)
|
||||
probe.assert_not_called()
|
||||
gitea = next(p for p in snapshot.dependencies if p.name == "gitea")
|
||||
self.assertEqual(gitea.status, STATUS_SKIPPED)
|
||||
|
||||
|
||||
class TestHealthRouteBackwardCompatibility(unittest.TestCase):
|
||||
"""`/health` is expanded additively; MVP consumers must keep working."""
|
||||
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_mvp_keys_are_unchanged(self):
|
||||
data = self.client.get("/health").json()
|
||||
self.assertEqual(data["status"], "ok")
|
||||
self.assertEqual(data["service"], "mcp-control-plane-webui")
|
||||
self.assertEqual(data["mode"], "read-only-mvp")
|
||||
self.assertIn("timestamp", data)
|
||||
self.assertEqual(data["deployment"]["mode"], "internal-operator-console")
|
||||
|
||||
def test_health_points_at_the_versioned_api(self):
|
||||
data = self.client.get("/health").json()
|
||||
self.assertEqual(data["system_health_api"], API_PATH)
|
||||
self.assertIn("uptime_seconds", data)
|
||||
self.assertIn("started_at", data)
|
||||
|
||||
def test_health_runs_no_dependency_probe(self):
|
||||
"""Liveness must stay cheap: no probe, no snapshot assembly."""
|
||||
with mock.patch("webui.app.load_system_health") as loader:
|
||||
response = self.client.get("/health")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
loader.assert_not_called()
|
||||
|
||||
|
||||
class TestSnapshotSerialisation(CleanParityMixin, unittest.TestCase):
|
||||
def test_snapshot_dict_is_json_serialisable(self):
|
||||
snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="abc123")
|
||||
encoded = json.dumps(snapshot_to_dict(snapshot))
|
||||
self.assertIn("readiness", encoded)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,458 @@
|
||||
"""Tests for the worker registry and configuration schema (#798, epic #797)."""
|
||||
import json
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from webui.worker_registry import (
|
||||
ALLOWED_ROLES,
|
||||
SCHEMA_VERSION,
|
||||
RegistryValidationError,
|
||||
WorkerRegistry,
|
||||
default_registry_path,
|
||||
find_provider,
|
||||
find_worker,
|
||||
history_dir,
|
||||
list_revisions,
|
||||
load_registry,
|
||||
registry_to_dict,
|
||||
registry_to_document,
|
||||
rollback_to_revision,
|
||||
save_registry,
|
||||
validate_payload,
|
||||
worker_to_dict,
|
||||
workers_for_provider,
|
||||
)
|
||||
|
||||
_EXPECTED_PROVIDER_IDS = ("claude", "grok", "codex", "agy", "kimi-k")
|
||||
|
||||
|
||||
def _provider(provider_id: str = "claude", **overrides) -> dict:
|
||||
payload = {
|
||||
"id": provider_id,
|
||||
"display_name": "Claude",
|
||||
"vendor": "Anthropic",
|
||||
"executable": "claude",
|
||||
"available": True,
|
||||
"models": ["claude-opus-4-8"],
|
||||
"notes": "",
|
||||
}
|
||||
payload.update(overrides)
|
||||
return payload
|
||||
|
||||
|
||||
def _worker(worker_id: str = "claude-author", **overrides) -> dict:
|
||||
payload = {
|
||||
"id": worker_id,
|
||||
"display_name": "Claude author",
|
||||
"provider": "claude",
|
||||
"model": "claude-opus-4-8",
|
||||
"project": "gitea-tools",
|
||||
"role": "author",
|
||||
"namespace": "gitea-author",
|
||||
"profile": "prgs-author",
|
||||
"workflow": "skills/llm-project-workflow/workflows/work-issue.md",
|
||||
"schedule": {"kind": "cron", "expression": "0 * * * *"},
|
||||
"timeout_seconds": 3600,
|
||||
"enabled": True,
|
||||
"scheduler": {"kind": "launchd", "label": "cc.prgs.claude.author"},
|
||||
"notes": "",
|
||||
}
|
||||
payload.update(overrides)
|
||||
return payload
|
||||
|
||||
|
||||
def _document(providers=None, workers=None, **overrides) -> dict:
|
||||
payload = {
|
||||
"version": SCHEMA_VERSION,
|
||||
"revision": 1,
|
||||
"updated_at": "2026-07-22T00:00:00Z",
|
||||
"providers": providers if providers is not None else [_provider()],
|
||||
"workers": workers if workers is not None else [_worker()],
|
||||
}
|
||||
payload.update(overrides)
|
||||
return payload
|
||||
|
||||
|
||||
class _TempRegistryCase(unittest.TestCase):
|
||||
"""Base case giving each test an isolated registry file."""
|
||||
|
||||
def setUp(self):
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self._tmp.cleanup)
|
||||
self.path = Path(self._tmp.name) / "workers.registry.json"
|
||||
|
||||
def write(self, document: dict) -> Path:
|
||||
self.path.write_text(json.dumps(document, indent=2) + "\n", encoding="utf-8")
|
||||
return self.path
|
||||
|
||||
def parse(self, document: dict) -> WorkerRegistry:
|
||||
return validate_payload(document, source_path=self.path)
|
||||
|
||||
|
||||
class TestPackagedRegistry(unittest.TestCase):
|
||||
"""AC: the declarative registry is the source of truth and ships with the app."""
|
||||
|
||||
def test_default_path_points_at_packaged_data(self):
|
||||
path = default_registry_path()
|
||||
self.assertEqual(path.name, "workers.registry.json")
|
||||
self.assertEqual(path.parent.name, "data")
|
||||
|
||||
def test_packaged_registry_loads_and_validates(self):
|
||||
registry = load_registry()
|
||||
self.assertEqual(registry.version, SCHEMA_VERSION)
|
||||
self.assertGreaterEqual(registry.revision, 1)
|
||||
|
||||
def test_packaged_registry_declares_all_five_providers(self):
|
||||
registry = load_registry()
|
||||
self.assertEqual(
|
||||
tuple(provider.id for provider in registry.providers),
|
||||
_EXPECTED_PROVIDER_IDS,
|
||||
)
|
||||
|
||||
def test_packaged_registry_carries_no_credentials(self):
|
||||
raw = default_registry_path().read_text(encoding="utf-8").lower()
|
||||
for marker in ("token", "password", "secret", "api_key", "credential"):
|
||||
self.assertNotIn(marker, raw)
|
||||
|
||||
|
||||
class TestSeparateEntities(_TempRegistryCase):
|
||||
"""AC: providers and configured workers are separate entities."""
|
||||
|
||||
def test_provider_may_exist_with_no_workers(self):
|
||||
registry = self.parse(
|
||||
_document(providers=[_provider("grok", display_name="Grok")], workers=[])
|
||||
)
|
||||
self.assertEqual(len(registry.providers), 1)
|
||||
self.assertEqual(registry.workers, ())
|
||||
self.assertEqual(workers_for_provider(registry, "grok"), ())
|
||||
|
||||
def test_many_workers_may_share_one_provider(self):
|
||||
registry = self.parse(
|
||||
_document(
|
||||
workers=[
|
||||
_worker("claude-author"),
|
||||
_worker(
|
||||
"claude-reviewer",
|
||||
role="reviewer",
|
||||
namespace="gitea-reviewer",
|
||||
profile="prgs-reviewer",
|
||||
scheduler={"kind": "launchd", "label": "cc.prgs.claude.reviewer"},
|
||||
),
|
||||
]
|
||||
)
|
||||
)
|
||||
self.assertEqual(len(workers_for_provider(registry, "claude")), 2)
|
||||
self.assertEqual(len(registry.providers), 1)
|
||||
|
||||
def test_worker_referencing_unknown_provider_is_refused(self):
|
||||
with self.assertRaises(RegistryValidationError) as ctx:
|
||||
self.parse(_document(workers=[_worker(provider="mystery")]))
|
||||
self.assertIn("unknown provider", str(ctx.exception))
|
||||
|
||||
def test_lookup_helpers(self):
|
||||
registry = self.parse(_document())
|
||||
self.assertIsNotNone(find_worker(registry, "claude-author"))
|
||||
self.assertIsNone(find_worker(registry, "absent"))
|
||||
self.assertIsNotNone(find_provider(registry, "claude"))
|
||||
self.assertIsNone(find_provider(registry, "absent"))
|
||||
|
||||
|
||||
class TestRecordedFields(_TempRegistryCase):
|
||||
"""AC: records provider, model, project, role, namespace/profile, workflow,
|
||||
schedule, timeout, enabled state, and scheduler metadata."""
|
||||
|
||||
def test_every_required_field_is_recorded(self):
|
||||
registry = self.parse(_document())
|
||||
worker = registry.workers[0]
|
||||
self.assertEqual(worker.provider, "claude")
|
||||
self.assertEqual(worker.model, "claude-opus-4-8")
|
||||
self.assertEqual(worker.project, "gitea-tools")
|
||||
self.assertEqual(worker.role, "author")
|
||||
self.assertEqual(worker.namespace, "gitea-author")
|
||||
self.assertEqual(worker.profile, "prgs-author")
|
||||
self.assertEqual(worker.workflow, "skills/llm-project-workflow/workflows/work-issue.md")
|
||||
self.assertEqual(worker.schedule.kind, "cron")
|
||||
self.assertEqual(worker.schedule.expression, "0 * * * *")
|
||||
self.assertEqual(worker.timeout_seconds, 3600)
|
||||
self.assertTrue(worker.enabled)
|
||||
self.assertEqual(worker.scheduler.kind, "launchd")
|
||||
self.assertEqual(worker.scheduler.label, "cc.prgs.claude.author")
|
||||
|
||||
def test_each_required_field_is_individually_required(self):
|
||||
for field in (
|
||||
"provider", "model", "project", "role", "namespace",
|
||||
"profile", "workflow", "schedule", "timeout_seconds",
|
||||
"enabled", "scheduler", "id", "display_name",
|
||||
):
|
||||
with self.subTest(field=field):
|
||||
worker = _worker()
|
||||
worker.pop(field)
|
||||
with self.assertRaises(RegistryValidationError):
|
||||
self.parse(_document(workers=[worker]))
|
||||
|
||||
def test_all_sanctioned_roles_are_accepted(self):
|
||||
for role in ALLOWED_ROLES:
|
||||
with self.subTest(role=role):
|
||||
registry = self.parse(_document(workers=[_worker(role=role)]))
|
||||
self.assertEqual(registry.workers[0].role, role)
|
||||
|
||||
def test_unsanctioned_role_is_refused(self):
|
||||
with self.assertRaises(RegistryValidationError) as ctx:
|
||||
self.parse(_document(workers=[_worker(role="admin")]))
|
||||
self.assertIn("role must be one of", str(ctx.exception))
|
||||
|
||||
def test_worker_dict_round_trips_every_field(self):
|
||||
registry = self.parse(_document())
|
||||
encoded = worker_to_dict(registry.workers[0])
|
||||
self.assertEqual(encoded, _worker())
|
||||
json.dumps(encoded) # must stay JSON-safe for the #799 API
|
||||
|
||||
|
||||
class TestSchemaValidation(_TempRegistryCase):
|
||||
"""AC: supports schema validation — and fails closed."""
|
||||
|
||||
def test_unsupported_version_is_refused(self):
|
||||
with self.assertRaises(RegistryValidationError):
|
||||
self.parse(_document(version=2))
|
||||
|
||||
def test_root_must_be_an_object(self):
|
||||
with self.assertRaises(RegistryValidationError):
|
||||
validate_payload([], source_path=self.path)
|
||||
|
||||
def test_providers_must_be_non_empty(self):
|
||||
with self.assertRaises(RegistryValidationError):
|
||||
self.parse(_document(providers=[]))
|
||||
|
||||
def test_unknown_top_level_field_is_refused(self):
|
||||
with self.assertRaises(RegistryValidationError) as ctx:
|
||||
self.parse(_document(fleet=[]))
|
||||
self.assertIn("unknown fields", str(ctx.exception))
|
||||
|
||||
def test_unknown_worker_field_is_refused_not_ignored(self):
|
||||
# A typo'd field must not be silently dropped: "timeout_second" would
|
||||
# otherwise read as "no timeout declared".
|
||||
worker = _worker()
|
||||
worker["timeout_second"] = 30
|
||||
with self.assertRaises(RegistryValidationError) as ctx:
|
||||
self.parse(_document(workers=[worker]))
|
||||
self.assertIn("timeout_second", str(ctx.exception))
|
||||
|
||||
def test_credentials_are_refused_anywhere_in_the_document(self):
|
||||
for label, mutate in (
|
||||
("provider.api_token", lambda doc: doc["providers"][0].__setitem__("api_token", "x")),
|
||||
("worker.password", lambda doc: doc["workers"][0].__setitem__("password", "x")),
|
||||
("root.secret", lambda doc: doc.__setitem__("secret", "x")),
|
||||
):
|
||||
with self.subTest(field=label):
|
||||
document = _document()
|
||||
mutate(document)
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
self.parse(document)
|
||||
self.assertIn("credential", str(ctx.exception).lower())
|
||||
|
||||
def test_duplicate_worker_id_is_refused(self):
|
||||
workers = [_worker("dup"), _worker("dup", scheduler={"kind": "manual"})]
|
||||
with self.assertRaises(RegistryValidationError) as ctx:
|
||||
self.parse(_document(workers=workers))
|
||||
self.assertIn("duplicate worker id", str(ctx.exception))
|
||||
|
||||
def test_duplicate_provider_id_is_refused(self):
|
||||
with self.assertRaises(RegistryValidationError) as ctx:
|
||||
self.parse(_document(providers=[_provider("claude"), _provider("claude")], workers=[]))
|
||||
self.assertIn("duplicate provider id", str(ctx.exception))
|
||||
|
||||
def test_duplicate_launchagent_label_is_refused(self):
|
||||
# Two workers sharing a label would silently overwrite each other's agent.
|
||||
workers = [
|
||||
_worker("a", scheduler={"kind": "launchd", "label": "cc.prgs.same"}),
|
||||
_worker("b", scheduler={"kind": "launchd", "label": "cc.prgs.same"}),
|
||||
]
|
||||
with self.assertRaises(RegistryValidationError) as ctx:
|
||||
self.parse(_document(workers=workers))
|
||||
self.assertIn("duplicate scheduler label", str(ctx.exception))
|
||||
|
||||
def test_manual_scheduler_needs_no_label_and_many_may_coexist(self):
|
||||
workers = [
|
||||
_worker("a", scheduler={"kind": "manual"}),
|
||||
_worker("b", scheduler={"kind": "manual"}),
|
||||
]
|
||||
registry = self.parse(_document(workers=workers))
|
||||
self.assertEqual([w.scheduler.label for w in registry.workers], [None, None])
|
||||
|
||||
def test_launchd_scheduler_requires_a_label(self):
|
||||
with self.assertRaises(RegistryValidationError) as ctx:
|
||||
self.parse(_document(workers=[_worker(scheduler={"kind": "launchd"})]))
|
||||
self.assertIn("label is required", str(ctx.exception))
|
||||
|
||||
def test_unknown_scheduler_kind_is_refused(self):
|
||||
with self.assertRaises(RegistryValidationError):
|
||||
self.parse(_document(workers=[_worker(scheduler={"kind": "systemd", "label": "x"})]))
|
||||
|
||||
def test_timeout_must_be_a_positive_bounded_integer(self):
|
||||
for bad in (0, -1, "3600", 1.5, True, 86_401):
|
||||
with self.subTest(timeout=bad):
|
||||
with self.assertRaises(RegistryValidationError):
|
||||
self.parse(_document(workers=[_worker(timeout_seconds=bad)]))
|
||||
|
||||
def test_enabled_must_be_a_real_boolean(self):
|
||||
for bad in ("true", 1, None):
|
||||
with self.subTest(enabled=bad):
|
||||
with self.assertRaises(RegistryValidationError):
|
||||
self.parse(_document(workers=[_worker(enabled=bad)]))
|
||||
|
||||
def test_identifier_shape_is_enforced(self):
|
||||
for bad in ("Claude Author", "-leading", "UPPER", ""):
|
||||
with self.subTest(worker_id=bad):
|
||||
with self.assertRaises(RegistryValidationError):
|
||||
self.parse(_document(workers=[_worker(bad)]))
|
||||
|
||||
|
||||
class TestScheduleValidation(_TempRegistryCase):
|
||||
"""Schedules are declarations; next-run computation belongs to #803."""
|
||||
|
||||
def test_interval_schedule_requires_positive_seconds(self):
|
||||
registry = self.parse(
|
||||
_document(workers=[_worker(schedule={"kind": "interval", "seconds": 900})])
|
||||
)
|
||||
self.assertEqual(registry.workers[0].schedule.seconds, 900)
|
||||
with self.assertRaises(RegistryValidationError):
|
||||
self.parse(_document(workers=[_worker(schedule={"kind": "interval"})]))
|
||||
with self.assertRaises(RegistryValidationError):
|
||||
self.parse(_document(workers=[_worker(schedule={"kind": "interval", "seconds": 0})]))
|
||||
|
||||
def test_cron_schedule_requires_five_fields(self):
|
||||
with self.assertRaises(RegistryValidationError) as ctx:
|
||||
self.parse(_document(workers=[_worker(schedule={"kind": "cron", "expression": "0 *"})]))
|
||||
self.assertIn("five crontab fields", str(ctx.exception))
|
||||
|
||||
def test_manual_schedule_needs_no_timing(self):
|
||||
registry = self.parse(_document(workers=[_worker(schedule={"kind": "manual"})]))
|
||||
schedule = registry.workers[0].schedule
|
||||
self.assertEqual(schedule.kind, "manual")
|
||||
self.assertIsNone(schedule.seconds)
|
||||
self.assertIsNone(schedule.expression)
|
||||
|
||||
def test_fields_from_the_wrong_kind_are_refused(self):
|
||||
with self.assertRaises(RegistryValidationError) as ctx:
|
||||
self.parse(_document(workers=[_worker(schedule={"kind": "manual", "seconds": 60})]))
|
||||
self.assertIn("not valid for kind", str(ctx.exception))
|
||||
|
||||
def test_unknown_schedule_kind_is_refused(self):
|
||||
with self.assertRaises(RegistryValidationError):
|
||||
self.parse(_document(workers=[_worker(schedule={"kind": "hourly"})]))
|
||||
|
||||
|
||||
class TestAtomicPersistence(_TempRegistryCase):
|
||||
"""AC: atomic persistence."""
|
||||
|
||||
def test_save_then_load_round_trips(self):
|
||||
registry = self.parse(_document())
|
||||
save_registry(registry, self.path)
|
||||
reloaded = load_registry(self.path)
|
||||
self.assertEqual(
|
||||
[worker_to_dict(w) for w in reloaded.workers],
|
||||
[worker_to_dict(w) for w in registry.workers],
|
||||
)
|
||||
|
||||
def test_save_leaves_no_temp_files_behind(self):
|
||||
registry = self.parse(_document())
|
||||
save_registry(registry, self.path)
|
||||
save_registry(registry, self.path)
|
||||
leftovers = [p.name for p in self.path.parent.iterdir() if p.name.startswith(".")]
|
||||
self.assertEqual(leftovers, [])
|
||||
|
||||
def test_save_refuses_to_persist_an_invalid_document(self):
|
||||
registry = self.parse(_document())
|
||||
broken = WorkerRegistry(
|
||||
version=registry.version,
|
||||
revision=registry.revision,
|
||||
updated_at=registry.updated_at,
|
||||
providers=registry.providers,
|
||||
# A worker whose provider is not declared in the registry.
|
||||
workers=tuple(
|
||||
type(worker)(**{**worker.__dict__, "provider": "vanished"})
|
||||
for worker in registry.workers
|
||||
),
|
||||
source_path=self.path,
|
||||
)
|
||||
with self.assertRaises(RegistryValidationError):
|
||||
save_registry(broken, self.path)
|
||||
self.assertFalse(self.path.exists(), "invalid save must not create the file")
|
||||
|
||||
def test_document_shape_excludes_local_paths_but_api_shape_includes_it(self):
|
||||
registry = self.parse(_document())
|
||||
self.assertNotIn("source_path", registry_to_document(registry))
|
||||
self.assertEqual(registry_to_dict(registry)["source_path"], str(self.path))
|
||||
|
||||
|
||||
class TestVersioningAndRollback(_TempRegistryCase):
|
||||
"""AC: versioning and rollback."""
|
||||
|
||||
def _seed(self) -> WorkerRegistry:
|
||||
self.write(_document())
|
||||
return load_registry(self.path)
|
||||
|
||||
def test_revision_increments_on_each_save(self):
|
||||
registry = self._seed()
|
||||
self.assertEqual(registry.revision, 1)
|
||||
second = save_registry(registry, self.path)
|
||||
self.assertEqual(second.revision, 2)
|
||||
third = save_registry(second, self.path)
|
||||
self.assertEqual(third.revision, 3)
|
||||
|
||||
def test_updated_at_is_refreshed_and_utc(self):
|
||||
registry = self._seed()
|
||||
saved = save_registry(registry, self.path)
|
||||
self.assertRegex(saved.updated_at, r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$")
|
||||
|
||||
def test_superseded_revisions_are_retained(self):
|
||||
registry = self._seed()
|
||||
second = save_registry(registry, self.path)
|
||||
save_registry(second, self.path)
|
||||
self.assertEqual(list_revisions(self.path), (1, 2))
|
||||
self.assertTrue(history_dir(self.path).is_dir())
|
||||
|
||||
def test_rollback_restores_prior_content_as_a_new_revision(self):
|
||||
self.write(_document(workers=[_worker("original")]))
|
||||
registry = load_registry(self.path)
|
||||
|
||||
changed = WorkerRegistry(
|
||||
version=registry.version,
|
||||
revision=registry.revision,
|
||||
updated_at=registry.updated_at,
|
||||
providers=registry.providers,
|
||||
workers=(), # operator deletes every worker
|
||||
source_path=self.path,
|
||||
)
|
||||
save_registry(changed, self.path)
|
||||
self.assertEqual(load_registry(self.path).workers, ())
|
||||
|
||||
restored = rollback_to_revision(1, self.path)
|
||||
self.assertEqual([w.id for w in restored.workers], ["original"])
|
||||
# Append-only: the rollback publishes a new head rather than rewinding.
|
||||
self.assertGreater(restored.revision, 2)
|
||||
self.assertEqual([w.id for w in load_registry(self.path).workers], ["original"])
|
||||
|
||||
def test_rollback_to_unknown_revision_fails_closed(self):
|
||||
self._seed()
|
||||
with self.assertRaises(RegistryValidationError) as ctx:
|
||||
rollback_to_revision(99, self.path)
|
||||
self.assertIn("not retained", str(ctx.exception))
|
||||
|
||||
def test_revision_must_be_a_positive_integer(self):
|
||||
for bad in (0, -1, "1", None):
|
||||
with self.subTest(revision=bad):
|
||||
with self.assertRaises(RegistryValidationError):
|
||||
self.parse(_document(revision=bad))
|
||||
|
||||
def test_history_is_empty_before_any_save(self):
|
||||
self.write(_document())
|
||||
self.assertEqual(list_revisions(self.path), ())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -29,6 +29,12 @@ from webui.worktree_scanner import load_hygiene_snapshot, snapshot_to_dict as wo
|
||||
from webui.worktree_views import render_worktrees_page
|
||||
from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
|
||||
from webui.runtime_views import render_runtime_page
|
||||
from webui.system_health import (
|
||||
API_PATH as SYSTEM_HEALTH_API_PATH,
|
||||
load_system_health,
|
||||
process_uptime,
|
||||
snapshot_to_dict as system_health_to_dict,
|
||||
)
|
||||
|
||||
_READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
|
||||
_AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
|
||||
@@ -62,16 +68,43 @@ async def home(_request: Request) -> HTMLResponse:
|
||||
|
||||
|
||||
async def health(_request: Request) -> JSONResponse:
|
||||
"""Liveness only — deliberately cheap, runs no dependency probe (#634).
|
||||
|
||||
Every MVP key is retained so existing pollers keep working; the additions
|
||||
are a pointer to the structured API and the in-memory process uptime.
|
||||
Readiness lives at that API because answering it costs real probes.
|
||||
"""
|
||||
bind_host = _request.app.state.webui_bind_host
|
||||
started_at, uptime_seconds = process_uptime()
|
||||
return JSONResponse({
|
||||
"status": "ok",
|
||||
"service": "mcp-control-plane-webui",
|
||||
"mode": "read-only-mvp",
|
||||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||||
"deployment": deployment_snapshot(bind_host=bind_host),
|
||||
"started_at": started_at,
|
||||
"uptime_seconds": uptime_seconds,
|
||||
"system_health_api": SYSTEM_HEALTH_API_PATH,
|
||||
})
|
||||
|
||||
|
||||
def _truthy_flag(value: str | None) -> bool:
|
||||
return (value or "").strip().lower() in {"1", "true", "yes", "on"}
|
||||
|
||||
|
||||
async def api_system_health(request: Request) -> JSONResponse:
|
||||
"""Structured read-only system health (#634).
|
||||
|
||||
`?deep=1` opts into the expensive network probe. The response status code
|
||||
reflects readiness so automated checks can branch on it without parsing the
|
||||
body: 200 when ready, 503 when a required dependency failed or never ran.
|
||||
"""
|
||||
deep = _truthy_flag(request.query_params.get("deep"))
|
||||
snapshot = load_system_health(deep=deep)
|
||||
payload = system_health_to_dict(snapshot)
|
||||
return JSONResponse(payload, status_code=200 if snapshot.ready else 503)
|
||||
|
||||
|
||||
async def queue(_request: Request) -> HTMLResponse:
|
||||
snapshot = load_queue_snapshot()
|
||||
return HTMLResponse(render_page(title="Queue", body_html=render_queue_page(snapshot)))
|
||||
@@ -263,6 +296,7 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
|
||||
routes=[
|
||||
Route("/", home, methods=["GET"]),
|
||||
Route("/health", health, methods=["GET"]),
|
||||
Route(SYSTEM_HEALTH_API_PATH, api_system_health, methods=["GET"]),
|
||||
Route("/queue", queue, methods=["GET"]),
|
||||
Route("/api/queue", api_queue, methods=["GET"]),
|
||||
Route("/projects", projects, methods=["GET"]),
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
{
|
||||
"version": 1,
|
||||
"revision": 1,
|
||||
"updated_at": "2026-07-22T00:00:00Z",
|
||||
"providers": [
|
||||
{
|
||||
"id": "claude",
|
||||
"display_name": "Claude",
|
||||
"vendor": "Anthropic",
|
||||
"executable": "claude",
|
||||
"available": true,
|
||||
"models": [
|
||||
"claude-opus-4-8",
|
||||
"claude-sonnet-5",
|
||||
"claude-haiku-4-5-20251001"
|
||||
],
|
||||
"notes": "Model list is a declaration. Live enumeration and version inspection belong to the provider adapter framework (#800)."
|
||||
},
|
||||
{
|
||||
"id": "grok",
|
||||
"display_name": "Grok",
|
||||
"vendor": "xAI",
|
||||
"executable": "grok",
|
||||
"available": true,
|
||||
"models": [],
|
||||
"notes": "Models enumerated by the provider adapter (#800); not declared here."
|
||||
},
|
||||
{
|
||||
"id": "codex",
|
||||
"display_name": "Codex",
|
||||
"vendor": "OpenAI",
|
||||
"executable": "codex",
|
||||
"available": true,
|
||||
"models": [],
|
||||
"notes": "Models enumerated by the provider adapter (#800); not declared here."
|
||||
},
|
||||
{
|
||||
"id": "agy",
|
||||
"display_name": "AGY",
|
||||
"vendor": "Antigravity",
|
||||
"executable": "agy",
|
||||
"available": true,
|
||||
"models": [],
|
||||
"notes": "MCP allowlist gating applies to this provider; confirm server-side allowlist before configuring a worker."
|
||||
},
|
||||
{
|
||||
"id": "kimi-k",
|
||||
"display_name": "Kimi K",
|
||||
"vendor": "Moonshot AI",
|
||||
"executable": "kimi",
|
||||
"available": true,
|
||||
"models": [],
|
||||
"notes": "Provider id is kimi-k; the executable on PATH is kimi. Models enumerated by the provider adapter (#800)."
|
||||
}
|
||||
],
|
||||
"workers": []
|
||||
}
|
||||
@@ -8,27 +8,7 @@ from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
_FORBIDDEN_EXACT_KEYS = frozenset({
|
||||
"token",
|
||||
"password",
|
||||
"secret",
|
||||
"credential",
|
||||
"auth",
|
||||
"api_key",
|
||||
"api-key",
|
||||
})
|
||||
_FORBIDDEN_KEY_PREFIXES = ("auth_", "api_key_", "api-key_")
|
||||
_FORBIDDEN_KEY_SUFFIXES = ("_token", "_secret", "_password", "_credential", "_auth")
|
||||
|
||||
|
||||
def _is_forbidden_key(key: str) -> bool:
|
||||
lowered = key.lower()
|
||||
if lowered in _FORBIDDEN_EXACT_KEYS:
|
||||
return True
|
||||
return (
|
||||
lowered.startswith(_FORBIDDEN_KEY_PREFIXES)
|
||||
or lowered.endswith(_FORBIDDEN_KEY_SUFFIXES)
|
||||
)
|
||||
from webui.registry_safety import reject_credential_keys as _reject_credential_keys
|
||||
|
||||
_REQUIRED_PROJECT_FIELDS = (
|
||||
"id",
|
||||
@@ -79,18 +59,6 @@ def default_registry_path() -> Path:
|
||||
return (Path(__file__).resolve().parent / "data" / "projects.registry.json").resolve()
|
||||
|
||||
|
||||
def _reject_credential_keys(obj: Any, *, path: str = "") -> None:
|
||||
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 ValueError(f"registry must not store credentials ({key_path})")
|
||||
_reject_credential_keys(value, path=key_path)
|
||||
elif isinstance(obj, list):
|
||||
for index, item in enumerate(obj):
|
||||
_reject_credential_keys(item, path=f"{path}[{index}]")
|
||||
|
||||
|
||||
def _parse_onboarding(raw: list[dict[str, Any]] | None) -> tuple[OnboardingStep, ...]:
|
||||
if not raw:
|
||||
return ()
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
"""Shared credential-rejection guard for web UI registries (#427, #798).
|
||||
|
||||
Registries are operator-editable declarative files that the web UI loads and,
|
||||
for the worker registry, writes back. None of them may ever carry a secret:
|
||||
credentials belong in the keychain and reach worker processes through
|
||||
environment injection, never through a file the browser layer can read.
|
||||
|
||||
The check is structural rather than value-based on purpose. A value scanner has
|
||||
to guess what a secret looks like; a key scanner refuses the *shape* of a
|
||||
credential field, so an operator cannot introduce one by accident and a later
|
||||
loader cannot silently pass one through.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
_FORBIDDEN_EXACT_KEYS = frozenset({
|
||||
"token",
|
||||
"password",
|
||||
"secret",
|
||||
"credential",
|
||||
"auth",
|
||||
"api_key",
|
||||
"api-key",
|
||||
})
|
||||
_FORBIDDEN_KEY_PREFIXES = ("auth_", "api_key_", "api-key_")
|
||||
_FORBIDDEN_KEY_SUFFIXES = ("_token", "_secret", "_password", "_credential", "_auth")
|
||||
|
||||
|
||||
def is_forbidden_key(key: str) -> bool:
|
||||
"""Return True when *key* names a credential field."""
|
||||
lowered = key.lower()
|
||||
if lowered in _FORBIDDEN_EXACT_KEYS:
|
||||
return True
|
||||
return (
|
||||
lowered.startswith(_FORBIDDEN_KEY_PREFIXES)
|
||||
or lowered.endswith(_FORBIDDEN_KEY_SUFFIXES)
|
||||
)
|
||||
|
||||
|
||||
def reject_credential_keys(obj: Any, *, path: str = "", subject: str = "registry") -> None:
|
||||
"""Raise ValueError when *obj* carries a credential-shaped key at any depth."""
|
||||
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 ValueError(f"{subject} must not store credentials ({key_path})")
|
||||
reject_credential_keys(value, path=key_path, subject=subject)
|
||||
elif isinstance(obj, list):
|
||||
for index, item in enumerate(obj):
|
||||
reject_credential_keys(item, path=f"{path}[{index}]", subject=subject)
|
||||
@@ -0,0 +1,682 @@
|
||||
"""Read-only system-health model for the operator console API (#634).
|
||||
|
||||
`/health` answers liveness only. Operators automating readiness checks need a
|
||||
structured view of *why* the control plane is or is not usable: which
|
||||
dependencies answered, how long they took, what version of the code is running,
|
||||
and whether the runtime is stale relative to its remote.
|
||||
|
||||
Three rules shape this module.
|
||||
|
||||
* **Read-only.** Every probe opens its subject read-only. The control-plane
|
||||
database is opened through a ``mode=ro`` URI so a health check can never
|
||||
create or migrate a schema, and no probe writes, restarts, or reloads
|
||||
anything — restart controls are Phase 2, and #630 forbids process-kill
|
||||
recovery outright.
|
||||
* **Fail-soft.** A dependency that is unreachable is a *status*, not an
|
||||
exception. Probes catch their own failures and report them as a degraded or
|
||||
down entry carrying a reason.
|
||||
* **Never claim more than was proven.** Readiness is derived only from probes
|
||||
that actually ran, ``mutation_safe`` stays false unless the parity commits are
|
||||
known and equal, and an MCP namespace is reported unproven because a web
|
||||
process cannot exercise the IDE-managed client path (#543).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable
|
||||
from urllib.parse import urlsplit, urlunsplit
|
||||
|
||||
import control_plane_db
|
||||
import mcp_namespace_health
|
||||
from gitea_auth import api_request, get_auth_header, gitea_url
|
||||
|
||||
from webui.project_registry import load_registry
|
||||
|
||||
SERVICE_NAME = "mcp-control-plane-webui"
|
||||
API_PATH = "/api/v1/system/health"
|
||||
|
||||
STATUS_OK = "ok"
|
||||
STATUS_DEGRADED = "degraded"
|
||||
STATUS_DOWN = "down"
|
||||
STATUS_SKIPPED = "skipped"
|
||||
STATUS_UNPROVEN = "unproven"
|
||||
|
||||
# Statuses that count as a healthy answer from a probe.
|
||||
_HEALTHY_STATUSES = frozenset({STATUS_OK})
|
||||
# Statuses meaning "this probe did not run", as opposed to "it ran and failed".
|
||||
_NOT_RUN_STATUSES = frozenset({STATUS_SKIPPED})
|
||||
|
||||
_DEEP_PROBE_TTL_ENV = "WEBUI_HEALTH_PROBE_TTL_SECONDS"
|
||||
_DEFAULT_DEEP_PROBE_TTL = 15.0
|
||||
_GITEA_PROBE_TIMEOUT_SECONDS = 5.0
|
||||
|
||||
_OFFLINE_ENV = "WEBUI_TEST_OFFLINE"
|
||||
|
||||
# Credential-shaped material that must never reach the browser, mirroring the
|
||||
# forbidden client patterns in webui/deployment_boundary.py.
|
||||
_SECRET_RE = re.compile(
|
||||
r"(?i)\b(token|password|passwd|secret|authorization|bearer)\b\s*[:=]?\s*\S+"
|
||||
)
|
||||
_LONG_OPAQUE_RE = re.compile(r"\b[A-Za-z0-9_\-]{32,}\b")
|
||||
|
||||
# Captured once at import so uptime measures this process, not the request.
|
||||
_STARTED_AT = datetime.now(timezone.utc)
|
||||
_STARTED_MONOTONIC = time.monotonic()
|
||||
|
||||
# TTL cache for the expensive (network) probe only.
|
||||
_deep_cache: dict[str, tuple[float, "DependencyProbe"]] = {}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DependencyProbe:
|
||||
"""One dependency check, fail-soft, with its own latency."""
|
||||
|
||||
name: str
|
||||
kind: str
|
||||
status: str
|
||||
detail: str
|
||||
required: bool
|
||||
latency_ms: float | None = None
|
||||
metadata: dict[str, Any] | None = None
|
||||
|
||||
@property
|
||||
def healthy(self) -> bool:
|
||||
return self.status in _HEALTHY_STATUSES
|
||||
|
||||
@property
|
||||
def ran(self) -> bool:
|
||||
return self.status not in _NOT_RUN_STATUSES
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class VersionInfo:
|
||||
git_sha: str | None
|
||||
git_describe: str | None
|
||||
control_plane_schema_version: int | None
|
||||
python_version: str
|
||||
known: bool
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class StaleRuntime:
|
||||
"""Parity between the running code, the checkout, and the remote.
|
||||
|
||||
``mutation_safe`` is deliberately conservative: unknown is not safe.
|
||||
"""
|
||||
|
||||
daemon_head: str | None
|
||||
checkout_head: str | None
|
||||
remote_head: str | None
|
||||
stale: bool
|
||||
determinable: bool
|
||||
mutation_safe: bool
|
||||
reasons: tuple[str, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SystemHealthSnapshot:
|
||||
status: str
|
||||
ready: bool
|
||||
readiness_complete: bool
|
||||
readiness_reasons: tuple[str, ...]
|
||||
service: str
|
||||
mode: str
|
||||
version: VersionInfo
|
||||
started_at: str
|
||||
uptime_seconds: float
|
||||
timestamp: str
|
||||
deep_probes_requested: bool
|
||||
dependencies: tuple[DependencyProbe, ...]
|
||||
mcp_namespaces: tuple[dict[str, Any], ...]
|
||||
stale_runtime: StaleRuntime
|
||||
probe_errors: tuple[str, ...] = ()
|
||||
|
||||
|
||||
def process_uptime() -> tuple[str, float]:
|
||||
"""Process start timestamp and uptime — in-memory, safe for `/health`."""
|
||||
return _STARTED_AT.isoformat(), round(time.monotonic() - _STARTED_MONOTONIC, 3)
|
||||
|
||||
|
||||
def _offline() -> bool:
|
||||
return (os.environ.get(_OFFLINE_ENV) or "").strip().lower() in {"1", "true", "yes"}
|
||||
|
||||
|
||||
def _repo_root() -> Path:
|
||||
override = (os.environ.get("WEBUI_REPO_ROOT") or "").strip()
|
||||
if override:
|
||||
return Path(override).resolve()
|
||||
return Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
def _deep_probe_ttl() -> float:
|
||||
raw = (os.environ.get(_DEEP_PROBE_TTL_ENV) or "").strip()
|
||||
if not raw:
|
||||
return _DEFAULT_DEEP_PROBE_TTL
|
||||
try:
|
||||
value = float(raw)
|
||||
except ValueError:
|
||||
return _DEFAULT_DEEP_PROBE_TTL
|
||||
return value if value >= 0 else _DEFAULT_DEEP_PROBE_TTL
|
||||
|
||||
|
||||
def redact(text: str) -> str:
|
||||
"""Strip credential-shaped material from operator-visible probe text.
|
||||
|
||||
Probe details carry exception strings, and an exception raised by an HTTP
|
||||
client can quote the request that failed. Redaction happens here, at the
|
||||
boundary where those strings become part of a browser-bound payload.
|
||||
"""
|
||||
if not text:
|
||||
return ""
|
||||
cleaned = _redact_urls(text)
|
||||
cleaned = _SECRET_RE.sub(lambda m: f"{m.group(1)}=[redacted]", cleaned)
|
||||
return _LONG_OPAQUE_RE.sub("[redacted]", cleaned)
|
||||
|
||||
|
||||
def _redact_urls(text: str) -> str:
|
||||
return re.sub(r"https?://\S+", lambda m: redact_url(m.group(0)), text)
|
||||
|
||||
|
||||
def redact_url(url: str) -> str:
|
||||
"""Reduce a URL to scheme://host/path — no userinfo, no query, no fragment."""
|
||||
try:
|
||||
parts = urlsplit(url)
|
||||
except ValueError:
|
||||
return "[redacted-url]"
|
||||
if not parts.scheme or not parts.hostname:
|
||||
return "[redacted-url]"
|
||||
netloc = parts.hostname
|
||||
if parts.port:
|
||||
netloc = f"{netloc}:{parts.port}"
|
||||
return urlunsplit((parts.scheme, netloc, parts.path, "", ""))
|
||||
|
||||
|
||||
def _git(repo: Path, *args: str) -> str | None:
|
||||
try:
|
||||
completed = subprocess.run(
|
||||
["git", "-C", str(repo), *args],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=10,
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
return None
|
||||
if completed.returncode != 0:
|
||||
return None
|
||||
return (completed.stdout or "").strip() or None
|
||||
|
||||
|
||||
def _load_version(repo: Path, *, schema_version: int | None) -> VersionInfo:
|
||||
import platform
|
||||
|
||||
git_sha = None if _offline() else _git(repo, "rev-parse", "HEAD")
|
||||
describe = None if _offline() else _git(repo, "describe", "--tags", "--always")
|
||||
return VersionInfo(
|
||||
git_sha=git_sha,
|
||||
git_describe=describe,
|
||||
control_plane_schema_version=schema_version,
|
||||
python_version=platform.python_version(),
|
||||
known=bool(git_sha),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dependency probes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _elapsed_ms(started: float) -> float:
|
||||
return round((time.monotonic() - started) * 1000, 3)
|
||||
|
||||
|
||||
def probe_control_plane_db(db_path: str | None = None) -> DependencyProbe:
|
||||
"""Read-only reachability check for the control-plane SQLite substrate.
|
||||
|
||||
Opened through a ``mode=ro`` URI on purpose: ``ControlPlaneDB.__init__``
|
||||
creates directories and runs schema migrations, which a health check must
|
||||
never do.
|
||||
"""
|
||||
path = (db_path or control_plane_db.default_db_path()).strip()
|
||||
started = time.monotonic()
|
||||
metadata: dict[str, Any] = {"path": path}
|
||||
|
||||
def _result(status: str, detail: str) -> DependencyProbe:
|
||||
return DependencyProbe(
|
||||
name="control_plane_db",
|
||||
kind="sqlite",
|
||||
status=status,
|
||||
detail=detail,
|
||||
required=True,
|
||||
latency_ms=_elapsed_ms(started),
|
||||
metadata=metadata,
|
||||
)
|
||||
|
||||
if not path or not os.path.exists(path):
|
||||
return _result(STATUS_DOWN, "control-plane database file does not exist yet")
|
||||
try:
|
||||
conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=5)
|
||||
try:
|
||||
row = conn.execute(
|
||||
"SELECT value FROM schema_meta WHERE key = 'schema_version'"
|
||||
).fetchone()
|
||||
leases = conn.execute(
|
||||
"SELECT COUNT(*) FROM leases WHERE status = 'active'"
|
||||
).fetchone()
|
||||
finally:
|
||||
conn.close()
|
||||
except sqlite3.Error as exc:
|
||||
return _result(STATUS_DOWN, redact(f"control-plane database unreadable: {exc}"))
|
||||
|
||||
schema_version = int(row[0]) if row and str(row[0]).isdigit() else None
|
||||
metadata["schema_version"] = schema_version
|
||||
metadata["active_leases"] = int(leases[0]) if leases else None
|
||||
if schema_version is None:
|
||||
return _result(
|
||||
STATUS_DEGRADED, "control-plane database has no recorded schema version"
|
||||
)
|
||||
if schema_version != control_plane_db.SCHEMA_VERSION:
|
||||
return _result(
|
||||
STATUS_DEGRADED,
|
||||
f"control-plane schema version {schema_version} does not match the "
|
||||
f"version this code expects ({control_plane_db.SCHEMA_VERSION})",
|
||||
)
|
||||
return _result(STATUS_OK, f"schema v{schema_version} readable")
|
||||
|
||||
|
||||
def probe_repository(repo: Path) -> DependencyProbe:
|
||||
"""Local checkout reachability — required, cheap, no network."""
|
||||
started = time.monotonic()
|
||||
metadata: dict[str, Any] = {"repo_root": str(repo)}
|
||||
if _offline():
|
||||
return DependencyProbe(
|
||||
name="repository",
|
||||
kind="git",
|
||||
status=STATUS_SKIPPED,
|
||||
detail=f"{_OFFLINE_ENV} is set; git probe skipped",
|
||||
required=True,
|
||||
latency_ms=_elapsed_ms(started),
|
||||
metadata=metadata,
|
||||
)
|
||||
head = _git(repo, "rev-parse", "HEAD")
|
||||
if not head:
|
||||
return DependencyProbe(
|
||||
name="repository",
|
||||
kind="git",
|
||||
status=STATUS_DOWN,
|
||||
detail=f"HEAD could not be read at {repo}",
|
||||
required=True,
|
||||
latency_ms=_elapsed_ms(started),
|
||||
metadata=metadata,
|
||||
)
|
||||
branch = _git(repo, "rev-parse", "--abbrev-ref", "HEAD")
|
||||
metadata["head"] = head
|
||||
metadata["branch"] = branch
|
||||
return DependencyProbe(
|
||||
name="repository",
|
||||
kind="git",
|
||||
status=STATUS_OK,
|
||||
detail=f"checkout readable at {branch or 'detached HEAD'}",
|
||||
required=True,
|
||||
latency_ms=_elapsed_ms(started),
|
||||
metadata=metadata,
|
||||
)
|
||||
|
||||
|
||||
def probe_gitea(host: str) -> DependencyProbe:
|
||||
"""Live Gitea reachability. Expensive (network), so opt-in via ``deep``.
|
||||
|
||||
Optional by design: the console stays useful for local inventory when the
|
||||
remote is unreachable, so a failure here degrades status without claiming
|
||||
the process itself is unready.
|
||||
"""
|
||||
started = time.monotonic()
|
||||
metadata: dict[str, Any] = {"host": host}
|
||||
|
||||
def _failure(status: str, detail: str) -> DependencyProbe:
|
||||
return DependencyProbe(
|
||||
name="gitea",
|
||||
kind="http",
|
||||
status=status,
|
||||
detail=detail,
|
||||
required=False,
|
||||
latency_ms=_elapsed_ms(started),
|
||||
metadata=metadata,
|
||||
)
|
||||
|
||||
if not host:
|
||||
return _failure(STATUS_DEGRADED, "no Gitea host is configured in the registry")
|
||||
try:
|
||||
auth = get_auth_header(host)
|
||||
except Exception as exc: # noqa: BLE001 — credential guards are a status here
|
||||
return _failure(STATUS_DEGRADED, redact(f"credential lookup refused: {exc}"))
|
||||
if not auth:
|
||||
return _failure(STATUS_DEGRADED, f"no credentials available for {host}")
|
||||
|
||||
url = gitea_url(host, "/api/v1/version")
|
||||
metadata["endpoint"] = redact_url(url)
|
||||
try:
|
||||
data = api_request("GET", url, auth, timeout=_GITEA_PROBE_TIMEOUT_SECONDS)
|
||||
except Exception as exc: # noqa: BLE001 — a down dependency is a status
|
||||
return _failure(STATUS_DOWN, redact(f"Gitea probe failed: {exc}"))
|
||||
|
||||
if isinstance(data, dict) and data.get("version"):
|
||||
metadata["gitea_version"] = str(data["version"])
|
||||
return DependencyProbe(
|
||||
name="gitea",
|
||||
kind="http",
|
||||
status=STATUS_OK,
|
||||
detail=f"{host} reachable",
|
||||
required=False,
|
||||
latency_ms=_elapsed_ms(started),
|
||||
metadata=metadata,
|
||||
)
|
||||
|
||||
|
||||
def _skipped_gitea(host: str) -> DependencyProbe:
|
||||
return DependencyProbe(
|
||||
name="gitea",
|
||||
kind="http",
|
||||
status=STATUS_SKIPPED,
|
||||
detail="network probe not requested; call with ?deep=1 to run it",
|
||||
required=False,
|
||||
latency_ms=None,
|
||||
metadata={"host": host},
|
||||
)
|
||||
|
||||
|
||||
def namespace_summaries() -> tuple[dict[str, Any], ...]:
|
||||
"""Declared MCP namespaces, each honestly reported as unproven.
|
||||
|
||||
The web process runs outside the IDE-managed MCP client, so it cannot
|
||||
invoke a namespace tool. Per #543 only a ``client_namespace`` probe proves
|
||||
that path, and inventing a healthy verdict here is exactly the false claim
|
||||
the mutation gates exist to prevent.
|
||||
"""
|
||||
rows: list[dict[str, Any]] = []
|
||||
for namespace, required_tool in sorted(
|
||||
mcp_namespace_health.REQUIRED_NAMESPACE_TOOLS.items()
|
||||
):
|
||||
classification = mcp_namespace_health.classify_namespace_probe(
|
||||
namespace,
|
||||
required_tool=required_tool,
|
||||
probe_result=None,
|
||||
probe_source=mcp_namespace_health.PROBE_SOURCE_UNKNOWN,
|
||||
)
|
||||
rows.append(
|
||||
{
|
||||
"namespace": namespace,
|
||||
"required_tool": required_tool,
|
||||
"status": STATUS_UNPROVEN,
|
||||
"ide_namespace_proven": bool(classification.get("ide_namespace_proven")),
|
||||
"reason": (
|
||||
"the web console cannot invoke the IDE-managed MCP client; "
|
||||
"namespace health must be proven with a client_namespace "
|
||||
"probe (#543)"
|
||||
),
|
||||
"error_type": classification.get("error_type"),
|
||||
}
|
||||
)
|
||||
return tuple(rows)
|
||||
|
||||
|
||||
def assess_stale_runtime(
|
||||
repo: Path,
|
||||
*,
|
||||
daemon_head: str | None = None,
|
||||
git_reader: Callable[..., str | None] | None = None,
|
||||
) -> StaleRuntime:
|
||||
"""Three-way parity view: running code, local checkout, remote-tracking ref.
|
||||
|
||||
``mutation_safe`` requires all three to be known and equal. Anything less —
|
||||
including "the remote ref was never fetched" — is reported as not safe with
|
||||
a reason, so an operator never reads an unproven green.
|
||||
"""
|
||||
reader = git_reader or (lambda *args: _git(repo, *args))
|
||||
reasons: list[str] = []
|
||||
# The offline switch suppresses real subprocess calls; an explicitly
|
||||
# injected reader is already a substitute for them and is always used.
|
||||
offline = _offline() and git_reader is None
|
||||
|
||||
checkout_head = None if offline else reader("rev-parse", "HEAD")
|
||||
remote_head = None if offline else reader("rev-parse", "@{upstream}")
|
||||
if offline:
|
||||
reasons.append(f"{_OFFLINE_ENV} is set; parity commits were not read")
|
||||
else:
|
||||
if checkout_head is None:
|
||||
reasons.append("local checkout HEAD could not be read")
|
||||
if remote_head is None:
|
||||
reasons.append(
|
||||
"no remote-tracking commit is known for the current branch; "
|
||||
"remote staleness is indeterminate (no fetch is performed here)"
|
||||
)
|
||||
|
||||
effective_daemon = daemon_head if daemon_head is not None else checkout_head
|
||||
if daemon_head is None:
|
||||
reasons.append(
|
||||
"the running MCP daemon's startup commit is not observable from the "
|
||||
"web process; the checkout commit is reported in its place"
|
||||
)
|
||||
|
||||
determinable = bool(checkout_head and remote_head and effective_daemon)
|
||||
stale = bool(
|
||||
determinable and len({checkout_head, remote_head, effective_daemon}) > 1
|
||||
)
|
||||
if stale:
|
||||
reasons.append(
|
||||
"runtime, checkout, and remote commits disagree; restart the MCP "
|
||||
"server after updating the checkout before trusting capability gates"
|
||||
)
|
||||
|
||||
return StaleRuntime(
|
||||
daemon_head=effective_daemon,
|
||||
checkout_head=checkout_head,
|
||||
remote_head=remote_head,
|
||||
stale=stale,
|
||||
determinable=determinable,
|
||||
mutation_safe=bool(determinable and not stale),
|
||||
reasons=tuple(reasons),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Snapshot assembly
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _default_host() -> str:
|
||||
registry = load_registry()
|
||||
if not registry.projects:
|
||||
return ""
|
||||
raw = registry.projects[0].remote_host
|
||||
parts = urlsplit(raw.strip())
|
||||
return parts.netloc or raw.strip().rstrip("/")
|
||||
|
||||
|
||||
def _aggregate(
|
||||
probes: tuple[DependencyProbe, ...],
|
||||
) -> tuple[str, bool, bool, tuple[str, ...]]:
|
||||
"""Fold probe results into overall status and readiness.
|
||||
|
||||
Required probes drive readiness; optional probes can only degrade status.
|
||||
A probe that did not run leaves readiness incomplete rather than passing.
|
||||
"""
|
||||
reasons: list[str] = []
|
||||
required = [probe for probe in probes if probe.required]
|
||||
unrun_required = [probe for probe in required if not probe.ran]
|
||||
failed_required = [probe for probe in required if probe.ran and not probe.healthy]
|
||||
failed_optional = [
|
||||
probe
|
||||
for probe in probes
|
||||
if not probe.required and probe.ran and not probe.healthy
|
||||
]
|
||||
|
||||
for probe in unrun_required:
|
||||
reasons.append(
|
||||
f"required dependency '{probe.name}' was not probed: {probe.detail}"
|
||||
)
|
||||
for probe in failed_required:
|
||||
reasons.append(
|
||||
f"required dependency '{probe.name}' is {probe.status}: {probe.detail}"
|
||||
)
|
||||
for probe in failed_optional:
|
||||
reasons.append(
|
||||
f"optional dependency '{probe.name}' is {probe.status}: {probe.detail}"
|
||||
)
|
||||
|
||||
readiness_complete = not unrun_required
|
||||
ready = readiness_complete and not failed_required
|
||||
|
||||
if any(probe.status == STATUS_DOWN for probe in failed_required):
|
||||
status = STATUS_DOWN
|
||||
elif failed_required or failed_optional or unrun_required:
|
||||
status = STATUS_DEGRADED
|
||||
else:
|
||||
status = STATUS_OK
|
||||
return status, ready, readiness_complete, tuple(reasons)
|
||||
|
||||
|
||||
def load_system_health(
|
||||
*,
|
||||
deep: bool = False,
|
||||
host: str | None = None,
|
||||
probes: tuple[DependencyProbe, ...] | None = None,
|
||||
daemon_head: str | None = None,
|
||||
use_cache: bool = True,
|
||||
) -> SystemHealthSnapshot:
|
||||
"""Assemble the read-only system-health snapshot.
|
||||
|
||||
``deep=True`` adds the network probe against Gitea; its result is cached for
|
||||
a short TTL so repeated dashboard polls do not amplify into remote load.
|
||||
"""
|
||||
repo = _repo_root()
|
||||
probe_errors: list[str] = []
|
||||
|
||||
if probes is None:
|
||||
collected: list[DependencyProbe] = []
|
||||
for probe_fn in (
|
||||
lambda: probe_control_plane_db(),
|
||||
lambda: probe_repository(repo),
|
||||
):
|
||||
try:
|
||||
collected.append(probe_fn())
|
||||
except Exception as exc: # noqa: BLE001 — a probe must not 500 the API
|
||||
probe_errors.append(redact(f"probe raised: {exc}"))
|
||||
resolved_host = host if host is not None else _default_host()
|
||||
if deep and not _offline():
|
||||
collected.append(_cached_gitea_probe(resolved_host, use_cache=use_cache))
|
||||
else:
|
||||
collected.append(_skipped_gitea(resolved_host))
|
||||
probes = tuple(collected)
|
||||
|
||||
status, ready, readiness_complete, reasons = _aggregate(probes)
|
||||
stale = assess_stale_runtime(repo, daemon_head=daemon_head)
|
||||
if stale.stale:
|
||||
if status == STATUS_OK:
|
||||
status = STATUS_DEGRADED
|
||||
reasons = reasons + (
|
||||
"runtime is stale relative to its remote-tracking commit",
|
||||
)
|
||||
|
||||
db_probe = next((p for p in probes if p.name == "control_plane_db"), None)
|
||||
schema_version = None
|
||||
if db_probe and db_probe.metadata:
|
||||
schema_version = db_probe.metadata.get("schema_version")
|
||||
|
||||
return SystemHealthSnapshot(
|
||||
status=status,
|
||||
ready=ready,
|
||||
readiness_complete=readiness_complete,
|
||||
readiness_reasons=reasons,
|
||||
service=SERVICE_NAME,
|
||||
mode="read-only",
|
||||
version=_load_version(repo, schema_version=schema_version),
|
||||
started_at=_STARTED_AT.isoformat(),
|
||||
uptime_seconds=round(time.monotonic() - _STARTED_MONOTONIC, 3),
|
||||
timestamp=datetime.now(timezone.utc).isoformat(),
|
||||
deep_probes_requested=deep,
|
||||
dependencies=probes,
|
||||
mcp_namespaces=namespace_summaries(),
|
||||
stale_runtime=stale,
|
||||
probe_errors=tuple(probe_errors),
|
||||
)
|
||||
|
||||
|
||||
def _cached_gitea_probe(host: str, *, use_cache: bool = True) -> DependencyProbe:
|
||||
ttl = _deep_probe_ttl()
|
||||
now = time.monotonic()
|
||||
if use_cache and ttl > 0:
|
||||
cached = _deep_cache.get(host)
|
||||
if cached and (now - cached[0]) < ttl:
|
||||
return cached[1]
|
||||
probe = probe_gitea(host)
|
||||
if use_cache and ttl > 0:
|
||||
_deep_cache[host] = (now, probe)
|
||||
return probe
|
||||
|
||||
|
||||
def clear_probe_cache() -> None:
|
||||
"""Drop cached deep-probe results (tests and operator-forced refresh)."""
|
||||
_deep_cache.clear()
|
||||
|
||||
|
||||
def probe_to_dict(probe: DependencyProbe) -> dict[str, Any]:
|
||||
return {
|
||||
"name": probe.name,
|
||||
"kind": probe.kind,
|
||||
"status": probe.status,
|
||||
"detail": probe.detail,
|
||||
"required": probe.required,
|
||||
"healthy": probe.healthy,
|
||||
"latency_ms": probe.latency_ms,
|
||||
"metadata": dict(probe.metadata or {}),
|
||||
}
|
||||
|
||||
|
||||
def snapshot_to_dict(snapshot: SystemHealthSnapshot) -> dict[str, Any]:
|
||||
return {
|
||||
"status": snapshot.status,
|
||||
"service": snapshot.service,
|
||||
"mode": snapshot.mode,
|
||||
"api": API_PATH,
|
||||
"timestamp": snapshot.timestamp,
|
||||
"readiness": {
|
||||
"ready": snapshot.ready,
|
||||
"complete": snapshot.readiness_complete,
|
||||
"reasons": list(snapshot.readiness_reasons),
|
||||
},
|
||||
"version": {
|
||||
"git_sha": snapshot.version.git_sha,
|
||||
"git_describe": snapshot.version.git_describe,
|
||||
"control_plane_schema_version": (
|
||||
snapshot.version.control_plane_schema_version
|
||||
),
|
||||
"python_version": snapshot.version.python_version,
|
||||
"known": snapshot.version.known,
|
||||
},
|
||||
"process": {
|
||||
"started_at": snapshot.started_at,
|
||||
"uptime_seconds": snapshot.uptime_seconds,
|
||||
},
|
||||
"deep_probes_requested": snapshot.deep_probes_requested,
|
||||
"dependencies": [probe_to_dict(probe) for probe in snapshot.dependencies],
|
||||
"mcp_namespaces": [dict(row) for row in snapshot.mcp_namespaces],
|
||||
"stale_runtime": {
|
||||
"daemon_head": snapshot.stale_runtime.daemon_head,
|
||||
"checkout_head": snapshot.stale_runtime.checkout_head,
|
||||
"remote_head": snapshot.stale_runtime.remote_head,
|
||||
"stale": snapshot.stale_runtime.stale,
|
||||
"determinable": snapshot.stale_runtime.determinable,
|
||||
"mutation_safe": snapshot.stale_runtime.mutation_safe,
|
||||
"reasons": list(snapshot.stale_runtime.reasons),
|
||||
},
|
||||
"probe_errors": list(snapshot.probe_errors),
|
||||
}
|
||||
@@ -0,0 +1,647 @@
|
||||
"""Declarative worker registry and configuration schema (#798, epic #797).
|
||||
|
||||
The registry is the single source of truth for the scheduled multi-LLM worker
|
||||
fleet. It is a versioned JSON document holding two *separate* entity kinds:
|
||||
|
||||
* **Providers** — the LLM runtimes a worker can be built on (Claude, Grok,
|
||||
Codex, AGY, Kimi K). A provider describes the runtime itself: vendor,
|
||||
executable name, models it can serve, and whether it is available on this
|
||||
machine. Providers exist whether or not any worker uses them.
|
||||
* **Workers** — a configured *instance*: one provider, one model, one project,
|
||||
one role, one MCP namespace/profile, one workflow, one schedule. Several
|
||||
workers may share a provider; a worker naming an undeclared provider is
|
||||
refused.
|
||||
|
||||
Keeping them separate is what lets #799 list all five providers even when a
|
||||
provider currently has no configured worker, and it stops provider facts from
|
||||
being copied into (and drifting across) every worker record.
|
||||
|
||||
Scope boundary. This module owns the data model, its validation, and its
|
||||
persistence. It does **not** schedule anything, launch anything, probe provider
|
||||
executables, or serve HTTP. Loading a registry never touches a process; the
|
||||
live fields a dashboard wants (PID, elapsed time, next run) are derived
|
||||
elsewhere (#799, #801, #803, #804) from these declarations.
|
||||
|
||||
Safety invariants:
|
||||
|
||||
* No credential may be stored (:mod:`webui.registry_safety`), so the registry
|
||||
stays safe to render and to hand to a browser layer.
|
||||
* Validation fails closed. Unknown fields are refused rather than ignored, so a
|
||||
typo cannot silently disable a timeout or a role binding.
|
||||
* Writes are atomic and every superseded document is retained as a numbered
|
||||
revision, so a bad edit is recoverable by rollback rather than hand-repair.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import tempfile
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from webui.registry_safety import reject_credential_keys
|
||||
|
||||
SCHEMA_VERSION = 1
|
||||
|
||||
#: Roles a worker may hold. These mirror the sanctioned MCP role kinds; a
|
||||
#: worker may not invent one, because the role selects the namespace/profile
|
||||
#: whose capability gates constrain it.
|
||||
ALLOWED_ROLES = ("author", "reviewer", "merger", "reconciler", "cleanup")
|
||||
|
||||
#: Scheduler backends the registry can describe. ``manual`` means the worker is
|
||||
#: only ever started on request and has no recurring trigger.
|
||||
ALLOWED_SCHEDULER_KINDS = ("launchd", "manual")
|
||||
|
||||
#: Schedule kinds. Next-run computation belongs to #803; this module only
|
||||
#: guarantees the declaration is well formed.
|
||||
ALLOWED_SCHEDULE_KINDS = ("interval", "cron", "manual")
|
||||
|
||||
_REQUIRED_PROVIDER_FIELDS = ("id", "display_name", "vendor", "executable", "available")
|
||||
_OPTIONAL_PROVIDER_FIELDS = ("models", "notes")
|
||||
|
||||
_REQUIRED_WORKER_FIELDS = (
|
||||
"id",
|
||||
"display_name",
|
||||
"provider",
|
||||
"model",
|
||||
"project",
|
||||
"role",
|
||||
"namespace",
|
||||
"profile",
|
||||
"workflow",
|
||||
"schedule",
|
||||
"timeout_seconds",
|
||||
"enabled",
|
||||
"scheduler",
|
||||
)
|
||||
_OPTIONAL_WORKER_FIELDS = ("notes",)
|
||||
|
||||
_ID_RE = re.compile(r"^[a-z0-9][a-z0-9._-]*$")
|
||||
|
||||
#: Guards against an operator writing a timeout that would let a worker hold a
|
||||
#: lease effectively forever. 24h is far above any sanctioned cycle.
|
||||
_MAX_TIMEOUT_SECONDS = 86_400
|
||||
|
||||
#: How many superseded revisions to retain beside the live file.
|
||||
_HISTORY_LIMIT = 20
|
||||
|
||||
_TOP_LEVEL_FIELDS = frozenset({"version", "revision", "updated_at", "providers", "workers"})
|
||||
|
||||
|
||||
class RegistryValidationError(ValueError):
|
||||
"""Raised when a registry document violates the schema."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProviderRecord:
|
||||
id: str
|
||||
display_name: str
|
||||
vendor: str
|
||||
executable: str
|
||||
available: bool
|
||||
models: tuple[str, ...]
|
||||
notes: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ScheduleSpec:
|
||||
kind: str
|
||||
#: Set for ``interval`` schedules.
|
||||
seconds: int | None
|
||||
#: Set for ``cron`` schedules — a five-field crontab expression.
|
||||
expression: str | None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SchedulerSpec:
|
||||
kind: str
|
||||
#: LaunchAgent label; required for ``launchd``, absent for ``manual``.
|
||||
label: str | None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class WorkerRecord:
|
||||
id: str
|
||||
display_name: str
|
||||
provider: str
|
||||
model: str
|
||||
project: str
|
||||
role: str
|
||||
namespace: str
|
||||
profile: str
|
||||
workflow: str
|
||||
schedule: ScheduleSpec
|
||||
timeout_seconds: int
|
||||
enabled: bool
|
||||
scheduler: SchedulerSpec
|
||||
notes: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class WorkerRegistry:
|
||||
version: int
|
||||
revision: int
|
||||
updated_at: str
|
||||
providers: tuple[ProviderRecord, ...]
|
||||
workers: tuple[WorkerRecord, ...]
|
||||
source_path: Path
|
||||
|
||||
|
||||
# ── paths ────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def default_registry_path() -> Path:
|
||||
"""Location of the packaged worker registry, overridable for tests/deploys."""
|
||||
override = os.environ.get("WEBUI_WORKER_REGISTRY", "").strip()
|
||||
if override:
|
||||
return Path(override).expanduser().resolve()
|
||||
return (Path(__file__).resolve().parent / "data" / "workers.registry.json").resolve()
|
||||
|
||||
|
||||
def history_dir(path: Path | None = None) -> Path:
|
||||
"""Directory holding superseded revisions of *path*."""
|
||||
source = (path or default_registry_path()).resolve()
|
||||
return source.parent / f"{source.name}.history"
|
||||
|
||||
|
||||
# ── field helpers ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _require_exact_fields(
|
||||
raw: Any,
|
||||
*,
|
||||
required: tuple[str, ...],
|
||||
optional: tuple[str, ...],
|
||||
subject: str,
|
||||
) -> dict[str, Any]:
|
||||
if not isinstance(raw, dict):
|
||||
raise RegistryValidationError(f"{subject} must be an object")
|
||||
missing = [field for field in required if field not in raw]
|
||||
if missing:
|
||||
raise RegistryValidationError(
|
||||
f"{subject} missing required fields: {', '.join(sorted(missing))}"
|
||||
)
|
||||
unknown = sorted(set(raw) - set(required) - set(optional))
|
||||
if unknown:
|
||||
# Fail closed: silently dropping an unrecognized key is how a typo'd
|
||||
# "timeout_second" ends up meaning "no timeout".
|
||||
raise RegistryValidationError(f"{subject} has unknown fields: {', '.join(unknown)}")
|
||||
return raw
|
||||
|
||||
|
||||
def _require_identifier(value: Any, *, subject: str) -> str:
|
||||
text = str(value).strip()
|
||||
if not _ID_RE.match(text):
|
||||
raise RegistryValidationError(
|
||||
f"{subject} must be lowercase alphanumeric with '.', '_', or '-' (got {value!r})"
|
||||
)
|
||||
return text
|
||||
|
||||
|
||||
def _require_text(value: Any, *, subject: str) -> str:
|
||||
if not isinstance(value, str):
|
||||
raise RegistryValidationError(f"{subject} must be a string (got {value!r})")
|
||||
text = value.strip()
|
||||
if not text:
|
||||
raise RegistryValidationError(f"{subject} must be a non-empty string")
|
||||
return text
|
||||
|
||||
|
||||
def _require_bool(value: Any, *, subject: str) -> bool:
|
||||
if not isinstance(value, bool):
|
||||
raise RegistryValidationError(f"{subject} must be a boolean (got {value!r})")
|
||||
return value
|
||||
|
||||
|
||||
def _require_positive_int(value: Any, *, subject: str, maximum: int | None = None) -> int:
|
||||
if isinstance(value, bool) or not isinstance(value, int):
|
||||
raise RegistryValidationError(f"{subject} must be an integer (got {value!r})")
|
||||
if value <= 0:
|
||||
raise RegistryValidationError(f"{subject} must be greater than zero (got {value})")
|
||||
if maximum is not None and value > maximum:
|
||||
raise RegistryValidationError(f"{subject} must not exceed {maximum} (got {value})")
|
||||
return value
|
||||
|
||||
|
||||
# ── parsing ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _parse_provider(raw: Any) -> ProviderRecord:
|
||||
data = _require_exact_fields(
|
||||
raw,
|
||||
required=_REQUIRED_PROVIDER_FIELDS,
|
||||
optional=_OPTIONAL_PROVIDER_FIELDS,
|
||||
subject="provider",
|
||||
)
|
||||
provider_id = _require_identifier(data["id"], subject="provider.id")
|
||||
|
||||
models_raw = data.get("models") or []
|
||||
if not isinstance(models_raw, list):
|
||||
raise RegistryValidationError(f"provider[{provider_id}].models must be an array")
|
||||
models = tuple(
|
||||
_require_text(item, subject=f"provider[{provider_id}].models[]") for item in models_raw
|
||||
)
|
||||
|
||||
return ProviderRecord(
|
||||
id=provider_id,
|
||||
display_name=_require_text(
|
||||
data["display_name"], subject=f"provider[{provider_id}].display_name"
|
||||
),
|
||||
vendor=_require_text(data["vendor"], subject=f"provider[{provider_id}].vendor"),
|
||||
executable=_require_text(data["executable"], subject=f"provider[{provider_id}].executable"),
|
||||
available=_require_bool(data["available"], subject=f"provider[{provider_id}].available"),
|
||||
models=models,
|
||||
notes=str(data.get("notes") or "").strip(),
|
||||
)
|
||||
|
||||
|
||||
def _parse_schedule(raw: Any, *, subject: str) -> ScheduleSpec:
|
||||
if not isinstance(raw, dict):
|
||||
raise RegistryValidationError(f"{subject} must be an object")
|
||||
kind = _require_text(raw.get("kind"), subject=f"{subject}.kind")
|
||||
if kind not in ALLOWED_SCHEDULE_KINDS:
|
||||
raise RegistryValidationError(
|
||||
f"{subject}.kind must be one of {', '.join(ALLOWED_SCHEDULE_KINDS)} (got {kind!r})"
|
||||
)
|
||||
|
||||
seconds: int | None = None
|
||||
expression: str | None = None
|
||||
|
||||
if kind == "interval":
|
||||
if "seconds" not in raw:
|
||||
raise RegistryValidationError(f"{subject}.seconds is required for interval schedules")
|
||||
seconds = _require_positive_int(raw["seconds"], subject=f"{subject}.seconds")
|
||||
elif kind == "cron":
|
||||
if "expression" not in raw:
|
||||
raise RegistryValidationError(f"{subject}.expression is required for cron schedules")
|
||||
expression = _require_text(raw["expression"], subject=f"{subject}.expression")
|
||||
if len(expression.split()) != 5:
|
||||
raise RegistryValidationError(
|
||||
f"{subject}.expression must have five crontab fields (got {expression!r})"
|
||||
)
|
||||
|
||||
allowed = {"kind"}
|
||||
if kind == "interval":
|
||||
allowed.add("seconds")
|
||||
elif kind == "cron":
|
||||
allowed.add("expression")
|
||||
unknown = sorted(set(raw) - allowed)
|
||||
if unknown:
|
||||
raise RegistryValidationError(
|
||||
f"{subject} has fields not valid for kind {kind!r}: {', '.join(unknown)}"
|
||||
)
|
||||
|
||||
return ScheduleSpec(kind=kind, seconds=seconds, expression=expression)
|
||||
|
||||
|
||||
def _parse_scheduler(raw: Any, *, subject: str) -> SchedulerSpec:
|
||||
if not isinstance(raw, dict):
|
||||
raise RegistryValidationError(f"{subject} must be an object")
|
||||
kind = _require_text(raw.get("kind"), subject=f"{subject}.kind")
|
||||
if kind not in ALLOWED_SCHEDULER_KINDS:
|
||||
raise RegistryValidationError(
|
||||
f"{subject}.kind must be one of {', '.join(ALLOWED_SCHEDULER_KINDS)} (got {kind!r})"
|
||||
)
|
||||
|
||||
label: str | None = None
|
||||
if kind == "launchd":
|
||||
if "label" not in raw:
|
||||
raise RegistryValidationError(f"{subject}.label is required for launchd schedulers")
|
||||
label = _require_text(raw["label"], subject=f"{subject}.label")
|
||||
|
||||
allowed = {"kind"}
|
||||
if kind == "launchd":
|
||||
allowed.add("label")
|
||||
unknown = sorted(set(raw) - allowed)
|
||||
if unknown:
|
||||
raise RegistryValidationError(
|
||||
f"{subject} has fields not valid for kind {kind!r}: {', '.join(unknown)}"
|
||||
)
|
||||
|
||||
return SchedulerSpec(kind=kind, label=label)
|
||||
|
||||
|
||||
def _parse_worker(raw: Any) -> WorkerRecord:
|
||||
data = _require_exact_fields(
|
||||
raw,
|
||||
required=_REQUIRED_WORKER_FIELDS,
|
||||
optional=_OPTIONAL_WORKER_FIELDS,
|
||||
subject="worker",
|
||||
)
|
||||
worker_id = _require_identifier(data["id"], subject="worker.id")
|
||||
|
||||
role = _require_text(data["role"], subject=f"worker[{worker_id}].role")
|
||||
if role not in ALLOWED_ROLES:
|
||||
raise RegistryValidationError(
|
||||
f"worker[{worker_id}].role must be one of {', '.join(ALLOWED_ROLES)} (got {role!r})"
|
||||
)
|
||||
|
||||
return WorkerRecord(
|
||||
id=worker_id,
|
||||
display_name=_require_text(
|
||||
data["display_name"], subject=f"worker[{worker_id}].display_name"
|
||||
),
|
||||
provider=_require_identifier(data["provider"], subject=f"worker[{worker_id}].provider"),
|
||||
model=_require_text(data["model"], subject=f"worker[{worker_id}].model"),
|
||||
project=_require_text(data["project"], subject=f"worker[{worker_id}].project"),
|
||||
role=role,
|
||||
namespace=_require_text(data["namespace"], subject=f"worker[{worker_id}].namespace"),
|
||||
profile=_require_text(data["profile"], subject=f"worker[{worker_id}].profile"),
|
||||
workflow=_require_text(data["workflow"], subject=f"worker[{worker_id}].workflow"),
|
||||
schedule=_parse_schedule(data["schedule"], subject=f"worker[{worker_id}].schedule"),
|
||||
timeout_seconds=_require_positive_int(
|
||||
data["timeout_seconds"],
|
||||
subject=f"worker[{worker_id}].timeout_seconds",
|
||||
maximum=_MAX_TIMEOUT_SECONDS,
|
||||
),
|
||||
enabled=_require_bool(data["enabled"], subject=f"worker[{worker_id}].enabled"),
|
||||
scheduler=_parse_scheduler(data["scheduler"], subject=f"worker[{worker_id}].scheduler"),
|
||||
notes=str(data.get("notes") or "").strip(),
|
||||
)
|
||||
|
||||
|
||||
def _require_unique(values: list[str], *, subject: str) -> None:
|
||||
seen: set[str] = set()
|
||||
for value in values:
|
||||
if value in seen:
|
||||
raise RegistryValidationError(f"duplicate {subject}: {value}")
|
||||
seen.add(value)
|
||||
|
||||
|
||||
def validate_payload(payload: Any, *, source_path: Path) -> WorkerRegistry:
|
||||
"""Validate a decoded registry document and return the typed registry.
|
||||
|
||||
Raises :class:`RegistryValidationError` on any violation; never partially
|
||||
accepts a document.
|
||||
"""
|
||||
if not isinstance(payload, dict):
|
||||
raise RegistryValidationError("registry root must be an object")
|
||||
|
||||
version = payload.get("version")
|
||||
if version != SCHEMA_VERSION:
|
||||
raise RegistryValidationError(f"unsupported registry version: {version!r}")
|
||||
|
||||
reject_credential_keys(payload, subject="worker registry")
|
||||
|
||||
unknown = sorted(set(payload) - _TOP_LEVEL_FIELDS)
|
||||
if unknown:
|
||||
raise RegistryValidationError(f"registry has unknown fields: {', '.join(unknown)}")
|
||||
|
||||
revision = _require_positive_int(payload.get("revision"), subject="revision")
|
||||
updated_at = _require_text(payload.get("updated_at"), subject="updated_at")
|
||||
|
||||
providers_raw = payload.get("providers")
|
||||
if not isinstance(providers_raw, list) or not providers_raw:
|
||||
raise RegistryValidationError("providers must be a non-empty array")
|
||||
providers = tuple(_parse_provider(item) for item in providers_raw)
|
||||
_require_unique([provider.id for provider in providers], subject="provider id")
|
||||
|
||||
workers_raw = payload.get("workers")
|
||||
if not isinstance(workers_raw, list):
|
||||
raise RegistryValidationError("workers must be an array")
|
||||
workers = tuple(_parse_worker(item) for item in workers_raw)
|
||||
_require_unique([worker.id for worker in workers], subject="worker id")
|
||||
|
||||
# Referential integrity: a worker naming an undeclared provider would look
|
||||
# configured while being unrunnable, which is exactly the ambiguous
|
||||
# ownership the epic requires to fail closed.
|
||||
known_providers = {provider.id for provider in providers}
|
||||
for worker in workers:
|
||||
if worker.provider not in known_providers:
|
||||
raise RegistryValidationError(
|
||||
f"worker[{worker.id}].provider references unknown provider {worker.provider!r}"
|
||||
)
|
||||
|
||||
# A LaunchAgent label identifies a job to launchd; two workers sharing one
|
||||
# would silently overwrite each other's agent.
|
||||
_require_unique(
|
||||
[worker.scheduler.label for worker in workers if worker.scheduler.label],
|
||||
subject="scheduler label",
|
||||
)
|
||||
|
||||
return WorkerRegistry(
|
||||
version=version,
|
||||
revision=revision,
|
||||
updated_at=updated_at,
|
||||
providers=providers,
|
||||
workers=workers,
|
||||
source_path=source_path,
|
||||
)
|
||||
|
||||
|
||||
def load_registry(path: Path | None = None) -> WorkerRegistry:
|
||||
"""Load and validate the worker registry from disk."""
|
||||
source = (path or default_registry_path()).resolve()
|
||||
payload = json.loads(source.read_text(encoding="utf-8"))
|
||||
return validate_payload(payload, source_path=source)
|
||||
|
||||
|
||||
# ── serialization ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def provider_to_dict(provider: ProviderRecord) -> dict[str, Any]:
|
||||
return {
|
||||
"id": provider.id,
|
||||
"display_name": provider.display_name,
|
||||
"vendor": provider.vendor,
|
||||
"executable": provider.executable,
|
||||
"available": provider.available,
|
||||
"models": list(provider.models),
|
||||
"notes": provider.notes,
|
||||
}
|
||||
|
||||
|
||||
def _schedule_to_dict(schedule: ScheduleSpec) -> dict[str, Any]:
|
||||
payload: dict[str, Any] = {"kind": schedule.kind}
|
||||
if schedule.kind == "interval":
|
||||
payload["seconds"] = schedule.seconds
|
||||
elif schedule.kind == "cron":
|
||||
payload["expression"] = schedule.expression
|
||||
return payload
|
||||
|
||||
|
||||
def _scheduler_to_dict(scheduler: SchedulerSpec) -> dict[str, Any]:
|
||||
payload: dict[str, Any] = {"kind": scheduler.kind}
|
||||
if scheduler.kind == "launchd":
|
||||
payload["label"] = scheduler.label
|
||||
return payload
|
||||
|
||||
|
||||
def worker_to_dict(worker: WorkerRecord) -> dict[str, Any]:
|
||||
return {
|
||||
"id": worker.id,
|
||||
"display_name": worker.display_name,
|
||||
"provider": worker.provider,
|
||||
"model": worker.model,
|
||||
"project": worker.project,
|
||||
"role": worker.role,
|
||||
"namespace": worker.namespace,
|
||||
"profile": worker.profile,
|
||||
"workflow": worker.workflow,
|
||||
"schedule": _schedule_to_dict(worker.schedule),
|
||||
"timeout_seconds": worker.timeout_seconds,
|
||||
"enabled": worker.enabled,
|
||||
"scheduler": _scheduler_to_dict(worker.scheduler),
|
||||
"notes": worker.notes,
|
||||
}
|
||||
|
||||
|
||||
def registry_to_document(registry: WorkerRegistry) -> dict[str, Any]:
|
||||
"""Serialize to the on-disk document shape (no local paths embedded)."""
|
||||
return {
|
||||
"version": registry.version,
|
||||
"revision": registry.revision,
|
||||
"updated_at": registry.updated_at,
|
||||
"providers": [provider_to_dict(provider) for provider in registry.providers],
|
||||
"workers": [worker_to_dict(worker) for worker in registry.workers],
|
||||
}
|
||||
|
||||
|
||||
def registry_to_dict(registry: WorkerRegistry) -> dict[str, Any]:
|
||||
"""Serialize for JSON API responses (adds the resolved source path)."""
|
||||
document = registry_to_document(registry)
|
||||
document["source_path"] = str(registry.source_path)
|
||||
return document
|
||||
|
||||
|
||||
def find_worker(registry: WorkerRegistry, worker_id: str) -> WorkerRecord | None:
|
||||
for worker in registry.workers:
|
||||
if worker.id == worker_id:
|
||||
return worker
|
||||
return None
|
||||
|
||||
|
||||
def find_provider(registry: WorkerRegistry, provider_id: str) -> ProviderRecord | None:
|
||||
for provider in registry.providers:
|
||||
if provider.id == provider_id:
|
||||
return provider
|
||||
return None
|
||||
|
||||
|
||||
def workers_for_provider(registry: WorkerRegistry, provider_id: str) -> tuple[WorkerRecord, ...]:
|
||||
return tuple(worker for worker in registry.workers if worker.provider == provider_id)
|
||||
|
||||
|
||||
# ── persistence ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _utc_now() -> str:
|
||||
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
|
||||
|
||||
def _atomic_write(path: Path, payload: str) -> None:
|
||||
"""Write *payload* to *path* atomically: temp file in the same dir, fsync, replace."""
|
||||
parent = path.parent
|
||||
parent.mkdir(parents=True, exist_ok=True)
|
||||
fd, temp_path = tempfile.mkstemp(prefix=f".{path.name}-", suffix=".tmp", dir=parent)
|
||||
try:
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
||||
handle.write(payload)
|
||||
handle.flush()
|
||||
os.fsync(handle.fileno())
|
||||
os.replace(temp_path, path)
|
||||
finally:
|
||||
if os.path.exists(temp_path):
|
||||
try:
|
||||
os.remove(temp_path)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def _revision_path(directory: Path, revision: int) -> Path:
|
||||
return directory / f"rev-{revision:06d}.json"
|
||||
|
||||
|
||||
def _prune_history(path: Path) -> None:
|
||||
directory = history_dir(path)
|
||||
revisions = list_revisions(path)
|
||||
excess = len(revisions) - _HISTORY_LIMIT
|
||||
for revision in revisions[: max(0, excess)]:
|
||||
_revision_path(directory, revision).unlink(missing_ok=True)
|
||||
|
||||
|
||||
def _archive_current(path: Path) -> int | None:
|
||||
"""Copy the live document into the history dir under its own revision number."""
|
||||
if not path.exists():
|
||||
return None
|
||||
try:
|
||||
existing = json.loads(path.read_text(encoding="utf-8"))
|
||||
revision = int(existing.get("revision", 0))
|
||||
except (json.JSONDecodeError, TypeError, ValueError, AttributeError):
|
||||
# An unreadable live file has no trustworthy revision number to file it
|
||||
# under, so it cannot join the history chain.
|
||||
return None
|
||||
if revision <= 0:
|
||||
return None
|
||||
_atomic_write(
|
||||
_revision_path(history_dir(path), revision),
|
||||
json.dumps(existing, indent=2, sort_keys=True) + "\n",
|
||||
)
|
||||
_prune_history(path)
|
||||
return revision
|
||||
|
||||
|
||||
def list_revisions(path: Path | None = None) -> tuple[int, ...]:
|
||||
"""Revision numbers retained in history for *path*, oldest first."""
|
||||
directory = history_dir(path)
|
||||
if not directory.is_dir():
|
||||
return ()
|
||||
revisions: list[int] = []
|
||||
for entry in directory.glob("rev-*.json"):
|
||||
try:
|
||||
revisions.append(int(entry.stem.split("-", 1)[1]))
|
||||
except (IndexError, ValueError):
|
||||
continue
|
||||
return tuple(sorted(revisions))
|
||||
|
||||
|
||||
def save_registry(
|
||||
registry: WorkerRegistry,
|
||||
path: Path | None = None,
|
||||
*,
|
||||
updated_at: str | None = None,
|
||||
) -> WorkerRegistry:
|
||||
"""Validate, archive the superseded revision, then atomically persist a new one.
|
||||
|
||||
The stored revision is always the previous revision plus one, so a reader
|
||||
can tell two documents apart even when their content is otherwise equal.
|
||||
Returns the registry exactly as persisted.
|
||||
"""
|
||||
target = (path or registry.source_path or default_registry_path()).resolve()
|
||||
|
||||
document = registry_to_document(registry)
|
||||
# Re-validate before writing: a registry assembled in memory has not
|
||||
# necessarily been through the loader.
|
||||
validate_payload(document, source_path=target)
|
||||
|
||||
archived = _archive_current(target)
|
||||
document["revision"] = (archived + 1) if archived is not None else registry.revision
|
||||
document["updated_at"] = updated_at or _utc_now()
|
||||
|
||||
persisted = validate_payload(document, source_path=target)
|
||||
_atomic_write(target, json.dumps(document, indent=2, sort_keys=True) + "\n")
|
||||
return persisted
|
||||
|
||||
|
||||
def rollback_to_revision(revision: int, path: Path | None = None) -> WorkerRegistry:
|
||||
"""Restore a retained *revision* as a new head revision.
|
||||
|
||||
History is append-only: rolling back does not delete the revisions in
|
||||
between, it republishes the chosen one under the next revision number, so a
|
||||
rollback is itself reversible.
|
||||
"""
|
||||
target = (path or default_registry_path()).resolve()
|
||||
snapshot_path = _revision_path(history_dir(target), revision)
|
||||
if not snapshot_path.exists():
|
||||
available = ", ".join(str(item) for item in list_revisions(target)) or "(none)"
|
||||
raise RegistryValidationError(
|
||||
f"revision {revision} is not retained for {target.name}; available: {available}"
|
||||
)
|
||||
|
||||
payload = json.loads(snapshot_path.read_text(encoding="utf-8"))
|
||||
restored = validate_payload(payload, source_path=target)
|
||||
return save_registry(restored, target)
|
||||
Reference in New Issue
Block a user