Compare commits

...
Author SHA1 Message Date
jcwalker3andClaude Opus 4.8 5494696227 feat(webui): read-only system-health API (Closes #634)
Adds `GET /api/v1/system/health`, a structured read-only health surface for
automated readiness checks, and keeps `/health` as the cheap liveness probe.

webui/system_health.py composes a DTO from fail-soft dependency probes: the
control-plane database, the local checkout, and — opt-in via `?deep=1` — live
Gitea reachability, each carrying status, reason, and probe latency. Required
probes drive readiness; the optional Gitea probe can only degrade overall
status, because local inventory stays serveable when the remote is
unreachable. A probe that did not run leaves readiness incomplete rather than
silently passing.

Read-only throughout: the control-plane database is opened through a `mode=ro`
URI because `ControlPlaneDB.__init__` creates directories and runs migrations,
which a health check must never do. No restart or reload control is exposed;
those are Phase 2 and #630 forbids process-kill recovery.

No unproven claims: `stale_runtime.mutation_safe` is true only when the
runtime, checkout, and remote commits are all known and equal, and MCP
namespaces always report `unproven` because a web process cannot exercise the
IDE-managed client path (#543). Probe details are redacted at the browser
boundary — URLs lose userinfo and query strings, credential-shaped text is
masked.

`/health` is expanded additively: every MVP key is retained, plus `started_at`,
`uptime_seconds`, and a pointer to the versioned API. The versioned route
returns 503 when not ready so automation can branch on the status code alone.

Verified at master 9eb0f29: focused file 40 passed / 11 subtests; `-k "webui or
health"` 230 passed / 159 subtests; full suite 4358 passed with the 11
pre-existing master-drift failures unchanged from the clean-master baseline.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-22 15:59:30 -05:00
sysadmin 9eb0f29cef Merge pull request 'feat(webui): worker registry and configuration schema (Closes #798)' (#810) from feat/issue-798-worker-registry-schema into master 2026-07-22 06:48:29 -05:00
sysadmin 0b29404031 Merge pull request 'docs(webui): MCP Control Plane Web Console architecture ADR (Closes #632)' (#796) from docs/issue-632-web-console-architecture into master 2026-07-22 06:22:11 -05:00
jcwalker3andClaude Opus 4.8 5463f58933 feat(webui): worker registry and configuration schema (Closes #798)
Add the declarative worker registry that epic #797 makes the source of
truth for the scheduled multi-LLM worker fleet.

Providers and configured workers are modelled as separate entities so a
provider can be listed with no worker configured, and so provider facts
are not copied into every worker record. A worker records provider,
model, project, role, namespace, profile, workflow, schedule, timeout,
enabled state, and scheduler metadata.

Validation fails closed: unknown fields are refused rather than ignored,
so a typo cannot silently disable a timeout; a worker naming an
undeclared provider is rejected; worker ids, provider ids, and
LaunchAgent labels must be unique.

Persistence is atomic (temp file in the same directory, fsync, replace).
Every superseded document is retained as a numbered revision, and
rollback republishes a chosen revision as a new head, so history stays
append-only and a rollback is itself reversible.

The credential-rejection guard is extracted to webui/registry_safety.py
so both registries share one implementation instead of two copies of a
security check; project_registry.py keeps identical behaviour.

Scope: data model, validation, persistence only. No routes, scheduler,
process control, or provider probing - those are #799/#800/#804/#805.
The workers array ships empty because populating it is #808.

Tests: tests/test_webui_worker_registry.py, 44 cases.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-22 06:10:34 -05:00
sysadmin 5032965e3a Merge pull request 'feat: make master-parity live-remote aware so a stale daemon fails closed (Closes #610)' (#788) from feat/issue-610-live-remote-parity into master 2026-07-22 05:51:56 -05:00
jcwalker3andGrok 4.5 57a52b1a99 fix(parity): hermetic live-remote master reads under pytest (Closes #610)
Remediate PR #788 review F1/F2: suite-wide hermetic mode prevents
git ls-remote from running in tests so feature worktrees no longer flip
legacy runtime-context assertions to live_stale, and unit tests stay offline.
Module flag survives patch.dict(clear=True); env override still wins.

Co-Authored-By: Grok 4.5 <[email protected]>
2026-07-22 05:40:15 -05:00
jcwalker3andClaude Opus 4.8 e33b8d3712 docs(webui): add MCP Control Plane Web Console architecture ADR (Closes #632)
Adds the durable architecture record epic #631 lacked: layer contract,
authority boundaries, the redaction boundary, /api/v1 versioning with a
compatibility rule for the existing unversioned MVP exports, a target page
map, per-child component ownership for all twenty children (#632-#651),
phase gates, and explicit forbidden paths.

Documentation only. No UI, API, or deployment change.

- docs/architecture/webui-control-plane-console-architecture-adr.md (new)
- docs/webui-local-dev.md: cross-link to the ADR
- tests/test_webui_architecture_docs.py: doc gate for the acceptance
  criteria and the cross-link (12 cases)

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-22 04:52:21 -05:00
jcwalker3 344dc41ce2 Merge branch 'master' into feat/issue-610-live-remote-parity 2026-07-22 04:30:44 -05:00
sysadmin 620ed6e9a9 Merge pull request 'fix(lock): allow exact-owner renewal of expired author issue locks (Closes #760)' (#791) from fix/issue-760-exact-owner-renewal into master 2026-07-22 01:02:39 -05:00
sysadminandClaude Opus 4.8 a30a3ce4c3 fix(lock): make the exact-owner renewal waiver survive downstream gates (#760)
Addresses review #499 on PR #791.

F1 — the renewal sanction was computed and then discarded twice.

`assess_issue_lock_worktree` waived base-equivalence only for
`recovery_sanctioned`, and `gitea_lock_issue` never passed the renewal waiver
into it. A branch being renewed always carries committed work, so it is never
base-equivalent, and #753 recovery refuses when the recorded PID is alive —
which is the defining condition of a renewal. Every real renewal was therefore
granted by the assessor and then rejected one gate later.

Thread `renewal_sanctioned` into `assess_issue_lock_worktree` alongside
`recovery_sanctioned`, waiving base-equivalence on the same grounds and nothing
else. Cleanliness is evaluated before the waiver and is never relaxed; the
assessment now reports which of the two waivers applied.

The MCP-level regression then exposed a second discard point: the duplicate-work
gate rejected the renewal with "open PR already covers issue" — the very PR the
lock being renewed already owns. Add `owning_pr_renewal_evidence`, the mirror of
`issue_lock_recovery.owning_pr_recovery_evidence` (#755), and carry it into the
gate only when renewal was granted. It re-checks that the PR, local, and remote
heads agree, so truncated or hand-built evidence cannot authorize an exemption.

Neither waiver is caller-supplied and `gitea_lock_issue` still gains no
parameter. Absolute wall-clock expiry is unchanged. No #790 heartbeat, sliding
expiry, fencing-token, or shared-lifecycle behavior is introduced, and #753
recovery behavior is untouched.

F2 — add tests/test_issue_760_mcp_renewal_path.py, 10 cases driving the native
`gitea_lock_issue` path against a real git repository and a real durable lock:
expired lease, live recorded PID, committed non-base-equivalent branch. Proves
the renewal completes, records prior and replacement lease evidence, advances
the generation exactly once, produces a live lock that satisfies
`verify_lock_for_mutation`, and does not claim dead-session recovery. Negative
companions prove a foreign claimant, foreign profile, unpublished branch, or
mismatched PR head cannot use the waiver, that a dirty worktree still fails
closed, and that base-equivalence still applies with no waiver at all.

Tests: new MCP suite 10 passed; combined #760 suites 50 passed; lock/lease
regression set 200 passed with 2 subtests; full suite 4240 passed, 11 failed,
6 skipped, 493 subtests passed — the same 11 pre-existing failures as the
master baseline at 3d0c13fa, no new failures.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01Ti8deB36iWcjHmE9cuxour
2026-07-22 01:10:48 -04:00
sysadminandClaude Opus 4.8 1a97ced133 feat(lock): allow exact-owner renewal of expired author issue locks (Closes #760)
An expired author issue lease could not be renewed by the exact session that
already owned it. `assess_same_issue_lease_conflict` computed same-owner
evidence and then returned on the expired branch before consulting it, and
`assess_expired_lock_reclaim` only permits takeover on a dead PID or a missing
worktree. Because the recorded PID is the long-lived MCP daemon rather than the
authoring task, a lease that expires under a live daemon is the ordinary case
for any author task outliving the TTL — and in that case the owner's own lock
became permanently unmodifiable through sanctioned tools.

Add `issue_lock_renewal`, a pure evidence assessor for that one case, and
evaluate its disposition before the expired foreign-takeover return.

Renewal requires an exact match of remote, org, repo, issue number, operation
type, branch, realpath-normalized worktree, claimant username, and claimant
profile; a registered worktree that exists, sits on the locked branch, and is
clean; local and remote heads that agree; and, when an owning PR exists, a PR
head that agrees too. No competing live lock, competing branch claim, or other
owning PR may exist. Any missing or contradictory evidence fails closed.

PID liveness is never authorization: it is recorded as evidence and is neither
necessary nor sufficient (AC16). Only an expired lease is ever a candidate, so
a live foreign lease stays non-recoverable (AC12) and dead-PID takeover keeps
its existing #601 conditions (AC11). Renewal is never caller-declarable — the
waiver is server-computed and `gitea_lock_issue` gains no parameter (AC14).

A sanctioned renewal records prior PID, prior expiry, replacement PID, new
expiry, claimant, and its supporting proof under `lease_renewal`, and reuses
the #772 compare-and-swap so two sessions observing the same expired lease
cannot both win.

Absolute wall-clock expiry is preserved. Sliding heartbeat renewal, fencing
tokens, and the shared cross-role lifecycle remain #790's scope and are
deliberately not implemented here; #790 stays sequenced behind this change.

Tests: 40 new cases covering the positive path, every near-match and
foreign-owner refusal, the non-candidate cases, the gate-ordering regression,
the renewal record, downstream mutation ownership, and AC14/AC17. Two #772 test
doubles now forward the new keyword.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01Ti8deB36iWcjHmE9cuxour
2026-07-22 00:39:51 -04:00
sysadmin 3d0c13fa5a Merge pull request 'fix(guard): quote-aware shell segmentation so only real daemon kills classify (Closes #787)' (#789) from fix/issue-787-kill-segment-separators into master 2026-07-21 21:40:02 -05:00
jcwalker3 0f19773076 Merge branch 'master' into feat/issue-610-live-remote-parity 2026-07-21 20:48:54 -05:00
sysadminandClaude Opus 4.8 324b4b3e93 feat: make master-parity live-remote aware so a stale daemon fails closed (Closes #610)
Master-parity previously compared only the daemon's startup commit against the
local on-disk HEAD. When the checkout was not pulled, parity reported green even
though the live remote master had advanced, so a stale daemon could claim a
mutation-safe result while running outdated capability gates (observed during
PR #592 recovery, where the resolver correctly required restart but parity said
in_parity=true).

Changes:
- master_parity_gate.assess_master_parity() gains an optional live_remote_head
  and reports the three commits distinctly (daemon_start_head, local_head,
  live_remote_head) plus live_known / live_stale / mutation_safe. A result is
  mutation_safe only when daemon, local checkout, and live remote all agree.
- parity_block_reasons() now blocks mutations on live-staleness too; read-only
  operations remain unblocked (non-goal: never block diagnostics offline).
- New parity_resolver_disagreement(): typed fail-closed blocker naming the
  capability resolver as authoritative when it requires restart but parity
  looks locally green.
- New read_remote_master_head(): best-effort `git ls-remote` for the live
  target, cached with a 60s TTL (bounded offline latency, no network probe per
  gate call); env override GITEA_TEST_LIVE_REMOTE_HEAD keeps tests hermetic.
- Server: _current_master_parity() reads the live remote head;
  gitea_assess_master_parity and runtime_context surface the distinguished
  SHAs, mutation_safe, and resolver-authoritative guidance.

Tests: 13 new cases (live-remote parity, live-stale blocking + typed blocker,
remote-head reader + TTL cache, server wiring). Full suite: 2423 passed; the 8
remaining failures (test_config TestAuthIntegration, test_credentials
TestGetCredentials) are baseline-proven keychain/env failures identical on the
unmodified base.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-09 18:52:59 -04:00
21 changed files with 4968 additions and 83 deletions
@@ -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 13 |
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
View File
@@ -37,12 +37,19 @@ Optional environment variables:
See [webui-deployment.md](webui-deployment.md) for internal-only serving, See [webui-deployment.md](webui-deployment.md) for internal-only serving,
Cloudflare Access/WARP/VPN guidance, and unsafe bind overrides (#435). 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) ## Routes (MVP)
| Path | Description | | Path | Description |
|------|-------------| |------|-------------|
| `/` | Home / operator overview | | `/` | 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) | | `/queue` | Live PR and issue queue dashboard (#429) |
| `/api/queue` | JSON queue export with pagination metadata | | `/api/queue` | JSON queue export with pagination metadata |
| `/projects` | Project registry list (#427) | | `/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 `read-only-mvp`, except `/audit` and `/api/audit` which accept POST for
local validator preview only (no Gitea mutations, no server-side storage). 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) ## Report audit (#431)
Paste an LLM final report at `/audit` or POST JSON to `/api/audit`. The UI Paste an LLM final report at `/audit` or POST JSON to `/api/audit`. The UI
+247 -22
View File
@@ -2017,6 +2017,7 @@ import issue_lock_provenance # noqa: E402
import issue_lock_store # noqa: E402 import issue_lock_store # noqa: E402
import issue_lock_adoption # noqa: E402 import issue_lock_adoption # noqa: E402
import issue_lock_recovery # noqa: E402 import issue_lock_recovery # noqa: E402
import issue_lock_renewal # noqa: E402
import stacked_pr_support # noqa: E402 import stacked_pr_support # noqa: E402
import merge_approval_gate # noqa: E402 import merge_approval_gate # noqa: E402
import review_quarantine # noqa: E402 # #695 contaminated formal-review quarantine import review_quarantine # noqa: E402 # #695 contaminated formal-review quarantine
@@ -2316,7 +2317,12 @@ def _resolve_issue_lock_for_pr(
return lock_data return lock_data
def _save_issue_lock(data: dict, *, expected_generation: int | None = None) -> str: def _save_issue_lock(
data: dict,
*,
expected_generation: int | None = None,
renewal_sanctioned: bool = False,
) -> str:
existing = issue_lock_store.load_issue_lock( existing = issue_lock_store.load_issue_lock(
remote=str(data.get("remote") or ""), remote=str(data.get("remote") or ""),
org=str(data.get("org") or ""), org=str(data.get("org") or ""),
@@ -2328,7 +2334,9 @@ def _save_issue_lock(data: dict, *, expected_generation: int | None = None) -> s
raise RuntimeError(overwrite_block) raise RuntimeError(overwrite_block)
try: try:
return issue_lock_store.bind_session_lock( return issue_lock_store.bind_session_lock(
data, expected_generation=expected_generation data,
expected_generation=expected_generation,
renewal_sanctioned=renewal_sanctioned,
) )
except Exception as e: except Exception as e:
raise RuntimeError(f"Could not write issue lock file: {e}") from e raise RuntimeError(f"Could not write issue lock file: {e}") from e
@@ -2448,6 +2456,79 @@ def _evaluate_issue_lock_recovery(
) )
def _evaluate_issue_lock_renewal(
existing_lock: dict,
*,
issue_number: int,
branch_name: str,
worktree_path: str,
remote: str,
h: str | None,
o: str,
r: str,
git_state: dict,
) -> dict:
"""Gather evidence and decide exact-owner renewal of an expired lease (#760).
Mirrors ``_evaluate_issue_lock_recovery``: every input is durable lock state
or a live server-side observation (Gitea branch/PR inventory, git in the
declared worktree, the local lock store). Nothing is reachable from an MCP
caller's parameters, so no caller can assert its way into a renewal
(#760 AC14).
"""
renewal_auth = _auth(h)
try:
renewal_branches = api_get_all(
f"{repo_api_url(h, o, r)}/branches", renewal_auth
)
except Exception as exc:
raise RuntimeError(
f"Could not list branches to verify exact-owner lease renewal: {exc}"
)
remote_head: str | None = None
candidates: list[str] = []
for entry in renewal_branches:
entry_name = _branch_entry_name(entry)
if entry_name == branch_name:
remote_head = _branch_entry_commit_sha(entry)
if issue_lock_adoption.branch_carries_issue_marker(entry_name, issue_number):
candidates.append(entry_name)
pr_head: str | None = None
pr_number: int | None = None
for pull in _list_open_pulls(h, o, r, renewal_auth):
pull_head = pull.get("head") or {}
if str(pull_head.get("ref") or "") == branch_name:
pr_head = pull_head.get("sha")
pr_number = pull.get("number")
break
claimant = _work_lease_claimant(h)
return issue_lock_renewal.assess_exact_owner_lease_renewal(
existing_lock,
issue_number=issue_number,
branch_name=branch_name,
worktree_path=worktree_path,
remote=remote,
org=o,
repo=r,
identity=claimant.get("username"),
profile=claimant.get("profile"),
operation_type=AUTHOR_ISSUE_WORK_LEASE,
current_branch=git_state.get("current_branch"),
porcelain_status=git_state.get("porcelain_status") or "",
worktree_exists=os.path.isdir(os.path.realpath(worktree_path)),
head_sha=git_state.get("head_sha"),
remote_head_sha=remote_head,
pr_head_sha=pr_head,
pr_number=pr_number,
competing_live_locks=issue_lock_store.list_live_locks(),
candidate_branches=candidates,
current_pid=os.getpid(),
)
def _work_lease_claimant(host: str | None) -> dict: def _work_lease_claimant(host: str | None) -> dict:
profile = get_profile() profile = get_profile()
username = _IDENTITY_CACHE.get(host) if host else None username = _IDENTITY_CACHE.get(host) if host else None
@@ -3893,15 +3974,12 @@ def gitea_lock_issue(
existing_issue_lock = _load_existing_issue_lock( existing_issue_lock = _load_existing_issue_lock(
remote=remote, org=o, repo=r, issue_number=issue_number remote=remote, org=o, repo=r, issue_number=issue_number
) )
active_lease_block = issue_lock_store.assess_same_issue_lease_conflict( # #760: the competing-lease disposition is decided below, once the worktree
existing_issue_lock, # and Gitea evidence an exact-owner renewal depends on has actually been
issue_number=issue_number, # observed. Deciding it here — before any of that exists — is what made the
branch_name=branch_name, # same-owner allowance unreachable for an expired lease. The authoritative
worktree_path=resolved_worktree, # check still runs inside bind_session_lock under the per-issue flock, so
operation_type=AUTHOR_ISSUE_WORK_LEASE, # moving this one later cannot widen the window for a competing writer.
)
if active_lease_block:
raise RuntimeError(active_lease_block)
# ── Stacked-PR base declaration (opt-in, #484) ── # ── Stacked-PR base declaration (opt-in, #484) ──
# Normal work leaves stacked_base_branch None → master-equivalent path. # Normal work leaves stacked_base_branch None → master-equivalent path.
@@ -3967,6 +4045,53 @@ def gitea_lock_issue(
recovery_sanctioned = bool( recovery_sanctioned = bool(
recovery_assessment and recovery_assessment.get("recovery_sanctioned") recovery_assessment and recovery_assessment.get("recovery_sanctioned")
) )
# ── Exact-owner renewal of an expired lease (#760) ──
# The opposite trigger from #753 above: there the lease is unexpired and the
# PID is dead; here the lease has expired while the recording daemon — which
# is the long-lived MCP server, not the authoring task — may well still be
# up. Only an expired lease is assessed, so a live foreign lease is never a
# candidate (AC12) and dead-PID takeover keeps its existing conditions
# (AC11). A refusal never raises: it withholds the waiver and leaves the
# conflict check below to fail closed exactly as before.
renewal_assessment: dict | None = None
if (
existing_issue_lock
and existing_issue_lock.get("issue_number") == issue_number
and issue_lock_store.is_lease_expired(existing_issue_lock)
):
renewal_assessment = _evaluate_issue_lock_renewal(
existing_issue_lock,
issue_number=issue_number,
branch_name=branch_name,
worktree_path=resolved_worktree,
remote=remote,
h=h,
o=o,
r=r,
git_state=git_state,
)
renewal_sanctioned = bool(
renewal_assessment and renewal_assessment.get("renewal_sanctioned")
)
active_lease_block = issue_lock_store.assess_same_issue_lease_conflict(
existing_issue_lock,
issue_number=issue_number,
branch_name=branch_name,
worktree_path=resolved_worktree,
operation_type=AUTHOR_ISSUE_WORK_LEASE,
renewal_sanctioned=renewal_sanctioned,
)
if active_lease_block:
reasons = [active_lease_block]
# Name the exact missing evidence when this looked like a renewal, so a
# blocked owner sees why rather than only the generic takeover text.
if renewal_assessment and renewal_assessment.get("is_candidate"):
reasons.append(
issue_lock_renewal.format_renewal_refusal(renewal_assessment)
)
raise RuntimeError("; ".join(reasons))
# #755: a sanctioned dead-session recovery always has an owning open PR — # #755: a sanctioned dead-session recovery always has an owning open PR —
# that is what makes it a recovery rather than a fresh claim. Carry the # that is what makes it a recovery rather than a fresh claim. Carry the
# server-derived owning-PR evidence into the duplicate-work gate below so # server-derived owning-PR evidence into the duplicate-work gate below so
@@ -3978,6 +4103,14 @@ def gitea_lock_issue(
if recovery_sanctioned if recovery_sanctioned
else None else None
) )
# #760: a sanctioned renewal owns its open PR for the same reason, so it
# needs the same exemption. Without it the duplicate-work gate rejects every
# renewal with "open PR already covers issue", which is the PR the lock
# being renewed already owns. Withheld unless renewal was granted.
if recovered_owning_pr is None and renewal_sanctioned:
recovered_owning_pr = issue_lock_renewal.owning_pr_renewal_evidence(
renewal_assessment
)
lock_assessment = issue_lock_worktree.assess_issue_lock_worktree( lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
worktree_path=resolved_worktree, worktree_path=resolved_worktree,
current_branch=git_state.get("current_branch"), current_branch=git_state.get("current_branch"),
@@ -3986,6 +4119,10 @@ def gitea_lock_issue(
inspected_git_root=git_state.get("inspected_git_root"), inspected_git_root=git_state.get("inspected_git_root"),
base_branch=git_state.get("base_branch"), base_branch=git_state.get("base_branch"),
recovery_sanctioned=recovery_sanctioned, recovery_sanctioned=recovery_sanctioned,
# #760: without this the renewal waiver was computed and then discarded
# here — the exact-owner branch always carries commits, so it can never
# be base-equivalent, and every real renewal failed at this gate.
renewal_sanctioned=renewal_sanctioned,
) )
if lock_assessment["block"]: if lock_assessment["block"]:
reasons = list(lock_assessment.get("reasons") or []) reasons = list(lock_assessment.get("reasons") or [])
@@ -3995,6 +4132,12 @@ def gitea_lock_issue(
reasons.append( reasons.append(
issue_lock_recovery.format_recovery_refusal(recovery_assessment) issue_lock_recovery.format_recovery_refusal(recovery_assessment)
) )
# #760: same courtesy for a refused renewal, so an exact owner blocked
# at this gate sees which piece of ownership evidence was missing.
if renewal_assessment and renewal_assessment.get("is_candidate"):
reasons.append(
issue_lock_renewal.format_renewal_refusal(renewal_assessment)
)
raise RuntimeError( raise RuntimeError(
issue_lock_worktree.format_issue_lock_worktree_error( issue_lock_worktree.format_issue_lock_worktree_error(
{**lock_assessment, "reasons": reasons} {**lock_assessment, "reasons": reasons}
@@ -4070,18 +4213,35 @@ def gitea_lock_issue(
recovery_assessment, recovery_assessment,
recovered_at=_work_lease_timestamp(_work_lease_now()), recovered_at=_work_lease_timestamp(_work_lease_now()),
) )
if renewal_sanctioned and renewal_assessment:
# #760 AC9: record both sides of the transition — prior PID and expiry,
# replacement PID and new expiry — so a renewed lock is auditable and
# never reads as an original claim.
data["lease_renewal"] = issue_lock_renewal.build_renewal_record(
renewal_assessment,
renewed_at=_work_lease_timestamp(_work_lease_now()),
new_expires_at=str(work_lease.get("expires_at") or ""),
)
# #772 AC5: a recovery replaces a claim another session already owned, so # #772 AC5: a recovery replaces a claim another session already owned, so
# its write is a compare-and-swap against the generation the assessment was # its write is a compare-and-swap against the generation the assessment was
# made on. Two replacement sessions that both observed the same dead owner # made on. Two replacement sessions that both observed the same dead owner
# cannot both succeed — the second finds a moved generation and fails # cannot both succeed — the second finds a moved generation and fails
# closed. Ordinary first-time claims keep the unconditional write. # closed. Ordinary first-time claims keep the unconditional write.
# #760 uses the same compare-and-swap: a renewal also replaces a claim that
# already existed on disk, so two sessions that both observed the same
# expired lease cannot both win — the second finds a moved generation and
# fails closed.
expected_generation = ( expected_generation = (
issue_lock_store.lock_generation(existing_issue_lock) issue_lock_store.lock_generation(existing_issue_lock)
if recovery_sanctioned if (recovery_sanctioned or renewal_sanctioned)
else None else None
) )
lock_file_path = _save_issue_lock(data, expected_generation=expected_generation) lock_file_path = _save_issue_lock(
data,
expected_generation=expected_generation,
renewal_sanctioned=renewal_sanctioned,
)
lock_record = issue_lock_store.read_lock_file(lock_file_path) or data lock_record = issue_lock_store.read_lock_file(lock_file_path) or data
freshness = issue_lock_store.assess_lock_freshness(lock_record) freshness = issue_lock_store.assess_lock_freshness(lock_record)
competing = [ competing = [
@@ -4150,6 +4310,16 @@ def gitea_lock_issue(
issue_number=issue_number, issue_number=issue_number,
branch_name=branch_name, branch_name=branch_name,
) )
if renewal_sanctioned:
# #760 AC13: the renewal is visible in the native tool result, so an
# owner never has to inspect the lock file to confirm what happened.
result["lease_renewal"] = lock_record.get("lease_renewal")
result["message"] = (
f"Renewed the expired {AUTHOR_ISSUE_WORK_LEASE} lease on issue "
f"#{issue_number} for its exact recorded owner, on branch "
f"'{branch_name}' from worktree '{resolved_worktree}' "
"(fail-closed check complete)."
)
if agent_artifacts: if agent_artifacts:
result["warnings"] = [ result["warnings"] = [
"Agent temp artifacts at repo root (delete before implementation): " "Agent temp artifacts at repo root (delete before implementation): "
@@ -12366,10 +12536,39 @@ def _try_auto_switch_for_operation(op: str, host: str | None = None) -> bool:
return False return False
def _git_default_remote_name(root: str) -> str:
"""First configured git remote name for *root*, defaulting to 'origin'.
Used to resolve the live remote master target for parity (#610). Best
effort: any failure falls back to 'origin' so callers never raise.
"""
try:
res = subprocess.run(
["git", "-C", root, "remote"],
capture_output=True, text=True, check=False,
)
except Exception:
return "origin"
if res.returncode != 0:
return "origin"
names = [n.strip() for n in (res.stdout or "").splitlines() if n.strip()]
return names[0] if names else "origin"
def _current_master_parity() -> dict: def _current_master_parity() -> dict:
"""Assess this process's code against the on-disk master HEAD (#420).""" """Assess this process's code against local and live remote master (#420/#610).
Compares the daemon's startup commit, the on-disk checkout HEAD, and the
live remote master target. A stale daemon relative to live master fails
closed for mutations even when the local checkout HEAD still matches the
startup commit. The live-remote read is best effort: an unresolved live
head leaves read-only diagnostics unblocked but is never mutation-safe.
"""
current_head = master_parity_gate.read_git_head(PROJECT_ROOT) current_head = master_parity_gate.read_git_head(PROJECT_ROOT)
return master_parity_gate.assess_master_parity(_STARTUP_PARITY, current_head) live_head = master_parity_gate.read_remote_master_head(
PROJECT_ROOT, remote=_git_default_remote_name(PROJECT_ROOT))
return master_parity_gate.assess_master_parity(
_STARTUP_PARITY, current_head, live_remote_head=live_head)
def _current_runtime_mode_report(refresh: bool = False) -> dict: def _current_runtime_mode_report(refresh: bool = False) -> dict:
@@ -16310,15 +16509,34 @@ def gitea_get_runtime_context(
"restart_required": parity["restart_required"], "restart_required": parity["restart_required"],
"startup_head": parity["startup_head"], "startup_head": parity["startup_head"],
"current_head": parity["current_head"], "current_head": parity["current_head"],
# #610 distinguished mutation-safety signals:
"daemon_start_head": parity["daemon_start_head"],
"local_head": parity["local_head"],
"live_remote_head": parity["live_remote_head"],
"live_known": parity["live_known"],
"live_stale": parity["live_stale"],
"mutation_safe": parity["mutation_safe"],
"summary": master_parity_gate.format_parity(parity), "summary": master_parity_gate.format_parity(parity),
"mutation_gate_enforced": not master_parity_gate.gate_disabled(), "mutation_gate_enforced": not master_parity_gate.gate_disabled(),
# #610: the capability resolver is authoritative for mutation safety;
# local parity alone must never authorize a mutation.
"resolver_authoritative_for_mutation_safety": True,
} }
if parity["stale"] and not master_parity_gate.gate_disabled(): if parity["restart_required"] and not master_parity_gate.gate_disabled():
safe_next_action = ( if parity["live_stale"]:
"Server code is stale relative to master; restart the Gitea MCP " safe_next_action = (
"server to load current capability gates before mutating. " "Daemon is stale relative to LIVE remote master "
f"({master_parity_gate.format_parity(parity)})" f"(started {parity['startup_head'][:12] if parity['startup_head'] else 'unknown'}, "
) f"live master {parity['live_remote_head'][:12] if parity['live_remote_head'] else 'unknown'}); "
"restart/reconnect the Gitea MCP server before mutating. The "
"capability resolver is authoritative for mutation safety."
)
else:
safe_next_action = (
"Server code is stale relative to master; restart the Gitea MCP "
"server to load current capability gates before mutating. "
f"({master_parity_gate.format_parity(parity)})"
)
result["safe_next_action"] = safe_next_action result["safe_next_action"] = safe_next_action
if reveal and h: if reveal and h:
@@ -16367,6 +16585,13 @@ def gitea_assess_master_parity(
"determinable": parity["determinable"], "determinable": parity["determinable"],
"startup_head": parity["startup_head"], "startup_head": parity["startup_head"],
"current_head": parity["current_head"], "current_head": parity["current_head"],
# #610 distinguished mutation-safety signals:
"daemon_start_head": parity["daemon_start_head"],
"local_head": parity["local_head"],
"live_remote_head": parity["live_remote_head"],
"live_known": parity["live_known"],
"live_stale": parity["live_stale"],
"mutation_safe": parity["mutation_safe"],
"mutation_gate_enforced": enforced, "mutation_gate_enforced": enforced,
"summary": master_parity_gate.format_parity(parity), "summary": master_parity_gate.format_parity(parity),
"reasons": parity["reasons"], "reasons": parity["reasons"],
@@ -16383,7 +16608,7 @@ def gitea_assess_master_parity(
source=canonical_source, source=canonical_source,
), ),
} }
if parity["stale"] and enforced: if parity["restart_required"] and enforced:
out["report"] = master_parity_gate.parity_report(parity) out["report"] = master_parity_gate.parity_report(parity)
return out return out
+481
View File
@@ -0,0 +1,481 @@
"""Exact-owner renewal of an expired author issue lease (#760).
An author issue lease carries an absolute wall-clock expiry stamped once at
lock time. The PID recorded alongside it is the long-lived MCP daemon, not the
authoring task, so a lease that expires while its daemon is still up is the
ordinary case for any author task that outlives the TTL — not an anomaly.
Before this module, that case was unreachable.
``issue_lock_store.assess_same_issue_lease_conflict`` computed same-owner
evidence and then returned on the expired branch before consulting it, and
``assess_expired_lock_reclaim`` only permits takeover on a dead PID or a
missing worktree. An exact owner whose daemon is alive and whose worktree is
present satisfied neither, so its own lock became permanently unmodifiable
through sanctioned tools.
This module is the pure evidence assessor for that one narrow case. It answers
a single question: may *this* session renew a lease it can prove it already
owns? It performs no mutation and no network I/O, and it never trusts a caller
assertion — every field is compared against durable lock state or a live
observation supplied by the caller and gathered server-side.
Deliberate boundaries:
* **Renewal is not takeover.** A refusal here never widens what
``assess_expired_lock_reclaim`` already allows; foreign expired locks keep
requiring a dead PID or missing worktree (#760 AC11), and a *live* foreign
lease stays non-recoverable by construction because only an expired lease is
ever a candidate (AC12).
* **PID liveness is never authorization.** A live recorded PID proves the
daemon is up, nothing more. It is recorded as evidence and is neither
necessary nor sufficient for renewal (AC16).
* **Absolute expiry is preserved.** Renewal issues a new absolute expiry from
the moment of the write. It does not introduce sliding heartbeat renewal,
lease generations as fencing tokens, or a shared cross-role lifecycle — that
is #790's scope and is deliberately not implemented here.
"""
from __future__ import annotations
import os
from typing import Any, Iterable, Mapping, Sequence
from issue_lock_store import AUTHOR_ISSUE_WORK_LEASE, is_lease_expired, is_process_alive
from reviewer_worktree import parse_dirty_tracked_files
# Outcome values.
RENEWAL_SANCTIONED = "RENEWAL_SANCTIONED"
NO_CANDIDATE = "NO_CANDIDATE"
REFUSED = "REFUSED"
# Durable fields a lock must carry before it can be considered at all.
REQUIRED_LOCK_FIELDS = ("issue_number", "branch_name", "worktree_path")
def _text(value: Any) -> str:
return str(value or "").strip()
def _same_realpath(left: str | None, right: str | None) -> bool:
if not left or not right:
return False
try:
return os.path.realpath(left) == os.path.realpath(right)
except OSError:
return left == right
def _lock_claimant(lock: Mapping[str, Any]) -> dict[str, Any]:
claimant = lock.get("claimant")
if not isinstance(claimant, Mapping):
lease = lock.get("work_lease")
claimant = lease.get("claimant") if isinstance(lease, Mapping) else None
return dict(claimant) if isinstance(claimant, Mapping) else {}
def _lock_lease(lock: Mapping[str, Any]) -> dict[str, Any]:
lease = lock.get("work_lease")
return dict(lease) if isinstance(lease, Mapping) else {}
def _lock_operation_type(lock: Mapping[str, Any]) -> str:
lease = _lock_lease(lock)
return _text(lease.get("operation_type")) or AUTHOR_ISSUE_WORK_LEASE
def _recorded_pid(lock: Mapping[str, Any]) -> Any:
pid = lock.get("session_pid")
if pid is None:
pid = lock.get("pid")
return pid
def _malformed_reasons(lock: Mapping[str, Any]) -> list[str]:
"""Names of durable fields that are missing or unusable."""
missing: list[str] = []
for field in REQUIRED_LOCK_FIELDS:
if not _text(lock.get(field)):
missing.append(field)
pid = _recorded_pid(lock)
if pid is None or _text(pid) == "":
missing.append("session_pid/pid")
else:
try:
if int(pid) <= 0:
missing.append("session_pid/pid")
except (TypeError, ValueError):
missing.append("session_pid/pid")
return missing
def _competing_lock_reasons(
competing_live_locks: Iterable[Mapping[str, Any]] | None,
*,
issue_number: int,
branch_name: str,
worktree_path: str,
) -> list[str]:
"""Live locks that would contend with this renewal (#760 AC7).
A live lock on the *same* issue cannot coexist with this expired lease, so
any live entry naming this issue, branch, or worktree belongs to somebody
else and refuses the renewal.
"""
reasons: list[str] = []
for entry in competing_live_locks or ():
if not isinstance(entry, Mapping):
continue
entry_issue = entry.get("issue_number")
entry_branch = _text(entry.get("branch_name"))
entry_worktree = _text(entry.get("worktree_path"))
if entry_issue == issue_number:
reasons.append(
f"a live lock already exists for issue #{issue_number} "
f"(pid {entry.get('pid')}); renewal would contend with it"
)
continue
if entry_branch and entry_branch == _text(branch_name):
reasons.append(
f"live lock for issue #{entry_issue} already holds branch "
f"'{branch_name}'"
)
if entry_worktree and _same_realpath(entry_worktree, worktree_path):
reasons.append(
f"live lock for issue #{entry_issue} already holds worktree "
f"'{worktree_path}'"
)
return reasons
def assess_exact_owner_lease_renewal(
existing_lock: Mapping[str, Any] | None,
*,
issue_number: int,
branch_name: str,
worktree_path: str,
remote: str,
org: str,
repo: str,
identity: str | None,
profile: str | None,
operation_type: str = AUTHOR_ISSUE_WORK_LEASE,
current_branch: str | None = None,
porcelain_status: str = "",
worktree_exists: bool = False,
head_sha: str | None = None,
remote_head_sha: str | None = None,
pr_head_sha: str | None = None,
pr_number: int | None = None,
competing_live_locks: Sequence[Mapping[str, Any]] | None = None,
candidate_branches: Sequence[str] | None = None,
current_pid: int | None = None,
now: Any = None,
) -> dict[str, Any]:
"""Decide whether an expired lease may be renewed by its exact owner.
Returns a disposition dict; it never raises and never mutates. A refusal
withholds permission, leaving every pre-existing guard to fail closed
exactly as before — this assessment can only ever *add* permission.
``NO_CANDIDATE`` means the situation is not an exact-owner renewal at all
(no lock, different issue, different operation, or an unexpired lease) and
the caller should carry on with its normal path. ``REFUSED`` means it looked
like one but the evidence did not hold, and ``reasons`` names exactly what
was missing.
"""
evidence: dict[str, Any] = {
"issue_number": issue_number,
"branch_name": branch_name,
"worktree_path": worktree_path,
"remote": remote,
"org": org,
"repo": repo,
"operation_type": operation_type,
"identity": identity,
"profile": profile,
}
def _result(outcome: str, reasons: list[str], **extra: Any) -> dict[str, Any]:
return {
"outcome": outcome,
"renewal_sanctioned": outcome == RENEWAL_SANCTIONED,
"is_candidate": outcome in (RENEWAL_SANCTIONED, REFUSED),
"reasons": reasons,
"evidence": {**evidence, **extra},
}
if not isinstance(existing_lock, Mapping) or not existing_lock:
return _result(NO_CANDIDATE, ["no existing lock to renew"])
if existing_lock.get("issue_number") != issue_number:
return _result(
NO_CANDIDATE,
[
f"existing lock is for issue #{existing_lock.get('issue_number')}, "
f"not #{issue_number}"
],
)
existing_operation = _lock_operation_type(existing_lock)
if existing_operation != operation_type:
return _result(
NO_CANDIDATE,
[
f"existing lease operation '{existing_operation}' is not "
f"'{operation_type}'"
],
)
# Only an *expired* lease is ever a renewal candidate. An unexpired lease —
# live, or stale by dead PID — is somebody else's problem: the first needs no
# renewal, and the second is #753's dead-session recovery. This is also what
# makes a live foreign lease non-recoverable here (#760 AC12).
if not is_lease_expired(existing_lock, now=now):
return _result(
NO_CANDIDATE,
["lease has not expired; renewal does not apply"],
)
malformed = _malformed_reasons(existing_lock)
if malformed:
return _result(
REFUSED,
["durable lock is missing or has unusable fields: " + ", ".join(malformed)],
)
lease = _lock_lease(existing_lock)
claimant = _lock_claimant(existing_lock)
recorded_pid = _recorded_pid(existing_lock)
prior_expires_at = _text(lease.get("expires_at"))
# #760 AC16: recorded purely as evidence. A live daemon PID is neither
# necessary nor sufficient for renewal, and nothing below branches on it.
recorded_pid_alive = is_process_alive(recorded_pid)
extra: dict[str, Any] = {
"prior_pid": recorded_pid,
"prior_pid_alive": recorded_pid_alive,
"prior_expires_at": prior_expires_at,
"replacement_pid": current_pid,
"recorded_claimant": claimant,
"head_sha": head_sha,
"remote_head_sha": remote_head_sha,
"pr_head_sha": pr_head_sha,
"pr_number": pr_number,
}
reasons: list[str] = []
# ── AC3: exact ownership identity ──
if _text(existing_lock.get("remote")) != _text(remote):
reasons.append(
f"recorded remote '{existing_lock.get('remote')}' does not match "
f"'{remote}'"
)
if _text(existing_lock.get("org")) != _text(org):
reasons.append(
f"recorded org '{existing_lock.get('org')}' does not match '{org}'"
)
if _text(existing_lock.get("repo")) != _text(repo):
reasons.append(
f"recorded repo '{existing_lock.get('repo')}' does not match '{repo}'"
)
if _text(existing_lock.get("branch_name")) != _text(branch_name):
reasons.append(
f"recorded branch '{existing_lock.get('branch_name')}' does not match "
f"'{branch_name}'"
)
if not _same_realpath(_text(existing_lock.get("worktree_path")), worktree_path):
reasons.append(
f"recorded worktree '{existing_lock.get('worktree_path')}' does not "
f"match '{worktree_path}'"
)
recorded_identity = _text(claimant.get("username"))
recorded_profile = _text(claimant.get("profile"))
if not recorded_identity or not recorded_profile:
reasons.append(
"durable lock does not record both a claimant username and profile"
)
if recorded_identity and recorded_identity != _text(identity):
reasons.append(
f"recorded claimant '{recorded_identity}' does not match active "
f"identity '{_text(identity) or 'unknown'}'"
)
if recorded_profile and recorded_profile != _text(profile):
reasons.append(
f"recorded profile '{recorded_profile}' does not match active profile "
f"'{_text(profile) or 'unknown'}'"
)
# ── AC4: the registered worktree still exists, is on the branch, and is clean ──
if not worktree_exists:
reasons.append(f"declared worktree '{worktree_path}' does not exist")
if _text(current_branch) != _text(branch_name):
reasons.append(
f"worktree is on branch '{_text(current_branch) or 'unknown'}', not "
f"'{branch_name}'"
)
dirty = parse_dirty_tracked_files(porcelain_status or "")
if dirty:
reasons.append(
"worktree has uncommitted tracked changes: " + ", ".join(sorted(dirty))
)
# ── AC5/AC6: published heads must agree ──
if not _text(head_sha):
reasons.append("local head could not be observed")
if not _text(remote_head_sha):
reasons.append(
"remote branch head could not be observed; an unpublished branch "
"cannot prove exact-owner renewal"
)
if _text(head_sha) and _text(remote_head_sha) and head_sha != remote_head_sha:
reasons.append(
f"local head {head_sha} does not equal remote head {remote_head_sha}"
)
if pr_number is not None:
if not _text(pr_head_sha):
reasons.append(f"owning PR #{pr_number} head could not be observed")
elif _text(head_sha) and pr_head_sha != head_sha:
reasons.append(
f"owning PR #{pr_number} head {pr_head_sha} does not equal local "
f"head {head_sha}"
)
# ── AC7: nothing else claims this work ──
reasons.extend(
_competing_lock_reasons(
competing_live_locks,
issue_number=issue_number,
branch_name=branch_name,
worktree_path=worktree_path,
)
)
other_branches = [
name
for name in (candidate_branches or ())
if _text(name) and _text(name) != _text(branch_name)
]
if other_branches:
reasons.append(
"other branches already carry this issue marker: "
+ ", ".join(sorted(other_branches))
)
if reasons:
return _result(REFUSED, reasons, **extra)
return _result(
RENEWAL_SANCTIONED,
[
f"exact owner '{recorded_identity}' ({recorded_profile}) proved "
f"ownership of issue #{issue_number} on branch '{branch_name}' from "
f"worktree '{worktree_path}'; local, remote"
+ (f", and PR #{pr_number}" if pr_number is not None else "")
+ f" heads all equal {head_sha}; lease expired at "
f"{prior_expires_at or 'unknown'}"
],
**extra,
)
def owning_pr_renewal_evidence(
assessment: Mapping[str, Any] | None,
) -> dict[str, Any] | None:
"""Server-derived proof of the open PR a sanctioned renewal already owns.
The mirror of ``issue_lock_recovery.owning_pr_recovery_evidence`` (#755) for
the renewal disposition. An exact-owner renewal of a published branch is, by
construction, renewal of work that already has an open PR — so the
duplicate-work gate's linked-open-PR blocker would otherwise discard every
sanctioned renewal, exactly as it once discarded every sanctioned recovery.
Returns ``None`` unless renewal was actually granted and the evidence names
one owning PR whose head agrees with both the local and remote heads the
assessor accepted. Nothing is caller-supplied: every field is copied from
evidence built out of durable lock state plus live git/Gitea observation.
Renewal has no descendant case — it requires the local, remote, and PR heads
to be equal — so there is only one head to report.
"""
if not isinstance(assessment, Mapping):
return None
if assessment.get("outcome") != RENEWAL_SANCTIONED:
return None
if not assessment.get("renewal_sanctioned"):
return None
evidence = assessment.get("evidence") or {}
branch_name = _text(evidence.get("branch_name"))
pr_head = _text(evidence.get("pr_head_sha"))
local_head = _text(evidence.get("head_sha"))
remote_head = _text(evidence.get("remote_head_sha"))
raw_pr_number = evidence.get("pr_number")
if raw_pr_number is None or not branch_name or not pr_head:
return None
# The assessor already required these to agree. Re-check, so a truncated or
# hand-built evidence map can never authorize an exemption.
if pr_head != local_head or pr_head != remote_head:
return None
try:
pr_number = int(raw_pr_number)
issue_number = int(evidence.get("issue_number"))
except (TypeError, ValueError):
return None
return {
"issue_number": issue_number,
"pr_number": pr_number,
"branch_name": branch_name,
"head_sha": pr_head,
"recorded_head": pr_head,
"accepted_head": pr_head,
"head_relation": "equal",
}
def build_renewal_record(
assessment: Mapping[str, Any] | None,
*,
renewed_at: str,
new_expires_at: str,
) -> dict[str, Any]:
"""Durable audit record for a sanctioned renewal (#760 AC9).
Records both sides of the transition — prior PID and expiry, replacement PID
and new expiry — so a renewed lock is never mistakable for an original
claim, and so the evidence the waiver was granted on stays inspectable.
"""
data = dict(assessment or {})
evidence = dict(data.get("evidence") or {})
recorded_claimant = dict(evidence.get("recorded_claimant") or {})
return {
"renewed": bool(data.get("renewal_sanctioned")),
"renewed_at": renewed_at,
"prior_pid": evidence.get("prior_pid"),
"prior_pid_alive": evidence.get("prior_pid_alive"),
"prior_expires_at": evidence.get("prior_expires_at"),
"replacement_pid": evidence.get("replacement_pid"),
"new_expires_at": new_expires_at,
"identity": recorded_claimant.get("username"),
"profile": recorded_claimant.get("profile"),
"branch_name": evidence.get("branch_name"),
"worktree_path": evidence.get("worktree_path"),
"head_sha": evidence.get("head_sha"),
"remote_head_sha": evidence.get("remote_head_sha"),
"pr_head_sha": evidence.get("pr_head_sha"),
"pr_number": evidence.get("pr_number"),
"reason": "expired lease renewed by its exact recorded owner",
"proof": list(data.get("reasons") or []),
}
def format_renewal_refusal(assessment: Mapping[str, Any] | None) -> str:
"""One-line refusal summary for a blocked caller."""
data = dict(assessment or {})
reasons = list(data.get("reasons") or [])
if not reasons:
return "exact-owner lease renewal was not available (no evidence recorded)"
return "exact-owner lease renewal refused: " + "; ".join(reasons)
+23 -1
View File
@@ -168,6 +168,7 @@ def bind_session_lock(
lock_dir: str | None = None, lock_dir: str | None = None,
*, *,
expected_generation: int | None = None, expected_generation: int | None = None,
renewal_sanctioned: bool = False,
) -> str: ) -> str:
"""Persist a keyed lock and bind it to the current process session. """Persist a keyed lock and bind it to the current process session.
@@ -220,6 +221,7 @@ def bind_session_lock(
issue_number=issue_number, issue_number=issue_number,
branch_name=str(record.get("branch_name") or ""), branch_name=str(record.get("branch_name") or ""),
worktree_path=str(record.get("worktree_path") or ""), worktree_path=str(record.get("worktree_path") or ""),
renewal_sanctioned=renewal_sanctioned,
) )
if lease_block: if lease_block:
raise RuntimeError(lease_block) raise RuntimeError(lease_block)
@@ -483,9 +485,19 @@ def assess_same_issue_lease_conflict(
branch_name: str, branch_name: str,
worktree_path: str, worktree_path: str,
operation_type: str = AUTHOR_ISSUE_WORK_LEASE, operation_type: str = AUTHOR_ISSUE_WORK_LEASE,
renewal_sanctioned: bool = False,
now: datetime | None = None, now: datetime | None = None,
) -> str | None: ) -> str | None:
"""Return a fail-closed error when a competing live lease blocks acquisition.""" """Return a fail-closed error when a competing live lease blocks acquisition.
``renewal_sanctioned`` is set only when
``issue_lock_renewal.assess_exact_owner_lease_renewal`` has already proven,
from the durable lock plus live server-side observation, that this session
is the exact recorded owner of an *expired* lease (#760). It is never a
caller-supplied parameter of any MCP tool (#760 AC14): the server computes
it and passes it down. Left False, every pre-existing disposition is
unchanged.
"""
if not existing_lock: if not existing_lock:
return None return None
@@ -506,6 +518,16 @@ def assess_same_issue_lease_conflict(
and _same_realpath(str(existing_worktree or ""), worktree_path) and _same_realpath(str(existing_worktree or ""), worktree_path)
) )
if is_lease_expired(existing_lock, now=now): if is_lease_expired(existing_lock, now=now):
# #760 AC1/AC2: exact-owner renewal is a different disposition from
# foreign takeover and is evaluated first. Before this, both branches
# below returned unconditionally, so the same_owner allowance further
# down was unreachable for every expired lease — an owner could never
# renew its own lock once the wall clock passed, no matter how complete
# its ownership evidence. Requires BOTH the locally recomputed
# same_owner match and the server-proven renewal waiver; either alone is
# insufficient.
if same_owner and renewal_sanctioned:
return None
reclaim = assess_expired_lock_reclaim(existing_lock, now=now) reclaim = assess_expired_lock_reclaim(existing_lock, now=now)
if reclaim.get("reclaim_allowed"): if reclaim.get("reclaim_allowed"):
# #601: expired + dead pid / missing worktree may be reclaimed # #601: expired + dead pid / missing worktree may be reclaimed
+25 -3
View File
@@ -285,6 +285,7 @@ def assess_issue_lock_worktree(
base_branch: str | None = None, base_branch: str | None = None,
base_branches: frozenset[str] | None = None, base_branches: frozenset[str] | None = None,
recovery_sanctioned: bool = False, recovery_sanctioned: bool = False,
renewal_sanctioned: bool = False,
) -> dict: ) -> dict:
"""Fail closed when lock preconditions are not met on the declared worktree. """Fail closed when lock preconditions are not met on the declared worktree.
@@ -296,6 +297,19 @@ def assess_issue_lock_worktree(
by construction and could never satisfy it. Every other precondition — by construction and could never satisfy it. Every other precondition —
notably worktree cleanliness — still applies unchanged, and brand-new issue notably worktree cleanliness — still applies unchanged, and brand-new issue
claims keep the full base-equivalence requirement. claims keep the full base-equivalence requirement.
``renewal_sanctioned`` waives base-equivalence on exactly the same grounds
for the other proven-ownership case (#760): ``issue_lock_renewal`` has shown
that an *expired* lease is being renewed by its exact recorded owner — same
remote, org, repo, issue, operation, branch, realpath-normalized worktree,
claimant username and profile — with the local head matching the remote head
and any owning PR head. Such a branch carries committed work for the same
reason a recovered one does, so it can never be base-equivalent either.
Both waivers relax this one requirement and nothing else. Neither is
caller-supplied: each is computed server-side from durable lock state plus
live observation. With both False every precondition applies exactly as
before.
""" """
bases = base_branches or BASE_BRANCHES bases = base_branches or BASE_BRANCHES
reasons: list[str] = [] reasons: list[str] = []
@@ -314,9 +328,12 @@ def assess_issue_lock_worktree(
f"(dirty files: {', '.join(dirty_files)})" f"(dirty files: {', '.join(dirty_files)})"
) )
if recovery_sanctioned: if recovery_sanctioned or renewal_sanctioned:
# Base-equivalence intentionally not evaluated: ownership was proven # Base-equivalence intentionally not evaluated: ownership was proven
# against the durable lock record instead (#753). # against the durable lock record instead — by dead-session recovery
# (#753) or by exact-owner renewal of an expired lease (#760). Every
# other precondition above and below still applies; cleanliness in
# particular is checked before this branch and is never waived.
pass pass
elif base_equivalent is False: elif base_equivalent is False:
reasons.append( reasons.append(
@@ -347,6 +364,7 @@ def assess_issue_lock_worktree(
base_branch=base_branch, base_branch=base_branch,
base_equivalent=base_equivalent, base_equivalent=base_equivalent,
recovery_sanctioned=recovery_sanctioned, recovery_sanctioned=recovery_sanctioned,
renewal_sanctioned=renewal_sanctioned,
) )
@@ -406,6 +424,7 @@ def _assessment(
base_branch: str | None = None, base_branch: str | None = None,
base_equivalent: bool | None = None, base_equivalent: bool | None = None,
recovery_sanctioned: bool = False, recovery_sanctioned: bool = False,
renewal_sanctioned: bool = False,
) -> dict: ) -> dict:
return { return {
"proven": proven, "proven": proven,
@@ -418,7 +437,10 @@ def _assessment(
"base_branch": base_branch, "base_branch": base_branch,
"base_equivalent": base_equivalent, "base_equivalent": base_equivalent,
"recovery_sanctioned": recovery_sanctioned, "recovery_sanctioned": recovery_sanctioned,
"base_equivalence_waived": bool(recovery_sanctioned), "renewal_sanctioned": renewal_sanctioned,
# Either proven-ownership waiver relaxes base-equivalence; the two are
# reported separately so an audit can tell which one applied.
"base_equivalence_waived": bool(recovery_sanctioned or renewal_sanctioned),
} }
+213 -17
View File
@@ -24,12 +24,50 @@ from __future__ import annotations
import os import os
import subprocess import subprocess
import time
# Live-remote head cache: the parity gate runs on every mutation and every
# runtime-context read, so the ``git ls-remote`` result is cached briefly to
# avoid a network round-trip per call (#610). Keyed by (root, remote, branch).
_REMOTE_HEAD_CACHE: dict[tuple[str, str, str], tuple[float, str | None]] = {}
_REMOTE_HEAD_TTL = 60.0
# When True, ``read_remote_master_head`` never performs ``git ls-remote`` unless
# ``GITEA_TEST_LIVE_REMOTE_HEAD`` is set. Conftest enables this suite-wide so
# feature worktrees (whose HEAD differs from live master) cannot flip legacy
# runtime-context assertions to live_stale, and so unit tests never depend on
# a live network (PR #788 F1/F2 / issue #610). Module-level (not env-only) so
# ``patch.dict(os.environ, …, clear=True)`` cannot re-enable the probe.
_HERMETIC_TEST_MODE: bool = False
def _clear_remote_head_cache() -> None:
"""Reset the live-remote head cache (test isolation / forced refresh)."""
_REMOTE_HEAD_CACHE.clear()
def set_hermetic_test_mode(enabled: bool) -> None:
"""Enable or disable suite-wide hermetic live-remote reads (tests only)."""
global _HERMETIC_TEST_MODE
_HERMETIC_TEST_MODE = bool(enabled)
_clear_remote_head_cache()
def hermetic_test_mode() -> bool:
"""Return whether hermetic live-remote reads are active."""
return bool(_HERMETIC_TEST_MODE)
# Environment escape hatches (ops + tests): # Environment escape hatches (ops + tests):
# GITEA_MCP_DISABLE_PARITY_GATE -> disable enforcement entirely (fail open). # GITEA_MCP_DISABLE_PARITY_GATE -> disable enforcement entirely (fail open).
# GITEA_TEST_CURRENT_HEAD -> force the "current" HEAD read, for tests. # GITEA_TEST_CURRENT_HEAD -> force the "current" HEAD read, for tests.
ENV_DISABLE = "GITEA_MCP_DISABLE_PARITY_GATE" ENV_DISABLE = "GITEA_MCP_DISABLE_PARITY_GATE"
ENV_TEST_CURRENT_HEAD = "GITEA_TEST_CURRENT_HEAD" ENV_TEST_CURRENT_HEAD = "GITEA_TEST_CURRENT_HEAD"
# GITEA_TEST_LIVE_REMOTE_HEAD -> force the live remote master read, for tests.
ENV_TEST_LIVE_REMOTE_HEAD = "GITEA_TEST_LIVE_REMOTE_HEAD"
# GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE -> opt a single test into a real ls-remote
# even when hermetic mode is on (rare; prefer ENV_TEST_LIVE_REMOTE_HEAD).
ENV_TEST_ALLOW_LIVE_REMOTE_PROBE = "GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE"
def read_git_head(root: str) -> str | None: def read_git_head(root: str) -> str | None:
@@ -58,6 +96,75 @@ def read_git_head(root: str) -> str | None:
return (res.stdout or "").strip() or None return (res.stdout or "").strip() or None
def read_remote_master_head(
root: str,
remote: str = "origin",
branch: str = "master",
ttl: float = _REMOTE_HEAD_TTL,
) -> str | None:
"""Return the live remote ``branch`` commit SHA, or ``None`` (#610).
Resolves the *live* target commit via ``git ls-remote`` so parity can tell
a daemon that is behind the live remote master apart from one whose local
checkout simply hasn't been pulled. ``None`` means the live head could not
be resolved (offline, no such remote, git unavailable, error) -- callers
must treat unknown live state as *not mutation-safe* while never blocking
read-only diagnostics. A ``GITEA_TEST_LIVE_REMOTE_HEAD`` override takes
precedence so the wiring can be exercised deterministically and offline.
The result is cached for *ttl* seconds per (root, remote, branch) so the
gate does not run a network probe on every mutation/read (``ttl=0`` forces
a live probe). Both hits and ``None`` misses are cached to bound offline
latency; the env override bypasses the cache and the subprocess entirely.
Under suite hermetic mode (``set_hermetic_test_mode(True)``, set by
conftest) a missing override returns ``None`` without network I/O so
feature-worktree test runs cannot observe live_stale against real master
(PR #788 F1) and unit tests stay offline (F2). Opt out with an explicit
``GITEA_TEST_LIVE_REMOTE_HEAD`` pin or ``GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE``.
"""
forced = os.environ.get(ENV_TEST_LIVE_REMOTE_HEAD)
if forced is not None:
return forced.strip() or None
if _HERMETIC_TEST_MODE and not (
os.environ.get(ENV_TEST_ALLOW_LIVE_REMOTE_PROBE) or ""
).strip():
# Hermetic default: live head unknown. live_stale stays False;
# mutation_safe is False when live is unknown (documented #610 note).
return None
# Defense in depth: even without the module flag, never probe while pytest
# is running unless the test opted into a real probe or set an override.
if (os.environ.get("PYTEST_CURRENT_TEST") or "").strip() and not (
os.environ.get(ENV_TEST_ALLOW_LIVE_REMOTE_PROBE) or ""
).strip():
return None
if not root:
return None
key = (root, remote, branch)
now = time.monotonic()
if ttl > 0:
cached = _REMOTE_HEAD_CACHE.get(key)
if cached is not None and (now - cached[0]) < ttl:
return cached[1]
sha: str | None = None
try:
res = subprocess.run(
["git", "-C", root, "ls-remote", remote, f"refs/heads/{branch}"],
capture_output=True,
text=True,
check=False,
timeout=5,
)
if res.returncode == 0:
lines = (res.stdout or "").strip().splitlines()
if lines:
sha = lines[0].split("\t", 1)[0].split()[0].strip() or None
except Exception:
sha = None
_REMOTE_HEAD_CACHE[key] = (now, sha)
return sha
def capture_startup_parity(root: str, head: str | None = None) -> dict: def capture_startup_parity(root: str, head: str | None = None) -> dict:
"""Capture the process source-tree baseline once at server startup. """Capture the process source-tree baseline once at server startup.
@@ -72,18 +179,38 @@ def _short(sha: str | None) -> str:
return sha[:12] if sha else "unknown" return sha[:12] if sha else "unknown"
def assess_master_parity(startup: dict | None, current_head: str | None) -> dict: def assess_master_parity(
startup: dict | None,
current_head: str | None,
live_remote_head: str | None = None,
) -> dict:
"""Compare the startup baseline against the current on-disk ``HEAD``. """Compare the startup baseline against the current on-disk ``HEAD``.
Pure: both HEADs are supplied by the caller. Returns a structured result: Pure: all HEADs are supplied by the caller. Returns a structured result:
- ``in_parity`` -- server code matches the on-disk master (or parity - ``in_parity`` -- server code matches the on-disk master (or parity
could not be determined, which is not treated as stale). could not be determined, which is not treated as stale).
- ``stale`` -- the on-disk master has definitively advanced past the - ``stale`` -- the on-disk master has definitively advanced past the
running process. running process.
- ``restart_required`` -- alias of ``stale``; the recovery action. - ``restart_required`` -- ``stale`` or ``live_stale``; the recovery action.
- ``determinable`` -- whether both HEADs were known well enough to compare. - ``determinable`` -- whether both local HEADs were known well enough to
compare.
- ``startup_head`` / ``current_head`` / ``reasons``. - ``startup_head`` / ``current_head`` / ``reasons``.
#610 adds live-remote awareness so a daemon that is stale relative to the
*live* remote master cannot report a mutation-safe result even when the
local checkout HEAD still matches the daemon's startup commit:
- ``daemon_start_head`` -- the commit the running process started at
(alias of ``startup_head``, named for clarity in reports).
- ``local_head`` -- the on-disk checkout HEAD (alias of ``current_head``).
- ``live_remote_head`` -- the live remote target commit, or ``None`` when it
could not be fetched.
- ``live_known`` -- whether the live remote target was resolved.
- ``live_stale`` -- the live remote master has advanced past the running
process (daemon is behind live master) even if local parity is green.
- ``mutation_safe`` -- the daemon code, local checkout, and live remote
target all agree; the only state in which a mutation may rely on parity.
""" """
startup_head = (startup or {}).get("startup_head") startup_head = (startup or {}).get("startup_head")
reasons: list[str] = [] reasons: list[str] = []
@@ -91,32 +218,56 @@ def assess_master_parity(startup: dict | None, current_head: str | None) -> dict
if startup_head is None: if startup_head is None:
reasons.append( reasons.append(
"startup commit was not captured; code parity cannot be enforced") "startup commit was not captured; code parity cannot be enforced")
return _result(True, False, False, startup_head, current_head, reasons) return _result(True, False, False, startup_head, current_head,
live_remote_head, False, reasons)
if current_head is None: if current_head is None:
reasons.append( reasons.append(
"current workspace HEAD could not be read; code parity cannot be " "current workspace HEAD could not be read; code parity cannot be "
"enforced") "enforced")
return _result(True, False, False, startup_head, current_head, reasons) return _result(True, False, False, startup_head, current_head,
live_remote_head, False, reasons)
if startup_head == current_head: local_in_parity = startup_head == current_head
return _result(True, False, True, startup_head, current_head, reasons) local_stale = not local_in_parity
if local_stale:
reasons.append(
f"MCP server started at commit {_short(startup_head)} but the "
f"workspace master is now {_short(current_head)}; restart the "
f"server to load the current capability gates")
reasons.append( live_known = live_remote_head is not None
f"MCP server started at commit {_short(startup_head)} but the workspace " live_stale = live_known and live_remote_head != startup_head
f"master is now {_short(current_head)}; restart the server to load the " if live_stale:
f"current capability gates") reasons.append(
return _result(False, True, True, startup_head, current_head, reasons) f"live remote master is {_short(live_remote_head)} but the MCP "
f"server started at {_short(startup_head)}; the daemon is stale "
f"relative to live master -- restart/reconnect before mutating")
return _result(
local_in_parity, local_stale, True, startup_head, current_head,
live_remote_head, live_stale, reasons)
def _result(in_parity, stale, determinable, startup_head, current_head, reasons): def _result(in_parity, stale, determinable, startup_head, current_head,
live_remote_head, live_stale, reasons):
live_known = live_remote_head is not None
mutation_safe = (
determinable and in_parity and live_known and not live_stale)
return { return {
"in_parity": in_parity, "in_parity": in_parity,
"stale": stale, "stale": stale,
"restart_required": stale, "restart_required": stale or live_stale,
"determinable": determinable, "determinable": determinable,
"startup_head": startup_head, "startup_head": startup_head,
"current_head": current_head, "current_head": current_head,
# #610 distinguished signals:
"daemon_start_head": startup_head,
"local_head": current_head,
"live_remote_head": live_remote_head,
"live_known": live_known,
"live_stale": live_stale,
"mutation_safe": mutation_safe,
"reasons": list(reasons), "reasons": list(reasons),
} }
@@ -130,11 +281,13 @@ def parity_block_reasons(assessment: dict) -> list[str]:
"""Block reasons for a mutation gate (empty when the mutation may proceed). """Block reasons for a mutation gate (empty when the mutation may proceed).
A disabled gate or an in-parity / non-determinable assessment yields no A disabled gate or an in-parity / non-determinable assessment yields no
reasons; only a definitively stale server blocks. reasons. A definitively stale server blocks, and (#610) a daemon that is
stale relative to the *live* remote master blocks even when the local
checkout HEAD still matches the daemon's startup commit.
""" """
if gate_disabled(): if gate_disabled():
return [] return []
if assessment.get("stale"): if assessment.get("stale") or assessment.get("live_stale"):
return list(assessment.get("reasons") or return list(assessment.get("reasons") or
["server code is stale relative to master (fail closed)"]) ["server code is stale relative to master (fail closed)"])
return [] return []
@@ -147,6 +300,10 @@ def parity_report(assessment: dict) -> dict:
"restart_required": True, "restart_required": True,
"startup_head": assessment.get("startup_head"), "startup_head": assessment.get("startup_head"),
"current_head": assessment.get("current_head"), "current_head": assessment.get("current_head"),
# #610: name the live remote target so the report distinguishes a
# local-code stale from a daemon-behind-live-master stale.
"live_remote_head": assessment.get("live_remote_head"),
"live_stale": bool(assessment.get("live_stale")),
"reasons": list(assessment.get("reasons") or []), "reasons": list(assessment.get("reasons") or []),
"recovery": [ "recovery": [
"The running MCP server is executing code older than the current " "The running MCP server is executing code older than the current "
@@ -157,6 +314,45 @@ def parity_report(assessment: dict) -> dict:
} }
def parity_resolver_disagreement(
assessment: dict,
resolver_restart_required: bool,
) -> dict | None:
"""Typed blocker when the resolver requires restart but parity looks green.
The capability resolver (``gitea_resolve_task_capability``) detects stale
runtime authoritatively for mutation safety (#610). When it requires a
restart, local-only parity must never override it: this returns a typed,
fail-closed blocker that names the resolver as authoritative. Returns
``None`` when the resolver does not require a restart.
"""
if not resolver_restart_required:
return None
parity_optimistic = bool(assessment.get("in_parity")) and not (
assessment.get("stale") or assessment.get("live_stale"))
return {
"kind": "parity_resolver_disagreement",
"restart_required": True,
"resolver_authoritative": True,
"parity_optimistic": parity_optimistic,
"daemon_start_head": assessment.get("daemon_start_head"),
"local_head": assessment.get("local_head"),
"live_remote_head": assessment.get("live_remote_head"),
"reasons": [
"The capability resolver requires a restart/reconnect (stale "
"runtime) but master-parity reported local code as in-parity. "
"The resolver is authoritative for mutation safety; do not mutate "
"on local parity alone. Restart/reconnect the Gitea MCP server "
"and re-verify before mutating.",
],
"recovery": [
"Trust the resolver: treat this session as stale.",
"Restart or /mcp reconnect the Gitea MCP namespace so it reloads "
"current master and live target state, then re-run preflight.",
],
}
def format_parity(assessment: dict) -> str: def format_parity(assessment: dict) -> str:
"""One-line human summary for logs / runtime context.""" """One-line human summary for logs / runtime context."""
if assessment.get("stale"): if assessment.get("stale"):
+29
View File
@@ -167,6 +167,35 @@ def _reset_mutation_authority(monkeypatch):
import pytest import pytest
@pytest.fixture(autouse=True)
def _hermetic_live_remote_master_head():
"""#610 / PR #788 F1/F2: keep live-remote parity reads offline in tests.
``read_remote_master_head`` would otherwise ``git ls-remote`` whenever
``GITEA_TEST_LIVE_REMOTE_HEAD`` is unset. Feature worktrees under
``branches/`` always differ from live master, so legacy suites that assert
runtime-context ``safe_next_action`` flip to live_stale. Module-level
hermetic mode survives ``patch.dict(os.environ, …, clear=True)``.
Tests that exercise the real probe path call
``master_parity_gate.set_hermetic_test_mode(False)`` and/or set
``GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE``.
"""
try:
import master_parity_gate as _mpg
_mpg.set_hermetic_test_mode(True)
except Exception:
_mpg = None
try:
yield
finally:
if _mpg is not None:
try:
_mpg.set_hermetic_test_mode(False)
except Exception:
pass
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def _deterministic_workspace_remotes(): def _deterministic_workspace_remotes():
try: try:
@@ -0,0 +1,447 @@
"""Exact-owner renewal of an expired author issue lease (#760).
Covers the renewal disposition that lets the exact recorded owner re-acquire
its own lock after the wall-clock lease expires — including while the recording
MCP daemon PID is still alive — plus every rejection condition that must keep
failing closed, and the pre-existing dead-PID and live-foreign dispositions
that must remain untouched.
"""
import inspect
import os
import subprocess
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import issue_lock_renewal # noqa: E402
import issue_lock_store # noqa: E402
ISSUE = 5150
BRANCH = f"fix/issue-{ISSUE}-demo"
WORKTREE = "/scratch/wt-5150"
HEAD = "c" * 40
OTHER_SHA = "d" * 40
IDENTITY = "example-user"
PROFILE = "example-author"
REMOTE = "prgs"
ORG = "ExampleOrg"
REPO = "ExampleRepo"
def dead_pid() -> int:
"""A PID that has certainly exited (spawned, then reaped)."""
proc = subprocess.Popen([sys.executable, "-c", "pass"])
proc.wait()
return proc.pid
def past_ts(hours: int = 1) -> str:
return (
(datetime.now(timezone.utc) - timedelta(hours=hours))
.isoformat()
.replace("+00:00", "Z")
)
def future_ts(hours: int = 4) -> str:
return (
(datetime.now(timezone.utc) + timedelta(hours=hours))
.isoformat()
.replace("+00:00", "Z")
)
def make_lock(*, expires_at: str | None = None, pid: int | None = None, **overrides):
"""An expired lock owned by a still-alive daemon PID — the #760 condition."""
lock = {
"issue_number": ISSUE,
"branch_name": BRANCH,
"worktree_path": WORKTREE,
"remote": REMOTE,
"org": ORG,
"repo": REPO,
# os.getpid() is unambiguously alive: the whole point of #760 is that
# daemon liveness is not evidence of an active author task.
"session_pid": os.getpid() if pid is None else pid,
"lock_generation": 3,
"work_lease": {
"operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
"issue_number": ISSUE,
"branch": BRANCH,
"worktree_path": WORKTREE,
"claimant": {"username": IDENTITY, "profile": PROFILE},
"created_at": past_ts(5),
"expires_at": expires_at or past_ts(),
},
}
lease_overrides = overrides.pop("work_lease", None)
if lease_overrides:
lock["work_lease"].update(lease_overrides)
lock.update(overrides)
return lock
def assess(lock=None, **overrides):
"""Run the assessor with all-passing evidence unless overridden."""
kwargs = {
"issue_number": ISSUE,
"branch_name": BRANCH,
"worktree_path": WORKTREE,
"remote": REMOTE,
"org": ORG,
"repo": REPO,
"identity": IDENTITY,
"profile": PROFILE,
"current_branch": BRANCH,
"porcelain_status": "",
"worktree_exists": True,
"head_sha": HEAD,
"remote_head_sha": HEAD,
"pr_head_sha": None,
"pr_number": None,
"competing_live_locks": [],
"candidate_branches": [BRANCH],
"current_pid": 4242,
}
kwargs.update(overrides)
return issue_lock_renewal.assess_exact_owner_lease_renewal(
make_lock() if lock is None else lock, **kwargs
)
class ExactOwnerRenewalGranted(unittest.TestCase):
"""AC1/AC3-AC7: the positive path."""
def test_expired_lease_alive_pid_exact_owner_is_renewable(self):
result = assess()
self.assertEqual(result["outcome"], issue_lock_renewal.RENEWAL_SANCTIONED)
self.assertTrue(result["renewal_sanctioned"])
self.assertTrue(result["is_candidate"])
def test_renewal_holds_when_owning_pr_head_matches(self):
result = assess(pr_number=999, pr_head_sha=HEAD)
self.assertTrue(result["renewal_sanctioned"])
def test_evidence_records_both_sides_of_the_transition(self):
result = assess()
evidence = result["evidence"]
self.assertEqual(evidence["prior_pid"], os.getpid())
self.assertTrue(evidence["prior_pid_alive"])
self.assertEqual(evidence["replacement_pid"], 4242)
self.assertTrue(evidence["prior_expires_at"])
class ExactOwnerRenewalRefused(unittest.TestCase):
"""AC3-AC8: every near-match must fail closed, one reason at a time."""
def _refused(self, **overrides):
result = assess(**overrides)
self.assertEqual(result["outcome"], issue_lock_renewal.REFUSED)
self.assertFalse(result["renewal_sanctioned"])
self.assertTrue(result["reasons"])
return result
def test_different_branch_refused(self):
result = self._refused(branch_name=f"fix/issue-{ISSUE}-other")
self.assertTrue(any("branch" in r for r in result["reasons"]))
def test_different_worktree_refused(self):
result = self._refused(worktree_path="/scratch/somewhere-else")
self.assertTrue(any("worktree" in r for r in result["reasons"]))
def test_different_claimant_refused(self):
result = self._refused(identity="someone-else")
self.assertTrue(any("claimant" in r for r in result["reasons"]))
def test_different_profile_refused(self):
result = self._refused(profile="other-author")
self.assertTrue(any("profile" in r for r in result["reasons"]))
def test_different_remote_org_or_repo_refused(self):
self._refused(remote="dadeschools")
self._refused(org="OtherOrg")
self._refused(repo="OtherRepo")
def test_dirty_worktree_refused(self):
result = self._refused(porcelain_status=" M gitea_mcp_server.py\n")
self.assertTrue(any("uncommitted" in r for r in result["reasons"]))
def test_missing_worktree_refused(self):
result = self._refused(worktree_exists=False)
self.assertTrue(any("does not exist" in r for r in result["reasons"]))
def test_worktree_on_wrong_branch_refused(self):
self._refused(current_branch="master")
def test_local_and_remote_head_mismatch_refused(self):
result = self._refused(remote_head_sha=OTHER_SHA)
self.assertTrue(
any("does not equal remote head" in r for r in result["reasons"])
)
def test_unpublished_branch_refused(self):
result = self._refused(remote_head_sha=None)
self.assertTrue(any("remote branch head" in r for r in result["reasons"]))
def test_pr_head_mismatch_refused(self):
result = self._refused(pr_number=999, pr_head_sha=OTHER_SHA)
self.assertTrue(any("does not equal local" in r for r in result["reasons"]))
def test_unobservable_pr_head_refused(self):
self._refused(pr_number=999, pr_head_sha=None)
def test_competing_live_lock_on_same_issue_refused(self):
result = self._refused(
competing_live_locks=[
{"issue_number": ISSUE, "branch_name": BRANCH, "pid": 777}
]
)
self.assertTrue(any("live lock" in r for r in result["reasons"]))
def test_competing_live_lock_holding_the_branch_refused(self):
self._refused(
competing_live_locks=[
{"issue_number": 111, "branch_name": BRANCH, "worktree_path": ""}
]
)
def test_competing_branch_claim_refused(self):
result = self._refused(candidate_branches=[BRANCH, f"feat/issue-{ISSUE}-rival"])
self.assertTrue(any("issue marker" in r for r in result["reasons"]))
def test_malformed_durable_lock_refused(self):
lock = make_lock()
lock["worktree_path"] = ""
result = assess(lock)
self.assertEqual(result["outcome"], issue_lock_renewal.REFUSED)
def test_lock_without_recorded_claimant_refused(self):
lock = make_lock()
lock["work_lease"]["claimant"] = {}
result = assess(lock)
self.assertEqual(result["outcome"], issue_lock_renewal.REFUSED)
class NotARenewalCandidate(unittest.TestCase):
"""AC12 and scope: situations renewal must decline to judge at all."""
def test_live_foreign_lease_is_never_a_candidate(self):
lock = make_lock(expires_at=future_ts())
result = assess(lock, identity="someone-else")
self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
self.assertFalse(result["renewal_sanctioned"])
def test_unexpired_lease_is_never_a_candidate(self):
lock = make_lock(expires_at=future_ts())
result = assess(lock)
self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
def test_dead_pid_under_unexpired_lease_stays_with_753(self):
"""The opposite trigger; #760 must not re-own it."""
lock = make_lock(expires_at=future_ts(), pid=dead_pid())
result = assess(lock)
self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
def test_absent_lock_is_not_a_candidate(self):
result = assess({})
self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
def test_different_issue_is_not_a_candidate(self):
lock = make_lock()
lock["issue_number"] = ISSUE + 1
result = assess(lock)
self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
def test_different_operation_type_is_not_a_candidate(self):
lock = make_lock()
lock["work_lease"]["operation_type"] = "review_pr_work"
result = assess(lock)
self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
class DaemonPidIsNotTaskLiveness(unittest.TestCase):
"""AC16: a live recorded PID is never, by itself, authorization."""
def test_alive_pid_alone_does_not_authorize_renewal(self):
# Every ownership fact except the live PID is wrong.
result = assess(identity="someone-else", branch_name="fix/issue-1-nope")
self.assertEqual(result["outcome"], issue_lock_renewal.REFUSED)
self.assertTrue(result["evidence"]["prior_pid_alive"])
def test_renewal_does_not_require_a_dead_pid(self):
result = assess()
self.assertTrue(result["evidence"]["prior_pid_alive"])
self.assertTrue(result["renewal_sanctioned"])
def test_dead_pid_does_not_block_an_otherwise_exact_owner(self):
lock = make_lock(pid=dead_pid())
result = assess(lock)
self.assertTrue(result["renewal_sanctioned"])
class ConflictGateOrdering(unittest.TestCase):
"""AC2: the same-owner allowance is reachable on an expired lease.
These cases need a worktree that genuinely exists on disk. The #601 reclaim
affordance already permits takeover when the recorded worktree is missing,
so a fictional path would satisfy the gate for the wrong reason and never
exercise the ordering defect this issue is about.
"""
@classmethod
def setUpClass(cls):
cls._tmp = tempfile.TemporaryDirectory()
cls.worktree = cls._tmp.name
@classmethod
def tearDownClass(cls):
cls._tmp.cleanup()
def present_lock(self, **overrides):
return make_lock(worktree_path=self.worktree, **overrides)
def test_expired_same_owner_is_allowed_when_renewal_is_sanctioned(self):
block = issue_lock_store.assess_same_issue_lease_conflict(
self.present_lock(),
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=self.worktree,
renewal_sanctioned=True,
)
self.assertIsNone(block)
def test_expired_same_owner_still_blocks_without_the_waiver(self):
"""Regression for the ordering defect: no waiver, no change in behavior.
Live PID and a present worktree, so the #601 reclaim affordance refuses;
before #760 this was the permanent dead end for an exact owner.
"""
lock = self.present_lock()
self.assertFalse(
issue_lock_store.assess_expired_lock_reclaim(lock)["reclaim_allowed"]
)
block = issue_lock_store.assess_same_issue_lease_conflict(
lock,
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=self.worktree,
)
self.assertIsNotNone(block)
self.assertIn("Recovery review is required", block)
def test_waiver_does_not_unlock_a_different_owner(self):
"""AC11: the waiver is scoped by same_owner, not merely by its own flag."""
block = issue_lock_store.assess_same_issue_lease_conflict(
self.present_lock(),
issue_number=ISSUE,
branch_name=f"fix/issue-{ISSUE}-someone-else",
worktree_path=self.worktree,
renewal_sanctioned=True,
)
self.assertIsNotNone(block)
self.assertIn("Recovery review is required", block)
def test_live_lease_disposition_is_unchanged(self):
"""AC12: a live foreign lease still blocks, waiver or not."""
block = issue_lock_store.assess_same_issue_lease_conflict(
self.present_lock(expires_at=future_ts()),
issue_number=ISSUE,
branch_name=f"fix/issue-{ISSUE}-someone-else",
worktree_path="/scratch/other",
renewal_sanctioned=True,
)
self.assertIsNotNone(block)
self.assertIn("already has an active", block)
def test_dead_pid_reclaim_path_is_unchanged(self):
"""AC11: expired + dead PID still reclaims through the #601 affordance."""
lock = self.present_lock(pid=dead_pid())
reclaim = issue_lock_store.assess_expired_lock_reclaim(lock)
self.assertTrue(reclaim["reclaim_allowed"])
block = issue_lock_store.assess_same_issue_lease_conflict(
lock,
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=self.worktree,
)
self.assertIsNone(block)
class RenewalRecordAndDownstream(unittest.TestCase):
"""AC9/AC10: durable audit trail, and a renewed lock that actually works."""
def test_record_captures_prior_and_replacement_state(self):
assessment = assess()
record = issue_lock_renewal.build_renewal_record(
assessment,
renewed_at="2026-01-01T00:00:00Z",
new_expires_at="2026-01-01T04:00:00Z",
)
self.assertTrue(record["renewed"])
self.assertEqual(record["prior_pid"], os.getpid())
self.assertEqual(record["new_expires_at"], "2026-01-01T04:00:00Z")
self.assertEqual(record["renewed_at"], "2026-01-01T00:00:00Z")
self.assertEqual(record["identity"], IDENTITY)
self.assertEqual(record["profile"], PROFILE)
self.assertTrue(record["prior_expires_at"])
self.assertTrue(record["proof"])
def test_renewed_lock_satisfies_verify_lock_for_mutation(self):
renewed = make_lock(expires_at=future_ts())
renewed["session_pid"] = os.getpid()
renewed["lease_renewal"] = {"renewed": True}
verdict = issue_lock_store.verify_lock_for_mutation(
renewed,
issue_number=ISSUE,
branch_name=BRANCH,
)
self.assertTrue(verdict["proven"])
self.assertFalse(verdict["block"])
def test_refusal_message_names_the_missing_evidence(self):
assessment = assess(porcelain_status=" M gitea_mcp_server.py\n")
message = issue_lock_renewal.format_renewal_refusal(assessment)
self.assertIn("refused", message)
self.assertIn("uncommitted", message)
class NoCallerControlledRenewalFlag(unittest.TestCase):
"""AC14: renewal eligibility is never declarable by a caller."""
def test_lock_issue_tool_exposes_no_renewal_parameter(self):
import gitea_mcp_server
target = gitea_mcp_server.gitea_lock_issue
target = getattr(target, "fn", getattr(target, "__wrapped__", target))
params = set(inspect.signature(target).parameters)
for forbidden in ("renewal_sanctioned", "renew", "allow_renewal", "is_owner"):
self.assertNotIn(forbidden, params)
def test_store_defaults_to_no_waiver(self):
params = inspect.signature(
issue_lock_store.assess_same_issue_lease_conflict
).parameters
self.assertIs(params["renewal_sanctioned"].default, False)
bind_params = inspect.signature(issue_lock_store.bind_session_lock).parameters
self.assertIs(bind_params["renewal_sanctioned"].default, False)
class NoIssueNumberSpecialCasing(unittest.TestCase):
"""AC17: no repository issue or PR number is special-cased."""
def test_module_contains_no_hardcoded_issue_special_cases(self):
source = inspect.getsource(issue_lock_renewal)
code = "\n".join(
line for line in source.splitlines() if not line.strip().startswith("#")
)
for literal in ("757", "759", "760"):
self.assertNotIn(f"== {literal}", code)
self.assertNotIn(f"issue_number == {literal}", code)
if __name__ == "__main__":
unittest.main()
+342
View File
@@ -0,0 +1,342 @@
"""MCP-level exact-owner lease renewal through ``gitea_lock_issue`` (#760).
The unit suite in ``test_issue_760_exact_owner_lease_renewal`` proves the
renewal *disposition*. It cannot prove the disposition survives the rest of the
tool, and it did not: the waiver was computed and then discarded before
``assess_issue_lock_worktree``, so every real renewal still failed on
base-equivalence. A branch being renewed always carries committed work, so it is
never base-equivalent by construction — exactly the argument #753 already makes
for recovery.
These tests drive the public tool end to end against a real git repository and a
real durable lock file, composing every gate in the production order.
"""
from __future__ import annotations
import os
import subprocess
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from mutation_profile_fixture import shared_mutation_env # noqa: E402
import issue_lock_provenance # noqa: E402
import issue_lock_store # noqa: E402
import mcp_server # noqa: E402
ISSUE = 9760
BRANCH = f"fix/issue-{ISSUE}-renewal-mcp"
IDENTITY = "example-user"
PROFILE = "test-author-prgs"
ORG = "Scaled-Tech-Consulting"
REPO = "Gitea-Tools"
def _past_ts(hours: int = 1) -> str:
return (
(datetime.now(timezone.utc) - timedelta(hours=hours))
.isoformat()
.replace("+00:00", "Z")
)
class _RenewalMcpBase(unittest.TestCase):
"""Real git repo + durable expired lock owned by a live PID.
The recorded PID is ``os.getpid()`` — unambiguously alive. That is the whole
point of #760: the PID belongs to the long-lived MCP daemon, so its liveness
says nothing about whether the authoring task still holds the work.
"""
def setUp(self):
self.lock_dir = tempfile.TemporaryDirectory()
self.addCleanup(self.lock_dir.cleanup)
self.repo = tempfile.mkdtemp(prefix="issue760-mcp-")
self.addCleanup(lambda: subprocess.run(["rm", "-rf", self.repo], check=False))
self._init_worktree()
self.remotes = patch.dict(
mcp_server.REMOTES,
{"prgs": {"host": "gitea.prgs.cc", "org": ORG, "repo": REPO}},
)
self.remotes.start()
self.addCleanup(patch.stopall)
mcp_server._IDENTITY_CACHE.clear()
def _git(self, *args):
return subprocess.run(
["git", "-C", self.repo, *args],
capture_output=True,
text=True,
check=True,
)
def _init_worktree(self):
self._git("init", "-q", "-b", "master")
self._git("config", "user.email", "[email protected]")
self._git("config", "user.name", "Test")
with open(os.path.join(self.repo, "seed.txt"), "w") as fh:
fh.write("seed\n")
self._git("add", "seed.txt")
self._git("commit", "-q", "-m", "seed")
self.base_sha = self._git("rev-parse", "HEAD").stdout.strip()
# The branch carries committed work, so it is NOT base-equivalent.
self._git("checkout", "-q", "-b", BRANCH)
with open(os.path.join(self.repo, "work.txt"), "w") as fh:
fh.write("author work\n")
self._git("add", "work.txt")
self._git("commit", "-q", "-m", "author work")
self.head_sha = self._git("rev-parse", "HEAD").stdout.strip()
self.worktree = os.path.realpath(self.repo)
def write_expired_lock(self, **overrides):
path = issue_lock_store.lock_file_path(
remote="prgs",
org=ORG,
repo=REPO,
issue_number=ISSUE,
lock_dir=self.lock_dir.name,
)
claimant = {"username": IDENTITY, "profile": PROFILE}
pid = overrides.pop("session_pid", os.getpid())
overrides.pop("pid", None)
lease_overrides = overrides.pop("work_lease", {})
data = {
"issue_number": ISSUE,
"branch_name": BRANCH,
"remote": "prgs",
"org": ORG,
"repo": REPO,
"worktree_path": self.worktree,
"session_pid": pid,
"pid": pid,
"lock_generation": 3,
"work_lease": {
"operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
"issue_number": ISSUE,
"pr_number": None,
"branch": BRANCH,
"worktree_path": self.worktree,
"claimant": claimant,
"created_at": _past_ts(5),
"last_heartbeat_at": _past_ts(5),
"expires_at": _past_ts(), # already expired
},
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
tool="gitea_lock_issue",
claimant=claimant,
),
}
data["work_lease"].update(lease_overrides)
data.update(overrides)
data["session_pid"] = pid
data["pid"] = pid
data["lock_file_path"] = path
issue_lock_store.save_lock_file(path, data)
return path
def _tool_env(self):
env = shared_mutation_env(
PROFILE,
include_example_repo=True,
GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
)
env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
return env
def _git_state(self, *, porcelain="", branch=BRANCH, head=None):
return {
"current_branch": branch,
"porcelain_status": porcelain,
# The decisive fact: a branch carrying work is never base-equivalent.
"base_equivalent": False,
"head_sha": head or self.head_sha,
"inspected_git_root": self.worktree,
"base_branch": "master",
}
def run_lock_issue(
self,
*,
branch_entries=None,
open_prs=None,
git_state=None,
identity=IDENTITY,
profile=PROFILE,
):
"""Drive the public tool for the published exact-owner renewal shape."""
if branch_entries is None:
branch_entries = [{"name": BRANCH, "commit": {"id": self.head_sha}}]
if open_prs is None:
open_prs = [{"number": 4242, "head": {"ref": BRANCH, "sha": self.head_sha}}]
if git_state is None:
git_state = self._git_state()
env = self._tool_env()
with patch(
"mcp_server.api_get_all", return_value=list(branch_entries)
), patch(
"mcp_server._list_open_pulls", return_value=list(open_prs)
), patch(
"mcp_server.get_auth_header", return_value="token x"
), patch(
"mcp_server._work_lease_claimant",
return_value={"username": identity, "profile": profile},
), patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=git_state,
), patch(
"mcp_server.issue_duplicate_context_fetcher",
side_effect=lambda h, o, r, auth, issue_number: (
list(open_prs),
[b.get("name") for b in branch_entries if isinstance(b, dict)],
{"status": "not_claimed"},
),
), patch.dict(os.environ, env, clear=True):
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
return mcp_server.gitea_lock_issue(
issue_number=ISSUE,
branch_name=BRANCH,
remote="prgs",
worktree_path=self.worktree,
)
class TestRenewalReachableThroughTool(_RenewalMcpBase):
"""F1: the sanctioned renewal must survive every downstream gate."""
def test_expired_lease_live_pid_exact_owner_renews_through_the_tool(self):
prior = issue_lock_store.read_lock_file(self.write_expired_lock())
self.assertTrue(issue_lock_store.is_lease_expired(prior))
self.assertTrue(issue_lock_store.is_process_alive(prior["session_pid"]))
result = self.run_lock_issue()
self.assertTrue(result["success"], result)
self.assertEqual(result["issue_number"], ISSUE)
self.assertEqual(result["branch_name"], BRANCH)
# The renewal is reported natively, so no lock-file inspection is needed.
self.assertIn("lease_renewal", result)
self.assertTrue(result["lease_renewal"]["renewed"])
self.assertIn("Renewed the expired", result["message"])
def test_renewed_lock_records_prior_and_replacement_evidence(self):
prior = issue_lock_store.read_lock_file(self.write_expired_lock())
prior_expiry = prior["work_lease"]["expires_at"]
prior_generation = issue_lock_store.lock_generation(prior)
result = self.run_lock_issue()
written = issue_lock_store.read_lock_file(result["lock_file_path"])
renewal = written["lease_renewal"]
self.assertTrue(renewal["renewed"])
self.assertEqual(renewal["prior_pid"], prior["session_pid"])
self.assertTrue(renewal["prior_pid_alive"])
self.assertEqual(renewal["prior_expires_at"], prior_expiry)
self.assertEqual(renewal["identity"], IDENTITY)
self.assertEqual(renewal["profile"], PROFILE)
self.assertEqual(renewal["head_sha"], self.head_sha)
self.assertTrue(renewal["proof"])
# New expiry is a fresh absolute stamp, later than the one it replaced.
self.assertEqual(renewal["new_expires_at"], written["work_lease"]["expires_at"])
self.assertGreater(renewal["new_expires_at"], prior_expiry)
# Compare-and-swap advanced the generation exactly once.
self.assertEqual(
issue_lock_store.lock_generation(written), prior_generation + 1
)
def test_renewed_lock_is_live_and_satisfies_mutation_ownership(self):
self.write_expired_lock()
result = self.run_lock_issue()
written = issue_lock_store.read_lock_file(result["lock_file_path"])
self.assertTrue(issue_lock_store.assess_lock_freshness(written)["live"])
verdict = issue_lock_store.verify_lock_for_mutation(
written,
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=self.worktree,
)
self.assertTrue(verdict["proven"], verdict)
self.assertFalse(verdict["block"])
def test_recovery_record_is_not_written_for_a_live_owner_renewal(self):
"""#753 recovery must not be claimed when the recorded PID is alive."""
self.write_expired_lock()
result = self.run_lock_issue()
written = issue_lock_store.read_lock_file(result["lock_file_path"])
self.assertNotIn("dead_session_recovery", written)
class TestRenewalWaiverIsNarrow(_RenewalMcpBase):
"""The waiver relaxes base-equivalence and nothing else."""
def test_dirty_worktree_still_blocks_a_would_be_renewal(self):
"""Cleanliness is never waived; the renewal assessor refuses first.
A dirty worktree makes the renewal refuse, so no waiver is issued and
the lease-conflict gate fails closed ahead of the worktree gate. The
refusal names the uncommitted files, so the owner still learns why.
"""
self.write_expired_lock()
with self.assertRaises(Exception) as ctx:
self.run_lock_issue(
git_state=self._git_state(porcelain=" M gitea_mcp_server.py\n")
)
message = str(ctx.exception)
self.assertIn("Recovery review is required before takeover", message)
self.assertIn("worktree has uncommitted tracked changes", message)
self.assertIn("gitea_mcp_server.py", message)
def test_foreign_claimant_cannot_use_the_waiver(self):
"""A near-match owner gets no renewal and no base-equivalence waiver."""
self.write_expired_lock()
with self.assertRaises(Exception) as ctx:
self.run_lock_issue(identity="someone-else")
message = str(ctx.exception)
self.assertIn("Recovery review is required before takeover", message)
# The refusal names the missing ownership evidence (#760 diagnostics).
self.assertIn("does not match active identity", message)
def test_foreign_profile_cannot_use_the_waiver(self):
self.write_expired_lock()
with self.assertRaises(Exception) as ctx:
self.run_lock_issue(profile="other-author")
self.assertIn(
"Recovery review is required before takeover", str(ctx.exception)
)
def test_unpublished_branch_cannot_use_the_waiver(self):
"""No remote head to agree with, so exact-owner renewal is refused."""
self.write_expired_lock()
with self.assertRaises(Exception) as ctx:
self.run_lock_issue(branch_entries=[], open_prs=[])
self.assertIn(
"Recovery review is required before takeover", str(ctx.exception)
)
def test_pr_head_mismatch_cannot_use_the_waiver(self):
self.write_expired_lock()
other = "9" * 40
with self.assertRaises(Exception) as ctx:
self.run_lock_issue(
open_prs=[{"number": 4242, "head": {"ref": BRANCH, "sha": other}}]
)
self.assertIn(
"Recovery review is required before takeover", str(ctx.exception)
)
def test_non_base_equivalent_branch_still_blocks_without_any_waiver(self):
"""No durable lock at all: the ordinary base-equivalence rule applies."""
with self.assertRaises(Exception) as ctx:
self.run_lock_issue()
self.assertIn("must be base-equivalent", str(ctx.exception))
if __name__ == "__main__":
unittest.main()
@@ -871,9 +871,13 @@ class TestAc6McpUnpublishedClaimRecovery(_UnpublishedMcpBase):
save_calls: list[dict] = [] save_calls: list[dict] = []
real_save = mcp_server._save_issue_lock real_save = mcp_server._save_issue_lock
def tracking_save(data, *, expected_generation=None): def tracking_save(data, *, expected_generation=None, renewal_sanctioned=False):
save_calls.append({"expected_generation": expected_generation, "data": dict(data)}) save_calls.append({"expected_generation": expected_generation, "data": dict(data)})
return real_save(data, expected_generation=expected_generation) return real_save(
data,
expected_generation=expected_generation,
renewal_sanctioned=renewal_sanctioned,
)
with patch("mcp_server._save_issue_lock", side_effect=tracking_save): with patch("mcp_server._save_issue_lock", side_effect=tracking_save):
result = self.run_lock_issue() result = self.run_lock_issue()
@@ -926,18 +930,33 @@ class TestAc6McpUnpublishedClaimRecovery(_UnpublishedMcpBase):
real_bind = issue_lock_store.bind_session_lock real_bind = issue_lock_store.bind_session_lock
bind_calls: list[int | None] = [] bind_calls: list[int | None] = []
def racing_bind(data, lock_dir=None, expected_generation=None): # #760 added the renewal waiver keyword; the double forwards it verbatim
# so this race still exercises the real compare-and-swap.
def racing_bind(
data, lock_dir=None, expected_generation=None, renewal_sanctioned=False
):
bind_calls.append(expected_generation) bind_calls.append(expected_generation)
if expected_generation is None: if expected_generation is None:
return real_bind(data, lock_dir=lock_dir, expected_generation=None) return real_bind(
data,
lock_dir=lock_dir,
expected_generation=None,
renewal_sanctioned=renewal_sanctioned,
)
# First concurrent writer wins. # First concurrent writer wins.
if len([c for c in bind_calls if c is not None]) == 1: if len([c for c in bind_calls if c is not None]) == 1:
return real_bind( return real_bind(
data, lock_dir=lock_dir, expected_generation=expected_generation data,
lock_dir=lock_dir,
expected_generation=expected_generation,
renewal_sanctioned=renewal_sanctioned,
) )
# Second concurrent writer still holds the pre-race generation. # Second concurrent writer still holds the pre-race generation.
return real_bind( return real_bind(
data, lock_dir=lock_dir, expected_generation=expected_generation data,
lock_dir=lock_dir,
expected_generation=expected_generation,
renewal_sanctioned=renewal_sanctioned,
) )
# First recovery succeeds and advances generation. # First recovery succeeds and advances generation.
+269
View File
@@ -78,6 +78,95 @@ class TestBlockReasonsAndReport(unittest.TestCase):
self.assertTrue(report["recovery"]) self.assertTrue(report["recovery"])
class TestLiveRemoteParity(unittest.TestCase):
"""#610: parity must account for the live remote master, not just local.
The daemon can be stale relative to the live remote target while the local
checkout HEAD still matches the daemon's startup commit, so local parity
reports green even though a mutation would run against outdated code.
"""
SHA_C = "c" * 40
def test_distinguishes_three_shas(self):
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
self.assertEqual(res["daemon_start_head"], SHA_A)
self.assertEqual(res["local_head"], SHA_A)
self.assertEqual(res["live_remote_head"], SHA_B)
def test_mutation_safe_only_when_all_three_match(self):
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_A)
self.assertTrue(res["mutation_safe"])
self.assertTrue(res["live_known"])
self.assertFalse(res["live_stale"])
def test_live_stale_when_remote_advanced_past_daemon(self):
# Local checkout still matches the daemon start (local parity green),
# but the live remote master has advanced -> daemon is live-stale.
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
self.assertTrue(res["in_parity"]) # local parity still green
self.assertTrue(res["live_stale"])
self.assertFalse(res["mutation_safe"])
self.assertTrue(any("live" in r.lower() for r in res["reasons"]))
def test_live_unknown_is_not_mutation_safe_but_not_stale(self):
# Non-goal: unfetchable live remote must not be treated as stale for
# read-only, but a mutation-safe claim fails closed.
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=None)
self.assertFalse(res["live_known"])
self.assertFalse(res["mutation_safe"])
self.assertFalse(res["live_stale"])
self.assertTrue(res["in_parity"])
def test_default_live_remote_preserves_legacy_shape(self):
# Callers that do not supply a live head keep the pre-#610 behavior:
# in-parity, not live-stale, no live-derived block.
res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
self.assertFalse(res["live_stale"])
self.assertEqual(mp.parity_block_reasons(res), [])
class TestLiveStaleBlockAndReport(unittest.TestCase):
"""#610: live-staleness must block mutations and surface a typed blocker."""
def test_live_stale_produces_block_reasons(self):
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
self.assertTrue(mp.parity_block_reasons(res))
def test_disable_env_suppresses_live_stale_block(self):
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
with patch.dict(os.environ, {mp.ENV_DISABLE: "1"}):
self.assertEqual(mp.parity_block_reasons(res), [])
def test_resolver_disagreement_returns_typed_blocker(self):
# Parity says local-green, resolver says restart required -> disagreement
# is a typed, fail-closed blocker naming the resolver as authoritative.
res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
blocker = mp.parity_resolver_disagreement(res, resolver_restart_required=True)
self.assertIsNotNone(blocker)
self.assertEqual(blocker["kind"], "parity_resolver_disagreement")
self.assertTrue(blocker["restart_required"])
self.assertTrue(blocker["resolver_authoritative"])
def test_no_disagreement_when_resolver_agrees(self):
res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
self.assertIsNone(
mp.parity_resolver_disagreement(res, resolver_restart_required=False))
def test_live_stale_report_names_live_remote(self):
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
report = mp.parity_report(res)
self.assertEqual(report["live_remote_head"], SHA_B)
self.assertTrue(report["restart_required"])
class TestReadGitHead(unittest.TestCase): class TestReadGitHead(unittest.TestCase):
def test_test_override_takes_precedence(self): def test_test_override_takes_precedence(self):
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}): with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
@@ -95,6 +184,149 @@ class TestReadGitHead(unittest.TestCase):
self.assertIsNone(mp.read_git_head("")) self.assertIsNone(mp.read_git_head(""))
class TestReadRemoteMasterHead(unittest.TestCase):
"""#610: live remote master head reader (env-overridable, fails to None)."""
def test_test_override_takes_precedence(self):
with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
self.assertEqual(mp.read_remote_master_head("/nonexistent"), SHA_B)
def test_blank_override_is_none(self):
with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: " "}):
self.assertIsNone(mp.read_remote_master_head("/nonexistent"))
def test_unfetchable_remote_is_none(self):
# No override; a bogus root/remote must fail closed to None, never raise.
env = {k: v for k, v in os.environ.items()
if k != mp.ENV_TEST_LIVE_REMOTE_HEAD}
with patch.dict(os.environ, env, clear=True):
self.assertIsNone(
mp.read_remote_master_head("/nonexistent", remote="nope"))
class TestRemoteHeadCache(unittest.TestCase):
"""#610: live remote reads are cached with a TTL to stay off the network.
The parity gate runs on every mutation and every runtime-context read, so an
unbounded ``git ls-remote`` per call would be a latency/flakiness regression.
"""
def setUp(self):
# These cases intentionally exercise the subprocess/cache path, so they
# opt out of suite-wide hermetic mode (PR #788 F1).
self._saved_hermetic = mp.hermetic_test_mode()
mp.set_hermetic_test_mode(False)
mp._clear_remote_head_cache()
env = {
k: v for k, v in os.environ.items()
if k not in (mp.ENV_TEST_LIVE_REMOTE_HEAD,
mp.ENV_TEST_ALLOW_LIVE_REMOTE_PROBE,
"PYTEST_CURRENT_TEST")
}
# Allow the probe path under hermetic defenses while still mocking
# subprocess so no real network call runs.
env[mp.ENV_TEST_ALLOW_LIVE_REMOTE_PROBE] = "1"
self._env = patch.dict(os.environ, env, clear=True)
self._env.start()
self.addCleanup(self._env.stop)
self.addCleanup(mp._clear_remote_head_cache)
self.addCleanup(
lambda: mp.set_hermetic_test_mode(self._saved_hermetic)
)
def _fake_run(self, sha):
class _R:
returncode = 0
stdout = f"{sha}\trefs/heads/master\n"
calls = {"n": 0}
def run(*args, **kwargs):
calls["n"] += 1
return _R()
return run, calls
def test_second_call_within_ttl_uses_cache(self):
run, calls = self._fake_run(SHA_B)
with patch.object(mp.subprocess, "run", run):
a = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
b = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
self.assertEqual(a, SHA_B)
self.assertEqual(b, SHA_B)
self.assertEqual(calls["n"], 1)
def test_zero_ttl_bypasses_cache(self):
run, calls = self._fake_run(SHA_B)
with patch.object(mp.subprocess, "run", run):
mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
self.assertEqual(calls["n"], 2)
def test_env_override_never_touches_subprocess(self):
run, calls = self._fake_run(SHA_B)
with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
with patch.object(mp.subprocess, "run", run):
self.assertEqual(
mp.read_remote_master_head("/repo", remote="prgs"), SHA_A)
self.assertEqual(calls["n"], 0)
class TestHermeticLiveRemoteReads(unittest.TestCase):
"""#610 / PR #788 F1/F2: suite hermetic mode never hits the network."""
def setUp(self):
self._saved = mp.hermetic_test_mode()
mp.set_hermetic_test_mode(True)
mp._clear_remote_head_cache()
self.addCleanup(lambda: mp.set_hermetic_test_mode(self._saved))
self.addCleanup(mp._clear_remote_head_cache)
def test_hermetic_mode_returns_none_without_subprocess(self):
run_calls = {"n": 0}
def boom(*args, **kwargs):
run_calls["n"] += 1
raise AssertionError("ls-remote must not run under hermetic mode")
env = {
k: v for k, v in os.environ.items()
if k not in (mp.ENV_TEST_LIVE_REMOTE_HEAD,
mp.ENV_TEST_ALLOW_LIVE_REMOTE_PROBE)
}
with patch.dict(os.environ, env, clear=True):
with patch.object(mp.subprocess, "run", boom):
self.assertIsNone(
mp.read_remote_master_head("/repo", remote="prgs")
)
self.assertEqual(run_calls["n"], 0)
def test_hermetic_mode_survives_clear_true_env(self):
"""Module flag, not env pin: clear=True cannot re-enable the probe."""
run_calls = {"n": 0}
def boom(*args, **kwargs):
run_calls["n"] += 1
raise AssertionError("ls-remote must not run after clear=True")
with patch.dict(os.environ, {}, clear=True):
with patch.object(mp.subprocess, "run", boom):
self.assertIsNone(mp.read_remote_master_head("/repo"))
self.assertEqual(run_calls["n"], 0)
def test_explicit_override_still_wins_under_hermetic(self):
run_calls = {"n": 0}
def boom(*args, **kwargs):
run_calls["n"] += 1
raise AssertionError("override must bypass subprocess")
with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
with patch.object(mp.subprocess, "run", boom):
self.assertEqual(
mp.read_remote_master_head("/repo"), SHA_B
)
self.assertEqual(run_calls["n"], 0)
class TestServerWiring(unittest.TestCase): class TestServerWiring(unittest.TestCase):
"""Integration with the gate choke point in the server namespace.""" """Integration with the gate choke point in the server namespace."""
@@ -105,6 +337,13 @@ class TestServerWiring(unittest.TestCase):
self._saved = self.srv._STARTUP_PARITY self._saved = self.srv._STARTUP_PARITY
self.srv._STARTUP_PARITY = {"root": self.srv.PROJECT_ROOT, self.srv._STARTUP_PARITY = {"root": self.srv.PROJECT_ROOT,
"startup_head": SHA_A} "startup_head": SHA_A}
# Keep the live-remote read hermetic (no real ls-remote network call):
# default the live master to the daemon start so parity is fully green
# unless a test overrides the live head explicitly (#610).
self._live_patch = patch.dict(
os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A})
self._live_patch.start()
self.addCleanup(self._live_patch.stop)
def tearDown(self): def tearDown(self):
self.srv._STARTUP_PARITY = self._saved self.srv._STARTUP_PARITY = self._saved
@@ -147,6 +386,36 @@ class TestServerWiring(unittest.TestCase):
self.assertTrue(out["in_parity"]) self.assertTrue(out["in_parity"])
self.assertNotIn("report", out) self.assertNotIn("report", out)
# --- #610: live-remote wiring -------------------------------------------
def test_live_stale_blocks_mutation_though_local_green(self):
# Local checkout matches the daemon start (local parity green) but the
# live remote master has advanced -> mutations must fail closed.
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
self.assertEqual(self.srv._master_parity_block("gitea.read"), [])
self.assertTrue(
self.srv._master_parity_block("gitea.pr.create"))
def test_assess_tool_exposes_three_distinct_shas(self):
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
out = self.srv.gitea_assess_master_parity(remote="prgs")
self.assertEqual(out["daemon_start_head"], SHA_A)
self.assertEqual(out["local_head"], SHA_A)
self.assertEqual(out["live_remote_head"], SHA_B)
self.assertTrue(out["live_stale"])
self.assertFalse(out["mutation_safe"])
self.assertIn("report", out)
def test_assess_tool_mutation_safe_when_all_three_match(self):
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
out = self.srv.gitea_assess_master_parity(remote="prgs")
self.assertTrue(out["mutation_safe"])
self.assertFalse(out["live_stale"])
self.assertNotIn("report", out)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+149
View File
@@ -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}"
+499
View File
@@ -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()
+458
View File
@@ -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()
+34
View File
@@ -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.worktree_views import render_worktrees_page
from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
from webui.runtime_views import render_runtime_page from webui.runtime_views import render_runtime_page
from webui.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"}) _READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
_AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"}) _AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
@@ -62,16 +68,43 @@ async def home(_request: Request) -> HTMLResponse:
async def health(_request: Request) -> JSONResponse: 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 bind_host = _request.app.state.webui_bind_host
started_at, uptime_seconds = process_uptime()
return JSONResponse({ return JSONResponse({
"status": "ok", "status": "ok",
"service": "mcp-control-plane-webui", "service": "mcp-control-plane-webui",
"mode": "read-only-mvp", "mode": "read-only-mvp",
"timestamp": datetime.now(timezone.utc).isoformat(), "timestamp": datetime.now(timezone.utc).isoformat(),
"deployment": deployment_snapshot(bind_host=bind_host), "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: async def queue(_request: Request) -> HTMLResponse:
snapshot = load_queue_snapshot() snapshot = load_queue_snapshot()
return HTMLResponse(render_page(title="Queue", body_html=render_queue_page(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=[ routes=[
Route("/", home, methods=["GET"]), Route("/", home, methods=["GET"]),
Route("/health", health, methods=["GET"]), Route("/health", health, methods=["GET"]),
Route(SYSTEM_HEALTH_API_PATH, api_system_health, methods=["GET"]),
Route("/queue", queue, methods=["GET"]), Route("/queue", queue, methods=["GET"]),
Route("/api/queue", api_queue, methods=["GET"]), Route("/api/queue", api_queue, methods=["GET"]),
Route("/projects", projects, methods=["GET"]), Route("/projects", projects, methods=["GET"]),
+57
View File
@@ -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": []
}
+1 -33
View File
@@ -8,27 +8,7 @@ from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
_FORBIDDEN_EXACT_KEYS = frozenset({ from webui.registry_safety import reject_credential_keys as _reject_credential_keys
"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)
)
_REQUIRED_PROJECT_FIELDS = ( _REQUIRED_PROJECT_FIELDS = (
"id", "id",
@@ -79,18 +59,6 @@ def default_registry_path() -> Path:
return (Path(__file__).resolve().parent / "data" / "projects.registry.json").resolve() 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, ...]: def _parse_onboarding(raw: list[dict[str, Any]] | None) -> tuple[OnboardingStep, ...]:
if not raw: if not raw:
return () return ()
+52
View File
@@ -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)
+682
View File
@@ -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),
}
+647
View File
@@ -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)