Compare commits

...
Author SHA1 Message Date
sysadmin f80e3b33b0 Merge remote-tracking branch 'prgs/master' into feat/issue-639-webui-system-health-dashboard 2026-07-23 21:14:50 -04:00
sysadmin 9301739910 Merge pull request 'feat(webui): workflow-event and conversation timeline model (Closes #637)' (#849) from feat/issue-637-timeline-model into master 2026-07-23 19:33:22 -05:00
sysadminandClaude Opus 4.8 edd5f813b2 Merge master into feat/issue-639-webui-system-health-dashboard
Resolve the #638 shell landing against the #639 dashboard:

- webui/layout.py: drop the flat NAV_ITEMS tuple in favor of master's
  grouped NAV_GROUPS nav-config module.
- webui/nav.py: register /system-health as a live item in the Health
  group, satisfying issue #639 AC5 through the canonical nav source.
- docs/webui-local-dev.md: keep both additive sections (#638 shell and
  #639 dashboard).
- tests/test_webui_system_health_dashboard.py: assert the nav entry via
  iter_nav_items() instead of the removed NAV_ITEMS tuple.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-23 19:56:57 -04:00
jcwalker3 b9ba43a5bf Merge branch 'master' into feat/issue-637-timeline-model 2026-07-23 17:47:29 -05:00
sysadminandClaude Opus 4.8 a1e5a4af8c fix(webui): validate externally influenced event_type at the timeline boundary (#637)
Review #526 found that event_type reached the serialized timeline payload
without crossing the redaction/validation boundary every other free-text
field on the same event crosses. A synthetic secret-shaped 40-hex canary
was redacted through message but survived verbatim through event_type on
the same control-plane record.

CTH heading path: CTH_TYPES is now the single authority for what a CTH
type may be. canonical_thread_handoff.is_known_cth_type() is that
authority, used by format_cth_body (write), assess_cth_comment (assess),
and now the read path too; parse_cth_comment reports membership as
cth_type_known and stays total. adapt_cth_comments serializes
handoff:<type> only for a declared type and otherwise emits the constant
handoff:unrecognized, so arbitrary, malformed, secret-shaped, or
whitespace-manipulated heading content never becomes an event_type.

Control-plane path: a stored event_type is treated as source data.
_safe_cp_event_type accepts only an ordinary identifier that is not a
bare secret-shaped hex run and that a redaction pass leaves unchanged;
anything else fails closed to the constant unsafe:redacted and marks the
event sensitive. The value is never emitted verbatim, never partially
sanitized, and never rewritten into a different valid-looking type.

Remaining serialized-field audit: event_key ids must be plain numeric
identifiers, the CTH adapter refuses a scope it cannot express, per-source
failure reasons are redacted (they can quote an authenticated fetch error),
and echoed scope/filter values are guarded so reflection is not a bypass.

Legitimate values are preserved: every declared CTH type, the real
producer types (assigned, lease_released, lease_adopted,
dependency_edge_state_change, allocation, pr.opened, lease.renew), and
the existing issue, PR, evidence, and full-SHA references.

Tests seed the canary independently through both event_type paths with a
benign message, so message redaction cannot be why they pass; each
inspects event_type directly, asserts the canary is absent from the
complete serialized payload, and asserts scan_for_secrets finds nothing.

tests/test_webui_timeline.py 64 passed, 15 subtests
pytest -k webui 362 passed, 285 subtests
pytest -k redact 79 passed
tests/test_canonical_thread_handoff.py tests/test_control_plane_db.py 29 passed

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-23 18:45:54 -04:00
sysadmin 188e83c4d6 Merge pull request 'fix: remove safe worktrees before reassessing remote delete ownership (#851)' (#852) from fix/issue-851-cleanup-worktree-before-remote-delete into master 2026-07-23 17:06:25 -05:00
sysadmin db5ed6042b Merge pull request 'feat(webui): evolve application shell for Phase 1 console IA (Closes #638)' (#818) from feat/issue-638-webui-app-shell-phase1 into master 2026-07-23 17:04:32 -05:00
jcwalker3 2d95e0fcc6 Merge branch 'master' into feat/issue-637-timeline-model 2026-07-23 16:21:09 -05:00
sysadminandClaude Opus 4.8 8ba1c5b87c fix(webui): make timeline session filter truthful and redact evidence refs (#637)
Remediates the two blocking findings in review #522 on PR #849.

F1 — the session filter dimension was dead end to end. No source could
produce an event carrying a session identifier, so filter_events dropped
every event whenever session was supplied and the API answered with a
green, empty page. An empty result reads to an operator as "no such
session activity", which is a stronger and false claim.

Each source now declares which filter dimensions its records can actually
carry. The control-plane events table is (event_id, work_item_id,
event_type, message, created_at) and records no session, so that source
declares the session dimension unsupported rather than pretending to
answer it; the dead read of a non-existent session_id column is removed.
A CTH handoff comment declares its own Session field, so the handoff
adapter populates session_id from that declared field — authoritative
source data, never inferred from an actor, work item, or message text.

When no source that ran can carry a requested dimension, load_timeline
refuses with ok=false and a structured error naming the unsupported
filters and the per-source reason, and the route answers 422. A source
that can answer the dimension and simply matched nothing still returns
200 with an honest empty page. Ordering, pagination, and the issue/PR
filters are unchanged.

F2 — evidence_refs bypassed redaction and could emit a credential
verbatim. proof and decision were passed to _extract_evidence_refs before
redaction, and the SHA pattern matched any 7-40 character lowercase hex
run, which is exactly the shape of a Gitea access token.

Redaction now runs first and every derived value is taken from the
redacted text. A commit reference is recognised only where the source
text declares one (commit, head, base, sha, ...), so an undeclared hex
run is never lifted out of prose into a structured field; this also drops
the ordinary-word noise the reviewer noted. Every reference is then
independently revalidated against an allowed shape and a second redaction
pass immediately before serialization, failing closed by dropping
anything unproven and flagging the event sensitive. Actor is redacted for
the same reason, and a secret-shaped session value is dropped rather than
emitted. Legitimate issue, PR, short-SHA and full 40-character SHA
references stay usable.

Tests: 16 added. Session filtering is now driven through the CTH adapter
and the composed load_timeline/API path rather than a hand-built
WorkflowEvent, covering a match, an honest empty result, pagination and
ordering under the filter, the 422 refusal, and the per-source support
declaration. Redaction coverage asserts a synthetic 40-character hex
value (not a real credential) appears nowhere in the complete serialized
payload including evidence_refs, that the independent revalidation drops
unproven references, and that legitimate references still resolve.

pytest tests/test_webui_timeline.py: 42 passed (was 26).
pytest -k webui: 340 passed, 270 subtests passed (was 324).
Full suite: 4607 passed, 12 failed, 6 skipped, 684 subtests passed — the
same 12 failures as the master baseline, none under webui/.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-23 16:45:43 -04:00
sysadminandClaude Opus 4.8 df58b5fb90 fix: remove safe worktrees before reassessing remote delete ownership (#851)
Post-merge cleanup previously continued past ownership-blocked remote
deletes, which skipped independently safe local worktree removal when the
only block was worktree_binding. Remove the clean owned worktree first,
reassess ownership, then delete the remote branch only if still safe.
Preserve fail-closed protection for dirty/foreign ownership categories.

Closes #851

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-23 16:41:27 -04:00
sysadminandClaude Opus 4.8 ecda200180 feat(webui): system-health dashboard (Closes #639)
Phase 1 child of the Web Console epic #631. Adds the operator-facing
system-health dashboard on top of the read-only system-health API landed
by #634, so runtime problems are visible on a surface instead of being
discovered late through failed LLM sessions.

- webui/system_health_views.py (new): renders the SystemHealthSnapshot as
  readiness, stale-runtime parity, version/uptime, dependency, MCP
  namespace, probe-error, and recovery cards.
- webui/app.py: GET /system-health, sharing load_system_health() with the
  JSON API so page and API cannot disagree. ?deep=1 behaves as on the API.
- webui/layout.py: nav entry and health card/badge styles.
- tests/test_webui_system_health_dashboard.py (new, 26 cases).
- docs/webui-local-dev.md: route, field authority, and redaction split.

Readiness honesty is preserved from the API: ready and readiness_complete
render separately, a probe that did not run is listed under "Not probed"
rather than counted healthy, and mutation safety is never claimed when the
runtime is stale or parity is indeterminate.

Redaction is split by field kind. Free text (probe details, reasons, probe
errors) passes through system_health.redact. Structured fields (commit
SHAs, probe names, statuses, timestamps) are HTML-escaped only: redact's
opaque-token rule matches any run of 32 or more characters, so routing a
40-character git SHA through it rendered "[redacted]" and blanked the
parity evidence the page exists to show.

Non-goals honored: no restart or reload controls (Phase 2, #642), no
manual process-kill guidance (#630). Read-only throughout.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-23 16:08:36 -04:00
sysadmin a20975688d Merge branch 'master' into feat/issue-637-timeline-model
# Conflicts:
#	webui/app.py
2026-07-23 14:53:12 -04:00
sysadminandClaude Opus 4.8 25bc2a3291 feat(webui): workflow-event and conversation timeline model (Closes #637)
Phase 1 child of the Web Console epic #631. Adds a durable, versioned
WorkflowEvent schema with per-source adapters and a read-only query API so
operators can browse a unified timeline of workflow events, decisions, tool
calls, and handoffs instead of scattered evidence.

- webui/timeline.py (new): versioned WorkflowEvent schema; control-plane
  event adapter and Gitea CTH handoff-comment adapter; read-only mode=ro
  control-plane reader; conjunctive filter by issue/PR/session; stable
  (timestamp, source_rank, event_key) ordering; bounded pagination;
  fail-soft per-source status; redaction at the boundary, fail closed.
- webui/app.py: GET /api/v1/timeline read-only route with thread-scoped,
  fail-soft handoff comment source.
- tests/test_webui_timeline.py (new): schema, adapters, redaction of
  secret-like payloads, filter/sort/pagination, scoped CP reader,
  fail-soft composition, and API integration.
- docs/webui-local-dev.md: timeline route and field-authority notes.

Read-only Phase 1; no mutation of historical events; no full chat replay;
no unredacted tool-argument storage.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-23 14:48:46 -04:00
13 changed files with 3490 additions and 101 deletions
+19 -2
View File
@@ -46,6 +46,17 @@ _FIELD_RE = re.compile(
) )
def is_known_cth_type(value: str | None) -> bool:
"""True when *value* is a declared member of the :data:`CTH_TYPES` contract.
``CTH_TYPES`` is the single authority for what a CTH type may be. The
heading a comment carries is free text, so a *read* path that turns a parsed
type into something durable — a serialized field, a routing decision — must
check membership here rather than trust the parse or keep a list of its own.
"""
return (value or "").strip() in CTH_TYPES
def format_cth_body( def format_cth_body(
*, *,
cth_type: str, cth_type: str,
@@ -60,7 +71,7 @@ def format_cth_body(
) -> str: ) -> str:
"""Render a canonical CTH comment body.""" """Render a canonical CTH comment body."""
normalized_type = (cth_type or "").strip() normalized_type = (cth_type or "").strip()
if normalized_type not in CTH_TYPES: if not is_known_cth_type(normalized_type):
raise ValueError( raise ValueError(
f"unknown CTH type '{cth_type}'; expected one of {sorted(CTH_TYPES)}" f"unknown CTH type '{cth_type}'; expected one of {sorted(CTH_TYPES)}"
) )
@@ -101,6 +112,12 @@ def parse_cth_comment(body: str) -> dict[str, Any] | None:
fields[key] = match.group(2).strip() fields[key] = match.group(2).strip()
return { return {
"cth_type": cth_type, "cth_type": cth_type,
# The heading capture is unconstrained free text, so the parse states
# whether it satisfies the CTH_TYPES contract instead of leaving every
# reader to decide (or forget). Parsing stays total — an unknown type is
# still parsed and reported, never raised on — but a reader that turns
# the type into a durable value can now tell the two apart.
"cth_type_known": is_known_cth_type(cth_type),
"fields": fields, "fields": fields,
"raw_body": text, "raw_body": text,
} }
@@ -119,7 +136,7 @@ def assess_cth_comment(body: str) -> dict[str, Any]:
} }
cth_type = parsed.get("cth_type") or "" cth_type = parsed.get("cth_type") or ""
if cth_type not in CTH_TYPES: if not is_known_cth_type(cth_type):
reasons.append( reasons.append(
f"unknown CTH type '{cth_type}'; expected one of {sorted(CTH_TYPES)}" f"unknown CTH type '{cth_type}'; expected one of {sorted(CTH_TYPES)}"
) )
+134
View File
@@ -54,6 +54,7 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
| `/` | Home / operator overview | | `/` | Home / operator overview |
| `/health` | JSON liveness (`status`, `service`, `mode`, `timestamp`, `uptime_seconds`) | | `/health` | JSON liveness (`status`, `service`, `mode`, `timestamp`, `uptime_seconds`) |
| `/api/v1/system/health` | Structured read-only system health (#634) | | `/api/v1/system/health` | Structured read-only system health (#634) |
| `/system-health` | System-health dashboard — readiness, version/uptime, dependencies, MCP namespaces, stale-runtime parity (#639) |
| `/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 with status and onboarding progress (#427, #635) | | `/projects` | Project registry list with status and onboarding progress (#427, #635) |
@@ -258,6 +259,37 @@ Not-yet-implemented surfaces (`/sessions`, `/inventory`, `/timeline`,
surfaces are backed by #636). Mutating methods on stub routes still fail closed surfaces are backed by #636). Mutating methods on stub routes still fail closed
with `read-only-mvp`. with `read-only-mvp`.
## System-health dashboard (#639)
`/system-health` renders the same snapshot the `/api/v1/system/health` API
returns, so the page and the API can never disagree. Cards: overall readiness,
stale-runtime parity, version and uptime, dependency probes, MCP namespaces,
probe errors (only when present), and recovery pointers. `?deep=1` opts into
the network probe exactly as the API does; the plain page load stays cheap.
Field authority and honesty rules:
* `ready` and `readiness_complete` are shown separately. A snapshot whose
required probes never ran is not the same as one that ran them and passed,
and the page never collapses the two into an unproven green.
* A probe that did not run appears under **Not probed**, never as healthy.
* `stale_runtime.mutation_safe` is displayed verbatim from the API. When the
runtime is stale, or when parity is indeterminate, the page warns and does
not claim mutation safety.
* MCP namespaces are reported `unproven`: the web process runs outside the
IDE-managed MCP client and cannot prove that path (#543).
Redaction is split by field kind. Free text — probe details, readiness and
parity reasons, probe errors — passes through `system_health.redact`.
Structured fields — commit SHAs, probe names, statuses, timestamps — are
HTML-escaped only, because `redact`'s opaque-token rule matches any run of 32
or more characters and would otherwise blank every 40-character git SHA, which
is precisely the evidence the parity view exists to show.
The dashboard is read-only: no restart, reload, or process-kill control. Those
arrive in Phase 2 (#642). Recovery guidance points at the sanctioned client
reconnect / operator restart path — never a manual daemon kill (#630).
## Deployment boundary (#435) ## Deployment boundary (#435)
MVP serves on loopback by default. Binding `0.0.0.0` or `::` is **refused** MVP serves on loopback by default. Binding `0.0.0.0` or `::` is **refused**
@@ -317,6 +349,108 @@ health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
checkout is behind merged safety-gate changes. Restart guidance links to #420; checkout is behind merged safety-gate changes. Restart guidance links to #420;
no tokens or MCP restart actions are exposed. no tokens or MCP restart actions are exposed.
## Workflow-event timeline (#637)
`GET /api/v1/timeline` is a read-only, versioned aggregation of workflow
events from every available source into one normalised, filterable stream. It
is the model layer for the Phase 1 timeline console view (a later child issue
of #631); this issue ships the schema, adapters, and read API only.
### Schema (versioned)
`webui/timeline.py` declares `TIMELINE_SCHEMA_VERSION` (currently `1`) and the
frozen `WorkflowEvent` record. Every response carries `schema_version` so a
consumer can branch on shape. One event:
```json
{
"source": "control_plane",
"event_type": "lease.renew",
"event_key": "cp:1421",
"timestamp": "2026-07-23T02:00:00Z",
"actor": null,
"role": null,
"issue_number": 637,
"pr_number": null,
"session_id": null,
"tool_name": null,
"decision": null,
"message": "lease renewed",
"correlation_id": "issue#637",
"evidence_refs": [],
"sensitive": true
}
```
`event_key` is stable and unique per source (`cp:<event_id>`,
`cth:<kind>:<number>:<comment_id>`), so pagination and dedup are deterministic.
### Sources and field authority
| Source | Adapter | Authority |
|---|---|---|
| Control-plane `events``work_items` | `adapt_cp_events` | `event_type`, `message`, `timestamp`, issue/PR scope come from the CP database, read through a `mode=ro` URI (never creates the DB or runs migrations) |
| Gitea Canonical Thread Handoff comments | `adapt_cth_comments` | `actor`, `role` (next owner), `decision`, `evidence_refs`, `timestamp` come from the parsed CTH comment body (`canonical_thread_handoff`) |
Handoff comments are thread-scoped: they are only read when the request filters
by a single `issue` or `pr`. Otherwise the handoff source reports `not run`
with a reason — it is never rendered as empty-and-healthy. Each source degrades
independently: an unavailable control-plane DB or a failed comment fetch is a
`sources[]` entry with `ok:false` and a `reason`, never a dropped timeline.
### Query parameters
`issue`, `pr`, `session` (conjunctive filters); `limit` (default 50, max 500)
and `offset` for pagination; `remote`, `org`, `repo` to override the default
registry-project scope. Events sort ascending by
`(timestamp, source_rank, event_key)`; missing timestamps sort last.
### Filter authority, and refusing what cannot be answered
A filter dimension is only meaningful for a source whose records carry it.
Each source declares its own support in `_SOURCE_FILTER_SUPPORT` and reports it
per response as `supported_filters` / `unsupported_filters`:
| Source | issue | pr | session |
|---|---|---|---|
| `control_plane` | yes | yes | **no** — the `events` table is `(event_id, work_item_id, event_type, message, created_at)` and records no session |
| `gitea_handoff` | yes | yes | yes — a CTH comment declares its own `Session:` field |
`session_id` is read only from that declared CTH field. It is never inferred
from a work item, an actor, or message text, and a value that is
redaction-altering or bare-secret-shaped is dropped rather than emitted.
When **no source that ran** can carry a requested dimension, the request is
refused rather than answered: the response is `422` with `ok:false` and a
structured `error` naming `unsupported_filters` and the per-source reason. A
`200` with zero events would tell an operator that no such activity exists,
which is a stronger — and false — claim than "this cannot be answered here".
A source that *can* answer the dimension and simply matched nothing still
returns `200` with `ok:true` and an empty page.
### Redaction
Every free-text field (event messages, decision/proof text, roles, actors) is
passed through the console redaction policy (`webui.console_redaction`, backed
by `gitea_audit.redact`) before it leaves the module, failing closed to the
placeholder. No unredacted tool arguments or secrets are ever emitted, and a
generation error never drops raw data to a caller or a log.
Redaction also runs *before* any structured value is derived from free text.
`evidence_refs` are extracted from already-redacted proof/decision text, and a
commit reference is recognised only where the text declares one (`commit`,
`head`, `base`, `sha`, …). An undeclared 40-character hex run has the exact
shape of a Gitea access token, so it is never lifted out of prose into a
structured field. Every reference is then independently revalidated against an
allowed shape and a second redaction pass immediately before serialization;
anything unproven is dropped and the event is flagged `sensitive`.
### Tests
```bash
pytest tests/test_webui_timeline.py -q
```
## Tests ## Tests
```bash ```bash
+135 -99
View File
@@ -11242,127 +11242,163 @@ def gitea_reconcile_merged_cleanups(
if dry_run: if dry_run:
report["dry_run"] = True report["dry_run"] = True
report["executed"] = False report["executed"] = False
# #851: surface planned lifecycle order so dry-run matches execute.
report["planned_execution_orders"] = {
str(entry.get("pr_number")): entry.get("planned_execution_order") or []
for entry in (report.get("entries") or [])
}
return {"success": True, "performed": False, **report} return {"success": True, "performed": False, **report}
verify_preflight_purity( verify_preflight_purity(
remote, task="reconcile_merged_cleanups", org=org, repo=repo remote, task="reconcile_merged_cleanups", org=org, repo=repo
) )
actions: list[dict] = [] actions: list[dict] = []
project_root = _canonical_local_git_root()
def _ownership_records_for_branch(
head_branch: str, pr_num_int: int | None
) -> list[dict]:
ownership_bundle = _collect_branch_ownership_records(
remote=remote,
host=h,
org=o,
repo=r,
branch=head_branch,
pr_number=pr_num_int,
project_root=project_root,
auth=auth,
base_api=base,
)
ownership_records = list(ownership_bundle.get("records") or [])
if ownership_bundle.get("inventory_error"):
ownership_records.append(
{
"category": (
branch_cleanup_guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR
),
"status": "unknown",
"remote": remote,
"host": h,
"org": o,
"repo": r,
"branch": head_branch,
"reclaim_allowed": False,
"role": "inventory",
}
)
return ownership_records
def _attempt_owned_remote_delete(
*,
head_branch: str,
pr_num_int: int | None,
after_worktree_removal: bool = False,
) -> dict:
"""Fail-closed remote delete with live ownership reassessment (#851)."""
import urllib.parse
ownership_records = _ownership_records_for_branch(head_branch, pr_num_int)
ownership = branch_cleanup_guard.assess_active_branch_ownership(
remote=remote,
org=o,
repo=r,
branch=head_branch,
host=h,
records=ownership_records,
)
if ownership.get("block"):
return {
"action": "delete_remote_branch",
"branch": head_branch,
"success": False,
"performed": False,
"delete_acknowledged": False,
"verified_absent": False,
"blocker_kind": "active_branch_ownership",
"reasons": ownership.get("reasons") or [],
"blocking_categories": ownership.get("blocking_categories") or [],
"after_worktree_removal": after_worktree_removal,
"ownership_reassessed": after_worktree_removal,
}
encoded = urllib.parse.quote(head_branch, safe="")
url = f"{base}/branches/{encoded}"
with _audited(
"delete_branch",
host=h,
remote=remote,
org=o,
repo=r,
target_branch=head_branch,
request_metadata={
"branch": head_branch,
"source": "reconcile_merged_cleanups",
"ownership_checked": True,
"after_worktree_removal": after_worktree_removal,
},
):
api_request("DELETE", url, auth)
readback = _probe_remote_branch(h, o, r, auth, head_branch)
readback_assessment = branch_cleanup_guard.assess_post_delete_readback(
readback
)
verified = bool(readback_assessment.get("verified_absent"))
return {
"action": "delete_remote_branch",
"branch": head_branch,
"success": bool(readback_assessment.get("ok")),
"performed": True,
"delete_acknowledged": True,
"verified_absent": verified,
"readback": readback_assessment.get("readback"),
"reasons": readback_assessment.get("reasons") or [],
"after_worktree_removal": after_worktree_removal,
"ownership_reassessed": after_worktree_removal,
}
for entry in report.get("entries") or []: for entry in report.get("entries") or []:
head_branch = entry.get("head_branch") or "" head_branch = entry.get("head_branch") or ""
remote_assessment = entry.get("remote_branch") or {} remote_assessment = entry.get("remote_branch") or {}
local_assessment = entry.get("local_worktree") or {} local_assessment = entry.get("local_worktree") or {}
pr_num = entry.get("pr_number")
try:
pr_num_int = int(pr_num) if pr_num is not None else None
except (TypeError, ValueError):
pr_num_int = None
if remote_assessment.get("safe_to_delete_remote"): # #851 lifecycle: when the target worktree is independently safe, remove
import urllib.parse # it first so worktree_binding ownership does not permanently strand
# both the worktree and the remote branch. Never skip worktree removal
pr_num = entry.get("pr_number") # merely because remote delete would be blocked by that binding.
try: # Ownership protection for remote delete remains fail-closed below.
pr_num_int = int(pr_num) if pr_num is not None else None worktree_removed = False
except (TypeError, ValueError):
pr_num_int = None
ownership_bundle = _collect_branch_ownership_records(
remote=remote,
host=h,
org=o,
repo=r,
branch=head_branch,
pr_number=pr_num_int,
project_root=_canonical_local_git_root(),
auth=auth,
base_api=base,
)
ownership_records = list(ownership_bundle.get("records") or [])
if ownership_bundle.get("inventory_error"):
ownership_records.append(
{
"category": (
branch_cleanup_guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR
),
"status": "unknown",
"remote": remote,
"host": h,
"org": o,
"repo": r,
"branch": head_branch,
"reclaim_allowed": False,
"role": "inventory",
}
)
ownership = branch_cleanup_guard.assess_active_branch_ownership(
remote=remote,
org=o,
repo=r,
branch=head_branch,
host=h,
records=ownership_records,
)
if ownership.get("block"):
actions.append(
{
"action": "delete_remote_branch",
"branch": head_branch,
"success": False,
"performed": False,
"delete_acknowledged": False,
"verified_absent": False,
"blocker_kind": "active_branch_ownership",
"reasons": ownership.get("reasons") or [],
"blocking_categories": ownership.get(
"blocking_categories"
)
or [],
}
)
continue
encoded = urllib.parse.quote(head_branch, safe="")
url = f"{base}/branches/{encoded}"
with _audited(
"delete_branch",
host=h,
remote=remote,
org=o,
repo=r,
target_branch=head_branch,
request_metadata={
"branch": head_branch,
"source": "reconcile_merged_cleanups",
"ownership_checked": True,
},
):
api_request("DELETE", url, auth)
readback = _probe_remote_branch(h, o, r, auth, head_branch)
readback_assessment = branch_cleanup_guard.assess_post_delete_readback(
readback
)
verified = bool(readback_assessment.get("verified_absent"))
actions.append(
{
"action": "delete_remote_branch",
"branch": head_branch,
"success": bool(readback_assessment.get("ok")),
"performed": True,
"delete_acknowledged": True,
"verified_absent": verified,
"readback": readback_assessment.get("readback"),
"reasons": readback_assessment.get("reasons") or [],
}
)
if local_assessment.get("safe_to_remove_worktree"): if local_assessment.get("safe_to_remove_worktree"):
result = merged_cleanup_reconcile.remove_local_worktree( result = merged_cleanup_reconcile.remove_local_worktree(
_canonical_local_git_root(), project_root,
head_branch, head_branch,
worktree_path=local_assessment.get("worktree_path"), worktree_path=local_assessment.get("worktree_path"),
) )
actions.append({"action": "remove_local_worktree", **result}) actions.append({"action": "remove_local_worktree", **result})
# Idempotent resume: absent worktree is already gone.
msg = (result.get("message") or "").lower()
worktree_removed = bool(result.get("success")) or (
"not found" in msg
)
if remote_assessment.get("safe_to_delete_remote"):
actions.append(
_attempt_owned_remote_delete(
head_branch=head_branch,
pr_num_int=pr_num_int,
after_worktree_removal=worktree_removed,
)
)
for scratch in report.get("reviewer_scratch_entries") or []: for scratch in report.get("reviewer_scratch_entries") or []:
if not scratch.get("safe_to_remove_worktree"): if not scratch.get("safe_to_remove_worktree"):
continue continue
result = merged_cleanup_reconcile.remove_reviewer_scratch_worktree( result = merged_cleanup_reconcile.remove_reviewer_scratch_worktree(
_canonical_local_git_root(), scratch.get("worktree_path") or "" project_root, scratch.get("worktree_path") or ""
) )
actions.append({"action": "remove_reviewer_scratch_worktree", **result}) actions.append({"action": "remove_reviewer_scratch_worktree", **result})
+58
View File
@@ -566,6 +566,10 @@ def build_pr_cleanup_entry(
worktree_state=worktree_state, worktree_state=worktree_state,
active_lock=active_lock, active_lock=active_lock,
) )
planned = plan_cleanup_execution_order(
remote_assessment=remote,
local_assessment=local,
)
return { return {
"pr_number": pr_number, "pr_number": pr_number,
"issue_number": issue_number, "issue_number": issue_number,
@@ -576,9 +580,63 @@ def build_pr_cleanup_entry(
"merged": merged, "merged": merged,
"remote_branch": remote, "remote_branch": remote,
"local_worktree": local, "local_worktree": local,
# #851: dry-run and execute share the same lifecycle order description.
"planned_execution_order": planned,
} }
def plan_cleanup_execution_order(
*,
remote_assessment: dict[str, Any] | None,
local_assessment: dict[str, Any] | None,
) -> list[dict[str, Any]]:
"""Describe independent worktree-then-reassess-then-remote cleanup order (#851).
Remote ownership protection remains fail-closed at execute time. A worktree
that is independently safe to remove is never skipped merely because remote
deletion may be blocked by that same ``worktree_binding``.
"""
remote = remote_assessment or {}
local = local_assessment or {}
steps: list[dict[str, Any]] = []
worktree_safe = bool(local.get("safe_to_remove_worktree"))
remote_safe = bool(remote.get("safe_to_delete_remote"))
if worktree_safe:
steps.append(
{
"action": "remove_local_worktree",
"reason": "independently_safe_to_remove",
"phase": 1,
}
)
if remote_safe:
if worktree_safe:
steps.append(
{
"action": "reassess_branch_ownership",
"reason": "after_worktree_removal_clear_worktree_binding",
"phase": 2,
}
)
steps.append(
{
"action": "delete_remote_branch",
"reason": "only_if_independently_safe_after_reassessment",
"phase": 3,
}
)
else:
steps.append(
{
"action": "delete_remote_branch",
"reason": "safe_to_delete_and_no_independent_worktree_removal",
"phase": 1,
}
)
return steps
def build_reconciliation_report( def build_reconciliation_report(
*, *,
project_root: str, project_root: str,
+372
View File
@@ -1266,6 +1266,378 @@ class TestSecondRemediationIntegration(unittest.TestCase):
self.assertIn("delete_acknowledged", delete_actions[0]) self.assertIn("delete_acknowledged", delete_actions[0])
self.assertTrue(delete_actions[0].get("verified_absent")) self.assertTrue(delete_actions[0].get("verified_absent"))
def test_issue_851_worktree_removed_when_remote_blocked_only_by_worktree_binding(self):
"""#851: remote blocked by worktree_binding must not skip safe worktree removal.
Lifecycle: remove clean owned worktree → reassess ownership → delete
remote only if independently safe. Unrelated entries stay untouched.
"""
from mcp_server import gitea_reconcile_merged_cleanups
target_branch = "fix/issue-844-exclude-epic-containers"
foreign_branch = "fix/issue-999-unrelated-active"
worktree_path = "/tmp/branches/fix-issue-844-exclude-epic-containers"
ownership_calls = []
remove_calls = []
delete_api_calls = []
def fake_collect(**kwargs):
ownership_calls.append(dict(kwargs))
# Ownership is reassessed *after* independent worktree removal (#851).
# Target worktree is already gone → no worktree_binding remains.
# Foreign branch keeps an active author lease → remote delete blocked.
if kwargs.get("branch") == foreign_branch:
# Match session-bound org/repo + host used by the tool resolve path.
return {
"records": [
{
"category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
"status": "active",
"remote": kwargs.get("remote") or "prgs",
"host": kwargs.get("host") or "gitea.example.com",
"org": kwargs.get("org") or "Scaled-Tech-Consulting",
"repo": kwargs.get("repo") or "Gitea-Tools",
"branch": foreign_branch,
"reclaim_allowed": False,
}
],
"inventory_error": False,
}
return {"records": [], "inventory_error": False}
def fake_remove(project_root, branch, worktree_path=None):
remove_calls.append(
{"branch": branch, "worktree_path": worktree_path}
)
return {
"success": True,
"performed": True,
"message": f"removed worktree {worktree_path}",
"worktree_path": worktree_path,
}
def fake_probe(h, o, r, auth, br):
return guard.classify_branch_readback_http_status(
404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH
)
def fake_api(method, url, auth, **kwargs):
if method == "DELETE":
delete_api_calls.append(url)
return {}
report = {
"entries": [
{
"pr_number": 848,
"head_branch": target_branch,
"remote_branch": {"safe_to_delete_remote": True},
"local_worktree": {
"safe_to_remove_worktree": True,
"worktree_path": worktree_path,
},
},
{
"pr_number": 999,
"head_branch": foreign_branch,
"remote_branch": {"safe_to_delete_remote": True},
"local_worktree": {
"safe_to_remove_worktree": False,
"worktree_path": None,
},
},
],
"reviewer_scratch_entries": [],
}
patch(
"mcp_server.get_profile",
return_value={
"profile_name": "prgs-reconciler",
"role": "reconciler",
"allowed_operations": [
"gitea.read",
"gitea.branch.delete",
"gitea.pr.close",
],
"forbidden_operations": [],
},
).start()
patch("mcp_server.api_get_all", return_value=[]).start()
patch(
"mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
return_value=report,
).start()
patch(
"mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
return_value=[],
).start()
patch(
"mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
return_value=(True, []),
).start()
patch("mcp_server.verify_preflight_purity", return_value=None).start()
patch(
"mcp_server._collect_branch_ownership_records",
side_effect=fake_collect,
).start()
patch("mcp_server._probe_remote_branch", side_effect=fake_probe).start()
patch(
"mcp_server.merged_cleanup_reconcile.remove_local_worktree",
side_effect=fake_remove,
).start()
self.mock_api.side_effect = fake_api
res = gitea_reconcile_merged_cleanups(
dry_run=False,
execute_confirmed=True,
remote="prgs",
)
self.assertTrue(res.get("performed") or res.get("executed"))
actions = res.get("actions") or []
remove_actions = [
a for a in actions if a.get("action") == "remove_local_worktree"
]
self.assertEqual(len(remove_actions), 1, actions)
self.assertTrue(remove_actions[0].get("success"))
self.assertEqual(remove_calls[0]["branch"], target_branch)
self.assertEqual(remove_calls[0]["worktree_path"], worktree_path)
# Target remote delete succeeds after worktree removal + reassessment.
target_deletes = [
a
for a in actions
if a.get("action") == "delete_remote_branch"
and a.get("branch") == target_branch
]
self.assertEqual(len(target_deletes), 1, actions)
self.assertTrue(target_deletes[0].get("success"))
self.assertTrue(target_deletes[0].get("after_worktree_removal"))
self.assertTrue(target_deletes[0].get("ownership_reassessed"))
self.assertTrue(target_deletes[0].get("verified_absent"))
# Foreign branch remains protected (author lease) and is not deleted.
foreign_deletes = [
a
for a in actions
if a.get("action") == "delete_remote_branch"
and a.get("branch") == foreign_branch
]
self.assertEqual(len(foreign_deletes), 1, actions)
self.assertFalse(foreign_deletes[0].get("success"))
self.assertEqual(
foreign_deletes[0].get("blocker_kind"), "active_branch_ownership"
)
self.assertIn(
guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
foreign_deletes[0].get("blocking_categories") or [],
)
# Only the target branch should hit the DELETE API.
self.assertEqual(len(delete_api_calls), 1)
# Ownership collected for target (post-removal) and foreign; worktree
# removal happened before target remote delete in the action log.
target_idx = next(
i
for i, a in enumerate(actions)
if a.get("action") == "remove_local_worktree"
)
delete_idx = next(
i
for i, a in enumerate(actions)
if a.get("action") == "delete_remote_branch"
and a.get("branch") == target_branch
and a.get("success")
)
self.assertLess(target_idx, delete_idx)
def test_issue_851_dirty_worktree_not_removed_and_remote_stays_protected(self):
"""#851: dirty/foreign worktrees remain protected; no unsafe cleanup."""
from mcp_server import gitea_reconcile_merged_cleanups
branch = "fix/issue-851-dirty"
remove_calls = []
def fake_collect(**kwargs):
return {
"records": [
{
"category": guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING,
"status": "active",
"remote": kwargs.get("remote") or "prgs",
"host": kwargs.get("host") or "gitea.example.com",
"org": kwargs.get("org") or "Scaled-Tech-Consulting",
"repo": kwargs.get("repo") or "Gitea-Tools",
"branch": branch,
"reclaim_allowed": False,
}
],
"inventory_error": False,
}
report = {
"entries": [
{
"pr_number": 851,
"head_branch": branch,
"remote_branch": {"safe_to_delete_remote": True},
"local_worktree": {
"safe_to_remove_worktree": False,
"worktree_path": "/tmp/dirty-wt",
},
}
],
"reviewer_scratch_entries": [],
}
patch(
"mcp_server.get_profile",
return_value={
"profile_name": "prgs-reconciler",
"role": "reconciler",
"allowed_operations": [
"gitea.read",
"gitea.branch.delete",
],
"forbidden_operations": [],
},
).start()
patch("mcp_server.api_get_all", return_value=[]).start()
patch(
"mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
return_value=report,
).start()
patch(
"mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
return_value=[],
).start()
patch(
"mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
return_value=(True, []),
).start()
patch("mcp_server.verify_preflight_purity", return_value=None).start()
patch(
"mcp_server._collect_branch_ownership_records",
side_effect=fake_collect,
).start()
patch(
"mcp_server.merged_cleanup_reconcile.remove_local_worktree",
side_effect=lambda *a, **k: remove_calls.append(k) or {
"success": True,
"performed": True,
},
).start()
self.mock_api.side_effect = lambda *a, **k: {}
res = gitea_reconcile_merged_cleanups(
dry_run=False,
execute_confirmed=True,
remote="prgs",
)
actions = res.get("actions") or []
self.assertEqual(remove_calls, [])
self.assertFalse(
any(a.get("action") == "remove_local_worktree" for a in actions)
)
deletes = [
a for a in actions if a.get("action") == "delete_remote_branch"
]
self.assertEqual(len(deletes), 1)
self.assertFalse(deletes[0].get("success"))
self.assertEqual(deletes[0].get("blocker_kind"), "active_branch_ownership")
self.assertIn(
guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING,
deletes[0].get("blocking_categories") or [],
)
def test_issue_851_idempotent_resume_when_worktree_already_absent(self):
"""#851: partial failures remain resumable and idempotent."""
from mcp_server import gitea_reconcile_merged_cleanups
branch = "fix/issue-851-resume"
ownership_calls = []
def fake_collect(**kwargs):
ownership_calls.append(kwargs)
return {"records": [], "inventory_error": False}
def fake_remove(project_root, branch, worktree_path=None):
return {
"success": False,
"performed": False,
"message": f"worktree not found: {worktree_path}",
}
def fake_probe(h, o, r, auth, br):
return guard.classify_branch_readback_http_status(
404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH
)
report = {
"entries": [
{
"pr_number": 851,
"head_branch": branch,
"remote_branch": {"safe_to_delete_remote": True},
"local_worktree": {
"safe_to_remove_worktree": True,
"worktree_path": "/tmp/already-gone",
},
}
],
"reviewer_scratch_entries": [],
}
patch(
"mcp_server.get_profile",
return_value={
"profile_name": "prgs-reconciler",
"role": "reconciler",
"allowed_operations": [
"gitea.read",
"gitea.branch.delete",
],
"forbidden_operations": [],
},
).start()
patch("mcp_server.api_get_all", return_value=[]).start()
patch(
"mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
return_value=report,
).start()
patch(
"mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
return_value=[],
).start()
patch(
"mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
return_value=(True, []),
).start()
patch("mcp_server.verify_preflight_purity", return_value=None).start()
patch(
"mcp_server._collect_branch_ownership_records",
side_effect=fake_collect,
).start()
patch("mcp_server._probe_remote_branch", side_effect=fake_probe).start()
patch(
"mcp_server.merged_cleanup_reconcile.remove_local_worktree",
side_effect=fake_remove,
).start()
self.mock_api.side_effect = lambda *a, **k: {}
res = gitea_reconcile_merged_cleanups(
dry_run=False,
execute_confirmed=True,
remote="prgs",
)
actions = res.get("actions") or []
removes = [a for a in actions if a.get("action") == "remove_local_worktree"]
deletes = [a for a in actions if a.get("action") == "delete_remote_branch"]
self.assertEqual(len(removes), 1)
self.assertFalse(removes[0].get("success"))
self.assertEqual(len(deletes), 1)
self.assertTrue(deletes[0].get("success"))
self.assertTrue(deletes[0].get("after_worktree_removal"))
self.assertTrue(ownership_calls)
if __name__ == "__main__": if __name__ == "__main__":
+53
View File
@@ -12,6 +12,59 @@ import merged_cleanup_reconcile as mcr # noqa: E402
class TestMergedCleanupAssessment(unittest.TestCase): class TestMergedCleanupAssessment(unittest.TestCase):
def test_issue_851_plan_order_worktree_then_reassess_then_remote(self):
"""#851 dry-run plan: remove worktree, reassess ownership, then remote."""
plan = mcr.plan_cleanup_execution_order(
remote_assessment={"safe_to_delete_remote": True},
local_assessment={"safe_to_remove_worktree": True},
)
actions = [s["action"] for s in plan]
self.assertEqual(
actions,
[
"remove_local_worktree",
"reassess_branch_ownership",
"delete_remote_branch",
],
)
self.assertEqual(plan[0]["phase"], 1)
self.assertEqual(plan[-1]["phase"], 3)
self.assertIn("independently_safe", plan[0]["reason"])
self.assertIn("reassessment", plan[-1]["reason"])
def test_issue_851_plan_remote_only_when_worktree_not_safe(self):
plan = mcr.plan_cleanup_execution_order(
remote_assessment={"safe_to_delete_remote": True},
local_assessment={"safe_to_remove_worktree": False},
)
self.assertEqual([s["action"] for s in plan], ["delete_remote_branch"])
self.assertNotIn("reassess_branch_ownership", [s["action"] for s in plan])
def test_issue_851_plan_worktree_only_when_remote_not_safe(self):
plan = mcr.plan_cleanup_execution_order(
remote_assessment={"safe_to_delete_remote": False},
local_assessment={"safe_to_remove_worktree": True},
)
self.assertEqual([s["action"] for s in plan], ["remove_local_worktree"])
def test_issue_851_entry_includes_planned_execution_order(self):
entry = mcr.build_pr_cleanup_entry(
pr={
"number": 848,
"title": "Closes #844",
"body": "",
"merged_at": "2026-07-23T00:00:00Z",
"head": {"ref": "fix/issue-844-x", "sha": "a" * 40},
},
project_root="/tmp/not-a-real-root",
open_pr_heads=set(),
remote_branch_exists=True,
head_on_master=True,
delete_capability_allowed=True,
)
self.assertIn("planned_execution_order", entry)
self.assertIsInstance(entry["planned_execution_order"], list)
def test_extract_linked_issue_from_closes(self): def test_extract_linked_issue_from_closes(self):
issue = mcr.extract_linked_issue( issue = mcr.extract_linked_issue(
"feat: cleanup (Closes #269)", "feat: cleanup (Closes #269)",
+345
View File
@@ -0,0 +1,345 @@
"""Tests for the system-health dashboard view (#639).
Covers the acceptance criteria directly: the page renders the health DTO
fields (AC1), degraded dependencies are visible (AC2), stale runtime is warned
prominently and never rendered as mutation-safe (AC3), healthy and degraded
fixtures both render (AC4), and the shell carries a nav entry (AC5).
"""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from starlette.testclient import TestClient
from webui.app import create_app
from webui.deployment_boundary import scan_text_for_client_secrets
from webui.layout import render_page
from webui.nav import iter_nav_items
from webui.system_health import (
STATUS_DEGRADED,
STATUS_DOWN,
STATUS_OK,
STATUS_SKIPPED,
STATUS_UNPROVEN,
DependencyProbe,
StaleRuntime,
SystemHealthSnapshot,
VersionInfo,
)
from webui.system_health_views import render_system_health_page
DASHBOARD_PATH = "/system-health"
def _version(*, known: bool = True) -> VersionInfo:
return VersionInfo(
git_sha="1c455b6ec0f9cb761fe6248de68c17e061fb5ecd" if known else None,
git_describe="v0.4.1-12-g1c455b6" if known else None,
control_plane_schema_version=4 if known else None,
python_version="3.13.1",
known=known,
)
def _parity(*, stale: bool = False, determinable: bool = True) -> StaleRuntime:
if stale:
return StaleRuntime(
daemon_head="aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
checkout_head="bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
remote_head="cccccccccccccccccccccccccccccccccccccccc",
stale=True,
determinable=True,
mutation_safe=False,
reasons=("runtime, checkout, and remote commits disagree",),
)
if not determinable:
return StaleRuntime(
daemon_head=None,
checkout_head=None,
remote_head=None,
stale=False,
determinable=False,
mutation_safe=False,
reasons=("local checkout HEAD could not be read",),
)
return StaleRuntime(
daemon_head="1c455b6ec0f9cb761fe6248de68c17e061fb5ecd",
checkout_head="1c455b6ec0f9cb761fe6248de68c17e061fb5ecd",
remote_head="1c455b6ec0f9cb761fe6248de68c17e061fb5ecd",
stale=False,
determinable=True,
mutation_safe=True,
reasons=(),
)
def _snapshot(
*,
status: str = STATUS_OK,
ready: bool = True,
readiness_complete: bool = True,
readiness_reasons: tuple[str, ...] = (),
dependencies: tuple[DependencyProbe, ...] | None = None,
parity: StaleRuntime | None = None,
namespaces: tuple[dict, ...] = (),
probe_errors: tuple[str, ...] = (),
version_known: bool = True,
) -> SystemHealthSnapshot:
if dependencies is None:
dependencies = (
DependencyProbe(
name="control_plane_db",
kind="sqlite",
status=STATUS_OK,
detail="schema version 4",
required=True,
latency_ms=1.25,
metadata={"schema_version": 4},
),
)
return SystemHealthSnapshot(
status=status,
ready=ready,
readiness_complete=readiness_complete,
readiness_reasons=readiness_reasons,
service="mcp-control-plane-webui",
mode="read-only",
version=_version(known=version_known),
started_at="2026-07-23T19:50:47+00:00",
uptime_seconds=3661.5,
timestamp="2026-07-23T20:51:48+00:00",
deep_probes_requested=False,
dependencies=dependencies,
mcp_namespaces=namespaces,
stale_runtime=parity if parity is not None else _parity(),
probe_errors=probe_errors,
)
class TestHealthyRender(unittest.TestCase):
"""AC1 / AC4 — every health DTO field reaches the page."""
def setUp(self):
self.html = render_system_health_page(_snapshot())
def test_readiness_fields_render(self):
self.assertIn("System health", self.html)
self.assertIn("Ready", self.html)
self.assertIn("mcp-control-plane-webui", self.html)
self.assertIn("read-only", self.html)
self.assertIn("2026-07-23T20:51:48+00:00", self.html)
def test_version_and_uptime_render(self):
self.assertIn("1c455b6ec0f9cb761fe6248de68c17e061fb5ecd", self.html)
self.assertIn("v0.4.1-12-g1c455b6", self.html)
self.assertIn("3.13.1", self.html)
self.assertIn("3661.500s", self.html)
self.assertIn("1.02h", self.html)
def test_dependency_row_renders_with_latency(self):
self.assertIn("control_plane_db", self.html)
self.assertIn("sqlite", self.html)
self.assertIn("schema version 4", self.html)
self.assertIn("1.2 ms", self.html)
def test_healthy_page_shows_no_stale_warning(self):
self.assertNotIn("Stale runtime:", self.html)
self.assertNotIn("Staleness", self.html)
def test_unknown_version_is_labelled_not_faked(self):
html = render_system_health_page(_snapshot(version_known=False))
self.assertIn("unknown", html)
self.assertIn("unresolved", html)
class TestDegradedRender(unittest.TestCase):
"""AC2 — a degraded or unrun dependency is visible, not swallowed."""
def setUp(self):
self.deps = (
DependencyProbe(
name="control_plane_db",
kind="sqlite",
status=STATUS_OK,
detail="schema version 4",
required=True,
latency_ms=0.9,
),
DependencyProbe(
name="repository",
kind="git",
status=STATUS_DOWN,
detail="repository root is not a git checkout",
required=True,
latency_ms=4.0,
),
DependencyProbe(
name="gitea",
kind="http",
status=STATUS_SKIPPED,
detail="deep probe not requested",
required=False,
),
)
self.html = render_system_health_page(
_snapshot(
status=STATUS_DEGRADED,
ready=False,
readiness_complete=False,
readiness_reasons=("required dependency 'repository' is down",),
dependencies=self.deps,
)
)
def test_degraded_banner_names_the_dependency(self):
self.assertIn("Degraded dependencies:", self.html)
self.assertIn("repository", self.html)
def test_not_run_probe_is_reported_separately(self):
self.assertIn("Not probed:", self.html)
self.assertIn("gitea", self.html)
self.assertIn("not counted", self.html)
def test_not_ready_headline_and_reason(self):
self.assertIn("Not ready", self.html)
self.assertIn("required dependency &#x27;repository&#x27; is down", self.html)
def test_degraded_status_badge_present(self):
self.assertIn("badge-health-degraded", self.html)
self.assertIn("badge-health-down", self.html)
def test_ready_but_incomplete_is_not_shown_as_plain_ready(self):
html = render_system_health_page(
_snapshot(ready=True, readiness_complete=False)
)
self.assertIn("Ready (incomplete evidence)", html)
class TestStaleRuntimeWarning(unittest.TestCase):
"""AC3 — staleness is prominent and never claims mutation safety."""
def test_stale_runtime_warns_and_denies_mutation_safety(self):
html = render_system_health_page(_snapshot(parity=_parity(stale=True)))
self.assertIn("Stale runtime:", html)
self.assertIn("do not treat this runtime as mutation-safe", html)
self.assertIn("<tr><th>Mutation safe</th><td>False</td></tr>", html)
def test_indeterminate_parity_is_not_reported_safe(self):
html = render_system_health_page(
_snapshot(parity=_parity(determinable=False))
)
self.assertIn("Staleness", html)
self.assertIn("<tr><th>Mutation safe</th><td>False</td></tr>", html)
self.assertIn("<tr><th>Determinable</th><td>False</td></tr>", html)
def test_healthy_parity_reports_mutation_safe_true(self):
html = render_system_health_page(_snapshot())
self.assertIn("<tr><th>Mutation safe</th><td>True</td></tr>", html)
class TestNamespacesAndErrors(unittest.TestCase):
def test_unproven_namespace_rows_render(self):
html = render_system_health_page(
_snapshot(
namespaces=(
{
"namespace": "gitea-author",
"required_tool": "gitea_lock_issue",
"status": STATUS_UNPROVEN,
"ide_namespace_proven": False,
"reason": "the web console cannot invoke the IDE-managed MCP client",
},
)
)
)
self.assertIn("gitea-author", html)
self.assertIn("gitea_lock_issue", html)
self.assertIn("badge-health-unproven", html)
def test_no_namespaces_degrades_gracefully(self):
html = render_system_health_page(_snapshot(namespaces=()))
self.assertIn("No MCP namespaces are declared.", html)
def test_probe_errors_render_when_present(self):
html = render_system_health_page(
_snapshot(probe_errors=("probe raised: disk offline",))
)
self.assertIn("Probe errors", html)
self.assertIn("disk offline", html)
def test_probe_error_card_absent_when_clean(self):
self.assertNotIn("Probe errors", render_system_health_page(_snapshot()))
class TestReadOnlyAndRedaction(unittest.TestCase):
def test_no_restart_or_kill_controls(self):
html = render_system_health_page(_snapshot())
self.assertNotIn("<button", html)
self.assertNotIn("<form", html)
self.assertNotIn("pkill", html)
self.assertIn("read-only", html)
def test_recovery_points_at_sanctioned_path(self):
html = render_system_health_page(_snapshot())
self.assertIn("Reconnect the MCP client", html)
self.assertIn("Never kill the daemon process manually", html)
def test_secret_shaped_detail_is_redacted(self):
leaky = DependencyProbe(
name="gitea",
kind="http",
status=STATUS_DOWN,
detail="auth failed for token=ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789",
required=False,
latency_ms=12.0,
)
html = render_system_health_page(_snapshot(dependencies=(leaky,)))
self.assertNotIn("ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789", html)
def test_html_in_detail_is_escaped(self):
hostile = DependencyProbe(
name="repository",
kind="git",
status=STATUS_DOWN,
detail="<script>alert(1)</script>",
required=True,
)
html = render_system_health_page(_snapshot(dependencies=(hostile,)))
self.assertNotIn("<script>", html)
self.assertIn("&lt;script&gt;", html)
class TestNavAndRoute(unittest.TestCase):
"""AC5 — the shell links the dashboard, and the route serves it."""
def setUp(self):
self.client = TestClient(create_app())
def test_nav_contains_system_health(self):
self.assertIn(
(DASHBOARD_PATH, "System health"),
[(item.href, item.label) for item in iter_nav_items()],
)
def test_rendered_shell_links_dashboard(self):
page = render_page(title="Home", body_html="<p>x</p>")
self.assertIn(f'href="{DASHBOARD_PATH}"', page)
def test_route_renders_dashboard(self):
response = self.client.get(DASHBOARD_PATH)
self.assertEqual(response.status_code, 200)
self.assertIn("System health", response.text)
self.assertIn("Stale-runtime parity", response.text)
def test_route_is_read_only(self):
self.assertEqual(self.client.post(DASHBOARD_PATH).status_code, 405)
def test_live_page_leaks_no_client_secret(self):
findings = scan_text_for_client_secrets(self.client.get(DASHBOARD_PATH).text)
self.assertEqual(findings, [])
if __name__ == "__main__": # pragma: no cover
unittest.main()
File diff suppressed because it is too large Load Diff
+120
View File
@@ -46,12 +46,14 @@ 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.timeline import load_timeline, snapshot_to_dict as timeline_snapshot_to_dict
from webui.system_health import ( from webui.system_health import (
API_PATH as SYSTEM_HEALTH_API_PATH, API_PATH as SYSTEM_HEALTH_API_PATH,
load_system_health, load_system_health,
process_uptime, process_uptime,
snapshot_to_dict as system_health_to_dict, snapshot_to_dict as system_health_to_dict,
) )
from webui.system_health_views import render_system_health_page
_READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"}) _READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
_AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"}) _AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
@@ -160,6 +162,24 @@ async def api_system_health(request: Request) -> JSONResponse:
return JSONResponse(payload, status_code=200 if snapshot.ready else 503) return JSONResponse(payload, status_code=200 if snapshot.ready else 503)
async def system_health(request: Request) -> HTMLResponse:
"""Read-only system-health dashboard (#639).
Shares the #634 snapshot loader with the JSON API so the page can never
disagree with it. `?deep=1` opts into the network probe exactly as the API
does; the default page load stays cheap. The response is always 200: this
is an operator view that must render the degraded state, not withhold it.
"""
deep = _truthy_flag(request.query_params.get("deep"))
snapshot = load_system_health(deep=deep)
return HTMLResponse(
render_page(
title="System health",
body_html=render_system_health_page(snapshot),
)
)
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)))
@@ -449,6 +469,104 @@ async def api_console_security_model(_request: Request) -> JSONResponse:
}) })
def _query_int(request: Request, key: str) -> int | None:
"""Parse an optional integer query parameter; None when absent/invalid."""
raw = request.query_params.get(key)
if raw is None or not str(raw).strip():
return None
try:
return int(str(raw).strip())
except (TypeError, ValueError):
return None
def _derive_remote(host: str) -> str:
"""Map a Gitea host to its known short remote name (control-plane scope key)."""
text = (host or "").lower()
if "prgs" in text:
return "prgs"
if "dadeschools" in text:
return "dadeschools"
return text.split(".")[0] if text else ""
def _timeline_comment_source(host: str, org: str, repo: str):
"""Build a fail-soft CTH-comment fetcher for one repo, or None when offline.
Returns a callable ``(kind, number) -> list[comment]``. Credentials or
network failures raise inside the callable so ``load_timeline`` degrades the
handoff source rather than the whole timeline. Offline test mode yields no
live source so the handoff section reports ``not run``.
"""
import os
from gitea_auth import api_fetch_page, get_auth_header, repo_api_url
offline = (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip().lower() in {"1", "true", "yes"}
if offline:
return None
auth = get_auth_header(host)
if not auth:
return None
def _fetch(kind: str, number: int) -> list:
segment = "pulls" if kind == "pr" else "issues"
url = f"{repo_api_url(host, org, repo)}/{segment}/{int(number)}/comments"
comments: list = []
page = 1
while page <= 20:
raw, meta = api_fetch_page(url, auth, page=page, limit=50)
comments.extend(raw)
if bool(meta["is_final_page"]):
break
page += 1
return comments
return _fetch
async def api_v1_timeline(request: Request) -> JSONResponse:
"""Read-only workflow-event timeline (#637). Filter by issue/PR/session."""
from webui.queue_loader import _host_from_url # host normalisation helper
registry, error = _load_project_registry()
if error is not None:
return JSONResponse(error.to_dict(), status_code=500)
project = registry.projects[0] if registry.projects else None
org = request.query_params.get("org") or (project.gitea_owner if project else "")
repo = request.query_params.get("repo") or (project.repo_name if project else "")
host = _host_from_url(project.remote_host) if project else ""
remote = request.query_params.get("remote") or _derive_remote(host)
if not (remote and org and repo):
return JSONResponse(
{
"error": "timeline_scope_unresolved",
"detail": "no project in registry and no remote/org/repo query params provided",
},
status_code=400,
)
comment_source = _timeline_comment_source(host, org, repo) if (host and org and repo) else None
snapshot = load_timeline(
remote=remote,
org=org,
repo=repo,
issue=_query_int(request, "issue"),
pr=_query_int(request, "pr"),
session=(request.query_params.get("session") or None),
limit=_query_int(request, "limit"),
offset=_query_int(request, "offset"),
comment_source=comment_source,
)
# A filter no surviving source can carry is refused, not answered empty:
# a 200 with zero events would tell the operator no such activity exists.
status_code = 200 if snapshot.ok else 422
return JSONResponse(timeline_snapshot_to_dict(snapshot), status_code=status_code)
async def method_not_allowed(request: Request, _exc: Exception) -> Response: async def method_not_allowed(request: Request, _exc: Exception) -> Response:
path = request.url.path path = request.url.path
if path in _AUDIT_MUTATION_PATHS and request.method == "POST": if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
@@ -472,6 +590,7 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
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(SYSTEM_HEALTH_API_PATH, api_system_health, methods=["GET"]),
Route("/system-health", 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"]),
@@ -488,6 +607,7 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
Route("/api/prompts", api_prompts, methods=["GET"]), Route("/api/prompts", api_prompts, methods=["GET"]),
Route("/runtime", runtime, methods=["GET"]), Route("/runtime", runtime, methods=["GET"]),
Route("/api/runtime", api_runtime, methods=["GET"]), Route("/api/runtime", api_runtime, methods=["GET"]),
Route("/api/v1/timeline", api_v1_timeline, methods=["GET"]),
Route("/audit", audit, methods=["GET", "POST"]), Route("/audit", audit, methods=["GET", "POST"]),
Route("/api/audit", api_audit, methods=["GET", "POST"]), Route("/api/audit", api_audit, methods=["GET", "POST"]),
Route("/worktrees", worktrees, methods=["GET"]), Route("/worktrees", worktrees, methods=["GET"]),
+19
View File
@@ -236,6 +236,25 @@ def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
.badge-in-review {{ color: #9ec8f0; border-color: #3d5f7a; }} .badge-in-review {{ color: #9ec8f0; border-color: #3d5f7a; }}
.badge-duplicate {{ color: #e0c27a; border-color: #6b5730; }} .badge-duplicate {{ color: #e0c27a; border-color: #6b5730; }}
.badge-stale {{ color: #c9b8e8; border-color: #5a4a78; }} .badge-stale {{ color: #c9b8e8; border-color: #5a4a78; }}
.badge-health-ok {{ color: #8fd19e; border-color: #3d6b4a; }}
.badge-health-degraded {{ color: #e0c27a; border-color: #6b5730; }}
.badge-health-down {{ color: #f0a8a8; border-color: #7a3b3b; }}
.badge-health-skipped {{ color: var(--muted); }}
.badge-health-unproven {{ color: #c9b8e8; border-color: #5a4a78; }}
.health-card {{
margin: 1.25rem 0;
padding: 0.85rem 1rem 1rem;
border: 1px solid var(--border);
border-radius: 8px;
background: var(--surface);
}}
.health-card h3 {{ margin: 0 0 0.5rem; font-size: 1.05rem; }}
.health-card h4 {{ margin: 1rem 0 0.35rem; font-size: 0.92rem; color: var(--muted); }}
.health-headline {{ color: var(--text); font-size: 1rem; margin: 0 0 0.5rem; }}
.health-degraded {{ border-left-color: #e0c27a; }}
.health-stale {{ border-left-color: #f0a8a8; }}
ul.reasons {{ margin: 0.35rem 0; padding-left: 1.15rem; color: var(--muted); font-size: 0.9rem; }}
ul.reasons li {{ margin-bottom: 0.3rem; }}
</style> </style>
{extra_head} {extra_head}
</head> </head>
+1
View File
@@ -38,6 +38,7 @@ class NavGroup:
NAV_GROUPS: tuple[NavGroup, ...] = ( NAV_GROUPS: tuple[NavGroup, ...] = (
NavGroup("Health", ( NavGroup("Health", (
NavItem("/health", "Liveness"), NavItem("/health", "Liveness"),
NavItem("/system-health", "System health"),
)), )),
NavGroup("Traffic", ( NavGroup("Traffic", (
NavItem("/queue", "Queue"), NavItem("/queue", "Queue"),
+307
View File
@@ -0,0 +1,307 @@
"""HTML views for the system-health dashboard (#639).
Renders the read-only :class:`~webui.system_health.SystemHealthSnapshot`
produced by the Phase 1 system-health API (#634). The page offers no restart,
reload, or process-kill control: those are Phase 2 work, and manual process
kills are the contamination path #630 exists to prevent.
Every free-text field passes through :func:`webui.system_health.redact` before
it reaches HTML, so a probe detail that captured a token or a credentialed URL
cannot leak through the dashboard even though the API redacts it already.
"""
from __future__ import annotations
import html
from webui.system_health import (
STATUS_DEGRADED,
STATUS_DOWN,
STATUS_OK,
STATUS_SKIPPED,
STATUS_UNPROVEN,
DependencyProbe,
SystemHealthSnapshot,
redact,
)
_STATUS_BADGE_CLASS = {
STATUS_OK: "badge-health-ok",
STATUS_DEGRADED: "badge-health-degraded",
STATUS_DOWN: "badge-health-down",
STATUS_SKIPPED: "badge-health-skipped",
STATUS_UNPROVEN: "badge-health-unproven",
}
def _safe(value: object) -> str:
"""Escape free text for HTML after redacting anything secret-shaped.
Use this for every value that can carry arbitrary text probe details,
reasons, probe errors because those are where a credential could ride
along.
"""
return html.escape(redact(str(value)))
def _esc(value: object) -> str:
"""Escape a structured field for HTML without redacting it.
Commit SHAs, probe names, statuses, and timestamps are enumerated or
machine-generated, never credential-bearing. They must not go through
:func:`redact`: its opaque-token rule matches any 32-plus-character run,
so a 40-character git SHA would render as ``[redacted]`` and the parity
view the one thing an operator reads this page for would be blank.
"""
return html.escape(str(value))
def _status_badge(status: str) -> str:
css = _STATUS_BADGE_CLASS.get(status, "badge-health-unproven")
return f'<span class="badge {css}">{_esc(status)}</span>'
def _reason_list(reasons: tuple[str, ...], *, empty: str) -> str:
if not reasons:
return f"<p class='muted'>{html.escape(empty)}</p>"
items = "".join(f"<li>{_safe(reason)}</li>" for reason in reasons)
return f"<ul class='reasons'>{items}</ul>"
def _readiness_card(snapshot: SystemHealthSnapshot) -> str:
"""Overall readiness.
``ready`` and ``readiness_complete`` are shown separately on purpose: a
snapshot whose required probes never ran is not the same as one that ran
them and passed, and collapsing the two would render an unproven green.
"""
if snapshot.ready and snapshot.readiness_complete:
headline = "Ready"
elif snapshot.ready:
headline = "Ready (incomplete evidence)"
else:
headline = "Not ready"
return (
"<section class='health-card'>"
f"<h3>Readiness {_status_badge(snapshot.status)}</h3>"
f"<p class='health-headline'>{html.escape(headline)}</p>"
"<table class='detail'>"
f"<tr><th>Service</th><td><code>{_esc(snapshot.service)}</code></td></tr>"
f"<tr><th>Mode</th><td>{_esc(snapshot.mode)}</td></tr>"
f"<tr><th>Ready</th><td>{_esc(snapshot.ready)}</td></tr>"
"<tr><th>Readiness evidence complete</th>"
f"<td>{_esc(snapshot.readiness_complete)}</td></tr>"
"<tr><th>Deep probes requested</th>"
f"<td>{_esc(snapshot.deep_probes_requested)}</td></tr>"
f"<tr><th>Observed at</th><td><code>{_esc(snapshot.timestamp)}</code></td></tr>"
"</table>"
"<h4>Readiness reasons</h4>"
f"{_reason_list(snapshot.readiness_reasons, empty='No readiness objections recorded.')}"
"</section>"
)
def _version_card(snapshot: SystemHealthSnapshot) -> str:
version = snapshot.version
uptime_hours = snapshot.uptime_seconds / 3600.0
known = (
"resolved"
if version.known
else "unresolved — version fields could not be read from the checkout"
)
schema = version.control_plane_schema_version
return (
"<section class='health-card'>"
"<h3>Version and uptime</h3>"
"<table class='detail'>"
f"<tr><th>Git SHA</th><td><code>{_esc(version.git_sha or 'unknown')}</code></td></tr>"
"<tr><th>Git describe</th>"
f"<td><code>{_esc(version.git_describe or 'unknown')}</code></td></tr>"
"<tr><th>Control-plane schema</th>"
f"<td>{_esc(schema if schema is not None else 'unknown')}</td></tr>"
f"<tr><th>Python</th><td><code>{_esc(version.python_version)}</code></td></tr>"
f"<tr><th>Version status</th><td>{html.escape(known)}</td></tr>"
f"<tr><th>Started at</th><td><code>{_esc(snapshot.started_at)}</code></td></tr>"
"<tr><th>Uptime</th>"
f"<td>{snapshot.uptime_seconds:.3f}s ({uptime_hours:.2f}h)</td></tr>"
"</table>"
"</section>"
)
def _dependency_rows(probes: tuple[DependencyProbe, ...]) -> str:
if not probes:
return "<p class='muted'>No dependency probes were reported.</p>"
rows = []
for probe in probes:
latency = (
f"{probe.latency_ms:.1f} ms" if probe.latency_ms is not None else "n/a"
)
rows.append(
"<tr>"
f"<td><code>{_esc(probe.name)}</code></td>"
f"<td>{_esc(probe.kind)}</td>"
f"<td>{_status_badge(probe.status)}</td>"
f"<td>{_esc('required' if probe.required else 'optional')}</td>"
f"<td>{html.escape(latency)}</td>"
f"<td>{_safe(probe.detail)}</td>"
"</tr>"
)
return (
"<table class='registry'><thead><tr>"
"<th>Dependency</th><th>Kind</th><th>Status</th><th>Requirement</th>"
"<th>Latency</th><th>Detail</th>"
"</tr></thead><tbody>"
f"{''.join(rows)}</tbody></table>"
)
def _dependency_card(snapshot: SystemHealthSnapshot) -> str:
degraded = [probe for probe in snapshot.dependencies if probe.ran and not probe.healthy]
not_run = [probe for probe in snapshot.dependencies if not probe.ran]
banner = ""
if degraded:
names = ", ".join(sorted(probe.name for probe in degraded))
banner += (
"<div class='stub health-degraded'><p><strong>Degraded dependencies:</strong> "
f"{_esc(names)}</p></div>"
)
if not_run:
names = ", ".join(sorted(probe.name for probe in not_run))
banner += (
"<div class='stub'><p><strong>Not probed:</strong> "
f"{_esc(names)} — these contribute no evidence and are not counted "
"as healthy.</p></div>"
)
return (
"<section class='health-card'>"
"<h3>Dependencies</h3>"
f"{banner}"
f"{_dependency_rows(snapshot.dependencies)}"
"<p class='muted'>Details are redacted at the API boundary and again "
"before rendering; credentials are never displayed.</p>"
"</section>"
)
def _namespace_card(snapshot: SystemHealthSnapshot) -> str:
if not snapshot.mcp_namespaces:
body = "<p class='muted'>No MCP namespaces are declared.</p>"
else:
rows = []
for entry in snapshot.mcp_namespaces:
rows.append(
"<tr>"
f"<td><code>{_esc(entry.get('namespace'))}</code></td>"
f"<td><code>{_esc(entry.get('required_tool'))}</code></td>"
f"<td>{_status_badge(str(entry.get('status') or STATUS_UNPROVEN))}</td>"
f"<td>{_esc(entry.get('ide_namespace_proven'))}</td>"
f"<td>{_safe(entry.get('reason'))}</td>"
"</tr>"
)
body = (
"<table class='registry'><thead><tr>"
"<th>Namespace</th><th>Required tool</th><th>Status</th>"
"<th>IDE-proven</th><th>Reason</th>"
"</tr></thead><tbody>"
f"{''.join(rows)}</tbody></table>"
)
return (
"<section class='health-card'>"
"<h3>MCP namespaces</h3>"
f"{body}"
"<p class='muted'>The web process runs outside the IDE-managed MCP "
"client, so namespace health is reported as unproven rather than "
"guessed (#543).</p>"
"</section>"
)
def _stale_runtime_card(snapshot: SystemHealthSnapshot) -> str:
stale = snapshot.stale_runtime
if stale.stale:
warning = (
"<div class='stub health-stale'><p><strong>Stale runtime:</strong> "
"the running code, the checkout, and the remote-tracking commit "
"disagree. Capability gates may be evaluating obsolete code — "
"do not treat this runtime as mutation-safe.</p></div>"
)
elif not stale.determinable:
warning = (
"<div class='stub health-stale'><p><strong>Staleness "
"indeterminate:</strong> parity could not be proven, so this "
"runtime is not reported as mutation-safe.</p></div>"
)
else:
warning = ""
return (
"<section class='health-card'>"
"<h3>Stale-runtime parity</h3>"
f"{warning}"
"<table class='detail'>"
"<tr><th>Daemon head</th>"
f"<td><code>{_esc(stale.daemon_head or 'unknown')}</code></td></tr>"
"<tr><th>Checkout head</th>"
f"<td><code>{_esc(stale.checkout_head or 'unknown')}</code></td></tr>"
"<tr><th>Remote head</th>"
f"<td><code>{_esc(stale.remote_head or 'unknown')}</code></td></tr>"
f"<tr><th>Stale</th><td>{_esc(stale.stale)}</td></tr>"
f"<tr><th>Determinable</th><td>{_esc(stale.determinable)}</td></tr>"
f"<tr><th>Mutation safe</th><td>{_esc(stale.mutation_safe)}</td></tr>"
"</table>"
f"{_reason_list(stale.reasons, empty='Runtime, checkout, and remote agree.')}"
"</section>"
)
def _probe_error_card(snapshot: SystemHealthSnapshot) -> str:
if not snapshot.probe_errors:
return ""
return (
"<section class='health-card'>"
"<h3>Probe errors</h3>"
f"{_reason_list(snapshot.probe_errors, empty='')}"
"</section>"
)
def _recovery_card() -> str:
"""Sanctioned recovery pointers only — never a manual process kill (#630)."""
return (
"<section class='health-card'>"
"<h3>Recovery</h3>"
"<p class='muted'>This dashboard is read-only. Restart and reload "
"controls arrive in Phase 2 (#642); until then recovery runs through "
"the sanctioned client reconnect / operator restart path.</p>"
"<ul class='reasons'>"
"<li><a href='/runtime'>Runtime and session view</a> — active profile, "
"workflow hashes, and shell health.</li>"
"<li>Reconnect the MCP client from the IDE, then re-run the blocked "
"cycle. Never kill the daemon process manually: unmanaged kills are "
"recorded as runtime contamination (#630).</li>"
"<li>See <code>docs/webui-local-dev.md</code> for the documented "
"recovery sequence.</li>"
"</ul>"
"</section>"
)
def render_system_health_page(snapshot: SystemHealthSnapshot) -> str:
"""Render the full system-health dashboard body."""
return (
"<h2>System health</h2>"
"<p class='meta'>Read-only view of the Phase 1 system-health API "
"(<code>/api/v1/system/health</code>). Reload this page to refresh; "
"nothing here polls or mutates on your behalf.</p>"
f"{_readiness_card(snapshot)}"
f"{_stale_runtime_card(snapshot)}"
f"{_version_card(snapshot)}"
f"{_dependency_card(snapshot)}"
f"{_namespace_card(snapshot)}"
f"{_probe_error_card(snapshot)}"
f"{_recovery_card()}"
)
+906
View File
@@ -0,0 +1,906 @@
"""Workflow-event and conversation timeline model (#637, Phase 1).
Operators cannot browse a unified timeline of workflow events, decisions,
tool calls, and handoffs: the evidence is scattered across control-plane
events, Gitea canonical handoff comments, and local logs. This module defines
one durable, versioned event schema and per-source adapters that normalise
those scattered records into a single ``WorkflowEvent`` stream, plus a
read-only query layer (filter by issue / PR / session, stable ordering,
pagination) that the ``/api/v1/timeline`` route serves.
Design rules honoured here:
- **Read-only.** Sources are read; nothing is mutated. The control-plane
database is opened through a ``mode=ro`` URI so a missing or unwritable DB
degrades to a reason instead of creating directories or running migrations.
- **Fail-soft per source.** An unavailable source degrades to a status with a
reason rather than raising, and a source that could not run is never
rendered as an empty-and-healthy timeline.
- **Answerable filters only.** Each source declares which filter dimensions it
can actually answer. A filter dimension no source that ran can carry is
refused with an explicit reason rather than silently matching nothing: an
empty page from an unanswerable filter reads to an operator as "no such
activity", which is a different — and false — statement.
- **Redaction at the boundary, fail closed.** Every free-text field (event
messages, redacted tool arguments, decision/proof text) is run through the
console redaction policy before it leaves this module, and *before* any
structured value is derived from it evidence references are extracted from
redacted text, then independently revalidated before serialization. An
unredactable value becomes the placeholder, and a value that cannot be proven
safe is dropped an unredacted payload is never emitted, and a generation
error never drops raw data to a caller or a log.
- **Stable ordering.** Events sort by ``(timestamp, source_rank, event_key)``
with a deterministic tiebreak, so pagination is stable across calls and
events with equal or missing timestamps keep a fixed order.
Non-goals (from the issue): no full chat replay, no mutation of historical
events, no unredacted tool-argument storage.
"""
from __future__ import annotations
import re
import sqlite3
from dataclasses import dataclass, replace
from datetime import datetime, timezone
from typing import Any, Callable, Iterable
import control_plane_db
from webui import console_redaction
# The schema is versioned so consumers can branch on shape. Bump on any
# breaking change to WorkflowEvent's serialized form.
TIMELINE_SCHEMA_VERSION = 1
# Known event sources and their deterministic ordering rank. When two events
# carry the same timestamp, the source rank breaks the tie before the
# per-source event key, so a control-plane event and a handoff comment minted
# in the same second always sort in a fixed order.
SOURCE_CONTROL_PLANE = "control_plane"
SOURCE_GITEA_HANDOFF = "gitea_handoff"
_SOURCE_RANK = {
SOURCE_CONTROL_PLANE: 0,
SOURCE_GITEA_HANDOFF: 1,
}
# The filter dimensions the query layer accepts.
FILTER_ISSUE = "issue"
FILTER_PR = "pr"
FILTER_SESSION = "session"
# Which dimensions each source can actually answer. This is a property of the
# underlying records, not of the query code: the control-plane ``events`` table
# is (event_id, work_item_id, event_type, message, created_at) and carries no
# session identity at all, so no control-plane event can ever match a session
# filter. A CTH handoff comment can declare its session as a field, so the
# handoff source answers all three. Filtering on a dimension the surviving
# sources cannot carry is refused in ``load_timeline`` rather than answered
# with an empty page.
_SOURCE_FILTER_SUPPORT: dict[str, tuple[str, ...]] = {
SOURCE_CONTROL_PLANE: (FILTER_ISSUE, FILTER_PR),
SOURCE_GITEA_HANDOFF: (FILTER_ISSUE, FILTER_PR, FILTER_SESSION),
}
# Why a source cannot answer a dimension, for the refusal reason an operator reads.
_SOURCE_FILTER_LIMITS: dict[tuple[str, str], str] = {
(SOURCE_CONTROL_PLANE, FILTER_SESSION): (
"control-plane events carry no session identity "
"(the events table has no session column)"
),
}
# A timestamp far in the future so events with no parseable timestamp sort
# last (after everything real) instead of first, without raising.
_MISSING_TS_SORT = "9999-12-31T23:59:59Z"
def _parse_ts(value: str | None) -> str | None:
"""Normalise a timestamp to ``...Z`` UTC, or None when unparseable."""
if not value:
return None
text = str(value).strip()
if not text:
return None
candidate = text[:-1] + "+00:00" if text.endswith("Z") else text
try:
parsed = datetime.fromisoformat(candidate)
except ValueError:
return None
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
return parsed.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
def _redact(value: Any) -> Any:
"""Redact a single free-text field, failing closed to the placeholder."""
if value is None:
return None
return console_redaction.redact_text(str(value))
@dataclass(frozen=True)
class WorkflowEvent:
"""One normalised timeline event.
Every field is optional except ``source``/``event_type``/``event_key``
because sources carry different subsets. The class is frozen so an adapted
event is an immutable record; a consumer that needs a variant builds a new
one rather than mutating history.
"""
source: str
event_type: str
event_key: str
timestamp: str | None = None
actor: str | None = None
role: str | None = None
issue_number: int | None = None
pr_number: int | None = None
session_id: str | None = None
tool_name: str | None = None
decision: str | None = None
message: str | None = None
correlation_id: str | None = None
evidence_refs: tuple[str, ...] = ()
sensitive: bool = False
def sort_key(self) -> tuple[str, int, str]:
return (
self.timestamp or _MISSING_TS_SORT,
_SOURCE_RANK.get(self.source, 99),
self.event_key,
)
def to_dict(self) -> dict[str, Any]:
return {
"source": self.source,
"event_type": self.event_type,
"event_key": self.event_key,
"timestamp": self.timestamp,
"actor": self.actor,
"role": self.role,
"issue_number": self.issue_number,
"pr_number": self.pr_number,
"session_id": self.session_id,
"tool_name": self.tool_name,
"decision": self.decision,
"message": self.message,
"correlation_id": self.correlation_id,
"evidence_refs": list(self.evidence_refs),
"sensitive": self.sensitive,
}
# --------------------------------------------------------------------------- #
# Adapters — pure functions from a source's raw records to WorkflowEvents. #
# Each is total: a malformed record is skipped, never raised on. #
# --------------------------------------------------------------------------- #
# Event types whose payload is treated as sensitive and always redaction-hard
# (they can carry lease/session provenance or tool arguments).
_SENSITIVE_EVENT_HINTS = ("lease", "capability", "token", "auth", "secret")
# Reference tokens (issue/PR/comment ids) and SHAs parsed out of proof text.
_EVIDENCE_REF_RE = re.compile(r"(?:#|PR\s*#?|issue\s*#?|comment\s*#?)(\d+)", re.IGNORECASE)
# A commit reference is only recognised when the text *declares* it as one.
# A bare lowercase hex run is not evidence of anything: at 40 characters it is
# exactly the shape of a Gitea personal access token, and at 7 it also matches
# ordinary words such as "defaced". Requiring an anchoring keyword keeps real
# references ("commit abc1234", "at head a209756...", "base caaae9b6") usable
# while refusing to lift an undeclared secret-shaped run out of free text.
_SHA_RE = re.compile(
r"(?i:\b(?:commit|sha|head|base|parent|revision|rev|merge[- ]base)\b[\s:=@#]*)"
r"([0-9a-f]{7,40})\b"
)
# Shapes a serialized evidence reference is allowed to take. Anything else is
# dropped rather than emitted.
_REF_ISSUE_SHAPE = re.compile(r"^#[0-9]{1,9}$")
_REF_SHA_SHAPE = re.compile(r"^[0-9a-f]{7,40}$")
# A long undelimited hex run with no declaring context is treated as credential
# material wherever it appears, never as an identifier.
_BARE_SECRET_SHAPE = re.compile(r"^[0-9a-f]{32,}$")
# An event type reads like an identifier, but a stored one is externally
# influenced: any producer that writes the control-plane ``events`` table
# chooses the string. It reaches ``to_dict`` verbatim, so it is validated here
# rather than trusted because of where it came from.
_CP_EVENT_TYPE_SHAPE = re.compile(r"^[A-Za-z][A-Za-z0-9._:+-]{0,63}$")
# Emitted in place of a value that cannot be proven safe. Deliberately not a
# plausible workflow type: an unsafe value is refused, never quietly rewritten
# into a different valid-looking one that would misdescribe the record.
UNSAFE_EVENT_TYPE = "unsafe:redacted"
# Emitted for a CTH heading that is not a declared member of ``CTH_TYPES``. The
# contract is enforced on write (``format_cth_body``) and on assess; the read
# path the timeline uses enforces it too rather than assuming it was.
UNKNOWN_HANDOFF_EVENT_TYPE = "handoff:unrecognized"
# A source record id is a plain integer in both sources it comes from: the
# control-plane ``events`` primary key and a Gitea comment id. ``event_key`` is
# serialized verbatim and is the pagination tiebreak, so anything else is
# refused rather than interpolated into it.
_RECORD_ID_SHAPE = re.compile(r"^[0-9]{1,19}$")
def _kind_to_numbers(kind: str | None, number: int | None) -> tuple[int | None, int | None]:
"""Map a control-plane work-item (kind, number) to (issue_no, pr_no)."""
if number is None:
return (None, None)
if kind == "pr":
return (None, int(number))
if kind == "issue":
return (int(number), None)
return (None, None)
def _correlation_for(kind: str | None, number: int | None) -> str | None:
if number is None or kind not in ("issue", "pr"):
return None
return f"{kind}#{number}"
def _extract_evidence_refs(*texts: str | None) -> tuple[str, ...]:
"""Extract issue/PR and declared-commit references from **redacted** text.
Callers must pass text that has already been through :func:`_redact`; this
function derives a structured field from its input, so extracting ahead of
redaction would republish whatever redaction was about to remove. Every
reference is revalidated by :func:`_validated_evidence_refs` before it is
serialized.
"""
refs: list[str] = []
for text in texts:
if not text:
continue
for match in _EVIDENCE_REF_RE.finditer(text):
token = f"#{match.group(1)}"
if token not in refs:
refs.append(token)
for match in _SHA_RE.finditer(text):
token = match.group(1)
if token not in refs:
refs.append(token)
return tuple(refs)
def _validated_evidence_refs(refs: Iterable[str]) -> tuple[tuple[str, ...], bool]:
"""Independently revalidate references immediately before serialization.
Extraction is not trusted on its own. A reference survives only when it has
a known reference shape and is unchanged by a second redaction pass a
value the redaction policy would alter is credential material that must not
be emitted as a structured field. A full 40-character SHA stays usable
because extraction only accepts a hex run the source text explicitly
declared as a commit. Returns ``(safe_refs, dropped_any)``; ``dropped_any``
marks the event sensitive so the drop is visible rather than silent.
"""
safe: list[str] = []
dropped = False
for ref in refs or ():
try:
token = str(ref).strip()
if not token:
continue
recognised = bool(_REF_ISSUE_SHAPE.match(token) or _REF_SHA_SHAPE.match(token))
if not recognised:
dropped = True
continue
if _redact(token) != token:
dropped = True
continue
if token not in safe:
safe.append(token)
except Exception:
# Fail closed: a reference that cannot be proven safe is dropped.
dropped = True
continue
return (tuple(safe), dropped)
def _safe_session_id(value: Any) -> str | None:
"""Return a session identifier only when it is safe to emit.
The value is authoritative source data a session the record names for
itself but it is still free text. It is dropped when redaction alters it
or when it is a bare secret-shaped hex run, so a credential parked in a
session field can never reach the payload or be echoed back by a filter.
"""
if value is None:
return None
text = str(value).strip()
if not text:
return None
if _BARE_SECRET_SHAPE.match(text):
return None
return text if _redact(text) == text else None
def _safe_record_id(value: Any) -> str | None:
"""Return a source record id only when it is a plain numeric identifier.
``event_key`` is serialized verbatim and is the deterministic pagination
tiebreak, so an id is interpolated into it only when it has the shape both
real sources actually produce. A record whose identity cannot be trusted is
refused by the caller rather than keyed on.
"""
if value is None or isinstance(value, bool):
return None
if isinstance(value, int):
return str(value)
text = str(value).strip()
return text if _RECORD_ID_SHAPE.match(text) else None
def _safe_cp_event_type(value: Any) -> tuple[str, bool]:
"""Validate a stored control-plane event type. Returns ``(type, unsafe)``.
The stored value is externally influenced whichever producer wrote the
``events`` row chose the string and ``to_dict`` serializes it verbatim, so
it passes a boundary of its own instead of relying on the one ``message``
passes. A value survives only when it is an ordinary identifier, is not a
bare secret-shaped hex run, and is unchanged by a redaction pass. Anything
else fails closed to :data:`UNSAFE_EVENT_TYPE`: the record stays visible as
an audit entry, but the value itself is never republished not verbatim,
not partially sanitized, and not rewritten into some other valid-looking
type that would misdescribe what happened.
"""
text = ("" if value is None else str(value)).strip()
if not text:
return ("", False)
if _BARE_SECRET_SHAPE.match(text):
return (UNSAFE_EVENT_TYPE, True)
if not _CP_EVENT_TYPE_SHAPE.match(text):
return (UNSAFE_EVENT_TYPE, True)
if _redact(text) != text:
return (UNSAFE_EVENT_TYPE, True)
return (text, False)
def _safe_echo(value: Any) -> Any:
"""Guard a scalar that is echoed back rather than derived from a record.
Query scope and filter values are caller-supplied and are reflected in the
response so an operator can see what was asked. Reflection is still
emission: a value redaction would alter, or a bare secret-shaped hex run, is
replaced by the placeholder instead of being echoed verbatim. Ordinary
scope and filter values pass through untouched.
"""
if value is None or isinstance(value, (int, bool)):
return value
text = str(value)
if _BARE_SECRET_SHAPE.match(text.strip()):
return console_redaction.REDACTED
return _redact(text)
def adapt_cp_events(rows: Iterable[dict[str, Any]]) -> list[WorkflowEvent]:
"""Adapt control-plane ``events`` rows (joined to work_items) into events.
Each row is expected to carry ``event_id``, ``event_type``, ``message``,
``created_at`` and the joined work-item ``kind``/``number``. Rows missing
an id or type are skipped so a partially written table never raises.
"""
events: list[WorkflowEvent] = []
for row in rows or []:
try:
event_id = _safe_record_id(row.get("event_id"))
raw_event_type = (row.get("event_type") or "").strip()
if event_id is None or not raw_event_type:
continue
# The stored type is source data, not a trusted constant: validate
# it before it is serialized, exactly as `message` below is redacted
# before it is serialized.
event_type, event_type_unsafe = _safe_cp_event_type(raw_event_type)
kind = row.get("kind")
number = row.get("number")
issue_no, pr_no = _kind_to_numbers(kind, number)
sensitive = event_type_unsafe or any(
hint in raw_event_type.lower() for hint in _SENSITIVE_EVENT_HINTS
)
events.append(
WorkflowEvent(
source=SOURCE_CONTROL_PLANE,
event_type=event_type,
event_key=f"cp:{event_id}",
timestamp=_parse_ts(row.get("created_at")),
issue_number=issue_no,
pr_number=pr_no,
# No session_id: the control-plane events table is
# (event_id, work_item_id, event_type, message, created_at)
# and records no session. Inventing one from the work item
# or the message text would be a guess, so this source
# declares the session dimension unsupported instead
# (_SOURCE_FILTER_SUPPORT) and the query layer refuses a
# session filter it cannot honestly answer.
message=_redact(row.get("message")),
correlation_id=_correlation_for(kind, number),
sensitive=sensitive,
)
)
except Exception:
# A single malformed row must not sink the whole adaptation.
continue
return events
def adapt_cth_comments(
comments: Iterable[dict[str, Any]],
*,
kind: str,
number: int,
) -> list[WorkflowEvent]:
"""Adapt Gitea Canonical Thread Handoff (CTH) comments into events.
Only comments that parse as a CTH (``canonical_thread_handoff.parse_cth_comment``)
become events; ordinary comments are ignored. ``kind``/``number`` scope the
events to the issue or PR the comments belong to.
"""
# Imported lazily so this module has no import-time dependency on the
# handoff parser when only the control-plane adapter is used.
from canonical_thread_handoff import is_known_cth_type, parse_cth_comment
# ``kind``/``number`` are interpolated into event_key and correlation_id, so
# they are normalised once here. A scope this adapter cannot express is
# refused outright rather than serialized into an identifier.
kind = (kind or "").strip().lower()
if kind not in ("issue", "pr"):
return []
try:
number = int(number)
except (TypeError, ValueError):
return []
issue_no, pr_no = _kind_to_numbers(kind, number)
correlation = _correlation_for(kind, number)
events: list[WorkflowEvent] = []
for comment in comments or []:
try:
body = comment.get("body") or ""
parsed = parse_cth_comment(body)
if not parsed:
continue
fields = parsed.get("fields") or {}
cth_type = parsed.get("cth_type") or ""
comment_id = _safe_record_id(comment.get("id"))
if comment_id is None:
continue
# The CTH heading is free text: the parser accepts whatever follows
# "## CTH:", and only the write and assess paths check it against
# the contract. Check it here too — an unrecognised heading is
# reported as such rather than serialized into event_type, so
# arbitrary, malformed, or secret-shaped heading content has no way
# through. Declared types are preserved exactly.
cth_type_known = is_known_cth_type(cth_type)
# Redaction runs first, and every derived value is taken from the
# redacted text — deriving evidence refs from the raw proof would
# re-emit exactly what redaction was about to remove.
decision = _redact(fields.get("decision"))
proof = _redact(fields.get("proof"))
next_action = _redact(fields.get("next action"))
refs, refs_dropped = _validated_evidence_refs(
_extract_evidence_refs(proof, decision)
)
events.append(
WorkflowEvent(
source=SOURCE_GITEA_HANDOFF,
event_type=(
f"handoff:{cth_type.strip()}"
if cth_type_known
else UNKNOWN_HANDOFF_EVENT_TYPE
),
event_key=f"cth:{kind}:{number}:{comment_id}",
timestamp=_parse_ts(comment.get("created_at")),
actor=_redact((comment.get("user") or {}).get("login")),
role=_redact(fields.get("next owner")),
issue_number=issue_no,
pr_number=pr_no,
# A CTH names its own session when the producer records one;
# it is read from that declared field, never inferred from
# unrelated text.
session_id=_safe_session_id(fields.get("session")),
decision=decision,
message=next_action or _redact(fields.get("status")),
correlation_id=correlation,
evidence_refs=refs,
sensitive=refs_dropped or not cth_type_known,
)
)
except Exception:
continue
return events
# --------------------------------------------------------------------------- #
# Read-only control-plane event source. #
# --------------------------------------------------------------------------- #
_CP_EVENTS_QUERY = """
SELECT e.event_id AS event_id,
e.event_type AS event_type,
e.message AS message,
e.created_at AS created_at,
w.kind AS kind,
w.number AS number
FROM events e
JOIN work_items w ON e.work_item_id = w.work_item_id
WHERE w.remote = ? AND w.org = ? AND w.repo = ?
"""
@dataclass(frozen=True)
class SourceStatus:
"""Fail-soft status for one timeline source.
``supported_filters`` states which filter dimensions this source's records
can carry; ``unsupported_filters`` names the requested dimensions it cannot,
so an operator can see *why* a source contributed nothing rather than being
left to read an empty list as an absence of activity.
"""
name: str
ok: bool
reason: str | None = None
count: int = 0
supported_filters: tuple[str, ...] = ()
unsupported_filters: tuple[str, ...] = ()
def to_dict(self) -> dict[str, Any]:
return {
"name": self.name,
"ok": self.ok,
"reason": self.reason,
"count": self.count,
"supported_filters": list(self.supported_filters),
"unsupported_filters": list(self.unsupported_filters),
}
def _cp_status(*, ok: bool, reason: str | None = None, count: int = 0) -> SourceStatus:
return SourceStatus(
SOURCE_CONTROL_PLANE,
ok=ok,
# A failure reason is serialized like any other field and is often an
# exception string carrying a path or a transport error, so it crosses
# the redaction boundary too. Static reasons pass through unchanged.
reason=_redact(reason),
count=count,
supported_filters=_SOURCE_FILTER_SUPPORT[SOURCE_CONTROL_PLANE],
)
def _handoff_status(*, ok: bool, reason: str | None = None, count: int = 0) -> SourceStatus:
return SourceStatus(
SOURCE_GITEA_HANDOFF,
ok=ok,
# Same boundary as the control-plane status: this reason can quote an
# error raised by a live authenticated fetch.
reason=_redact(reason),
count=count,
supported_filters=_SOURCE_FILTER_SUPPORT[SOURCE_GITEA_HANDOFF],
)
def read_cp_events(
*,
remote: str,
org: str,
repo: str,
db_path: str | None = None,
) -> tuple[list[WorkflowEvent], SourceStatus]:
"""Read scoped control-plane events read-only. Never creates the DB.
Opens the SQLite file through a ``mode=ro`` URI: a health/timeline read
must never create directories or run the schema migration that
``ControlPlaneDB()`` performs on construction. A missing or unreadable DB
degrades to a status with a reason.
"""
path = (db_path or control_plane_db.default_db_path()).strip()
conn: sqlite3.Connection | None = None
try:
conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
conn.row_factory = sqlite3.Row
cursor = conn.execute(_CP_EVENTS_QUERY, (remote, org, repo))
rows = [dict(r) for r in cursor.fetchall()]
except sqlite3.OperationalError as exc:
return ([], _cp_status(ok=False, reason=f"control-plane DB unavailable: {exc}"))
except sqlite3.Error as exc:
return ([], _cp_status(ok=False, reason=f"control-plane read failed: {exc}"))
finally:
if conn is not None:
conn.close()
events = adapt_cp_events(rows)
return (events, _cp_status(ok=True, count=len(events)))
# --------------------------------------------------------------------------- #
# Filter, sort, paginate. #
# --------------------------------------------------------------------------- #
def filter_events(
events: Iterable[WorkflowEvent],
*,
issue: int | None = None,
pr: int | None = None,
session: str | None = None,
) -> list[WorkflowEvent]:
"""Filter events by issue number, PR number, and/or session id.
Filters are conjunctive. A filter that names a dimension an event does not
carry excludes that event (an issue filter excludes PR-only events).
"""
out: list[WorkflowEvent] = []
for ev in events:
if issue is not None and ev.issue_number != issue:
continue
if pr is not None and ev.pr_number != pr:
continue
if session is not None and ev.session_id != session:
continue
out.append(ev)
return out
def sort_events(events: Iterable[WorkflowEvent]) -> list[WorkflowEvent]:
"""Return events in stable timeline order (ascending)."""
return sorted(events, key=lambda ev: ev.sort_key())
@dataclass(frozen=True)
class TimelinePage:
"""One page of the sorted, filtered timeline."""
events: tuple[WorkflowEvent, ...]
total: int
limit: int
offset: int
@property
def next_offset(self) -> int | None:
nxt = self.offset + len(self.events)
return nxt if nxt < self.total else None
def to_dict(self) -> dict[str, Any]:
return {
"events": [ev.to_dict() for ev in self.events],
"pagination": {
"total": self.total,
"limit": self.limit,
"offset": self.offset,
"returned": len(self.events),
"next_offset": self.next_offset,
"has_more": self.next_offset is not None,
},
}
_MAX_LIMIT = 500
_DEFAULT_LIMIT = 50
def _coerce_bounds(limit: int | None, offset: int | None) -> tuple[int, int]:
try:
lim = int(limit) if limit is not None else _DEFAULT_LIMIT
except (TypeError, ValueError):
lim = _DEFAULT_LIMIT
try:
off = int(offset) if offset is not None else 0
except (TypeError, ValueError):
off = 0
lim = max(1, min(lim, _MAX_LIMIT))
off = max(0, off)
return (lim, off)
def paginate(events: list[WorkflowEvent], *, limit: int | None, offset: int | None) -> TimelinePage:
lim, off = _coerce_bounds(limit, offset)
window = events[off : off + lim]
return TimelinePage(events=tuple(window), total=len(events), limit=lim, offset=off)
# --------------------------------------------------------------------------- #
# Composition — load_timeline aggregates all sources, fail-soft. #
# --------------------------------------------------------------------------- #
# A comment source is a callable that, given (kind, number), returns the raw
# Gitea comment list for that issue/PR. The route supplies a live fail-soft
# fetcher; tests supply a fixture. When None, the handoff source is reported as
# not-run (never silently empty-and-healthy).
CommentSource = Callable[[str, int], list[dict[str, Any]]]
@dataclass(frozen=True)
class TimelineSnapshot:
"""One answered timeline query.
``ok`` is False when the query could not be answered as asked currently
when a requested filter dimension no surviving source can carry was
supplied. The page is then empty *and* the snapshot says so, because an
``ok`` empty page is a claim that no such activity exists.
"""
schema_version: int
remote: str
org: str
repo: str
filters: dict[str, Any]
page: TimelinePage
sources: tuple[SourceStatus, ...]
ok: bool = True
error: dict[str, Any] | None = None
def to_dict(self) -> dict[str, Any]:
return {
"ok": self.ok,
"error": self.error,
"schema_version": self.schema_version,
# Scope and filters are echoed caller input, not derived record
# data. Reflecting a value is still emitting it, so both cross the
# same boundary; ordinary scope and filter values are unchanged.
"scope": {
"remote": _safe_echo(self.remote),
"org": _safe_echo(self.org),
"repo": _safe_echo(self.repo),
},
"filters": {key: _safe_echo(value) for key, value in self.filters.items()},
"sources": [s.to_dict() for s in self.sources],
**self.page.to_dict(),
}
def _unanswerable_reasons(
statuses: Iterable[SourceStatus], unanswerable: Iterable[str]
) -> list[dict[str, str]]:
"""Explain, per source, why each unanswerable dimension went unanswered."""
out: list[dict[str, str]] = []
for status in statuses:
for dim in unanswerable:
if dim not in status.supported_filters:
reason = _SOURCE_FILTER_LIMITS.get(
(status.name, dim), f"this source's records carry no {dim} identity"
)
elif not status.ok:
reason = (
f"this source can carry {dim} but did not run: "
f"{status.reason or 'unavailable'}"
)
else:
continue
out.append({"source": status.name, "filter": dim, "reason": reason})
return out
def load_timeline(
*,
remote: str,
org: str,
repo: str,
issue: int | None = None,
pr: int | None = None,
session: str | None = None,
limit: int | None = None,
offset: int | None = None,
db_path: str | None = None,
comment_source: CommentSource | None = None,
) -> TimelineSnapshot:
"""Aggregate every timeline source into one filtered, paginated snapshot.
Sources are read independently and fail soft: an unavailable source
contributes a ``SourceStatus`` with ``ok=False`` and a reason, and never
collapses the whole timeline. The handoff source only runs when a specific
issue or PR is requested (a handoff comment belongs to one thread) and a
``comment_source`` is available; otherwise it is reported as ``not run``
rather than as an empty-and-healthy source.
A filter dimension that no surviving source can carry a ``session``
filter when the only source that ran is the control plane, whose events
record no session is refused with ``ok=False`` and a structured error
instead of being answered with an empty page.
"""
all_events: list[WorkflowEvent] = []
statuses: list[SourceStatus] = []
cp_events, cp_status = read_cp_events(remote=remote, org=org, repo=repo, db_path=db_path)
all_events.extend(cp_events)
statuses.append(cp_status)
# Gitea handoff comments are thread-scoped: only fetch when the caller
# narrowed to one issue or PR, and only when a source was provided.
handoff_target: tuple[str, int] | None = None
if pr is not None:
handoff_target = ("pr", pr)
elif issue is not None:
handoff_target = ("issue", issue)
if handoff_target is None:
statuses.append(
_handoff_status(
ok=False,
reason="not run: handoff comments are thread-scoped; filter by issue or pr to include them",
)
)
elif comment_source is None:
statuses.append(
_handoff_status(
ok=False,
reason="not run: no comment source configured for this timeline read",
)
)
else:
kind, number = handoff_target
try:
comments = comment_source(kind, number) or []
handoff_events = adapt_cth_comments(comments, kind=kind, number=number)
all_events.extend(handoff_events)
statuses.append(_handoff_status(ok=True, count=len(handoff_events)))
except Exception as exc: # fail soft: a fetch/parse error degrades this source only
statuses.append(_handoff_status(ok=False, reason=f"handoff source failed: {exc}"))
requested = tuple(
name
for name, value in ((FILTER_ISSUE, issue), (FILTER_PR, pr), (FILTER_SESSION, session))
if value is not None
)
statuses = [
replace(
status,
unsupported_filters=tuple(
dim for dim in requested if dim not in status.supported_filters
),
)
for status in statuses
]
filters = {"issue": issue, "pr": pr, "session": session}
# A dimension is answerable only if a source that actually ran can carry it.
# If none can, refuse: an empty page would assert "no such activity", which
# is a claim this timeline is not in a position to make.
answerable: set[str] = set()
for status in statuses:
if status.ok:
answerable.update(status.supported_filters)
unanswerable = tuple(dim for dim in requested if dim not in answerable)
if unanswerable:
return TimelineSnapshot(
schema_version=TIMELINE_SCHEMA_VERSION,
remote=remote,
org=org,
repo=repo,
filters=filters,
page=paginate([], limit=limit, offset=offset),
sources=tuple(statuses),
ok=False,
error={
"code": "filter_not_supported",
"unsupported_filters": list(unanswerable),
"detail": (
"no timeline source that ran can answer "
+ ", ".join(f"'{dim}'" for dim in unanswerable)
+ "; the result is refused rather than returned empty"
),
"sources": _unanswerable_reasons(statuses, unanswerable),
},
)
filtered = filter_events(all_events, issue=issue, pr=pr, session=session)
ordered = sort_events(filtered)
page = paginate(ordered, limit=limit, offset=offset)
return TimelineSnapshot(
schema_version=TIMELINE_SCHEMA_VERSION,
remote=remote,
org=org,
repo=repo,
filters=filters,
page=page,
sources=tuple(statuses),
)
def snapshot_to_dict(snapshot: TimelineSnapshot) -> dict[str, Any]:
return snapshot.to_dict()