feat(webui): read-only system-health API (Closes #634 )

Adds `GET /api/v1/system/health`, a structured read-only health surface for automated readiness checks, and keeps `/health` as the cheap liveness probe. webui/system_health.py composes a DTO from fail-soft dependency probes: the control-plane database, the local checkout, and — opt-in via `?deep=1` — live Gitea reachability, each carrying status, reason, and probe latency. Required probes drive readiness; the optional Gitea probe can only degrade overall status, because local inventory stays serveable when the remote is unreachable. A probe that did not run leaves readiness incomplete rather than silently passing. Read-only throughout: the control-plane database is opened through a `mode=ro` URI because `ControlPlaneDB.__init__` creates directories and runs migrations, which a health check must never do. No restart or reload control is exposed; those are Phase 2 and #630 forbids process-kill recovery. No unproven claims: `stale_runtime.mutation_safe` is true only when the runtime, checkout, and remote commits are all known and equal, and MCP namespaces always report `unproven` because a web process cannot exercise the IDE-managed client path (#543). Probe details are redacted at the browser boundary — URLs lose userinfo and query strings, credential-shaped text is masked. `/health` is expanded additively: every MVP key is retained, plus `started_at`, `uptime_seconds`, and a pointer to the versioned API. The versioned route returns 503 when not ready so automation can branch on the status code alone. Verified at master 9eb0f29: focused file 40 passed / 11 subtests; `-k "webui or health"` 230 passed / 159 subtests; full suite 4358 passed with the 11 pre-existing master-drift failures unchanged from the clean-master baseline. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-22 15:59:30 -05:00
10 changed files with 1297 additions and 2070 deletions
@@ -1,295 +0,0 @@
 # Web console authorization, RBAC, redaction, and audit model (#633)
 **Phase 1. Read-only. This document defines the model that future console
 writes must pass through; it enables none of them.**
 The MVP deployment boundary ([`webui-deployment.md`](webui-deployment.md), #435)
 documents internal-only serving and states plainly that MVP authentication is
 *none* — protection comes from network placement. That is adequate while every
 route is a GET, and inadequate the moment a gated write ships. This document
 and the three modules it describes land **before** any write exists, so no
 Phase 2 action can be added without an authority to check it against.
 | Concern | Module |
 |---------|--------|
 | Identity, roles, authorization decision | `webui/console_authz.py` |
 | Secret redaction for every surface | `webui/console_redaction.py` |
 | Audit event schema, retention, sink | `webui/console_audit.py` |
 | Machine-readable publication | `GET /api/console/security-model` |
 Two invariants hold everywhere and are non-negotiable for every child of #631:
 1. **No secrets reach the browser.** Credentials are resolved server-side and
   redacted before any payload, page, log line, or audit record leaves.
 2. **No ungated mutations.** Authorization is necessary but never sufficient;
   execution stays disabled until the Phase 2 framework ships.
 ## Identity sources
 The console performs *authorization*. Authentication is delegated, because a
 console that mints its own sessions is a credential store, and this one must
 not be.
 | Source | Mode value | Authenticated | Shared host | Phase |
 |--------|-----------|---------------|-------------|-------|
 | None | `none` (default) | No — anonymous, capped at `viewer` | No | 1 |
 | Local dev | `local-dev` / `local_dev` | Yes, **asserted not verified** | No | 1 |
 | Access proxy | `access-proxy` / `access_proxy` | Yes, asserted by trusted proxy | Yes | 2 |
 Selected by `WEBUI_AUTH_MODE`. An unrecognised value falls back to `none`
 rather than erroring open.
 **Access-proxy mode** reads the subject from the
 `Cf-Access-Authenticated-User-Email` header, set by Cloudflare Access, WARP, or
 an equivalent org portal that terminates authentication in front of the
 console. If the header is absent the request did not traverse the proxy, so the
 principal degrades to anonymous — it is never trusted by default.
 The **role is always server-side configuration**, never a client assertion. It
 comes from `WEBUI_ROLE_MAP`, a JSON object of subject → role:
 ```json
 {"[email protected]": "operator", "[email protected]": "controller"}
 ```
 An unmapped subject gets `viewer`. Malformed JSON yields an empty map, so
 everyone gets `viewer` — a parse failure loses authority rather than granting
 it.
 Full SSO is explicitly a non-goal of this issue.
 ## Role matrix
 Four roles, ordered least to most authority. Each role inherits every lower
 role's actions; the table states the *minimum* rank required.
 | Role | Authority |
 |------|-----------|
 | `viewer` | Read every console view. No write, ever, in any phase. |
 | `operator` | Viewer, plus author-class work: claim, comment, open a PR. |
 | `controller` | Operator, plus reviewer/merger-class decisions on a PR. |
 | `admin` | Controller, plus destructive and policy-editing actions. |
 `viewer` holds the empty write set by construction, and a test asserts it stays
 empty.
 ## Privileged actions
 Every console action maps to a `task_key` in `task_capability_map.py`, the same
 single source of truth `gitea_resolve_task_capability` and the MCP tool gates
 use. The console therefore cannot invent an authority the MCP layer does not
 already define, and a regression test asserts each mapping matches.
 | Action | Minimum role | Class | MCP permission | Confirm | Dual control | Break-glass | Phase |
 |--------|--------------|-------|----------------|---------|--------------|-------------|-------|
 | `claim_issue` | operator | gated_write | `gitea.issue.comment` | Yes | No | No | 2 |
 | `comment_issue` | operator | gated_write | `gitea.issue.comment` | Yes | No | No | 2 |
 | `create_issue` | operator | gated_write | `gitea.issue.create` | Yes | No | No | 2 |
 | `comment_pr` | operator | gated_write | `gitea.pr.comment` | Yes | No | No | 2 |
 | `create_pr` | operator | gated_write | `gitea.pr.create` | Yes | No | No | 2 |
 | `review_pr` | controller | privileged | `gitea.pr.review` | Yes | No | No | 3 |
 | `close_pr` | controller | privileged | `gitea.pr.close` | Yes | No | No | 3 |
 | `merge_pr` | controller | privileged | `gitea.pr.merge` | Yes | **Yes** | **Yes** | 3 |
 | `delete_branch` | admin | destructive | `gitea.branch.delete` | Yes | **Yes** | **Yes** | 3 |
 **Dual control** means the acting principal may not be the sole authority: a
 second distinct principal must confirm. **Break-glass** means the action is
 expected to be unavailable in normal operation and its use is retained for two
 years. Both are declared here and enforced by the Phase 2 framework; Phase 1
 records the requirement on every decision so the framework cannot ship without
 honouring it.
 `delete_branch` is admin-only rather than controller because it is the one
 irreversible action in the set.
 ### Authorization decision
 `authorize(action_id, principal, for_execution=False)` returns a decision
 record and **denies by default**. The deny reasons are closed and enumerated:
 | Reason code | Meaning |
 |-------------|---------|
 | `unknown_action` | No such console action is registered. |
 | `unauthenticated` | The principal is anonymous. |
 | `unknown_role` | The role is not in the matrix. |
 | `insufficient_role` | The role ranks below the action's minimum. |
 | `phase_not_active` | Execution requested for an action whose phase is not open. |
 | `allowed_preview_only` | Authorized — preview only, execution still disabled. |
 There is no implicit allow branch. Even the allow result reports
 `execution_enabled: false` while the console is in Phase 1, so no caller can
 read an allow as permission to mutate.
 ## Secret redaction
 One pass applies to **API payloads, rendered HTML, server logs, and audit
 records** — the four surfaces where a credential could escape.
 Redaction reuses `gitea_audit.redact` rather than forking it: that remains the
 authority for secret-looking dict keys, `Authorization` material, and raw URLs.
 The console layer then applies its own patterns:
 Each rule below matches an *assignment form*: the named key, followed by `=` or
 `:`, followed by the value. The keys are listed bare rather than spelled out as
 complete assignments, because this document is itself scanned by
 `scan_for_secrets` — writing the examples in full assignment form would make the
 documentation trip the very detectors it documents.
 | Rule | Catches (as an assignment) |
 |------|----------------------------|
 | `credential_assignment` | `token`, `password`, `passwd`, `secret`, `api_key`, `access_key`, `client_secret`, `private_key` |
 | `credential_env_assignment` | `GITEA_TOKEN`, `GITEA_PASS`, `GITEA_PASSWORD` and suffixed variants |
 | `keychain_reference` | `keychain:` entry references |
 | `keychain_command` | macOS `security` keychain lookups (`find-generic-password`, `find-internet-password`) |
 | `private_key_block` | PEM `BEGIN ... PRIVATE KEY` blocks |
 | `json_web_token` | Three-segment `eyJ...` JWTs |
 | `bearer_credential` | `Bearer` / `Basic` credentials |
 Assignments keep the key and replace only the value, so an operator can still
 see *what* was removed. Two behaviours are deliberate:
 - **Fail closed.** A value that cannot be redacted becomes `[REDACTED]`
  outright rather than being emitted raw. Redaction never raises.
 - **Redact before persist.** `console_audit.build_event` redacts before
  serialization, and `write_event` re-scans and **drops** any record that still
  trips a detector. An unredacted record is never durable.
 `scan_for_secrets` is the assertion helper: it returns the detector names still
 matching a payload, and already-redacted hits are not findings. Tests use it to
 prove the published policy, the security-model endpoint, and this document
 itself carry no secret material.
 ## Audit event schema
 `gitea_audit` records MCP-side *mutations* — which profile and Gitea user
 performed which tool call. It has no console actor, no identity source, no
 correlation identifier, and no retention class, and an authorization **denial**
 is not a mutation, so it would never appear there at all. The console record is
 additive, not a replacement: a Phase 2 action emits both, joined on
 `correlation.request_id`.
 Required fields, all asserted by tests so an edit cannot quietly drop one:
 | Field | Content |
 |-------|---------|
 | `schema_version` | Currently `1`. |
 | `event_id` | Unique per record. |
 | `timestamp` | Timezone-aware ISO-8601, UTC. |
 | `actor` | `subject`, `role`, `identity_source`, `authenticated`. |
 | `action` | Console action id. |
 | `action_class` | `gated_write`, `privileged`, `destructive`, or `unknown`. |
 | `target` | `{kind, ref}`, e.g. `{"kind": "pr", "ref": "#123"}`. |
 | `result` | `allowed`, `denied`, `previewed`, `failed`, `succeeded`. |
 | `reason_code` | The authorization reason code above. |
 | `correlation` | `request_id`, `session_id`, `mcp_task`, `mcp_permission`. |
 | `retention` | `class`, `days`, `expires_at`. |
 | `redacted` | Always `true`; records are redacted at build time. |
 An unrecognised `result` degrades to `failed` rather than being stored
 verbatim.
 The sink is an append-only JSON Lines file named by
 `WEBUI_CONSOLE_AUDIT_LOG`. It is **off by default**: with the variable unset,
 events are still built — so callers and tests exercise the schema — but nothing
 is written. Auditing never raises; a failed write returns `False` rather than
 breaking the request it describes.
 ## Retention
 | Class | Applies to | Default |
 |-------|-----------|---------|
 | `standard` | Routine gated writes | 90 days |
 | `privileged` | `review_pr`, `close_pr`, and any unclassifiable action | 365 days |
 | `break_glass` | `merge_pr`, `delete_branch` | 730 days |
 Each record carries its own class, day count, and computed `expires_at`, so
 retention is auditable per record rather than inferred from file age. An
 **unknown action is retained as privileged, not standard** — for a safety
 control the conservative direction is to keep the record longer.
 Nothing in this module updates or deletes. Expiry is enforced by an
 operator-run policy against `expires_at`, never by the console silently
 rewriting its own history.
 ## Phase 2 integration
 Phase 2 opens gated writes. It must reuse this model rather than introduce a
 second one. The integration points are already wired and observable:
 - **`GET /api/actions/{action_id}/preview`** attaches an `authorization` block
  to the existing preview payload and records a `previewed` audit event.
 - **`POST /api/actions/{action_id}/attempt`** attaches the same block and
  records a `denied` event. The terminal outcome is unchanged — the MVP
  registry in `webui/gated_actions.py` still fails closed for every action — so
  Phase 1 cannot loosen anything. Phase 2 enforces on this same decision
  instead of adding a parallel check.
 - **`GET /api/console/security-model`** publishes the RBAC matrix, redaction
  policy, and audit policy as JSON for operators and tests.
 To open Phase 2, a child issue must: raise `ACTIVE_PHASE`, implement the
 confirmation and dual-control flow the matrix already declares, emit a
 `succeeded` or `failed` record alongside the `gitea_audit` mutation record, and
 keep `viewer` unable to reach any of it. Turning on execution without the
 confirmation flow contradicts a declared requirement and is a review failure,
 not a shortcut.
 ## Local-dev mode
 `WEBUI_AUTH_MODE=local-dev` reads the principal straight from the environment:
 | Variable | Purpose |
 |----------|---------|
 | `WEBUI_DEV_SUBJECT` | Subject string; absent ⇒ anonymous |
 | `WEBUI_DEV_ROLE` | One of `viewer`, `operator`, `controller`, `admin`; unrecognised ⇒ `viewer` |
 **INSECURE — this mode is for loopback development only.** The subject and role
 are *asserted by the developer running the process and verified by nothing*.
 Anyone able to set an environment variable on the host is an `admin`, and
 anyone able to reach the port inherits that principal. It provides no
 authentication whatsoever; it exists so Phase 2 authorization paths can be
 exercised without standing up a proxy.
 Never enable local-dev mode on a non-loopback bind. Combining it with
 `WEBUI_ALLOW_PUBLIC_BIND=1` or `WEBUI_ALLOW_REMOTE_BIND=1` publishes an
 unauthenticated admin console.
 For anything beyond a laptop use `access-proxy` mode behind Cloudflare Access,
 WARP, or a VPN, as [`webui-deployment.md`](webui-deployment.md) requires.
 ### Probe authentication
 `WEBUI_REQUIRE_PROBE_AUTH=1` declares that non-public probes should require an
 authenticated principal. It is **opt-in**: the default is off so the MVP
 `/health` contract is unchanged.
 **This flag is declarative in Phase 1 and enforces nothing today.**
 `console_authz.probe_auth_required()` reports the operator's intent, and no
 route consults it — setting the variable does not currently change the
 behaviour of `/health` or any other endpoint. It is published here so the Phase
 2 action framework has a declared policy to honour rather than inventing a
 second one, exactly as `ACTIVE_PHASE` gates execution while the matrix is
 already declared. A regression test pins this "declared, not enforced" status,
 so wiring it later is a deliberate change rather than a silent one.
 Until Phase 2 wires it, probe protection rests on network placement alone, as
 [`webui-deployment.md`](webui-deployment.md) (#435) states.
 ## Environment variables
 | Variable | Default | Purpose |
 |----------|---------|---------|
 | `WEBUI_AUTH_MODE` | `none` | Identity source selection |
 | `WEBUI_DEV_SUBJECT` | unset | Local-dev subject (insecure) |
 | `WEBUI_DEV_ROLE` | `viewer` | Local-dev role (insecure) |
 | `WEBUI_ROLE_MAP` | unset | JSON subject → role map |
 | `WEBUI_REQUIRE_PROBE_AUTH` | unset | Require auth for non-public probes |
 | `WEBUI_CONSOLE_AUDIT_LOG` | unset | Append-only audit sink path |
 All are read server-side only. None is ever rendered into a page or returned by
 an API.
 ## Non-goals
 - No full SSO product; authentication stays delegated to the proxy.
 - No browser-initiated merges or approvals in any phase covered here.
 - No tokens in the frontend, in browser storage, or in committed config.
@@ -7,10 +7,7 @@ only.
 ## MVP deployment model
 - **Default bind:** `127.0.0.1:8765` (`WEBUI_HOST` / `WEBUI_PORT`)
- **Authentication:** none in MVP — protection comes from network placement.
+- **Authentication:** none in MVP — protection comes from network placement
  The authorization, RBAC, redaction, and audit model that future gated writes
  must pass through is defined in
  [`webui-authz-audit.md`](webui-authz-audit.md) (#633).
 - **Mutations:** read-only routes; gated write actions remain disabled (#434)
 - **Secrets:** resolved server-side via `gitea_auth` / `GITEA_MCP_CONFIG`; never
  embedded in HTML, JavaScript, or browser storage
@@ -48,7 +48,8 @@ that govern when a write path may open (#632, epic #631).
 | Path | Description |
 |------|-------------|
 | `/` | Home / operator overview |
-| `/health` | JSON liveness (`status`, `service`, `mode`, `timestamp`) |
+| `/health` | JSON liveness (`status`, `service`, `mode`, `timestamp`, `uptime_seconds`) |
 | `/api/v1/system/health` | Structured read-only system health (#634) |
 | `/queue` | Live PR and issue queue dashboard (#429) |
 | `/api/queue` | JSON queue export with pagination metadata |
 | `/projects` | Project registry list (#427) |
@@ -72,6 +73,85 @@ Most routes are GET-only. POST/PUT/PATCH/DELETE return `405` with
 `read-only-mvp`, except `/audit` and `/api/audit` which accept POST for
 local validator preview only (no Gitea mutations, no server-side storage).
 ## System health API (#634)
 `GET /api/v1/system/health` is the structured, read-only health surface for
 automated readiness checks. It is the first console API under the `/api/v1`
 prefix; the unversioned MVP exports remain as compatibility aliases.
 `/health` is unchanged for existing consumers — every MVP key is still present
 — and now also carries `started_at`, `uptime_seconds`, and a
 `system_health_api` pointer. It stays deliberately cheap and runs no dependency
 probe, because answering readiness costs real work.
 **Status codes.** `200` when ready, `503` when a required dependency failed or
 was never probed. Automation can branch on the code without parsing the body.
 **Query flags.** The Gitea check is a network call, so it is opt-in:
 `GET /api/v1/system/health?deep=1` runs it and caches the result for
 `WEBUI_HEALTH_PROBE_TTL_SECONDS` (default 15s) so dashboard polling does not
 amplify into remote load. Without the flag that probe reports `skipped`.
 **Dependencies.** `control_plane_db` and `repository` are required and drive
 readiness. `gitea` is optional: when it fails the overall `status` degrades but
 `readiness.ready` stays true, because local inventory is still serveable. Each
 entry carries `status`, `detail`, `required`, and `latency_ms`.
 Two honesty rules are worth knowing before reading the payload:
 * `stale_runtime.mutation_safe` is true only when the runtime, checkout, and
  remote-tracking commits are all known and equal. An unfetched remote is
  reported as indeterminate, never as safe.
 * `mcp_namespaces` entries are always `unproven`. A web process runs outside
  the IDE-managed MCP client and cannot invoke a namespace tool, so per #543
  only a `client_namespace` probe can prove that path.
 Sample response (abridged, healthy):
 ```json
 {
  "status": "ok",
  "service": "mcp-control-plane-webui",
  "mode": "read-only",
  "api": "/api/v1/system/health",
  "timestamp": "2026-07-22T11:04:18.512034+00:00",
  "readiness": { "ready": true, "complete": true, "reasons": [] },
  "version": {
    "git_sha": "620ed6e9a9550b8da2ceb82d9ab8744e8920490f",
    "git_describe": "v1.1.0-898-g620ed6e",
    "control_plane_schema_version": 4,
    "python_version": "3.14.5",
    "known": true
  },
  "process": { "started_at": "2026-07-22T10:58:02.114+00:00", "uptime_seconds": 376.4 },
  "deep_probes_requested": false,
  "dependencies": [
    {
      "name": "control_plane_db",
      "kind": "sqlite",
      "status": "ok",
      "detail": "schema v4 readable",
      "required": true,
      "healthy": true,
      "latency_ms": 1.482,
      "metadata": { "schema_version": 4, "active_leases": 3 }
    },
    { "name": "repository", "kind": "git", "status": "ok", "required": true, "healthy": true },
    { "name": "gitea", "kind": "http", "status": "skipped", "required": false, "healthy": false }
  ],
  "mcp_namespaces": [
    { "namespace": "gitea-author", "required_tool": "gitea_whoami", "status": "unproven" }
  ],
  "stale_runtime": { "stale": false, "determinable": true, "mutation_safe": true, "reasons": [] },
  "probe_errors": []
 }
 ```
 No restart, reload, or process-kill control is exposed here: those are Phase 2
 at the earliest, and #630 forbids process-kill recovery outright. Every probe
 opens its subject read-only — the control-plane database is opened through a
 `mode=ro` URI so a health check can never create or migrate a schema.
 ## Report audit (#431)
 Paste an LLM final report at `/audit` or POST JSON to `/api/audit`. The UI
@@ -1,703 +0,0 @@
 """Console authorization, redaction, and audit model tests (#633).
 Covers each acceptance criterion and each required test named in the issue:
 * AC1 — RBAC matrix and privileged-action list.
 * AC2 — redaction rules, unit-tested against sample payloads.
 * AC3 — audit event schema with required fields and retention defaults.
 * AC4 — Phase 2 integration points.
 * AC5 — local-dev mode with explicit insecurity warnings.
 Required tests: redaction units (token, keychain, password patterns),
 default-deny for unauthenticated write stubs, and audit record creation for a
 simulated privileged preview.
 """
 from __future__ import annotations
 import datetime
 import json
 import os
 import pathlib
 import sys
 import tempfile
 import unittest
 from starlette.testclient import TestClient
 sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))
 from task_capability_map import TASK_CAPABILITY_MAP  # noqa: E402
 from webui import console_audit, console_authz  # noqa: E402
 from webui.app import create_app  # noqa: E402
 from webui.console_redaction import (  # noqa: E402
    REDACTED,
    redact_payload,
    redact_text,
    redaction_policy,
    scan_for_secrets,
 )
 DOCS = pathlib.Path(__file__).resolve().parents[1] / "docs"
 AUTHZ_DOC = DOCS / "webui-authz-audit.md"
 def _principal(role: str) -> console_authz.Principal:
    return console_authz.Principal(
        subject=f"{role}@example.com",
        role=role,
        identity_source=console_authz.IDENTITY_ACCESS_PROXY,
        authenticated=True,
    )
 class TestRoleMatrix(unittest.TestCase):
    """AC1 — the written RBAC matrix and privileged-action list."""
    def test_roles_are_ordered_least_to_most_authority(self):
        self.assertEqual(
            console_authz.ROLE_ORDER,
            ("viewer", "operator", "controller", "admin"),
        )
    def test_every_role_has_a_description(self):
        for role in console_authz.ROLE_ORDER:
            with self.subTest(role=role):
                self.assertTrue(console_authz.ROLE_DESCRIPTIONS[role].strip())
    def test_higher_roles_inherit_lower_role_actions(self):
        matrix = {
            entry["role"]: set(entry["permitted_actions"])
            for entry in console_authz.rbac_matrix()["roles"]
        }
        for lower, higher in zip(
            console_authz.ROLE_ORDER, console_authz.ROLE_ORDER[1:]
        ):
            with self.subTest(lower=lower, higher=higher):
                self.assertTrue(matrix[lower].issubset(matrix[higher]))
    def test_viewer_holds_no_write_action(self):
        matrix = {
            entry["role"]: set(entry["permitted_actions"])
            for entry in console_authz.rbac_matrix()["roles"]
        }
        self.assertEqual(matrix["viewer"], set())
    def test_privileged_action_list_is_non_empty_and_classified(self):
        privileged = console_authz.privileged_actions()
        self.assertTrue(privileged)
        ids = {action.action_id for action in privileged}
        # Merge and branch deletion are the canonical privileged pair.
        self.assertIn("merge_pr", ids)
        self.assertIn("delete_branch", ids)
    def test_merge_and_delete_require_dual_control_and_break_glass(self):
        for action_id in ("merge_pr", "delete_branch"):
            with self.subTest(action=action_id):
                action = console_authz.get_action(action_id)
                self.assertTrue(action.dual_control)
                self.assertTrue(action.break_glass)
                self.assertTrue(action.requires_confirmation)
    def test_every_write_action_requires_confirmation(self):
        for action in console_authz.ACTIONS.values():
            with self.subTest(action=action.action_id):
                self.assertTrue(action.requires_confirmation)
    def test_delete_branch_is_admin_only(self):
        self.assertEqual(
            console_authz.get_action("delete_branch").minimum_role,
            console_authz.ADMIN,
        )
    def test_actions_map_to_real_mcp_capability_vocabulary(self):
        """The console must not invent an authority the MCP layer lacks."""
        for action in console_authz.ACTIONS.values():
            with self.subTest(action=action.action_id):
                self.assertIn(action.task_key, TASK_CAPABILITY_MAP)
                self.assertEqual(
                    action.mcp_permission,
                    TASK_CAPABILITY_MAP[action.task_key]["permission"],
                )
                self.assertEqual(
                    action.mcp_role,
                    TASK_CAPABILITY_MAP[action.task_key]["role"],
                )
    def test_matrix_declares_deny_by_default_and_execution_disabled(self):
        matrix = console_authz.rbac_matrix()
        self.assertEqual(matrix["default_decision"], "deny")
        self.assertFalse(matrix["execution_enabled"])
 class TestAuthorizeDefaultDeny(unittest.TestCase):
    """Fail-closed behaviour of the authorization decision."""
    def test_anonymous_is_denied_every_action(self):
        for action_id in console_authz.ACTIONS:
            with self.subTest(action=action_id):
                decision = console_authz.authorize(action_id)
                self.assertFalse(decision.allowed)
                self.assertEqual(
                    decision.reason_code, console_authz.DENY_UNAUTHENTICATED
                )
    def test_unknown_action_is_denied(self):
        decision = console_authz.authorize(
            "not_a_real_action", _principal("admin")
        )
        self.assertFalse(decision.allowed)
        self.assertEqual(decision.reason_code, console_authz.DENY_UNKNOWN_ACTION)
    def test_unknown_role_is_denied(self):
        rogue = console_authz.Principal(
            subject="[email protected]",
            role="superuser",
            identity_source=console_authz.IDENTITY_ACCESS_PROXY,
            authenticated=True,
        )
        decision = console_authz.authorize("comment_issue", rogue)
        self.assertFalse(decision.allowed)
        self.assertEqual(decision.reason_code, console_authz.DENY_UNKNOWN_ROLE)
    def test_insufficient_role_is_denied(self):
        decision = console_authz.authorize("merge_pr", _principal("operator"))
        self.assertFalse(decision.allowed)
        self.assertEqual(
            decision.reason_code, console_authz.DENY_INSUFFICIENT_ROLE
        )
    def test_sufficient_role_allows_preview_only(self):
        decision = console_authz.authorize("merge_pr", _principal("controller"))
        self.assertTrue(decision.allowed)
        self.assertFalse(decision.execution_enabled)
    def test_execution_is_refused_while_phase_is_not_active(self):
        decision = console_authz.authorize(
            "merge_pr", _principal("controller"), for_execution=True
        )
        self.assertFalse(decision.allowed)
        self.assertEqual(
            decision.reason_code, console_authz.DENY_PHASE_NOT_ACTIVE
        )
    def test_allowed_decision_never_reports_execution_enabled(self):
        for action_id in console_authz.ACTIONS:
            with self.subTest(action=action_id):
                decision = console_authz.authorize(
                    action_id, _principal("admin")
                )
                self.assertFalse(decision.execution_enabled)
 class TestIdentityResolution(unittest.TestCase):
    """AC5 — identity sources, including the insecure local-dev mode."""
    def test_no_auth_mode_yields_anonymous_viewer(self):
        principal = console_authz.resolve_principal(env={})
        self.assertFalse(principal.authenticated)
        self.assertEqual(principal.role, console_authz.VIEWER)
        self.assertEqual(principal.identity_source, console_authz.IDENTITY_NONE)
    def test_local_dev_mode_warns_that_identity_is_unverified(self):
        principal = console_authz.resolve_principal(
            env={
                console_authz.AUTH_MODE_ENV: "local-dev",
                console_authz.DEV_SUBJECT_ENV: "[email protected]",
                console_authz.DEV_ROLE_ENV: "admin",
            }
        )
        self.assertTrue(principal.authenticated)
        self.assertEqual(principal.role, "admin")
        self.assertTrue(principal.warnings)
        self.assertIn("asserted", " ".join(principal.warnings).lower())
    def test_local_dev_without_subject_falls_back_to_anonymous(self):
        principal = console_authz.resolve_principal(
            env={console_authz.AUTH_MODE_ENV: "local-dev"}
        )
        self.assertFalse(principal.authenticated)
    def test_local_dev_unknown_role_degrades_to_viewer(self):
        principal = console_authz.resolve_principal(
            env={
                console_authz.AUTH_MODE_ENV: "local_dev",
                console_authz.DEV_SUBJECT_ENV: "[email protected]",
                console_authz.DEV_ROLE_ENV: "root",
            }
        )
        self.assertEqual(principal.role, console_authz.VIEWER)
    def test_access_proxy_without_header_fails_closed(self):
        """A proxy-mode request that did not traverse the proxy is anonymous."""
        principal = console_authz.resolve_principal(
            headers={},
            env={console_authz.AUTH_MODE_ENV: "access_proxy"},
        )
        self.assertFalse(principal.authenticated)
    def test_access_proxy_role_comes_from_server_config_not_client(self):
        env = {
            console_authz.AUTH_MODE_ENV: "access_proxy",
            console_authz.ROLE_MAP_ENV: json.dumps(
                {"[email protected]": "controller"}
            ),
        }
        principal = console_authz.resolve_principal(
            headers={
                console_authz.ACCESS_SUBJECT_HEADER: "[email protected]",
                "x-role": "admin",  # client-supplied role must be ignored
            },
            env=env,
        )
        self.assertEqual(principal.role, "controller")
    def test_access_proxy_unmapped_subject_defaults_to_viewer(self):
        principal = console_authz.resolve_principal(
            headers={
                console_authz.ACCESS_SUBJECT_HEADER: "[email protected]"
            },
            env={console_authz.AUTH_MODE_ENV: "access_proxy"},
        )
        self.assertEqual(principal.role, console_authz.VIEWER)
    def test_malformed_role_map_does_not_raise_and_denies(self):
        principal = console_authz.resolve_principal(
            headers={console_authz.ACCESS_SUBJECT_HEADER: "[email protected]"},
            env={
                console_authz.AUTH_MODE_ENV: "access_proxy",
                console_authz.ROLE_MAP_ENV: "{not json",
            },
        )
        self.assertEqual(principal.role, console_authz.VIEWER)
    def test_probe_auth_is_opt_in(self):
        self.assertFalse(console_authz.probe_auth_required(env={}))
        self.assertTrue(
            console_authz.probe_auth_required(
                env={console_authz.REQUIRE_PROBE_AUTH_ENV: "1"}
            )
        )
    def test_probe_auth_is_declared_but_not_yet_enforced(self):
        """Phase 1 declares the probe-auth policy; no route enforces it yet.
        The flag exists so the Phase 2 action framework has a declared policy
        to honour instead of inventing a second one. Pinning the current
        not-enforced status here means wiring it later is a deliberate change
        that updates this test and the documentation together, rather than a
        silent behaviour shift. The documentation must say so plainly, because
        an operator who sets the variable believing it protects a probe is
        worse off than one who knows it does not.
        """
        import inspect
        from webui import app as webui_app
        source = inspect.getsource(webui_app)
        self.assertNotIn(
            "probe_auth_required",
            source,
            msg=(
                "webui.app now consults probe_auth_required, so probe auth is "
                "no longer merely declared. Update the 'Probe authentication' "
                "section of docs/webui-authz-audit.md, which states it "
                "enforces nothing, and replace this test with real "
                "enforcement coverage."
            ),
        )
        self.assertIn(
            "enforces nothing today",
            AUTHZ_DOC.read_text(encoding="utf-8"),
        )
 class TestRedaction(unittest.TestCase):
    """AC2 — required redaction units: token, keychain, password patterns."""
    def test_token_assignment_is_redacted(self):
        out = redact_text("GITEA_TOKEN=abcd1234efgh5678ijkl")
        self.assertIn(REDACTED, out)
        self.assertNotIn("abcd1234efgh5678ijkl", out)
    def test_password_assignment_is_redacted(self):
        out = redact_text("password: hunter2supersecret")
        self.assertIn(REDACTED, out)
        self.assertNotIn("hunter2supersecret", out)
    def test_keychain_reference_is_redacted(self):
        out = redact_text("keychain:gitea-prgs-token")
        self.assertIn(REDACTED, out)
        self.assertNotIn("gitea-prgs-token", out)
    def test_keychain_command_is_redacted(self):
        out = redact_text("security find-generic-password -s gitea -w")
        self.assertIn(REDACTED, out)
        self.assertNotIn("find-generic-password -s gitea", out)
    def test_bearer_credential_is_redacted(self):
        out = redact_text("Authorization: Bearer abcdef1234567890abcdef")
        self.assertNotIn("abcdef1234567890abcdef", out)
    def test_jwt_is_redacted(self):
        token = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxIn0.abcdefghijklmnop"
        out = redact_text(f"session={token}")
        self.assertNotIn(token, out)
    def test_private_key_block_is_redacted(self):
        pem = (
            "-----BEGIN RSA PRIVATE KEY-----\n"
            "MIIEowIBAAKCAQEAsecretmaterial\n"
            "-----END RSA PRIVATE KEY-----"
        )
        out = redact_text(pem)
        self.assertNotIn("MIIEowIBAAKCAQEAsecretmaterial", out)
    def test_api_key_assignment_is_redacted(self):
        out = redact_text('api_key = "sk-live-9f8e7d6c5b4a3210"')
        self.assertNotIn("sk-live-9f8e7d6c5b4a3210", out)
    def test_nested_payload_is_redacted_recursively(self):
        payload = {
            "token": "abc123456789",
            "nested": {"note": "password=letmein12345"},
            "list": ["keychain:some-entry"],
            "safe": "plain text",
        }
        out = redact_payload(payload)
        self.assertEqual(out["token"], REDACTED)
        self.assertNotIn("letmein12345", json.dumps(out))
        self.assertNotIn("some-entry", json.dumps(out))
        self.assertEqual(out["safe"], "plain text")
    def test_scan_reports_findings_before_and_none_after(self):
        dirty = "password: hunter2supersecret"
        self.assertTrue(scan_for_secrets(dirty))
        self.assertEqual(scan_for_secrets(redact_text(dirty)), [])
    def test_non_strings_pass_through_untouched(self):
        self.assertEqual(redact_text(42), 42)
        self.assertEqual(
            redact_payload({"n": 1, "b": True}), {"n": 1, "b": True}
        )
    def test_policy_is_documented_and_declares_redact_before_persist(self):
        policy = redaction_policy()
        self.assertTrue(policy["redact_before_persist"])
        self.assertIn("audit_records", policy["applies_to"])
        self.assertTrue(policy["console_rules"])
    def test_policy_statement_contains_no_secret_material(self):
        self.assertEqual(scan_for_secrets(redaction_policy()), [])
 class TestAuditSchema(unittest.TestCase):
    """AC3 — audit event schema, required fields, and retention defaults."""
    def _event(self, action_id="merge_pr", **kwargs):
        return console_audit.build_event(
            action_id=action_id,
            result=console_audit.RESULT_DENIED,
            decision=console_authz.authorize(action_id, _principal("operator")),
            target={"kind": "pr", "ref": "#123"},
            request_id="req-test",
            **kwargs,
        )
    def test_every_required_field_is_present(self):
        event = self._event()
        for field in console_audit.REQUIRED_FIELDS:
            with self.subTest(field=field):
                self.assertIn(field, event)
    def test_actor_carries_who_and_how_they_were_identified(self):
        event = self._event()
        for field in console_audit.REQUIRED_ACTOR_FIELDS:
            with self.subTest(field=field):
                self.assertIn(field, event["actor"])
    def test_correlation_ids_are_present(self):
        event = self._event()
        for field in console_audit.REQUIRED_CORRELATION_FIELDS:
            with self.subTest(field=field):
                self.assertIn(field, event["correlation"])
        self.assertEqual(event["correlation"]["request_id"], "req-test")
        self.assertEqual(event["correlation"]["mcp_task"], "merge_pr")
    def test_timestamp_is_timezone_aware_utc_iso8601(self):
        now = datetime.datetime(
            2026, 7, 22, 10, 16, 42, tzinfo=datetime.timezone.utc
        )
        event = self._event(now=now)
        self.assertEqual(event["timestamp"], "2026-07-22T10:16:42+00:00")
        parsed = datetime.datetime.fromisoformat(event["timestamp"])
        self.assertIsNotNone(parsed.tzinfo)
    def test_retention_defaults_by_class(self):
        self.assertEqual(
            console_audit.RETENTION_DAYS[console_audit.RETENTION_STANDARD], 90
        )
        self.assertEqual(
            console_audit.RETENTION_DAYS[console_audit.RETENTION_PRIVILEGED],
            365,
        )
        self.assertEqual(
            console_audit.RETENTION_DAYS[console_audit.RETENTION_BREAK_GLASS],
            730,
        )
    def test_break_glass_action_retains_longest(self):
        event = self._event("merge_pr")
        self.assertEqual(
            event["retention"]["class"], console_audit.RETENTION_BREAK_GLASS
        )
    def test_routine_write_uses_standard_retention(self):
        event = self._event("comment_issue")
        self.assertEqual(
            event["retention"]["class"], console_audit.RETENTION_STANDARD
        )
    def test_unknown_action_retains_as_privileged_not_standard(self):
        """Conservative direction: keep an unclassifiable record longer."""
        self.assertEqual(
            console_audit.retention_class_for(None),
            console_audit.RETENTION_PRIVILEGED,
        )
    def test_retention_expiry_matches_declared_days(self):
        now = datetime.datetime(2026, 7, 22, tzinfo=datetime.timezone.utc)
        event = self._event("comment_issue", now=now)
        expires = datetime.datetime.fromisoformat(
            event["retention"]["expires_at"]
        )
        self.assertEqual((expires - now).days, 90)
    def test_invalid_result_degrades_to_failed(self):
        event = console_audit.build_event(action_id="merge_pr", result="banana")
        self.assertEqual(event["result"], console_audit.RESULT_FAILED)
    def test_denied_result_is_representable(self):
        """An authorization denial has no MCP-side mutation record."""
        self.assertIn(console_audit.RESULT_DENIED, console_audit.RESULTS)
    def test_event_is_redacted_before_it_is_returned(self):
        event = console_audit.build_event(
            action_id="merge_pr",
            result=console_audit.RESULT_DENIED,
            detail="failed with token=abcdef1234567890",
            metadata={"password": "hunter2supersecret"},
        )
        serialized = json.dumps(event)
        self.assertNotIn("abcdef1234567890", serialized)
        self.assertNotIn("hunter2supersecret", serialized)
        self.assertTrue(event["redacted"])
    def test_audit_policy_reports_schema_and_retention(self):
        policy = console_audit.audit_policy()
        self.assertTrue(policy["append_only"])
        self.assertTrue(policy["redact_before_persist"])
        self.assertEqual(
            policy["retention_defaults_days"], console_audit.RETENTION_DAYS
        )
 class TestAuditSink(unittest.TestCase):
    """Append-only persistence behaviour."""
    def test_write_is_a_noop_when_sink_is_unconfigured(self):
        saved = os.environ.pop(console_audit.AUDIT_LOG_ENV, None)
        try:
            self.assertFalse(console_audit.audit_enabled())
            self.assertFalse(console_audit.write_event({"schema_version": 1}))
        finally:
            if saved is not None:
                os.environ[console_audit.AUDIT_LOG_ENV] = saved
    def test_records_append_one_json_line_each(self):
        with tempfile.TemporaryDirectory() as tmp:
            sink = os.path.join(tmp, "console-audit.jsonl")
            for _ in range(3):
                event = console_audit.build_event(
                    action_id="merge_pr", result=console_audit.RESULT_DENIED
                )
                self.assertTrue(console_audit.write_event(event, path=sink))
            with open(sink, encoding="utf-8") as handle:
                lines = [json.loads(line) for line in handle if line.strip()]
            self.assertEqual(len(lines), 3)
            self.assertEqual(len({line["event_id"] for line in lines}), 3)
    def test_a_record_that_still_carries_a_secret_is_not_persisted(self):
        with tempfile.TemporaryDirectory() as tmp:
            sink = os.path.join(tmp, "console-audit.jsonl")
            leaky = {
                "schema_version": 1,
                "detail": "password: hunter2supersecret",
            }
            self.assertFalse(console_audit.write_event(leaky, path=sink))
            self.assertFalse(os.path.exists(sink))
    def test_write_never_raises_on_a_bad_path(self):
        self.assertFalse(
            console_audit.write_event(
                {"schema_version": 1}, path="/nonexistent-dir/audit.jsonl"
            )
        )
    def test_simulated_privileged_preview_creates_an_audit_record(self):
        """Required test: audit record creation for a privileged preview."""
        with tempfile.TemporaryDirectory() as tmp:
            sink = os.path.join(tmp, "console-audit.jsonl")
            os.environ[console_audit.AUDIT_LOG_ENV] = sink
            try:
                decision = console_authz.authorize(
                    "merge_pr", _principal("controller")
                )
                outcome = console_audit.record_event(
                    action_id="merge_pr",
                    result=console_audit.RESULT_PREVIEWED,
                    decision=decision,
                    target={"kind": "pr", "ref": "#123"},
                    request_id="req-preview",
                )
            finally:
                os.environ.pop(console_audit.AUDIT_LOG_ENV, None)
            self.assertTrue(outcome["written"])
            with open(sink, encoding="utf-8") as handle:
                record = json.loads(handle.read().strip())
            self.assertEqual(record["action"], "merge_pr")
            self.assertEqual(record["result"], console_audit.RESULT_PREVIEWED)
            self.assertEqual(record["action_class"], "privileged")
            self.assertTrue(record["decision"]["allowed"])
            self.assertFalse(record["decision"]["execution_enabled"])
            self.assertEqual(record["actor"]["role"], "controller")
    def test_decision_block_survives_redaction(self):
        """Regression: naming it 'authorization' collided with a secret hint.
        ``gitea_audit._SECRET_KEY_HINTS`` contains "authorization" (for the
        HTTP header), so a block under that key was replaced wholesale by the
        placeholder and the record lost its decision entirely.
        """
        event = console_audit.build_event(
            action_id="merge_pr",
            result=console_audit.RESULT_DENIED,
            decision=console_authz.authorize("merge_pr", _principal("admin")),
        )
        self.assertIsInstance(event["decision"], dict)
        self.assertIn("allowed", event["decision"])
 class TestConsoleRoutes(unittest.TestCase):
    """AC4 — the wired Phase 2 integration points, still fail-closed."""
    def setUp(self):
        self.client = TestClient(create_app(bind_host="127.0.0.1"))
    def test_unauthenticated_write_stub_is_denied(self):
        """Required test: default-deny for unauthenticated write stubs."""
        response = self.client.post(
            "/api/actions/merge_pr/attempt", json={"pr_number": 99}
        )
        self.assertEqual(response.status_code, 403)
        body = response.json()
        self.assertFalse(body["success"])
        authorization = body["authorization"]
        self.assertFalse(authorization["allowed"])
        self.assertEqual(
            authorization["reason_code"], console_authz.DENY_UNAUTHENTICATED
        )
        self.assertFalse(authorization["execution_enabled"])
    def test_preview_reports_an_authorization_decision(self):
        response = self.client.get("/api/actions/merge_pr/preview?pr_number=7")
        self.assertEqual(response.status_code, 200)
        authorization = response.json()["authorization"]
        self.assertFalse(authorization["allowed"])
        self.assertTrue(authorization["dual_control"])
        self.assertEqual(authorization["required_role"], "controller")
    def test_unknown_action_preview_still_404s(self):
        response = self.client.get("/api/actions/no_such_action/preview")
        self.assertEqual(response.status_code, 404)
    def test_security_model_endpoint_publishes_all_three_policies(self):
        response = self.client.get("/api/console/security-model")
        self.assertEqual(response.status_code, 200)
        body = response.json()
        self.assertIn("rbac", body)
        self.assertIn("redaction", body)
        self.assertIn("audit", body)
        self.assertEqual(body["rbac"]["default_decision"], "deny")
    def test_security_model_endpoint_leaks_no_secrets(self):
        response = self.client.get("/api/console/security-model")
        self.assertEqual(scan_for_secrets(response.json()), [])
    def test_security_model_rejects_writes(self):
        response = self.client.post("/api/console/security-model", json={})
        self.assertEqual(response.status_code, 405)
    def test_existing_read_routes_are_unaffected(self):
        for path in ("/", "/health", "/actions", "/api/actions"):
            with self.subTest(path=path):
                self.assertEqual(self.client.get(path).status_code, 200)
 class TestAuthzAuditDoc(unittest.TestCase):
    """The model must be written down, not only coded."""
    @classmethod
    def setUpClass(cls):
        cls.text = (
            AUTHZ_DOC.read_text(encoding="utf-8") if AUTHZ_DOC.exists() else ""
        )
    def test_doc_exists(self):
        self.assertTrue(AUTHZ_DOC.exists(), f"missing {AUTHZ_DOC}")
    def test_doc_covers_each_required_section(self):
        for heading in (
            "Identity sources",
            "Role matrix",
            "Privileged actions",
            "Secret redaction",
            "Audit event schema",
            "Retention",
            "Phase 2 integration",
            "Local-dev mode",
        ):
            with self.subTest(heading=heading):
                self.assertIn(heading, self.text)
    def test_doc_names_every_role(self):
        for role in console_authz.ROLE_ORDER:
            with self.subTest(role=role):
                self.assertIn(role, self.text)
    def test_doc_names_every_console_action(self):
        for action_id in console_authz.ACTIONS:
            with self.subTest(action=action_id):
                self.assertIn(action_id, self.text)
    def test_doc_states_retention_defaults(self):
        for days in console_audit.RETENTION_DAYS.values():
            with self.subTest(days=days):
                self.assertIn(str(days), self.text)
    def test_doc_warns_local_dev_is_insecure(self):
        self.assertIn("INSECURE", self.text.upper())
    def test_doc_states_default_deny(self):
        self.assertIn("deny", self.text.lower())
    def test_doc_contains_no_secret_material(self):
        self.assertEqual(scan_for_secrets(self.text), [])
    def test_deployment_doc_links_to_the_model(self):
        deployment = (DOCS / "webui-deployment.md").read_text(encoding="utf-8")
        self.assertIn("webui-authz-audit", deployment)
 if __name__ == "__main__":  # pragma: no cover
    unittest.main()
@@ -0,0 +1,499 @@
 """Tests for the read-only system-health API (#634).
 Covers the acceptance criteria directly: a structured payload with readiness
 and a dependency list (AC1), version and uptime when knowable (AC2), stale
 runtime reported without a false mutation-safe claim (AC3), and the healthy /
 degraded-dependency / redaction cases (AC4).
 """
 import json
 import os
 import sqlite3
 import sys
 import tempfile
 import unittest
 from pathlib import Path
 from unittest import mock
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 from starlette.testclient import TestClient
 import control_plane_db
 from webui.app import create_app
 from webui.deployment_boundary import scan_text_for_client_secrets
 from webui.system_health import (
    API_PATH,
    STATUS_DEGRADED,
    STATUS_DOWN,
    STATUS_OK,
    STATUS_SKIPPED,
    DependencyProbe,
    StaleRuntime,
    assess_stale_runtime,
    clear_probe_cache,
    load_system_health,
    namespace_summaries,
    probe_control_plane_db,
    probe_gitea,
    process_uptime,
    redact,
    redact_url,
    snapshot_to_dict,
 )
 def _probe(name, status, *, required=True, detail="detail", kind="test"):
    return DependencyProbe(
        name=name,
        kind=kind,
        status=status,
        detail=detail,
        required=required,
        latency_ms=1.5,
        metadata={},
    )
 _ALL_HEALTHY = (
    _probe("control_plane_db", STATUS_OK, kind="sqlite"),
    _probe("repository", STATUS_OK, kind="git"),
    _probe("gitea", STATUS_OK, required=False, kind="http"),
 )
 _CLEAN_PARITY = StaleRuntime(
    daemon_head="abc123",
    checkout_head="abc123",
    remote_head="abc123",
    stale=False,
    determinable=True,
    mutation_safe=True,
    reasons=(),
 )
 class CleanParityMixin:
    """Pin parity for tests about aggregation rather than staleness.
    Without this the assertions depend on the real checkout: a worktree whose
    branch is ahead of its upstream is genuinely stale, which would degrade the
    overall status and make these cases fail for an unrelated reason.
    """
    def setUp(self):
        super().setUp()
        patcher = mock.patch(
            "webui.system_health.assess_stale_runtime",
            return_value=_CLEAN_PARITY,
        )
        patcher.start()
        self.addCleanup(patcher.stop)
 class TestDependencyAggregation(CleanParityMixin, unittest.TestCase):
    """AC1 — readiness and dependency list derived from probe results."""
    def test_all_healthy_is_ok_and_ready(self):
        snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="abc123")
        self.assertEqual(snapshot.status, STATUS_OK)
        self.assertTrue(snapshot.ready)
        self.assertTrue(snapshot.readiness_complete)
        self.assertEqual(snapshot.readiness_reasons, ())
        self.assertEqual(len(snapshot.dependencies), 3)
    def test_required_dependency_down_blocks_readiness(self):
        probes = (
            _probe("control_plane_db", STATUS_DOWN, detail="file missing", kind="sqlite"),
            _probe("repository", STATUS_OK, kind="git"),
            _probe("gitea", STATUS_OK, required=False, kind="http"),
        )
        snapshot = load_system_health(probes=probes, daemon_head="abc123")
        self.assertEqual(snapshot.status, STATUS_DOWN)
        self.assertFalse(snapshot.ready)
        self.assertTrue(
            any("control_plane_db" in reason for reason in snapshot.readiness_reasons)
        )
    def test_optional_dependency_down_degrades_but_stays_ready(self):
        """A failing optional probe must not claim the process itself is unready."""
        probes = (
            _probe("control_plane_db", STATUS_OK, kind="sqlite"),
            _probe("repository", STATUS_OK, kind="git"),
            _probe("gitea", STATUS_DOWN, required=False, detail="timeout", kind="http"),
        )
        snapshot = load_system_health(probes=probes, daemon_head="abc123")
        self.assertEqual(snapshot.status, STATUS_DEGRADED)
        self.assertTrue(snapshot.ready)
        self.assertTrue(any("gitea" in reason for reason in snapshot.readiness_reasons))
    def test_unrun_required_probe_leaves_readiness_incomplete(self):
        """Not probed is not the same as passing."""
        probes = (
            _probe("control_plane_db", STATUS_OK, kind="sqlite"),
            _probe("repository", STATUS_SKIPPED, detail="offline", kind="git"),
        )
        snapshot = load_system_health(probes=probes, daemon_head="abc123")
        self.assertFalse(snapshot.ready)
        self.assertFalse(snapshot.readiness_complete)
        self.assertEqual(snapshot.status, STATUS_DEGRADED)
    def test_skipped_optional_probe_does_not_block_readiness(self):
        probes = (
            _probe("control_plane_db", STATUS_OK, kind="sqlite"),
            _probe("repository", STATUS_OK, kind="git"),
            _probe("gitea", STATUS_SKIPPED, required=False, kind="http"),
        )
        snapshot = load_system_health(probes=probes, daemon_head="abc123")
        self.assertTrue(snapshot.ready)
        self.assertTrue(snapshot.readiness_complete)
 class TestVersionAndUptime(CleanParityMixin, unittest.TestCase):
    """AC2 — version and uptime present when knowable."""
    def test_uptime_and_start_time_present(self):
        snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="abc123")
        self.assertGreaterEqual(snapshot.uptime_seconds, 0.0)
        self.assertIn("T", snapshot.started_at)
    def test_process_uptime_helper_matches_shape(self):
        started_at, uptime = process_uptime()
        self.assertIn("T", started_at)
        self.assertGreaterEqual(uptime, 0.0)
    def test_version_reports_python_and_schema_version(self):
        probes = (
            DependencyProbe(
                name="control_plane_db",
                kind="sqlite",
                status=STATUS_OK,
                detail="ok",
                required=True,
                latency_ms=1.0,
                metadata={"schema_version": control_plane_db.SCHEMA_VERSION},
            ),
            _probe("repository", STATUS_OK, kind="git"),
        )
        snapshot = load_system_health(probes=probes, daemon_head="abc123")
        self.assertEqual(
            snapshot.version.control_plane_schema_version,
            control_plane_db.SCHEMA_VERSION,
        )
        self.assertTrue(snapshot.version.python_version)
    def test_version_known_flag_false_when_sha_unavailable(self):
        with mock.patch("webui.system_health._git", return_value=None):
            snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="abc")
        self.assertIsNone(snapshot.version.git_sha)
        self.assertFalse(snapshot.version.known)
 class TestStaleRuntime(unittest.TestCase):
    """AC3 — stale runtime reflected without a false mutation-safe claim."""
    def test_matching_commits_are_mutation_safe(self):
        assessment = assess_stale_runtime(
            Path("/tmp"),
            daemon_head="aaa",
            git_reader=lambda *args: "aaa",
        )
        self.assertFalse(assessment.stale)
        self.assertTrue(assessment.determinable)
        self.assertTrue(assessment.mutation_safe)
    def test_diverged_commits_are_stale_and_not_mutation_safe(self):
        reads = {"HEAD": "aaa", "@{upstream}": "bbb"}
        assessment = assess_stale_runtime(
            Path("/tmp"),
            daemon_head="aaa",
            git_reader=lambda *args: reads.get(args[-1]),
        )
        self.assertTrue(assessment.stale)
        self.assertFalse(assessment.mutation_safe)
        self.assertTrue(assessment.reasons)
    def test_unknown_remote_is_not_mutation_safe(self):
        """Indeterminate must never read as safe."""
        reads = {"HEAD": "aaa", "@{upstream}": None}
        assessment = assess_stale_runtime(
            Path("/tmp"),
            daemon_head="aaa",
            git_reader=lambda *args: reads.get(args[-1]),
        )
        self.assertFalse(assessment.determinable)
        self.assertFalse(assessment.mutation_safe)
        self.assertFalse(assessment.stale)
        self.assertTrue(
            any("indeterminate" in reason for reason in assessment.reasons)
        )
    def test_unobservable_daemon_head_is_disclosed(self):
        assessment = assess_stale_runtime(
            Path("/tmp"),
            git_reader=lambda *args: "aaa",
        )
        self.assertTrue(
            any("not observable" in reason for reason in assessment.reasons)
        )
    def test_stale_runtime_degrades_overall_status(self):
        reads = {"HEAD": "aaa", "@{upstream}": "bbb"}
        # Pinned rather than inherited: this path uses the default git reader,
        # so the assertion must hold whether or not the suite runs offline.
        with mock.patch.dict(os.environ, {"WEBUI_TEST_OFFLINE": ""}), mock.patch(
            "webui.system_health._git",
            side_effect=lambda repo, *args: reads.get(args[-1]),
        ):
            snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="aaa")
        self.assertTrue(snapshot.stale_runtime.stale)
        self.assertFalse(snapshot.stale_runtime.mutation_safe)
        self.assertEqual(snapshot.status, STATUS_DEGRADED)
 class TestControlPlaneDbProbe(unittest.TestCase):
    """The required local dependency, probed read-only."""
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.db_path = str(Path(self.tmp.name) / "control-plane.db")
    def _build_db(self, schema_version):
        conn = sqlite3.connect(self.db_path)
        conn.execute("CREATE TABLE schema_meta (key TEXT PRIMARY KEY, value TEXT)")
        conn.execute("CREATE TABLE leases (lease_id TEXT PRIMARY KEY, status TEXT)")
        conn.execute(
            "INSERT INTO schema_meta(key, value) VALUES ('schema_version', ?)",
            (str(schema_version),),
        )
        conn.execute("INSERT INTO leases(lease_id, status) VALUES ('l1', 'active')")
        conn.commit()
        conn.close()
    def test_missing_database_is_down(self):
        probe = probe_control_plane_db(str(Path(self.tmp.name) / "absent.db"))
        self.assertEqual(probe.status, STATUS_DOWN)
        self.assertTrue(probe.required)
        self.assertIsNotNone(probe.latency_ms)
    def test_matching_schema_is_ok(self):
        self._build_db(control_plane_db.SCHEMA_VERSION)
        probe = probe_control_plane_db(self.db_path)
        self.assertEqual(probe.status, STATUS_OK)
        self.assertEqual(
            probe.metadata["schema_version"], control_plane_db.SCHEMA_VERSION
        )
        self.assertEqual(probe.metadata["active_leases"], 1)
    def test_mismatched_schema_is_degraded(self):
        self._build_db(control_plane_db.SCHEMA_VERSION + 99)
        probe = probe_control_plane_db(self.db_path)
        self.assertEqual(probe.status, STATUS_DEGRADED)
    def test_probe_does_not_create_a_database(self):
        """A health check must never initialise the substrate it inspects."""
        absent = str(Path(self.tmp.name) / "never-created.db")
        probe_control_plane_db(absent)
        self.assertFalse(Path(absent).exists())
    def test_unreadable_database_is_down_not_raised(self):
        Path(self.db_path).write_text("this is not a sqlite database")
        probe = probe_control_plane_db(self.db_path)
        self.assertEqual(probe.status, STATUS_DOWN)
 class TestRedaction(unittest.TestCase):
    """AC4 — redaction. No credential-shaped text crosses the boundary."""
    def test_redacts_token_assignment(self):
        cleaned = redact("failed with token=ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ012345")
        self.assertNotIn("ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ012345", cleaned)
        self.assertIn("[redacted]", cleaned)
    def test_redacts_authorization_header_text(self):
        cleaned = redact("Authorization: Bearer abcdefghijklmnopqrstuvwxyz123456")
        self.assertNotIn("abcdefghijklmnopqrstuvwxyz123456", cleaned)
    def test_redacts_long_opaque_strings(self):
        cleaned = redact("value 0123456789abcdef0123456789abcdef here")
        self.assertNotIn("0123456789abcdef0123456789abcdef", cleaned)
    def test_url_userinfo_and_query_are_stripped(self):
        cleaned = redact_url("https://user:[email protected]/api/v1?token=xyz")
        self.assertNotIn("secretpass", cleaned)
        self.assertNotIn("token=xyz", cleaned)
        self.assertEqual(cleaned, "https://gitea.example.com/api/v1")
    def test_url_inside_free_text_is_redacted(self):
        cleaned = redact("GET https://u:[email protected]/x?token=abc failed")
        self.assertNotIn("u:p@", cleaned)
        self.assertNotIn("token=abc", cleaned)
    def test_gitea_probe_failure_detail_is_redacted(self):
        boom = RuntimeError(
            "connection refused for https://user:[email protected]/api/v1/version"
        )
        with mock.patch("webui.system_health.get_auth_header", return_value="token x"), \
                mock.patch("webui.system_health.api_request", side_effect=boom):
            probe = probe_gitea("gitea.example.com")
        self.assertEqual(probe.status, STATUS_DOWN)
        self.assertNotIn("hunter2", probe.detail)
        self.assertEqual(scan_text_for_client_secrets(probe.detail), [])
    def test_credential_guard_refusal_is_a_status_not_a_crash(self):
        with mock.patch(
            "webui.system_health.get_auth_header",
            side_effect=RuntimeError("daemon guard refused"),
        ):
            probe = probe_gitea("gitea.example.com")
        self.assertEqual(probe.status, STATUS_DEGRADED)
        self.assertFalse(probe.required)
 class TestNamespaceSummaries(unittest.TestCase):
    """A web process cannot prove IDE namespace health, and must not claim to."""
    def test_every_namespace_reports_unproven(self):
        rows = namespace_summaries()
        self.assertTrue(rows)
        for row in rows:
            with self.subTest(namespace=row["namespace"]):
                self.assertEqual(row["status"], "unproven")
                self.assertFalse(row["ide_namespace_proven"])
                self.assertIn("client_namespace", row["reason"])
 class TestSystemHealthRoutes(CleanParityMixin, unittest.TestCase):
    """The HTTP surface: versioned path, status codes, read-only guard."""
    def setUp(self):
        super().setUp()
        clear_probe_cache()
        self.addCleanup(clear_probe_cache)
        self.client = TestClient(create_app())
    def _patch_snapshot(self, probes, daemon_head="abc123"):
        snapshot = load_system_health(probes=probes, daemon_head=daemon_head)
        patcher = mock.patch(
            "webui.app.load_system_health",
            return_value=snapshot,
        )
        patcher.start()
        self.addCleanup(patcher.stop)
        return snapshot
    def test_versioned_route_is_registered(self):
        self.assertEqual(API_PATH, "/api/v1/system/health")
        self._patch_snapshot(_ALL_HEALTHY)
        response = self.client.get(API_PATH)
        self.assertEqual(response.status_code, 200)
    def test_healthy_payload_shape(self):
        self._patch_snapshot(_ALL_HEALTHY)
        data = self.client.get(API_PATH).json()
        self.assertEqual(data["status"], STATUS_OK)
        self.assertTrue(data["readiness"]["ready"])
        self.assertTrue(data["readiness"]["complete"])
        self.assertEqual(data["api"], API_PATH)
        self.assertEqual(len(data["dependencies"]), 3)
        for key in ("version", "process", "stale_runtime", "mcp_namespaces"):
            self.assertIn(key, data)
        self.assertIn("uptime_seconds", data["process"])
        self.assertIn("mutation_safe", data["stale_runtime"])
    def test_degraded_dependency_returns_503(self):
        probes = (
            _probe("control_plane_db", STATUS_DOWN, detail="missing", kind="sqlite"),
            _probe("repository", STATUS_OK, kind="git"),
        )
        self._patch_snapshot(probes)
        response = self.client.get(API_PATH)
        self.assertEqual(response.status_code, 503)
        data = response.json()
        self.assertFalse(data["readiness"]["ready"])
        self.assertTrue(data["readiness"]["reasons"])
    def test_dependency_entries_expose_status_and_latency(self):
        self._patch_snapshot(_ALL_HEALTHY)
        data = self.client.get(API_PATH).json()
        names = {entry["name"] for entry in data["dependencies"]}
        self.assertEqual(names, {"control_plane_db", "repository", "gitea"})
        for entry in data["dependencies"]:
            with self.subTest(dependency=entry["name"]):
                self.assertIn("status", entry)
                self.assertIn("required", entry)
                self.assertIn("latency_ms", entry)
    def test_response_body_carries_no_client_secrets(self):
        self._patch_snapshot(_ALL_HEALTHY)
        body = self.client.get(API_PATH).text
        self.assertEqual(scan_text_for_client_secrets(body), [])
    def test_deep_flag_is_forwarded(self):
        snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="abc")
        with mock.patch(
            "webui.app.load_system_health", return_value=snapshot
        ) as loader:
            self.client.get(f"{API_PATH}?deep=1")
            loader.assert_called_once_with(deep=True)
    def test_shallow_is_the_default(self):
        snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="abc")
        with mock.patch(
            "webui.app.load_system_health", return_value=snapshot
        ) as loader:
            self.client.get(API_PATH)
            loader.assert_called_once_with(deep=False)
    def test_route_rejects_mutation_methods(self):
        for method in ("POST", "PUT", "PATCH", "DELETE"):
            with self.subTest(method=method):
                response = self.client.request(method, API_PATH)
                self.assertEqual(response.status_code, 405)
                self.assertEqual(response.json()["error"], "read-only-mvp")
    def test_default_shallow_call_skips_the_network_probe(self):
        """The expensive probe must not run unless it was asked for."""
        with mock.patch("webui.system_health.probe_gitea") as probe:
            snapshot = load_system_health(deep=False)
        probe.assert_not_called()
        gitea = next(p for p in snapshot.dependencies if p.name == "gitea")
        self.assertEqual(gitea.status, STATUS_SKIPPED)
 class TestHealthRouteBackwardCompatibility(unittest.TestCase):
    """`/health` is expanded additively; MVP consumers must keep working."""
    def setUp(self):
        self.client = TestClient(create_app())
    def test_mvp_keys_are_unchanged(self):
        data = self.client.get("/health").json()
        self.assertEqual(data["status"], "ok")
        self.assertEqual(data["service"], "mcp-control-plane-webui")
        self.assertEqual(data["mode"], "read-only-mvp")
        self.assertIn("timestamp", data)
        self.assertEqual(data["deployment"]["mode"], "internal-operator-console")
    def test_health_points_at_the_versioned_api(self):
        data = self.client.get("/health").json()
        self.assertEqual(data["system_health_api"], API_PATH)
        self.assertIn("uptime_seconds", data)
        self.assertIn("started_at", data)
    def test_health_runs_no_dependency_probe(self):
        """Liveness must stay cheap: no probe, no snapshot assembly."""
        with mock.patch("webui.app.load_system_health") as loader:
            response = self.client.get("/health")
        self.assertEqual(response.status_code, 200)
        loader.assert_not_called()
 class TestSnapshotSerialisation(CleanParityMixin, unittest.TestCase):
    def test_snapshot_dict_is_json_serialisable(self):
        snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="abc123")
        encoded = json.dumps(snapshot_to_dict(snapshot))
        self.assertIn("readiness", encoded)
 if __name__ == "__main__":
    unittest.main()
@@ -2,7 +2,6 @@
 from __future__ import annotations
 import uuid
 from datetime import datetime, timezone
 from starlette.applications import Starlette
@@ -20,9 +19,6 @@ from final_report_validator import FINAL_REPORT_TASK_KINDS
 from webui.gated_actions import attempt_action, load_action_registry, preview_action
 from webui.gated_action_views import render_actions_page
 from webui import console_audit
 from webui.console_authz import authorize, rbac_matrix, resolve_principal
 from webui.console_redaction import redaction_policy
 from webui.audit_validator import audit_report, audit_to_dict
 from webui.audit_views import render_audit_page
 from webui.lease_loader import load_lease_snapshot, snapshot_to_dict as lease_snapshot_to_dict
@@ -33,6 +29,12 @@ from webui.worktree_scanner import load_hygiene_snapshot, snapshot_to_dict as wo
 from webui.worktree_views import render_worktrees_page
 from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
 from webui.runtime_views import render_runtime_page
 from webui.system_health import (
    API_PATH as SYSTEM_HEALTH_API_PATH,
    load_system_health,
    process_uptime,
    snapshot_to_dict as system_health_to_dict,
 )
 _READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
 _AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
@@ -66,16 +68,43 @@ async def home(_request: Request) -> HTMLResponse:
 async def health(_request: Request) -> JSONResponse:
    """Liveness only — deliberately cheap, runs no dependency probe (#634).
    Every MVP key is retained so existing pollers keep working; the additions
    are a pointer to the structured API and the in-memory process uptime.
    Readiness lives at that API because answering it costs real probes.
    """
    bind_host = _request.app.state.webui_bind_host
    started_at, uptime_seconds = process_uptime()
    return JSONResponse({
        "status": "ok",
        "service": "mcp-control-plane-webui",
        "mode": "read-only-mvp",
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "deployment": deployment_snapshot(bind_host=bind_host),
        "started_at": started_at,
        "uptime_seconds": uptime_seconds,
        "system_health_api": SYSTEM_HEALTH_API_PATH,
    })
 def _truthy_flag(value: str | None) -> bool:
    return (value or "").strip().lower() in {"1", "true", "yes", "on"}
 async def api_system_health(request: Request) -> JSONResponse:
    """Structured read-only system health (#634).
    `?deep=1` opts into the expensive network probe. The response status code
    reflects readiness so automated checks can branch on it without parsing the
    body: 200 when ready, 503 when a required dependency failed or never ran.
    """
    deep = _truthy_flag(request.query_params.get("deep"))
    snapshot = load_system_health(deep=deep)
    payload = system_health_to_dict(snapshot)
    return JSONResponse(payload, status_code=200 if snapshot.ready else 503)
 async def queue(_request: Request) -> HTMLResponse:
    snapshot = load_queue_snapshot()
    return HTMLResponse(render_page(title="Queue", body_html=render_queue_page(snapshot)))
@@ -219,49 +248,6 @@ async def api_actions(_request: Request) -> JSONResponse:
    return JSONResponse(load_action_registry().to_dict())
 def _request_id() -> str:
    return f"req-{uuid.uuid4().hex}"
 def _audit_target(action_id: str, params: dict[str, object]) -> dict[str, object]:
    """Describe the action target for the audit record (never secrets)."""
    if "pr_number" in params:
        return {"kind": "pr", "ref": f"#{params['pr_number']}"}
    if "issue_number" in params:
        return {"kind": "issue", "ref": f"#{params['issue_number']}"}
    if "branch_name" in params:
        return {"kind": "branch", "ref": str(params["branch_name"])}
    return {"kind": "unspecified", "ref": action_id}
 def _authorize_request(
    request: Request,
    action_id: str,
    params: dict[str, object],
    *,
    for_execution: bool,
    result: str,
 ) -> dict[str, object]:
    """Resolve principal, decide, and audit. Returns the decision payload.
    Phase 1 records the decision rather than enforcing it as the terminal
    outcome: ``webui.gated_actions`` already fails closed for every action, so
    this layer cannot loosen anything. Phase 2 enforces on this same decision.
    """
    principal = resolve_principal(headers=dict(request.headers))
    decision = authorize(action_id, principal, for_execution=for_execution)
    console_audit.record_event(
        action_id=action_id,
        result=result,
        decision=decision,
        principal=principal,
        target=_audit_target(action_id, params),
        request_id=_request_id(),
        detail=decision.detail,
    )
    return decision.to_dict()
 async def api_action_preview(request: Request) -> JSONResponse:
    action_id = request.path_params["action_id"]
    params = dict(request.query_params)
@@ -271,13 +257,6 @@ async def api_action_preview(request: Request) -> JSONResponse:
    result = preview_action(action_id, **params)
    if "error" in result:
        return JSONResponse(result, status_code=404)
    result["authorization"] = _authorize_request(
        request,
        action_id,
        params,
        for_execution=False,
        result=console_audit.RESULT_PREVIEWED,
    )
    return JSONResponse(result)
@@ -291,31 +270,10 @@ async def api_action_attempt(request: Request) -> JSONResponse:
    if not isinstance(body, dict):
        body = {}
    result = attempt_action(action_id, **body)
    authorization = _authorize_request(
        request,
        action_id,
        body,
        for_execution=True,
        result=(
            console_audit.RESULT_DENIED
            if not result.get("success")
            else console_audit.RESULT_ALLOWED
        ),
    )
    result["authorization"] = authorization
    status = 403 if not result.get("success") else 200
    return JSONResponse(result, status_code=status)
 async def api_console_security_model(_request: Request) -> JSONResponse:
    """Read-only publication of the #633 authorization/redaction/audit model."""
    return JSONResponse({
        "rbac": rbac_matrix(),
        "redaction": redaction_policy(),
        "audit": console_audit.audit_policy(),
    })
 async def method_not_allowed(request: Request, _exc: Exception) -> Response:
    path = request.url.path
    if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
@@ -338,6 +296,7 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
        routes=[
            Route("/", home, methods=["GET"]),
            Route("/health", health, methods=["GET"]),
            Route(SYSTEM_HEALTH_API_PATH, api_system_health, methods=["GET"]),
            Route("/queue", queue, methods=["GET"]),
            Route("/api/queue", api_queue, methods=["GET"]),
            Route("/projects", projects, methods=["GET"]),
@@ -366,11 +325,6 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
                methods=["POST"],
            ),
            Route("/api/leases", api_leases, methods=["GET"]),
            Route(
                "/api/console/security-model",
                api_console_security_model,
                methods=["GET"],
            ),
        ],
        exception_handlers={405: method_not_allowed},
    )
@@ -1,281 +0,0 @@
 """Console audit event schema, retention, and append-only sink (#633).
 ``gitea_audit`` records MCP-side *mutations*: which profile and Gitea user
 performed which tool call. It carries no console actor, no identity source, no
 correlation identifier, and no retention class, so it cannot answer the
 question #633 exists to answer — *who sat at the console, what did they
 attempt, and was it authorized?* An authorization denial is not a mutation and
 would never appear there at all.
 This module adds the console-side record. It does not replace ``gitea_audit``:
 when a Phase 2 action eventually reaches MCP, both fire, correlated by
 ``correlation.request_id``.
 Design constraints:
 - **Redact before persist.** Every record passes through
  ``webui.console_redaction.redact_payload`` before serialization, so an
  unredacted field is never durable.
 - **Append-only.** Records are appended as JSON lines. Nothing here updates or
  deletes; retention is metadata on each record, enforced by an operator-run
  policy, never by silent rewriting.
 - **Never raises.** Auditing must not break the request it describes. A failed
  write returns ``False``.
 - **Off by default.** With ``WEBUI_CONSOLE_AUDIT_LOG`` unset, events are still
  *built* (so callers and tests see the schema) but nothing is written.
 A record looks like this (synthetic values):
    {"schema_version": 1, "event_id": "evt-0001",
     "timestamp": "2026-07-22T10:16:42+00:00",
     "actor": {"subject": "[email protected]", "role": "operator",
               "identity_source": "access_proxy", "authenticated": true},
     "action": "merge_pr", "action_class": "privileged",
     "target": {"kind": "pr", "ref": "#123"},
     "result": "denied", "reason_code": "insufficient_role",
     "correlation": {"request_id": "req-abc", "session_id": null,
                     "mcp_task": "merge_pr", "mcp_permission": "gitea.pr.merge"},
     "retention": {"class": "privileged", "days": 365,
                   "expires_at": "2027-07-22T10:16:42+00:00"},
     "redacted": true}
 Timestamps are timezone-aware ISO-8601 in UTC.
 """
 from __future__ import annotations
 import datetime
 import json
 import os
 import uuid
 from typing import Any
 from webui import console_authz
 from webui.console_redaction import redact_payload, scan_for_secrets
 SCHEMA_VERSION = 1
 AUDIT_LOG_ENV = "WEBUI_CONSOLE_AUDIT_LOG"
 # Result vocabulary. ``denied`` is the one ``gitea_audit`` has no equivalent
 # for: an authorization refusal never reaches the MCP layer.
 RESULT_ALLOWED = "allowed"
 RESULT_DENIED = "denied"
 RESULT_PREVIEWED = "previewed"
 RESULT_FAILED = "failed"
 RESULT_SUCCEEDED = "succeeded"
 RESULTS = frozenset(
    {
        RESULT_ALLOWED,
        RESULT_DENIED,
        RESULT_PREVIEWED,
        RESULT_FAILED,
        RESULT_SUCCEEDED,
    }
 )
 # Retention classes and default lifetimes in days. Privileged and break-glass
 # records outlive routine ones because they are what an incident review needs.
 RETENTION_STANDARD = "standard"
 RETENTION_PRIVILEGED = "privileged"
 RETENTION_BREAK_GLASS = "break_glass"
 RETENTION_DAYS: dict[str, int] = {
    RETENTION_STANDARD: 90,
    RETENTION_PRIVILEGED: 365,
    RETENTION_BREAK_GLASS: 730,
 }
 # Fields every record must carry. Asserted by the test suite so a future edit
 # cannot quietly drop one.
 REQUIRED_FIELDS: tuple[str, ...] = (
    "schema_version",
    "event_id",
    "timestamp",
    "actor",
    "action",
    "action_class",
    "target",
    "result",
    "reason_code",
    "correlation",
    "retention",
    "redacted",
 )
 REQUIRED_ACTOR_FIELDS: tuple[str, ...] = (
    "subject",
    "role",
    "identity_source",
    "authenticated",
 )
 REQUIRED_CORRELATION_FIELDS: tuple[str, ...] = (
    "request_id",
    "session_id",
    "mcp_task",
    "mcp_permission",
 )
 def audit_log_path() -> str | None:
    """Configured sink path, or ``None`` when console auditing is off."""
    return (os.environ.get(AUDIT_LOG_ENV) or "").strip() or None
 def audit_enabled() -> bool:
    return audit_log_path() is not None
 def retention_class_for(action: console_authz.ConsoleAction | None) -> str:
    """Classify retention from the action, defaulting to the longest-lived.
    An unknown action is treated as privileged rather than standard: for a
    safety control the conservative direction is to keep the record longer.
    """
    if action is None:
        return RETENTION_PRIVILEGED
    if action.break_glass:
        return RETENTION_BREAK_GLASS
    if action.privileged:
        return RETENTION_PRIVILEGED
    return RETENTION_STANDARD
 def _retention_block(
    retention_class: str, now: datetime.datetime
 ) -> dict[str, Any]:
    days = RETENTION_DAYS.get(
        retention_class, RETENTION_DAYS[RETENTION_PRIVILEGED]
    )
    return {
        "class": retention_class,
        "days": days,
        "expires_at": (now + datetime.timedelta(days=days)).isoformat(),
    }
 def build_event(
    *,
    action_id: str,
    result: str,
    decision: console_authz.AuthorizationDecision | None = None,
    principal: console_authz.Principal | None = None,
    target: dict[str, Any] | None = None,
    reason_code: str | None = None,
    request_id: str | None = None,
    session_id: str | None = None,
    detail: str | None = None,
    metadata: dict[str, Any] | None = None,
    now: datetime.datetime | None = None,
    event_id: str | None = None,
 ) -> dict[str, Any]:
    """Build one redacted, JSON-able console audit record.
    Redaction runs here rather than at write time so an in-memory record handed
    to a template or an API response is already clean.
    """
    ts = now or datetime.datetime.now(datetime.timezone.utc)
    action = console_authz.get_action(action_id)
    who = principal or (
        decision.principal if decision else console_authz.ANONYMOUS
    )
    resolved_result = result if result in RESULTS else RESULT_FAILED
    resolved_reason = reason_code or (
        decision.reason_code if decision else "unspecified"
    )
    retention_class = retention_class_for(action)
    event: dict[str, Any] = {
        "schema_version": SCHEMA_VERSION,
        "event_id": event_id or f"evt-{uuid.uuid4().hex}",
        "timestamp": ts.isoformat(),
        "actor": who.to_dict(),
        "action": action_id,
        "action_class": action.action_class if action else "unknown",
        "target": dict(target or {}),
        "result": resolved_result,
        "reason_code": resolved_reason,
        "correlation": {
            "request_id": request_id,
            "session_id": session_id,
            "mcp_task": action.task_key if action else None,
            "mcp_permission": action.mcp_permission if action else None,
        },
        "retention": _retention_block(retention_class, ts),
        "redacted": True,
        "detail": detail,
        "metadata": dict(metadata or {}),
    }
    if decision is not None:
        # Deliberately *not* named "authorization": ``gitea_audit`` treats that
        # substring as a secret key hint (it matches the HTTP Authorization
        # header) and would replace this whole block with the placeholder.
        event["decision"] = {
            "allowed": decision.allowed,
            "required_role": decision.required_role,
            "requires_confirmation": decision.requires_confirmation,
            "dual_control": decision.dual_control,
            "break_glass": decision.break_glass,
            "execution_enabled": decision.execution_enabled,
        }
    redacted = redact_payload(event)
    if not isinstance(redacted, dict):  # pragma: no cover - defensive
        return {"schema_version": SCHEMA_VERSION, "redacted": True}
    return redacted
 def write_event(event: dict[str, Any], path: str | None = None) -> bool:
    """Append *event* as one JSON line. Never raises.
    Returns ``True`` when a line was written, ``False`` when auditing is off or
    the write failed. A record that still trips a secret detector is dropped
    rather than persisted.
    """
    sink = path or audit_log_path()
    if not sink:
        return False
    try:
        if scan_for_secrets(event):
            return False
        line = json.dumps(event, default=str, sort_keys=True)
        with open(sink, "a", encoding="utf-8") as handle:
            handle.write(line + "\n")
        return True
    except Exception:
        return False
 def record_event(**kwargs: Any) -> dict[str, Any]:
    """Build and persist one record; return the record either way.
    Callers get the record back so it can be surfaced in a response or a test
    regardless of whether a sink is configured.
    """
    event = build_event(**kwargs)
    written = write_event(event)
    return {"event": event, "written": written}
 def audit_policy() -> dict[str, Any]:
    """Machine-readable audit schema and retention defaults (never secrets)."""
    return {
        "schema_version": SCHEMA_VERSION,
        "required_fields": list(REQUIRED_FIELDS),
        "required_actor_fields": list(REQUIRED_ACTOR_FIELDS),
        "required_correlation_fields": list(REQUIRED_CORRELATION_FIELDS),
        "results": sorted(RESULTS),
        "retention_defaults_days": dict(RETENTION_DAYS),
        "sink_env": AUDIT_LOG_ENV,
        "enabled": audit_enabled(),
        "append_only": True,
        "redact_before_persist": True,
        "timestamp_format": "ISO-8601, timezone-aware, UTC",
        "relationship_to_mcp_audit": (
            "webui.console_audit records console intent and authorization "
            "outcomes; gitea_audit records MCP mutations. A Phase 2 action "
            "emits both, correlated by correlation.request_id."
        ),
    }
@@ -1,537 +0,0 @@
 """Console authorization and RBAC model (#633, Phase 1).
 The read-only MVP (#426–#436) ships with no authentication: protection comes
 from network placement alone (#435). That is adequate while every route is a
 GET, and inadequate the moment Phase 2 wires a gated write. This module is the
 authorization model those writes must go through, landed *before* any of them
 exists so no write can be added without an authority to check against.
 Phase 1 scope is the model itself: identity resolution, the role matrix, the
 privileged-action list, and a fail-closed :func:`authorize`. It deliberately
 does **not** enable any write. ``webui.gated_actions`` stays globally disabled,
 so an allow decision here is necessary but never sufficient.
 Two invariants hold for every caller:
 - **Default deny.** An unrecognised action, an unknown role, or an absent
  principal denies. There is no implicit allow branch and no "unless" clause.
 - **Authorization is not execution.** :func:`authorize` returns a decision
  record. It never calls MCP, never mutates, and never consults credentials.
 """
 from __future__ import annotations
 import json
 import os
 from dataclasses import asdict, dataclass, field
 from typing import Any
 from task_capability_map import required_permission, required_role
 # --- Roles ------------------------------------------------------------------
 # Ordered least to most authority. Higher ranks inherit every lower rank's
 # permitted actions; the matrix below is expressed as a minimum required rank.
 VIEWER = "viewer"
 OPERATOR = "operator"
 CONTROLLER = "controller"
 ADMIN = "admin"
 ROLE_ORDER: tuple[str, ...] = (VIEWER, OPERATOR, CONTROLLER, ADMIN)
 _ROLE_RANK: dict[str, int] = {role: idx for idx, role in enumerate(ROLE_ORDER)}
 ROLE_DESCRIPTIONS: dict[str, str] = {
    VIEWER: "Read every console view. No write, ever, in any phase.",
    OPERATOR: "Viewer, plus author-class work: claim, comment, open a PR.",
    CONTROLLER: "Operator, plus reviewer/merger-class decisions on a PR.",
    ADMIN: "Controller, plus destructive and policy-editing actions.",
 }
 # --- Identity sources -------------------------------------------------------
 IDENTITY_NONE = "none"
 IDENTITY_LOCAL_DEV = "local_dev"
 IDENTITY_ACCESS_PROXY = "access_proxy"
 IDENTITY_SOURCES: dict[str, dict[str, Any]] = {
    IDENTITY_NONE: {
        "description": (
            "No authentication configured. Every request is anonymous and "
            "capped at viewer. This is the MVP default and the only mode "
            "whose safety rests entirely on network placement (#435)."
        ),
        "authenticated": False,
        "safe_for_shared_host": False,
        "phase_available": 1,
    },
    IDENTITY_LOCAL_DEV: {
        "description": (
            "Developer-supplied principal read from the environment. INSECURE: "
            "the subject and role are asserted, never verified. Loopback only."
        ),
        "authenticated": True,
        "safe_for_shared_host": False,
        "phase_available": 1,
    },
    IDENTITY_ACCESS_PROXY: {
        "description": (
            "Subject asserted by a trusted access proxy (Cloudflare Access, "
            "WARP, or an org VPN portal) via a verified request header. The "
            "proxy performs authentication; the console performs authorization."
        ),
        "authenticated": True,
        "safe_for_shared_host": True,
        "phase_available": 2,
    },
 }
 # Environment configuration. All are read server-side and never rendered.
 AUTH_MODE_ENV = "WEBUI_AUTH_MODE"
 DEV_SUBJECT_ENV = "WEBUI_DEV_SUBJECT"
 DEV_ROLE_ENV = "WEBUI_DEV_ROLE"
 ROLE_MAP_ENV = "WEBUI_ROLE_MAP"
 REQUIRE_PROBE_AUTH_ENV = "WEBUI_REQUIRE_PROBE_AUTH"
 ACCESS_SUBJECT_HEADER = "cf-access-authenticated-user-email"
 # --- Action classes ---------------------------------------------------------
 CLASS_READ = "read"
 CLASS_WRITE = "gated_write"
 CLASS_PRIVILEGED = "privileged"
 CLASS_DESTRUCTIVE = "destructive"
 # --- Privileged action list -------------------------------------------------
 # ``task_key`` ties each console action back to ``task_capability_map``, so the
 # console cannot invent an authority the MCP layer does not already define.
@dataclass(frozen=True)
 class ConsoleAction:
    """One console action and the authority required to invoke it."""
    action_id: str
    task_key: str
    action_class: str
    minimum_role: str
    requires_confirmation: bool
    dual_control: bool
    break_glass: bool
    phase: int
    summary: str
    @property
    def mcp_permission(self) -> str:
        return required_permission(self.task_key)
    @property
    def mcp_role(self) -> str:
        return required_role(self.task_key)
    @property
    def privileged(self) -> bool:
        return self.action_class in {CLASS_PRIVILEGED, CLASS_DESTRUCTIVE}
    def to_dict(self) -> dict[str, Any]:
        data = asdict(self)
        data["mcp_permission"] = self.mcp_permission
        data["mcp_role"] = self.mcp_role
        data["privileged"] = self.privileged
        return data
 _ACTION_SPECS: tuple[ConsoleAction, ...] = (
    ConsoleAction(
        action_id="claim_issue",
        task_key="claim_issue",
        action_class=CLASS_WRITE,
        minimum_role=OPERATOR,
        requires_confirmation=True,
        dual_control=False,
        break_glass=False,
        phase=2,
        summary="Apply status:in-progress to an issue.",
    ),
    ConsoleAction(
        action_id="comment_issue",
        task_key="comment_issue",
        action_class=CLASS_WRITE,
        minimum_role=OPERATOR,
        requires_confirmation=True,
        dual_control=False,
        break_glass=False,
        phase=2,
        summary="Post an issue comment.",
    ),
    ConsoleAction(
        action_id="create_issue",
        task_key="create_issue",
        action_class=CLASS_WRITE,
        minimum_role=OPERATOR,
        requires_confirmation=True,
        dual_control=False,
        break_glass=False,
        phase=2,
        summary="Open a new tracking issue.",
    ),
    ConsoleAction(
        action_id="comment_pr",
        task_key="comment_pr",
        action_class=CLASS_WRITE,
        minimum_role=OPERATOR,
        requires_confirmation=True,
        dual_control=False,
        break_glass=False,
        phase=2,
        summary="Post a PR thread comment.",
    ),
    ConsoleAction(
        action_id="create_pr",
        task_key="create_pr",
        action_class=CLASS_WRITE,
        minimum_role=OPERATOR,
        requires_confirmation=True,
        dual_control=False,
        break_glass=False,
        phase=2,
        summary="Open a PR from a locked feature branch.",
    ),
    ConsoleAction(
        action_id="review_pr",
        task_key="review_pr",
        action_class=CLASS_PRIVILEGED,
        minimum_role=CONTROLLER,
        requires_confirmation=True,
        dual_control=False,
        break_glass=False,
        phase=3,
        summary="Submit an approve / request-changes verdict.",
    ),
    ConsoleAction(
        action_id="close_pr",
        task_key="close_pr",
        action_class=CLASS_PRIVILEGED,
        minimum_role=CONTROLLER,
        requires_confirmation=True,
        dual_control=False,
        break_glass=False,
        phase=3,
        summary="Close a pull request without merging.",
    ),
    ConsoleAction(
        action_id="merge_pr",
        task_key="merge_pr",
        action_class=CLASS_PRIVILEGED,
        minimum_role=CONTROLLER,
        requires_confirmation=True,
        dual_control=True,
        break_glass=True,
        phase=3,
        summary="Merge an approved pull request.",
    ),
    ConsoleAction(
        action_id="delete_branch",
        task_key="delete_branch",
        action_class=CLASS_DESTRUCTIVE,
        minimum_role=ADMIN,
        requires_confirmation=True,
        dual_control=True,
        break_glass=True,
        phase=3,
        summary="Remove a remote feature branch.",
    ),
 )
 ACTIONS: dict[str, ConsoleAction] = {a.action_id: a for a in _ACTION_SPECS}
 def privileged_actions() -> tuple[ConsoleAction, ...]:
    """Actions requiring dual control, break-glass, or controller+ authority."""
    return tuple(a for a in _ACTION_SPECS if a.privileged)
 def get_action(action_id: str) -> ConsoleAction | None:
    return ACTIONS.get(action_id)
 # --- Principals -------------------------------------------------------------
@dataclass(frozen=True)
 class Principal:
    """Who is making a request, and how strongly that is known."""
    subject: str
    role: str
    identity_source: str
    authenticated: bool
    warnings: tuple[str, ...] = field(default_factory=tuple)
    @property
    def rank(self) -> int:
        return _ROLE_RANK.get(self.role, -1)
    def to_dict(self) -> dict[str, Any]:
        return {
            "subject": self.subject,
            "role": self.role,
            "identity_source": self.identity_source,
            "authenticated": self.authenticated,
            "warnings": list(self.warnings),
        }
 ANONYMOUS = Principal(
    subject="anonymous",
    role=VIEWER,
    identity_source=IDENTITY_NONE,
    authenticated=False,
    warnings=("No authentication configured; capped at viewer.",),
 )
 def auth_mode(env: dict[str, str] | None = None) -> str:
    """Resolve the configured identity source, defaulting to ``none``."""
    source = env if env is not None else os.environ
    raw = (source.get(AUTH_MODE_ENV) or "").strip().lower().replace("-", "_")
    if raw in IDENTITY_SOURCES:
        return raw
    return IDENTITY_NONE
 def _role_map(env: dict[str, str]) -> dict[str, str]:
    """Parse ``WEBUI_ROLE_MAP`` (JSON subject→role). Invalid config yields {}."""
    raw = (env.get(ROLE_MAP_ENV) or "").strip()
    if not raw:
        return {}
    try:
        parsed = json.loads(raw)
    except Exception:
        return {}
    if not isinstance(parsed, dict):
        return {}
    return {
        str(k): str(v).strip().lower()
        for k, v in parsed.items()
        if str(v).strip().lower() in _ROLE_RANK
    }
 def resolve_principal(
    headers: dict[str, str] | None = None,
    env: dict[str, str] | None = None,
 ) -> Principal:
    """Resolve the requesting principal. Unknown or unconfigured → anonymous.
    Never raises and never trusts a client-supplied role: the role always comes
    from server-side configuration keyed by the resolved subject.
    """
    source_env = dict(env) if env is not None else dict(os.environ)
    lowered = {str(k).lower(): str(v) for k, v in (headers or {}).items()}
    mode = auth_mode(source_env)
    if mode == IDENTITY_LOCAL_DEV:
        subject = (source_env.get(DEV_SUBJECT_ENV) or "").strip()
        if not subject:
            return ANONYMOUS
        role = (source_env.get(DEV_ROLE_ENV) or VIEWER).strip().lower()
        if role not in _ROLE_RANK:
            role = VIEWER
        return Principal(
            subject=subject,
            role=role,
            identity_source=IDENTITY_LOCAL_DEV,
            authenticated=True,
            warnings=(
                "local-dev identity is asserted, not verified; never use "
                "outside loopback.",
            ),
        )
    if mode == IDENTITY_ACCESS_PROXY:
        subject = (lowered.get(ACCESS_SUBJECT_HEADER) or "").strip()
        if not subject:
            # Proxy mode with no proxy header means the request did not
            # traverse the proxy. Fail closed rather than trust it.
            return ANONYMOUS
        role = _role_map(source_env).get(subject, VIEWER)
        return Principal(
            subject=subject,
            role=role,
            identity_source=IDENTITY_ACCESS_PROXY,
            authenticated=True,
        )
    return ANONYMOUS
 def probe_auth_required(env: dict[str, str] | None = None) -> bool:
    """Whether non-public probes must be authenticated. Default False.
    #633 requires the console to *fail closed on missing auth for non-public
    health probes if configured*. The default stays off so the MVP ``/health``
    contract is unchanged; an operator opts in explicitly.
    """
    source = env if env is not None else os.environ
    return (source.get(REQUIRE_PROBE_AUTH_ENV) or "").strip().lower() in {
        "1",
        "true",
        "yes",
    }
 # --- Authorization ----------------------------------------------------------
 DENY_UNKNOWN_ACTION = "unknown_action"
 DENY_UNAUTHENTICATED = "unauthenticated"
 DENY_INSUFFICIENT_ROLE = "insufficient_role"
 DENY_UNKNOWN_ROLE = "unknown_role"
 DENY_PHASE_NOT_ACTIVE = "phase_not_active"
 ALLOW_PREVIEW = "allowed_preview_only"
 # Phase 1 is the only active console phase. Phase 2 opens gated writes and is
 # gated on this model landing; nothing here enables it.
 ACTIVE_PHASE = 1
@dataclass(frozen=True)
 class AuthorizationDecision:
    """Result of an authorization check. Never an execution grant."""
    allowed: bool
    reason_code: str
    detail: str
    action_id: str
    principal: Principal
    required_role: str | None = None
    action_class: str | None = None
    requires_confirmation: bool = False
    dual_control: bool = False
    break_glass: bool = False
    execution_enabled: bool = False
    def to_dict(self) -> dict[str, Any]:
        return {
            "allowed": self.allowed,
            "reason_code": self.reason_code,
            "detail": self.detail,
            "action_id": self.action_id,
            "principal": self.principal.to_dict(),
            "required_role": self.required_role,
            "action_class": self.action_class,
            "requires_confirmation": self.requires_confirmation,
            "dual_control": self.dual_control,
            "break_glass": self.break_glass,
            "execution_enabled": self.execution_enabled,
            "active_phase": ACTIVE_PHASE,
        }
 def authorize(
    action_id: str,
    principal: Principal | None = None,
    *,
    for_execution: bool = False,
 ) -> AuthorizationDecision:
    """Decide whether *principal* may invoke *action_id*. Deny by default.
    ``for_execution`` distinguishes a read-only preview from a real invocation.
    Even an allowed decision reports ``execution_enabled=False`` while the
    console is in Phase 1, so no caller can read an allow as permission to
    mutate.
    """
    who = principal if principal is not None else ANONYMOUS
    action = get_action(action_id)
    if action is None:
        return AuthorizationDecision(
            allowed=False,
            reason_code=DENY_UNKNOWN_ACTION,
            detail=f"No console action registered as {action_id!r}.",
            action_id=action_id,
            principal=who,
        )
    base: dict[str, Any] = {
        "action_id": action_id,
        "principal": who,
        "required_role": action.minimum_role,
        "action_class": action.action_class,
        "requires_confirmation": action.requires_confirmation,
        "dual_control": action.dual_control,
        "break_glass": action.break_glass,
        "execution_enabled": False,
    }
    if not who.authenticated:
        return AuthorizationDecision(
            allowed=False,
            reason_code=DENY_UNAUTHENTICATED,
            detail=(
                "Write actions require an authenticated principal; this "
                "request is anonymous."
            ),
            **base,
        )
    if who.rank < 0:
        return AuthorizationDecision(
            allowed=False,
            reason_code=DENY_UNKNOWN_ROLE,
            detail=f"Role {who.role!r} is not in the console role matrix.",
            **base,
        )
    if who.rank < _ROLE_RANK[action.minimum_role]:
        return AuthorizationDecision(
            allowed=False,
            reason_code=DENY_INSUFFICIENT_ROLE,
            detail=(
                f"Action {action_id!r} requires {action.minimum_role!r}; "
                f"principal holds {who.role!r}."
            ),
            **base,
        )
    if for_execution and action.phase > ACTIVE_PHASE:
        return AuthorizationDecision(
            allowed=False,
            reason_code=DENY_PHASE_NOT_ACTIVE,
            detail=(
                f"Action {action_id!r} belongs to phase {action.phase}; the "
                f"console is in phase {ACTIVE_PHASE}. Execution is not wired."
            ),
            **base,
        )
    return AuthorizationDecision(
        allowed=True,
        reason_code=ALLOW_PREVIEW,
        detail=(
            "Principal holds the required role. Preview only — execution "
            "remains disabled until the Phase 2 action framework ships."
        ),
        **base,
    )
 def rbac_matrix() -> dict[str, Any]:
    """Machine-readable RBAC matrix and privileged-action list."""
    return {
        "model_version": 1,
        "active_phase": ACTIVE_PHASE,
        "roles": [
            {
                "role": role,
                "rank": _ROLE_RANK[role],
                "description": ROLE_DESCRIPTIONS[role],
                "permitted_actions": sorted(
                    a.action_id
                    for a in _ACTION_SPECS
                    if _ROLE_RANK[role] >= _ROLE_RANK[a.minimum_role]
                ),
            }
            for role in ROLE_ORDER
        ],
        "identity_sources": IDENTITY_SOURCES,
        "actions": [a.to_dict() for a in _ACTION_SPECS],
        "privileged_actions": [a.action_id for a in privileged_actions()],
        "default_decision": "deny",
        "execution_enabled": False,
    }
@@ -1,169 +0,0 @@
 """Secret redaction policy for every console surface (#633).
 The MVP already redacts MCP-side mutation records through ``gitea_audit``.
 This module is the console-facing policy: one redaction pass applied to API
 payloads, rendered HTML, log lines, and audit records *before* they leave the
 server or reach persistent storage.
 Design constraints:
 - **Reuse, never fork.** ``gitea_audit.redact`` remains the authority for
  secret-looking dict keys, ``Authorization`` material, and raw URLs. This
  module runs that pass first and then applies console-specific patterns for
  keychain references, key/value assignments, private-key blocks, and JWTs.
 - **Never raises.** Redaction is a safety control; a malformed payload must
  degrade to a redacted placeholder rather than propagate an exception.
 - **Redact before persist.** ``webui.console_audit`` calls this module before
  writing, so an unredacted record is never durable.
 """
 from __future__ import annotations
 import json
 import re
 from typing import Any
 import gitea_audit
 REDACTED = gitea_audit.REDACTED
 # Console-specific patterns applied after the shared ``gitea_audit`` pass.
 # Each keeps the identifying key so an operator can still tell *what* was
 # removed, and replaces only the secret run itself.
 _KEYCHAIN_REF = re.compile(r"(?i)\bkeychain:[\w.\-/@]+")
 _KEYCHAIN_CMD = re.compile(
    r"(?i)\bsecurity\s+find-(?:generic|internet)-password\b[^\n]*"
 )
 _ASSIGNMENT = re.compile(
    r"(?i)\b(token|password|passwd|secret|api[_-]?key|access[_-]?key|"
    r"client[_-]?secret|private[_-]?key)\b(\s*[:=]\s*)"
    r"(\"[^\"]*\"|'[^']*'|\S+)"
 )
 _ENV_ASSIGNMENT = re.compile(
    r"(?i)\b(GITEA_(?:TOKEN|PASS|PASSWORD)[A-Z0-9_]*)(\s*=\s*)"
    r"(\"[^\"]*\"|'[^']*'|\S+)"
 )
 _PRIVATE_KEY_BLOCK = re.compile(
    r"-----BEGIN [A-Z ]*PRIVATE KEY-----.*?-----END [A-Z ]*PRIVATE KEY-----",
    re.S,
 )
 _JWT = re.compile(
    r"\beyJ[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}\b"
 )
 # Shapes that mean a payload still carries a secret. ``scan_for_secrets`` uses
 # these to assert a surface is clean.
 _DETECTORS: tuple[tuple[str, re.Pattern[str]], ...] = (
    ("keychain_reference", _KEYCHAIN_REF),
    ("keychain_command", _KEYCHAIN_CMD),
    ("credential_assignment", _ASSIGNMENT),
    ("credential_env_assignment", _ENV_ASSIGNMENT),
    ("private_key_block", _PRIVATE_KEY_BLOCK),
    ("json_web_token", _JWT),
    ("bearer_credential", re.compile(r"(?i)\b(?:bearer|basic)\s+\S{8,}")),
 )
 def _mask_assignment(match: re.Match[str]) -> str:
    """Keep the key and separator, replace the value."""
    return f"{match.group(1)}{match.group(2)}{REDACTED}"
 def redact_text(text: Any) -> Any:
    """Redact secret material from a single string.
    Non-strings are returned unchanged so this is safe to map over mixed
    payloads. Runs the shared ``gitea_audit`` pass first, then the
    console-specific patterns.
    """
    if not isinstance(text, str) or not text:
        return text
    try:
        out = gitea_audit.redact(text)
        if not isinstance(out, str):  # defensive; redact() returns str for str
            return REDACTED
        out = _PRIVATE_KEY_BLOCK.sub(f"{REDACTED}_PRIVATE_KEY", out)
        out = _ENV_ASSIGNMENT.sub(_mask_assignment, out)
        out = _ASSIGNMENT.sub(_mask_assignment, out)
        out = _KEYCHAIN_CMD.sub(f"{REDACTED}_KEYCHAIN_COMMAND", out)
        out = _KEYCHAIN_REF.sub(f"{REDACTED}_KEYCHAIN_REF", out)
        out = _JWT.sub(f"{REDACTED}_JWT", out)
        return out
    except Exception:
        # Fail closed: an unredactable string is dropped rather than emitted raw.
        return REDACTED
 def redact_payload(value: Any) -> Any:
    """Recursively redact a JSON-able payload for any console surface.
    Secret-looking dict keys are replaced wholesale by the shared
    ``gitea_audit`` policy; every remaining string is run through
    :func:`redact_text`.
    """
    try:
        shared = gitea_audit.redact(value)
    except Exception:
        return REDACTED
    return _walk(shared)
 def _walk(value: Any) -> Any:
    if isinstance(value, dict):
        return {k: _walk(v) for k, v in value.items()}
    if isinstance(value, (list, tuple)):
        return [_walk(v) for v in value]
    if isinstance(value, str):
        return redact_text(value)
    return value
 def scan_for_secrets(value: Any) -> list[str]:
    """Return detector names that still match *value* after serialization.
    Used to assert an outbound payload or rendered page is clean. An empty
    list means no known secret shape was found. Already-redacted hits are not
    findings.
    """
    if isinstance(value, str):
        text = value
    else:
        try:
            text = json.dumps(value, default=str)
        except Exception:
            text = str(value)
    findings: list[str] = []
    for name, pattern in _DETECTORS:
        for match in pattern.finditer(text):
            if REDACTED in match.group(0):
                continue
            findings.append(name)
            break
    return findings
 def redaction_policy() -> dict[str, Any]:
    """Machine-readable statement of the redaction rules (never secrets)."""
    return {
        "policy_version": 1,
        "applies_to": [
            "json_api_responses",
            "rendered_html",
            "server_logs",
            "audit_records",
        ],
        "ordering": "shared gitea_audit pass, then console patterns",
        "redact_before_persist": True,
        "shared_rules": {
            "source": "gitea_audit.redact",
            "secret_key_hints": list(gitea_audit._SECRET_KEY_HINTS),
            "secret_value_prefixes": list(gitea_audit._SECRET_VALUE_PREFIXES),
            "urls": "credentials, secret query parameters, and real hosts redacted",
        },
        "console_rules": [
            {"name": name, "pattern": pattern.pattern}
            for name, pattern in _DETECTORS
        ],
        "placeholder": REDACTED,
        "failure_mode": "fail closed — unredactable values become the placeholder",
    }
@@ -0,0 +1,682 @@
 """Read-only system-health model for the operator console API (#634).
 `/health` answers liveness only. Operators automating readiness checks need a
 structured view of *why* the control plane is or is not usable: which
 dependencies answered, how long they took, what version of the code is running,
 and whether the runtime is stale relative to its remote.
 Three rules shape this module.
 * **Read-only.** Every probe opens its subject read-only. The control-plane
  database is opened through a ``mode=ro`` URI so a health check can never
  create or migrate a schema, and no probe writes, restarts, or reloads
  anything — restart controls are Phase 2, and #630 forbids process-kill
  recovery outright.
 * **Fail-soft.** A dependency that is unreachable is a *status*, not an
  exception. Probes catch their own failures and report them as a degraded or
  down entry carrying a reason.
 * **Never claim more than was proven.** Readiness is derived only from probes
  that actually ran, ``mutation_safe`` stays false unless the parity commits are
  known and equal, and an MCP namespace is reported unproven because a web
  process cannot exercise the IDE-managed client path (#543).
 """
 from __future__ import annotations
 import os
 import re
 import sqlite3
 import subprocess
 import time
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Callable
 from urllib.parse import urlsplit, urlunsplit
 import control_plane_db
 import mcp_namespace_health
 from gitea_auth import api_request, get_auth_header, gitea_url
 from webui.project_registry import load_registry
 SERVICE_NAME = "mcp-control-plane-webui"
 API_PATH = "/api/v1/system/health"
 STATUS_OK = "ok"
 STATUS_DEGRADED = "degraded"
 STATUS_DOWN = "down"
 STATUS_SKIPPED = "skipped"
 STATUS_UNPROVEN = "unproven"
 # Statuses that count as a healthy answer from a probe.
 _HEALTHY_STATUSES = frozenset({STATUS_OK})
 # Statuses meaning "this probe did not run", as opposed to "it ran and failed".
 _NOT_RUN_STATUSES = frozenset({STATUS_SKIPPED})
 _DEEP_PROBE_TTL_ENV = "WEBUI_HEALTH_PROBE_TTL_SECONDS"
 _DEFAULT_DEEP_PROBE_TTL = 15.0
 _GITEA_PROBE_TIMEOUT_SECONDS = 5.0
 _OFFLINE_ENV = "WEBUI_TEST_OFFLINE"
 # Credential-shaped material that must never reach the browser, mirroring the
 # forbidden client patterns in webui/deployment_boundary.py.
 _SECRET_RE = re.compile(
    r"(?i)\b(token|password|passwd|secret|authorization|bearer)\b\s*[:=]?\s*\S+"
 )
 _LONG_OPAQUE_RE = re.compile(r"\b[A-Za-z0-9_\-]{32,}\b")
 # Captured once at import so uptime measures this process, not the request.
 _STARTED_AT = datetime.now(timezone.utc)
 _STARTED_MONOTONIC = time.monotonic()
 # TTL cache for the expensive (network) probe only.
 _deep_cache: dict[str, tuple[float, "DependencyProbe"]] = {}
@dataclass(frozen=True)
 class DependencyProbe:
    """One dependency check, fail-soft, with its own latency."""
    name: str
    kind: str
    status: str
    detail: str
    required: bool
    latency_ms: float | None = None
    metadata: dict[str, Any] | None = None
    @property
    def healthy(self) -> bool:
        return self.status in _HEALTHY_STATUSES
    @property
    def ran(self) -> bool:
        return self.status not in _NOT_RUN_STATUSES
@dataclass(frozen=True)
 class VersionInfo:
    git_sha: str | None
    git_describe: str | None
    control_plane_schema_version: int | None
    python_version: str
    known: bool
@dataclass(frozen=True)
 class StaleRuntime:
    """Parity between the running code, the checkout, and the remote.
    ``mutation_safe`` is deliberately conservative: unknown is not safe.
    """
    daemon_head: str | None
    checkout_head: str | None
    remote_head: str | None
    stale: bool
    determinable: bool
    mutation_safe: bool
    reasons: tuple[str, ...]
@dataclass(frozen=True)
 class SystemHealthSnapshot:
    status: str
    ready: bool
    readiness_complete: bool
    readiness_reasons: tuple[str, ...]
    service: str
    mode: str
    version: VersionInfo
    started_at: str
    uptime_seconds: float
    timestamp: str
    deep_probes_requested: bool
    dependencies: tuple[DependencyProbe, ...]
    mcp_namespaces: tuple[dict[str, Any], ...]
    stale_runtime: StaleRuntime
    probe_errors: tuple[str, ...] = ()
 def process_uptime() -> tuple[str, float]:
    """Process start timestamp and uptime — in-memory, safe for `/health`."""
    return _STARTED_AT.isoformat(), round(time.monotonic() - _STARTED_MONOTONIC, 3)
 def _offline() -> bool:
    return (os.environ.get(_OFFLINE_ENV) or "").strip().lower() in {"1", "true", "yes"}
 def _repo_root() -> Path:
    override = (os.environ.get("WEBUI_REPO_ROOT") or "").strip()
    if override:
        return Path(override).resolve()
    return Path(__file__).resolve().parent.parent
 def _deep_probe_ttl() -> float:
    raw = (os.environ.get(_DEEP_PROBE_TTL_ENV) or "").strip()
    if not raw:
        return _DEFAULT_DEEP_PROBE_TTL
    try:
        value = float(raw)
    except ValueError:
        return _DEFAULT_DEEP_PROBE_TTL
    return value if value >= 0 else _DEFAULT_DEEP_PROBE_TTL
 def redact(text: str) -> str:
    """Strip credential-shaped material from operator-visible probe text.
    Probe details carry exception strings, and an exception raised by an HTTP
    client can quote the request that failed. Redaction happens here, at the
    boundary where those strings become part of a browser-bound payload.
    """
    if not text:
        return ""
    cleaned = _redact_urls(text)
    cleaned = _SECRET_RE.sub(lambda m: f"{m.group(1)}=[redacted]", cleaned)
    return _LONG_OPAQUE_RE.sub("[redacted]", cleaned)
 def _redact_urls(text: str) -> str:
    return re.sub(r"https?://\S+", lambda m: redact_url(m.group(0)), text)
 def redact_url(url: str) -> str:
    """Reduce a URL to scheme://host/path — no userinfo, no query, no fragment."""
    try:
        parts = urlsplit(url)
    except ValueError:
        return "[redacted-url]"
    if not parts.scheme or not parts.hostname:
        return "[redacted-url]"
    netloc = parts.hostname
    if parts.port:
        netloc = f"{netloc}:{parts.port}"
    return urlunsplit((parts.scheme, netloc, parts.path, "", ""))
 def _git(repo: Path, *args: str) -> str | None:
    try:
        completed = subprocess.run(
            ["git", "-C", str(repo), *args],
            capture_output=True,
            text=True,
            check=False,
            timeout=10,
        )
    except (OSError, subprocess.SubprocessError):
        return None
    if completed.returncode != 0:
        return None
    return (completed.stdout or "").strip() or None
 def _load_version(repo: Path, *, schema_version: int | None) -> VersionInfo:
    import platform
    git_sha = None if _offline() else _git(repo, "rev-parse", "HEAD")
    describe = None if _offline() else _git(repo, "describe", "--tags", "--always")
    return VersionInfo(
        git_sha=git_sha,
        git_describe=describe,
        control_plane_schema_version=schema_version,
        python_version=platform.python_version(),
        known=bool(git_sha),
    )
 # ---------------------------------------------------------------------------
 # Dependency probes
 # ---------------------------------------------------------------------------
 def _elapsed_ms(started: float) -> float:
    return round((time.monotonic() - started) * 1000, 3)
 def probe_control_plane_db(db_path: str | None = None) -> DependencyProbe:
    """Read-only reachability check for the control-plane SQLite substrate.
    Opened through a ``mode=ro`` URI on purpose: ``ControlPlaneDB.__init__``
    creates directories and runs schema migrations, which a health check must
    never do.
    """
    path = (db_path or control_plane_db.default_db_path()).strip()
    started = time.monotonic()
    metadata: dict[str, Any] = {"path": path}
    def _result(status: str, detail: str) -> DependencyProbe:
        return DependencyProbe(
            name="control_plane_db",
            kind="sqlite",
            status=status,
            detail=detail,
            required=True,
            latency_ms=_elapsed_ms(started),
            metadata=metadata,
        )
    if not path or not os.path.exists(path):
        return _result(STATUS_DOWN, "control-plane database file does not exist yet")
    try:
        conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=5)
        try:
            row = conn.execute(
                "SELECT value FROM schema_meta WHERE key = 'schema_version'"
            ).fetchone()
            leases = conn.execute(
                "SELECT COUNT(*) FROM leases WHERE status = 'active'"
            ).fetchone()
        finally:
            conn.close()
    except sqlite3.Error as exc:
        return _result(STATUS_DOWN, redact(f"control-plane database unreadable: {exc}"))
    schema_version = int(row[0]) if row and str(row[0]).isdigit() else None
    metadata["schema_version"] = schema_version
    metadata["active_leases"] = int(leases[0]) if leases else None
    if schema_version is None:
        return _result(
            STATUS_DEGRADED, "control-plane database has no recorded schema version"
        )
    if schema_version != control_plane_db.SCHEMA_VERSION:
        return _result(
            STATUS_DEGRADED,
            f"control-plane schema version {schema_version} does not match the "
            f"version this code expects ({control_plane_db.SCHEMA_VERSION})",
        )
    return _result(STATUS_OK, f"schema v{schema_version} readable")
 def probe_repository(repo: Path) -> DependencyProbe:
    """Local checkout reachability — required, cheap, no network."""
    started = time.monotonic()
    metadata: dict[str, Any] = {"repo_root": str(repo)}
    if _offline():
        return DependencyProbe(
            name="repository",
            kind="git",
            status=STATUS_SKIPPED,
            detail=f"{_OFFLINE_ENV} is set; git probe skipped",
            required=True,
            latency_ms=_elapsed_ms(started),
            metadata=metadata,
        )
    head = _git(repo, "rev-parse", "HEAD")
    if not head:
        return DependencyProbe(
            name="repository",
            kind="git",
            status=STATUS_DOWN,
            detail=f"HEAD could not be read at {repo}",
            required=True,
            latency_ms=_elapsed_ms(started),
            metadata=metadata,
        )
    branch = _git(repo, "rev-parse", "--abbrev-ref", "HEAD")
    metadata["head"] = head
    metadata["branch"] = branch
    return DependencyProbe(
        name="repository",
        kind="git",
        status=STATUS_OK,
        detail=f"checkout readable at {branch or 'detached HEAD'}",
        required=True,
        latency_ms=_elapsed_ms(started),
        metadata=metadata,
    )
 def probe_gitea(host: str) -> DependencyProbe:
    """Live Gitea reachability. Expensive (network), so opt-in via ``deep``.
    Optional by design: the console stays useful for local inventory when the
    remote is unreachable, so a failure here degrades status without claiming
    the process itself is unready.
    """
    started = time.monotonic()
    metadata: dict[str, Any] = {"host": host}
    def _failure(status: str, detail: str) -> DependencyProbe:
        return DependencyProbe(
            name="gitea",
            kind="http",
            status=status,
            detail=detail,
            required=False,
            latency_ms=_elapsed_ms(started),
            metadata=metadata,
        )
    if not host:
        return _failure(STATUS_DEGRADED, "no Gitea host is configured in the registry")
    try:
        auth = get_auth_header(host)
    except Exception as exc:  # noqa: BLE001 — credential guards are a status here
        return _failure(STATUS_DEGRADED, redact(f"credential lookup refused: {exc}"))
    if not auth:
        return _failure(STATUS_DEGRADED, f"no credentials available for {host}")
    url = gitea_url(host, "/api/v1/version")
    metadata["endpoint"] = redact_url(url)
    try:
        data = api_request("GET", url, auth, timeout=_GITEA_PROBE_TIMEOUT_SECONDS)
    except Exception as exc:  # noqa: BLE001 — a down dependency is a status
        return _failure(STATUS_DOWN, redact(f"Gitea probe failed: {exc}"))
    if isinstance(data, dict) and data.get("version"):
        metadata["gitea_version"] = str(data["version"])
    return DependencyProbe(
        name="gitea",
        kind="http",
        status=STATUS_OK,
        detail=f"{host} reachable",
        required=False,
        latency_ms=_elapsed_ms(started),
        metadata=metadata,
    )
 def _skipped_gitea(host: str) -> DependencyProbe:
    return DependencyProbe(
        name="gitea",
        kind="http",
        status=STATUS_SKIPPED,
        detail="network probe not requested; call with ?deep=1 to run it",
        required=False,
        latency_ms=None,
        metadata={"host": host},
    )
 def namespace_summaries() -> tuple[dict[str, Any], ...]:
    """Declared MCP namespaces, each honestly reported as unproven.
    The web process runs outside the IDE-managed MCP client, so it cannot
    invoke a namespace tool. Per #543 only a ``client_namespace`` probe proves
    that path, and inventing a healthy verdict here is exactly the false claim
    the mutation gates exist to prevent.
    """
    rows: list[dict[str, Any]] = []
    for namespace, required_tool in sorted(
        mcp_namespace_health.REQUIRED_NAMESPACE_TOOLS.items()
    ):
        classification = mcp_namespace_health.classify_namespace_probe(
            namespace,
            required_tool=required_tool,
            probe_result=None,
            probe_source=mcp_namespace_health.PROBE_SOURCE_UNKNOWN,
        )
        rows.append(
            {
                "namespace": namespace,
                "required_tool": required_tool,
                "status": STATUS_UNPROVEN,
                "ide_namespace_proven": bool(classification.get("ide_namespace_proven")),
                "reason": (
                    "the web console cannot invoke the IDE-managed MCP client; "
                    "namespace health must be proven with a client_namespace "
                    "probe (#543)"
                ),
                "error_type": classification.get("error_type"),
            }
        )
    return tuple(rows)
 def assess_stale_runtime(
    repo: Path,
    *,
    daemon_head: str | None = None,
    git_reader: Callable[..., str | None] | None = None,
 ) -> StaleRuntime:
    """Three-way parity view: running code, local checkout, remote-tracking ref.
    ``mutation_safe`` requires all three to be known and equal. Anything less —
    including "the remote ref was never fetched" — is reported as not safe with
    a reason, so an operator never reads an unproven green.
    """
    reader = git_reader or (lambda *args: _git(repo, *args))
    reasons: list[str] = []
    # The offline switch suppresses real subprocess calls; an explicitly
    # injected reader is already a substitute for them and is always used.
    offline = _offline() and git_reader is None
    checkout_head = None if offline else reader("rev-parse", "HEAD")
    remote_head = None if offline else reader("rev-parse", "@{upstream}")
    if offline:
        reasons.append(f"{_OFFLINE_ENV} is set; parity commits were not read")
    else:
        if checkout_head is None:
            reasons.append("local checkout HEAD could not be read")
        if remote_head is None:
            reasons.append(
                "no remote-tracking commit is known for the current branch; "
                "remote staleness is indeterminate (no fetch is performed here)"
            )
    effective_daemon = daemon_head if daemon_head is not None else checkout_head
    if daemon_head is None:
        reasons.append(
            "the running MCP daemon's startup commit is not observable from the "
            "web process; the checkout commit is reported in its place"
        )
    determinable = bool(checkout_head and remote_head and effective_daemon)
    stale = bool(
        determinable and len({checkout_head, remote_head, effective_daemon}) > 1
    )
    if stale:
        reasons.append(
            "runtime, checkout, and remote commits disagree; restart the MCP "
            "server after updating the checkout before trusting capability gates"
        )
    return StaleRuntime(
        daemon_head=effective_daemon,
        checkout_head=checkout_head,
        remote_head=remote_head,
        stale=stale,
        determinable=determinable,
        mutation_safe=bool(determinable and not stale),
        reasons=tuple(reasons),
    )
 # ---------------------------------------------------------------------------
 # Snapshot assembly
 # ---------------------------------------------------------------------------
 def _default_host() -> str:
    registry = load_registry()
    if not registry.projects:
        return ""
    raw = registry.projects[0].remote_host
    parts = urlsplit(raw.strip())
    return parts.netloc or raw.strip().rstrip("/")
 def _aggregate(
    probes: tuple[DependencyProbe, ...],
 ) -> tuple[str, bool, bool, tuple[str, ...]]:
    """Fold probe results into overall status and readiness.
    Required probes drive readiness; optional probes can only degrade status.
    A probe that did not run leaves readiness incomplete rather than passing.
    """
    reasons: list[str] = []
    required = [probe for probe in probes if probe.required]
    unrun_required = [probe for probe in required if not probe.ran]
    failed_required = [probe for probe in required if probe.ran and not probe.healthy]
    failed_optional = [
        probe
        for probe in probes
        if not probe.required and probe.ran and not probe.healthy
    ]
    for probe in unrun_required:
        reasons.append(
            f"required dependency '{probe.name}' was not probed: {probe.detail}"
        )
    for probe in failed_required:
        reasons.append(
            f"required dependency '{probe.name}' is {probe.status}: {probe.detail}"
        )
    for probe in failed_optional:
        reasons.append(
            f"optional dependency '{probe.name}' is {probe.status}: {probe.detail}"
        )
    readiness_complete = not unrun_required
    ready = readiness_complete and not failed_required
    if any(probe.status == STATUS_DOWN for probe in failed_required):
        status = STATUS_DOWN
    elif failed_required or failed_optional or unrun_required:
        status = STATUS_DEGRADED
    else:
        status = STATUS_OK
    return status, ready, readiness_complete, tuple(reasons)
 def load_system_health(
    *,
    deep: bool = False,
    host: str | None = None,
    probes: tuple[DependencyProbe, ...] | None = None,
    daemon_head: str | None = None,
    use_cache: bool = True,
 ) -> SystemHealthSnapshot:
    """Assemble the read-only system-health snapshot.
    ``deep=True`` adds the network probe against Gitea; its result is cached for
    a short TTL so repeated dashboard polls do not amplify into remote load.
    """
    repo = _repo_root()
    probe_errors: list[str] = []
    if probes is None:
        collected: list[DependencyProbe] = []
        for probe_fn in (
            lambda: probe_control_plane_db(),
            lambda: probe_repository(repo),
        ):
            try:
                collected.append(probe_fn())
            except Exception as exc:  # noqa: BLE001 — a probe must not 500 the API
                probe_errors.append(redact(f"probe raised: {exc}"))
        resolved_host = host if host is not None else _default_host()
        if deep and not _offline():
            collected.append(_cached_gitea_probe(resolved_host, use_cache=use_cache))
        else:
            collected.append(_skipped_gitea(resolved_host))
        probes = tuple(collected)
    status, ready, readiness_complete, reasons = _aggregate(probes)
    stale = assess_stale_runtime(repo, daemon_head=daemon_head)
    if stale.stale:
        if status == STATUS_OK:
            status = STATUS_DEGRADED
        reasons = reasons + (
            "runtime is stale relative to its remote-tracking commit",
        )
    db_probe = next((p for p in probes if p.name == "control_plane_db"), None)
    schema_version = None
    if db_probe and db_probe.metadata:
        schema_version = db_probe.metadata.get("schema_version")
    return SystemHealthSnapshot(
        status=status,
        ready=ready,
        readiness_complete=readiness_complete,
        readiness_reasons=reasons,
        service=SERVICE_NAME,
        mode="read-only",
        version=_load_version(repo, schema_version=schema_version),
        started_at=_STARTED_AT.isoformat(),
        uptime_seconds=round(time.monotonic() - _STARTED_MONOTONIC, 3),
        timestamp=datetime.now(timezone.utc).isoformat(),
        deep_probes_requested=deep,
        dependencies=probes,
        mcp_namespaces=namespace_summaries(),
        stale_runtime=stale,
        probe_errors=tuple(probe_errors),
    )
 def _cached_gitea_probe(host: str, *, use_cache: bool = True) -> DependencyProbe:
    ttl = _deep_probe_ttl()
    now = time.monotonic()
    if use_cache and ttl > 0:
        cached = _deep_cache.get(host)
        if cached and (now - cached[0]) < ttl:
            return cached[1]
    probe = probe_gitea(host)
    if use_cache and ttl > 0:
        _deep_cache[host] = (now, probe)
    return probe
 def clear_probe_cache() -> None:
    """Drop cached deep-probe results (tests and operator-forced refresh)."""
    _deep_cache.clear()
 def probe_to_dict(probe: DependencyProbe) -> dict[str, Any]:
    return {
        "name": probe.name,
        "kind": probe.kind,
        "status": probe.status,
        "detail": probe.detail,
        "required": probe.required,
        "healthy": probe.healthy,
        "latency_ms": probe.latency_ms,
        "metadata": dict(probe.metadata or {}),
    }
 def snapshot_to_dict(snapshot: SystemHealthSnapshot) -> dict[str, Any]:
    return {
        "status": snapshot.status,
        "service": snapshot.service,
        "mode": snapshot.mode,
        "api": API_PATH,
        "timestamp": snapshot.timestamp,
        "readiness": {
            "ready": snapshot.ready,
            "complete": snapshot.readiness_complete,
            "reasons": list(snapshot.readiness_reasons),
        },
        "version": {
            "git_sha": snapshot.version.git_sha,
            "git_describe": snapshot.version.git_describe,
            "control_plane_schema_version": (
                snapshot.version.control_plane_schema_version
            ),
            "python_version": snapshot.version.python_version,
            "known": snapshot.version.known,
        },
        "process": {
            "started_at": snapshot.started_at,
            "uptime_seconds": snapshot.uptime_seconds,
        },
        "deep_probes_requested": snapshot.deep_probes_requested,
        "dependencies": [probe_to_dict(probe) for probe in snapshot.dependencies],
        "mcp_namespaces": [dict(row) for row in snapshot.mcp_namespaces],
        "stale_runtime": {
            "daemon_head": snapshot.stale_runtime.daemon_head,
            "checkout_head": snapshot.stale_runtime.checkout_head,
            "remote_head": snapshot.stale_runtime.remote_head,
            "stale": snapshot.stale_runtime.stale,
            "determinable": snapshot.stale_runtime.determinable,
            "mutation_safe": snapshot.stale_runtime.mutation_safe,
            "reasons": list(snapshot.stale_runtime.reasons),
        },
        "probe_errors": list(snapshot.probe_errors),
    }