Compare commits

..
Author SHA1 Message Date
sysadminandClaude Opus 4.8 b7a63a5579 feat(webui): unified session/lease/lock/worktree inventory API (Closes #636)
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-22 21:13:54 -04:00
22 changed files with 1554 additions and 6004 deletions
+40 -294
View File
@@ -78,33 +78,6 @@ VALID_ROLES = frozenset(
{ROLE_AUTHOR, ROLE_REVIEWER, ROLE_MERGER, ROLE_RECONCILER, ROLE_CONTROLLER}
)
# Allocation modes (#840).
# role_scoped: only candidates whose expected role matches the caller role.
# cross_role: controller-owned generic queue selection — inspect full queue,
# rank/eligibility canonically, return one selection naming the required
# downstream role/profile. Controller routes; it does not perform mutations.
ALLOCATION_MODE_ROLE_SCOPED = "role_scoped"
ALLOCATION_MODE_CROSS_ROLE = "cross_role"
VALID_ALLOCATION_MODES = frozenset(
{ALLOCATION_MODE_ROLE_SCOPED, ALLOCATION_MODE_CROSS_ROLE}
)
# Default execution-profile / MCP-namespace names for each role.
DEFAULT_ROLE_PROFILES: dict[str, str] = {
ROLE_AUTHOR: "prgs-author",
ROLE_REVIEWER: "prgs-reviewer",
ROLE_MERGER: "prgs-merger",
ROLE_RECONCILER: "prgs-reconciler",
ROLE_CONTROLLER: "prgs-controller",
}
DEFAULT_ROLE_NAMESPACES: dict[str, str] = {
ROLE_AUTHOR: "gitea-author",
ROLE_REVIEWER: "gitea-reviewer",
ROLE_MERGER: "gitea-merger",
ROLE_RECONCILER: "gitea-reconciler",
ROLE_CONTROLLER: "gitea-controller",
}
# Default action matrices by role (mutation gate will re-check).
ROLE_ACTIONS: dict[str, tuple[tuple[str, ...], tuple[str, ...]]] = {
ROLE_AUTHOR: (
@@ -286,126 +259,6 @@ def normalize_role(role: str | None, *, profile_name: str | None = None) -> str:
)
def resolve_allocation_mode(
role: str,
allocation_mode: str | None = None,
) -> str:
"""Resolve allocation mode; controller defaults to cross_role (#840)."""
raw = (allocation_mode or "").strip().lower()
if raw:
if raw not in VALID_ALLOCATION_MODES:
raise ControlPlaneError(
f"unknown allocation_mode {allocation_mode!r}; expected one of "
f"{sorted(VALID_ALLOCATION_MODES)}"
)
return raw
if role == ROLE_CONTROLLER:
return ALLOCATION_MODE_CROSS_ROLE
return ALLOCATION_MODE_ROLE_SCOPED
def required_profile_for_role(
role: str,
*,
profile_name: str | None = None,
) -> str:
"""Map a required role to the canonical execution profile name."""
role_norm = (role or "").strip().lower()
# Preserve remote/env prefix from the active profile when present
# (e.g. dadeschools-author → dadeschools-reviewer).
active = (profile_name or "").strip()
if active:
lower = active.lower()
for token in ("author", "reviewer", "merger", "reconciler", "controller"):
if lower.endswith(f"-{token}") or lower == token:
prefix = active[: -len(token)].rstrip("-")
if prefix:
return f"{prefix}-{role_norm}"
return role_norm
return DEFAULT_ROLE_PROFILES.get(role_norm, f"prgs-{role_norm}")
def required_namespace_for_role(
role: str,
*,
profile_name: str | None = None,
) -> str:
"""Map a required role to the canonical MCP namespace name."""
role_norm = (role or "").strip().lower()
profile = required_profile_for_role(role_norm, profile_name=profile_name)
# Namespace is typically gitea-<role>; keep stable mapping when profile is
# non-prgs (still gitea-<role> for isolation).
return DEFAULT_ROLE_NAMESPACES.get(role_norm, f"gitea-{role_norm}")
def selected_action_for_candidate(c: WorkCandidate, required_role: str) -> str:
"""Canonical next action for the selected work under *required_role*."""
role = (required_role or "").strip().lower()
if role == ROLE_AUTHOR:
if c.kind == "pr" and c.request_changes_current_head:
return "address_pr_change_requests"
if c.kind == "pr":
return "update_pr"
return "implement"
if role == ROLE_REVIEWER:
if c.approval_stale:
return "re_review"
return "review"
if role == ROLE_MERGER:
return "merge"
if role == ROLE_RECONCILER:
if c.approval_contaminated:
return "reconcile_contaminated_approval"
return "reconcile"
if role == ROLE_CONTROLLER:
return "diagnose"
return "process"
def build_selection_dict(
selected: WorkCandidate,
*,
active_role: str,
required_role: str,
profile_name: str | None = None,
allocation_mode: str,
) -> dict[str, Any]:
"""Authoritative single selection payload for allocator results (#840)."""
action = selected_action_for_candidate(selected, required_role)
req_profile = required_profile_for_role(
required_role, profile_name=profile_name
)
req_ns = required_namespace_for_role(
required_role, profile_name=profile_name
)
return {
"kind": selected.kind,
"number": selected.number,
"title": selected.title,
"labels": list(selected.labels),
"head_sha": selected.head_sha,
"priority": selected.priority,
"expected_role_next": required_role,
"required_role": required_role,
"selected_action": action,
"action": action,
"required_profile": req_profile,
"required_namespace": req_ns,
"pinned": {
"kind": selected.kind,
"number": selected.number,
"head_sha": selected.head_sha,
"issue_number": selected.number if selected.kind == "issue" else None,
"pr_number": selected.number if selected.kind == "pr" else None,
},
"reason_selected": (
f"highest-priority eligible candidate under allocation_mode="
f"'{allocation_mode}' (active_role={active_role}, "
f"required_role={required_role}, action={action})"
),
}
def expected_role_for_candidate(c: WorkCandidate) -> str:
"""ADR §5.3 routing: which role should take this work next."""
if c.kind == "pr":
@@ -436,7 +289,6 @@ def classify_skip(
role: str,
terminal_pr: int | None,
claim_ownership: str | None = None,
allocation_mode: str | None = None,
) -> str | None:
"""Return skip reason, or None if candidate is selectable for *role*.
@@ -445,12 +297,7 @@ def classify_skip(
and unknown claims are excluded so one session's in-progress task can never
blockade the queue for a different controller; ``own`` stays selectable so
a controller can resume its own work.
*allocation_mode* (#840): ``cross_role`` (controller default) ranks the full
queue and selects the highest-priority eligible item for any downstream
role. ``role_scoped`` retains prior role-match filtering.
"""
mode = resolve_allocation_mode(role, allocation_mode)
if c.state in ("merged", "closed"):
return f"{c.kind}#{c.number} is {c.state}; never assign"
if c.blocked or "status:blocked" in c.labels:
@@ -475,58 +322,34 @@ def classify_skip(
if c.kind == "pr" and not (c.head_sha or "").strip():
return f"pr#{c.number} missing head_sha pin"
expected = expected_role_for_candidate(c)
# Terminal path first: when an active terminal PR exists, only that PR
# is assignable for review-path roles (or for work whose expected role is
# review/merge under cross_role selection).
# (or controller diagnosis) is assignable for review-path roles.
if terminal_pr is not None and c.kind == "pr" and c.number != terminal_pr:
terminal_roles = (ROLE_REVIEWER, ROLE_MERGER)
if mode == ALLOCATION_MODE_CROSS_ROLE:
if expected in terminal_roles:
return (
f"pr#{c.number} skipped: active terminal-review lock on "
f"PR #{terminal_pr} must be resolved first"
)
elif role in terminal_roles:
if role in (ROLE_REVIEWER, ROLE_MERGER):
return (
f"pr#{c.number} skipped: active terminal-review lock on "
f"PR #{terminal_pr} must be resolved first"
)
if mode == ALLOCATION_MODE_CROSS_ROLE:
# Cross-role controller selection: eligibility only — no active-role
# match filter. The selection payload names required_role.
pass
elif role == ROLE_CONTROLLER:
# Legacy diagnosis-only controller path (role_scoped): only reconciler-
# needed targets. Prefer cross_role for generic queue allocation.
expected = expected_role_for_candidate(c)
if role == ROLE_CONTROLLER:
# Controller may inspect anything but only assigns diagnosis targets
# when contaminated / blocked.
if expected == ROLE_RECONCILER or c.blocked:
return None
return (
f"{c.kind}#{c.number} does not require controller "
f"(expected {expected})"
)
elif role != expected:
return f"{c.kind}#{c.number} does not require controller (expected {expected})"
if role != expected:
return (
f"{c.kind}#{c.number} expects role '{expected}', active role is '{role}'"
)
# Ready-gate for issues: prefer status:ready when labels present.
# Applies for author-bound work in both modes (cross_role only gates
# author-expected issues so reconciler/reviewer PRs stay selectable).
if c.kind == "issue" and c.labels:
gate_role = expected if mode == ALLOCATION_MODE_CROSS_ROLE else role
if gate_role in (ROLE_AUTHOR, ROLE_CONTROLLER):
if (
"status:ready" not in c.labels
and "status:in-progress" not in c.labels
):
if any(l.startswith("status:") for l in c.labels):
return (
f"issue#{c.number} not status:ready "
f"({','.join(c.labels)})"
)
if "status:ready" not in c.labels and "status:in-progress" not in c.labels:
# Allow unlabeled open issues; only skip explicit non-ready states.
if any(l.startswith("status:") for l in c.labels):
return f"issue#{c.number} not status:ready ({','.join(c.labels)})"
return None
@@ -678,19 +501,12 @@ def allocate_next_work(
claims: Mapping[tuple[str, int], dict[str, Any]] | None = None,
exclude_issue_numbers: Sequence[int] | None = None,
expected_candidate_set_fingerprint: str | None = None,
allocation_mode: str | None = None,
) -> dict[str, Any]:
"""Select and optionally reserve the next work unit via control-plane DB.
*apply=False* (default): dry-run selection only — no lease/assignment.
*apply=True*: atomic ``assign_and_lease`` for the selected candidate.
*allocation_mode* (#840): ``cross_role`` (default for controller) inspects
the complete queue and returns one authoritative selection naming the
required downstream role/profile/action. ``role_scoped`` keeps prior
per-role filtering. Controller routes only — never grants author/reviewer/
merger/reconciler mutation rights to the controller session.
*exclude_issue_numbers* (#776): numbers removed before ranking. Omitted /
empty preserves prior behavior.
@@ -725,19 +541,6 @@ def allocate_next_work(
"substrate": "control_plane_db",
}
try:
mode = resolve_allocation_mode(role_norm, allocation_mode)
except ControlPlaneError as exc:
return {
"success": False,
"outcome": OUTCOME_ROLE_INELIGIBLE,
"reasons": [str(exc)],
"skipped": [],
"assignment": None,
"substrate": "control_plane_db",
"allocation_mode": (allocation_mode or "").strip() or None,
}
session_id = (session_id or "").strip() or f"alloc-{uuid.uuid4().hex[:12]}"
try:
db.upsert_session(
@@ -945,7 +748,6 @@ def allocate_next_work(
role=role_norm,
terminal_pr=terminal_pr,
claim_ownership=ownership,
allocation_mode=mode,
)
if reason:
is_claim_skip = SKIP_CLAIMED_BY_OTHER_SESSION in reason
@@ -1026,10 +828,6 @@ def allocate_next_work(
"outcome": outcome,
"apply": bool(apply),
"role": role_norm,
"allocation_mode": mode,
"routing_role": role_norm,
"required_role": None,
"selected_action": None,
"profile_name": profile_name,
"username": username,
"session_id": session_id,
@@ -1042,12 +840,6 @@ def allocate_next_work(
"skipped": [s.as_dict() for s in skipped],
"terminal_pr": terminal_pr,
"assignment": None,
"allocation_evidence": {
"mode": "empty",
"allocation_mode": mode,
"lease_created": False,
"selection_policy": SELECTION_POLICY,
},
"substrate": "control_plane_db",
"file_lock_only": False,
"comment_lease_only": False,
@@ -1060,37 +852,25 @@ def allocate_next_work(
"owner_session_id": owner_session_id,
"downstream_note": (
"#612 incident bridge remains downstream of #600; "
"allocator never assigns raw monitoring incidents; "
"controller routes only under cross_role (#840)"
"allocator never assigns raw monitoring incidents"
),
}
expected_role = expected_role_for_candidate(selected)
# Cross-role: lease/action matrix follows the required downstream role so
# evidence names the worker that must act. Controller session still owns
# the routing decision; mutation isolation is enforced by role gates on
# mutation tools (controller profile lacks author/review/merge ops).
lease_role = (
expected_role if mode == ALLOCATION_MODE_CROSS_ROLE else role_norm
)
allowed, forbidden = role_actions(lease_role)
# Controller must never receive mutation-class rights via cross-role apply.
if role_norm == ROLE_CONTROLLER:
ctrl_allowed, ctrl_forbidden = role_actions(ROLE_CONTROLLER)
# Keep controller session capability evidence separate from lease_role.
controller_allowed_actions = ctrl_allowed
controller_forbidden_actions = ctrl_forbidden
else:
controller_allowed_actions = allowed
controller_forbidden_actions = forbidden
selection = build_selection_dict(
selected,
active_role=role_norm,
required_role=expected_role,
profile_name=profile_name,
allocation_mode=mode,
)
allowed, forbidden = role_actions(role_norm)
selection = {
"kind": selected.kind,
"number": selected.number,
"title": selected.title,
"labels": list(selected.labels),
"head_sha": selected.head_sha,
"priority": selected.priority,
"expected_role_next": expected_role,
"reason_selected": (
f"highest-priority candidate for role '{role_norm}' "
f"(expected_role={expected_role})"
),
}
if not apply:
return {
@@ -1098,12 +878,6 @@ def allocate_next_work(
"outcome": OUTCOME_PREVIEW,
"apply": False,
"role": role_norm,
"allocation_mode": mode,
"routing_role": role_norm,
"required_role": expected_role,
"selected_action": selection["selected_action"],
"required_profile": selection["required_profile"],
"required_namespace": selection["required_namespace"],
"profile_name": profile_name,
"username": username,
"session_id": session_id,
@@ -1119,12 +893,6 @@ def allocate_next_work(
"skipped": [s.as_dict() for s in skipped],
"terminal_pr": terminal_pr,
"assignment": None,
"allocation_evidence": {
"mode": "preview",
"allocation_mode": mode,
"lease_created": False,
"selection_policy": SELECTION_POLICY,
},
"substrate": "control_plane_db",
"file_lock_only": False,
"comment_lease_only": False,
@@ -1134,12 +902,9 @@ def allocate_next_work(
"controller_excluded": list(controller_excluded),
"exclude_issue_numbers": list(exclude_nums),
"candidate_set_fingerprint": cas_fp,
"controller_allowed_actions": list(controller_allowed_actions),
"controller_forbidden_actions": list(controller_forbidden_actions),
"downstream_note": (
"#612 incident bridge remains downstream of #600; "
"allocator never assigns raw monitoring incidents; "
"controller routes only under cross_role (#840)"
"allocator never assigns raw monitoring incidents"
),
}
@@ -1148,7 +913,7 @@ def allocate_next_work(
try:
kwargs: dict[str, Any] = {
"session_id": session_id,
"role": lease_role,
"role": role_norm,
"remote": remote,
"org": org,
"repo": repo,
@@ -1227,27 +992,11 @@ def allocate_next_work(
}
# assigned
lease_proof = {
"assignment_id": result.assignment_id,
"lease_id": result.lease_id,
"expires_at": result.expires_at,
"expected_head_sha": result.expected_head_sha,
"allowed_actions": list(result.allowed_actions),
"forbidden_actions": list(result.forbidden_actions),
"lease_role": lease_role,
"source": "control_plane_db.assign_and_lease",
}
return {
"success": True,
"outcome": OUTCOME_ASSIGNED,
"apply": True,
"role": role_norm,
"allocation_mode": mode,
"routing_role": role_norm,
"required_role": expected_role,
"selected_action": selection["selected_action"],
"required_profile": selection["required_profile"],
"required_namespace": selection["required_namespace"],
"profile_name": profile_name,
"username": username,
"session_id": session_id,
@@ -1263,16 +1012,16 @@ def allocate_next_work(
"skipped": [s.as_dict() for s in skipped],
"terminal_pr": terminal_pr,
"assignment": result.as_dict(),
"lease_proof": lease_proof,
"allocation_evidence": {
"mode": "assigned",
"allocation_mode": mode,
"lease_created": True,
"lease_role": lease_role,
"lease_proof": lease_proof,
"selection_policy": SELECTION_POLICY,
"lease_proof": {
"assignment_id": result.assignment_id,
"lease_id": result.lease_id,
"expires_at": result.expires_at,
"expected_head_sha": result.expected_head_sha,
"allowed_actions": list(result.allowed_actions),
"forbidden_actions": list(result.forbidden_actions),
"source": "control_plane_db.assign_and_lease",
},
"next_valid_command": _next_command(lease_role, selected),
"next_valid_command": _next_command(role_norm, selected),
"substrate": "control_plane_db",
"file_lock_only": False,
"comment_lease_only": False,
@@ -1282,12 +1031,9 @@ def allocate_next_work(
"controller_excluded": list(controller_excluded),
"exclude_issue_numbers": list(exclude_nums),
"candidate_set_fingerprint": cas_fp,
"controller_allowed_actions": list(controller_allowed_actions),
"controller_forbidden_actions": list(controller_forbidden_actions),
"downstream_note": (
"#612 incident bridge remains downstream of #600; "
"allocator never assigns raw monitoring incidents; "
"controller routes only under cross_role (#840)"
"allocator never assigns raw monitoring incidents"
),
}
-892
View File
@@ -1,892 +0,0 @@
"""ARCH-01 Foundation Slice A — atomic platform installation + authority kernel (#822).
Parents: #820, #821. **First implementation leaf of the ARCH-01 program.**
This module implements the smallest executable ARCH-01 foundation:
* a connection-bound authenticated actor context (``cp_actor_*`` /
``cp_operation_mode`` / ``cp_context_epoch`` SQLite scalar functions that SQL
may *read* but can never *set* — ``[TRUSTED-SERVICE]`` authenticity);
* an immutable authority-dominance lattice with an exact seeded tuple set
(``[SCHEMA]``);
* the principal-equivalence root (a class exists *before* its first principal;
``principals.current_class_id`` is ``NOT NULL``; ``[SCHEMA]``);
* a single-transaction platform installation that seeds the initial
``platform.bootstrap`` grant and an immutable ``installed`` marker, validated
by a fail-closed ``install_state`` ``BEFORE INSERT`` trigger (``[SCHEMA]``).
Everything else in the ARCH-01/02/04 program (evidence stores, repository
bindings, workspaces, PostgreSQL parity, full grant succession, full principal
merge) is out of scope here and tracked in its own issue — see #822 §5/§17.
**Readiness / production posture.** This subsystem is *disabled by default*.
Nothing in the running MCP server imports or enables it. It becomes a security
boundary only once its readiness checks (the ACs in #822) pass in the target
environment. Instantiating :class:`PlatformKernel` creates an isolated SQLite
database and never touches the operational control-plane store.
Enforcement classification (per #820 vocabulary):
* ``[TRUSTED-SERVICE]`` — actor-context authenticity: the scalar functions are
registered by the trusted Python process; SQL cannot define or redefine them.
* ``[SCHEMA]`` — fail-closed aborts, the dominance/immutability/NOT-NULL-class/
last-active-grant invariants, enforced by CHECK/FK/trigger.
* ``[RUNTIME-ADAPTER]`` — *none* in this slice.
SQLite-first. ``BEGIN IMMEDIATE`` serializes concurrent installs and concurrent
grant/revoke on the singleton invariant row. PostgreSQL parity is a distinct
issue (#827); this module does **not** claim it.
"""
from __future__ import annotations
import os
import sqlite3
import threading
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Iterator, Optional
# --------------------------------------------------------------------------- #
# Closed enumerations (#822 §4).
# --------------------------------------------------------------------------- #
ACTOR_KINDS = ("operator", "supervisor", "service", "installer")
OPERATION_MODES = ("normal", "install", "merge", "internal_service")
# Exact seeded authority-dominance tuple set (#822 §4). This set is normative:
# the install-state trigger rejects any missing, additional, or malformed tuple.
DOMINANCE_TUPLES = (
("platform.bootstrap", "platform.bootstrap"),
("platform.bootstrap", "project.admin"),
("platform.bootstrap", "supervisor.root.establish"),
("supervisor.root", "supervisor.register"),
("supervisor.root", "supervisor.verify"),
("supervisor.root", "supervisor.recover"),
)
# The distinguished operator-key issuer seeded during install.
DISTINGUISHED_ISSUER_KIND = "operator-key"
DISTINGUISHED_ISSUER_ID = "platform.bootstrap.operator-key"
# Structured result codes (#822 §10).
INSTALLED = "INSTALLED"
ALREADY_INSTALLED = "ALREADY_INSTALLED"
INVALID_ACTOR_CONTEXT = "INVALID_ACTOR_CONTEXT"
INVALID_BOOTSTRAP_STATE = "INVALID_BOOTSTRAP_STATE"
DOMINANCE_SET_MISMATCH = "DOMINANCE_SET_MISMATCH"
AUTHORIZATION_DENIED = "AUTHORIZATION_DENIED"
CONCURRENT_INSTALLATION_LOST = "CONCURRENT_INSTALLATION_LOST"
# Required audit events (#822 §14).
EVT_PLATFORM_INSTALLED = "platform_installed"
EVT_GRANT_CREATED = "platform_grant_created"
EVT_GRANT_REVOKED = "platform_grant_revoked"
EVT_PRINCIPAL_REGISTERED = "principal_registered"
SCHEMA_VERSION = 1
DB_PATH_ENV = "ARCH01_PLATFORM_DB"
class PlatformKernelError(RuntimeError):
"""Base class for structured, code-bearing kernel failures."""
def __init__(self, code: str, message: str = "") -> None:
super().__init__(message or code)
self.code = code
class ActorContextError(PlatformKernelError):
"""Raised when a mutation is attempted without a valid actor context."""
# --------------------------------------------------------------------------- #
# Schema (#822 §6). Tables + fail-closed triggers.
#
# Every *mutating* trigger opens with the actor protocol: read the context
# epoch, read the actor fields, and abort unless the context is present,
# non-null, mode/kind well-formed, and epoch-consistent with the active
# transaction. The scalar functions ``cp_*`` are registered from Python only;
# SQL has no statement that can set them, which is the trusted-service boundary.
# --------------------------------------------------------------------------- #
_ACTOR_KINDS_SQL = ", ".join("'%s'" % k for k in ACTOR_KINDS)
_OP_MODES_SQL = ", ".join("'%s'" % m for m in OPERATION_MODES)
# Actor-protocol predicate: TRUE when the context is INVALID and the trigger
# must abort. ``cp_actor_context_valid()`` folds "present + non-expired +
# live-epoch == bound-epoch" (the read/re-read epoch equality of #822 §4) into
# one trusted-service answer; the remaining reads assert field well-formedness.
_INVALID_ACTOR = (
"cp_actor_context_valid() IS NOT 1 "
"OR cp_context_epoch() IS NULL "
"OR cp_actor_principal() IS NULL "
"OR cp_actor_kind() NOT IN (%s) "
"OR cp_operation_mode() NOT IN (%s)" % (_ACTOR_KINDS_SQL, _OP_MODES_SQL)
)
_ACTOR_GUARD = (
"SELECT CASE WHEN (%s) "
"THEN RAISE(ABORT, 'INVALID_ACTOR_CONTEXT') END;" % _INVALID_ACTOR
)
# require_installed: abort a privileged mutation when there is no install
# marker and we are not currently installing (#822 §4).
_REQUIRE_INSTALLED = (
"SELECT CASE WHEN ((SELECT COUNT(*) FROM install_state) = 0 "
"AND cp_operation_mode() <> 'install') "
"THEN RAISE(ABORT, 'NOT_INSTALLED') END;"
)
_SCHEMA_SQL = f"""
PRAGMA foreign_keys = ON;
CREATE TABLE IF NOT EXISTS arch01_meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
-- Equivalence classes are created BEFORE their first principal (#822 §4).
CREATE TABLE IF NOT EXISTS principal_equivalence_classes (
class_id INTEGER PRIMARY KEY AUTOINCREMENT,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS authoritative_issuers (
issuer_id INTEGER PRIMARY KEY AUTOINCREMENT,
issuer_kind TEXT NOT NULL,
issuer_ref TEXT NOT NULL,
created_at TEXT NOT NULL,
UNIQUE (issuer_kind, issuer_ref)
);
-- current_class_id is NOT NULL: a principal cannot exist without a class
-- (#822 AC6). issuer_id is nullable ONLY for the installer during install
-- (#822 AC7), enforced by trg_principals_null_issuer below.
CREATE TABLE IF NOT EXISTS principals (
principal_id TEXT PRIMARY KEY,
actor_kind TEXT NOT NULL CHECK (actor_kind IN ({_ACTOR_KINDS_SQL})),
current_class_id INTEGER NOT NULL REFERENCES principal_equivalence_classes(class_id),
issuer_id INTEGER REFERENCES authoritative_issuers(issuer_id),
registered_by TEXT REFERENCES principals(principal_id),
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS authority_dominance (
dominant TEXT NOT NULL,
subordinate TEXT NOT NULL,
PRIMARY KEY (dominant, subordinate)
);
CREATE TABLE IF NOT EXISTS platform_bootstrap_seed (
seed_id INTEGER PRIMARY KEY CHECK (seed_id = 1),
installer_principal_id TEXT NOT NULL REFERENCES principals(principal_id),
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS platform_bootstrap_grants (
grant_id INTEGER PRIMARY KEY AUTOINCREMENT,
grantee_principal_id TEXT NOT NULL REFERENCES principals(principal_id),
granted_by TEXT REFERENCES principals(principal_id),
active INTEGER NOT NULL DEFAULT 1 CHECK (active IN (0, 1)),
created_at TEXT NOT NULL,
revoked_at TEXT
);
-- Singleton row; active_count floored at 1 by CHECK so the last active grant
-- can never be revoked (#822 AC11).
CREATE TABLE IF NOT EXISTS platform_active_invariant (
id INTEGER PRIMARY KEY CHECK (id = 1),
active_count INTEGER NOT NULL CHECK (active_count >= 1)
);
-- The immutable install marker; inserted LAST in the install transaction.
CREATE TABLE IF NOT EXISTS install_state (
id INTEGER PRIMARY KEY CHECK (id = 1),
marker TEXT NOT NULL CHECK (marker = 'installed'),
installed_at TEXT NOT NULL
);
-- Append-only (#822 AC14).
CREATE TABLE IF NOT EXISTS audit_records (
audit_id INTEGER PRIMARY KEY AUTOINCREMENT,
event TEXT NOT NULL,
principal_id TEXT,
detail TEXT,
created_at TEXT NOT NULL
);
-- ------------------------------------------------------------------------- --
-- Actor protocol on every mutating trigger (#822 §4, [SCHEMA] fail-closed).
-- ------------------------------------------------------------------------- --
CREATE TRIGGER IF NOT EXISTS trg_classes_actor
BEFORE INSERT ON principal_equivalence_classes
BEGIN
{_ACTOR_GUARD}
END;
CREATE TRIGGER IF NOT EXISTS trg_issuers_actor
BEFORE INSERT ON authoritative_issuers
BEGIN
{_ACTOR_GUARD}
END;
CREATE TRIGGER IF NOT EXISTS trg_principals_actor
BEFORE INSERT ON principals
BEGIN
{_ACTOR_GUARD}
END;
CREATE TRIGGER IF NOT EXISTS trg_dominance_actor
BEFORE INSERT ON authority_dominance
BEGIN
{_ACTOR_GUARD}
END;
CREATE TRIGGER IF NOT EXISTS trg_seed_actor
BEFORE INSERT ON platform_bootstrap_seed
BEGIN
{_ACTOR_GUARD}
END;
CREATE TRIGGER IF NOT EXISTS trg_grants_actor_insert
BEFORE INSERT ON platform_bootstrap_grants
BEGIN
{_ACTOR_GUARD}
{_REQUIRE_INSTALLED}
END;
CREATE TRIGGER IF NOT EXISTS trg_grants_actor_update
BEFORE UPDATE ON platform_bootstrap_grants
BEGIN
{_ACTOR_GUARD}
END;
CREATE TRIGGER IF NOT EXISTS trg_invariant_actor_insert
BEFORE INSERT ON platform_active_invariant
BEGIN
{_ACTOR_GUARD}
END;
CREATE TRIGGER IF NOT EXISTS trg_invariant_actor_update
BEFORE UPDATE ON platform_active_invariant
BEGIN
{_ACTOR_GUARD}
END;
CREATE TRIGGER IF NOT EXISTS trg_audit_actor
BEFORE INSERT ON audit_records
BEGIN
{_ACTOR_GUARD}
END;
-- ------------------------------------------------------------------------- --
-- NOT-NULL-issuer exception for the installer only (#822 AC7).
-- A NULL issuer_id is accepted solely for an installer principal during
-- install mode, before the marker exists; any other NULL-issuer principal is
-- rejected. install-time issuer linkage (installer -> distinguished issuer)
-- is applied by a later UPDATE, permitted while no marker exists.
-- ------------------------------------------------------------------------- --
CREATE TRIGGER IF NOT EXISTS trg_principals_null_issuer
BEFORE INSERT ON principals
WHEN NEW.issuer_id IS NULL
BEGIN
SELECT CASE WHEN NOT (
NEW.actor_kind = 'installer'
AND cp_operation_mode() = 'install'
AND (SELECT COUNT(*) FROM install_state) = 0
AND (SELECT COUNT(*) FROM principals WHERE issuer_id IS NULL) = 0
) THEN RAISE(ABORT, 'INVALID_BOOTSTRAP_STATE') END;
END;
-- ------------------------------------------------------------------------- --
-- Post-install immutability of the authority root (#822 §4, AC9).
-- Registration fields freeze only AFTER the marker exists, so the install
-- transaction's own installer issuer-linkage UPDATE is permitted.
-- ------------------------------------------------------------------------- --
CREATE TRIGGER IF NOT EXISTS trg_principals_frozen_update
BEFORE UPDATE ON principals
WHEN (SELECT COUNT(*) FROM install_state) > 0
BEGIN
SELECT RAISE(ABORT, 'IMMUTABLE_PRINCIPAL');
END;
CREATE TRIGGER IF NOT EXISTS trg_principals_frozen_delete
BEFORE DELETE ON principals
BEGIN
SELECT RAISE(ABORT, 'IMMUTABLE_PRINCIPAL');
END;
-- Distinguished issuer identity is immutable once written.
CREATE TRIGGER IF NOT EXISTS trg_issuers_immutable_update
BEFORE UPDATE ON authoritative_issuers
BEGIN
SELECT RAISE(ABORT, 'IMMUTABLE_ISSUER');
END;
CREATE TRIGGER IF NOT EXISTS trg_issuers_immutable_delete
BEFORE DELETE ON authoritative_issuers
BEGIN
SELECT RAISE(ABORT, 'IMMUTABLE_ISSUER');
END;
-- The dominance lattice is immutable once seeded.
CREATE TRIGGER IF NOT EXISTS trg_dominance_immutable_update
BEFORE UPDATE ON authority_dominance
BEGIN
SELECT RAISE(ABORT, 'IMMUTABLE_DOMINANCE');
END;
CREATE TRIGGER IF NOT EXISTS trg_dominance_immutable_delete
BEFORE DELETE ON authority_dominance
BEGIN
SELECT RAISE(ABORT, 'IMMUTABLE_DOMINANCE');
END;
-- The bootstrap seed is immutable once written.
CREATE TRIGGER IF NOT EXISTS trg_seed_immutable_update
BEFORE UPDATE ON platform_bootstrap_seed
BEGIN
SELECT RAISE(ABORT, 'IMMUTABLE_SEED');
END;
CREATE TRIGGER IF NOT EXISTS trg_seed_immutable_delete
BEFORE DELETE ON platform_bootstrap_seed
BEGIN
SELECT RAISE(ABORT, 'IMMUTABLE_SEED');
END;
-- The install marker is immutable once written.
CREATE TRIGGER IF NOT EXISTS trg_install_state_immutable_update
BEFORE UPDATE ON install_state
BEGIN
SELECT RAISE(ABORT, 'IMMUTABLE_INSTALL_STATE');
END;
CREATE TRIGGER IF NOT EXISTS trg_install_state_immutable_delete
BEFORE DELETE ON install_state
BEGIN
SELECT RAISE(ABORT, 'IMMUTABLE_INSTALL_STATE');
END;
-- Grants: identity is immutable; the ONLY permitted mutation is a single
-- active 1 -> 0 revocation (#822 §4 initial-grant identity immutability +
-- grant/revoke). Reactivation and identity edits are rejected.
CREATE TRIGGER IF NOT EXISTS trg_grants_identity_frozen
BEFORE UPDATE ON platform_bootstrap_grants
WHEN NOT (
NEW.grant_id = OLD.grant_id
AND NEW.grantee_principal_id = OLD.grantee_principal_id
AND NEW.granted_by IS OLD.granted_by
AND NEW.created_at = OLD.created_at
AND OLD.active = 1
AND NEW.active = 0
)
BEGIN
SELECT RAISE(ABORT, 'IMMUTABLE_GRANT');
END;
CREATE TRIGGER IF NOT EXISTS trg_grants_no_delete
BEFORE DELETE ON platform_bootstrap_grants
BEGIN
SELECT RAISE(ABORT, 'IMMUTABLE_GRANT');
END;
-- audit_records is append-only.
CREATE TRIGGER IF NOT EXISTS trg_audit_immutable_update
BEFORE UPDATE ON audit_records
BEGIN
SELECT RAISE(ABORT, 'IMMUTABLE_AUDIT');
END;
CREATE TRIGGER IF NOT EXISTS trg_audit_immutable_delete
BEFORE DELETE ON audit_records
BEGIN
SELECT RAISE(ABORT, 'IMMUTABLE_AUDIT');
END;
-- ------------------------------------------------------------------------- --
-- install_state BEFORE INSERT: validate the whole bootstrap atomically
-- (#822 §4, AC4). Each dominance tuple is checked individually; a missing,
-- additional, or malformed tuple -> DOMINANCE_SET_MISMATCH. The seed<->installer
-- link, the single active NULL-grantor installer grant, the installer's
-- non-NULL issuer, the active invariant, and "no extra principal created under
-- the NULL-issuer exception" -> INVALID_BOOTSTRAP_STATE.
-- ------------------------------------------------------------------------- --
CREATE TRIGGER IF NOT EXISTS trg_install_state_validate
BEFORE INSERT ON install_state
BEGIN
SELECT CASE WHEN NOT (
(SELECT COUNT(*) FROM authority_dominance) = {len(DOMINANCE_TUPLES)}
AND EXISTS (SELECT 1 FROM authority_dominance WHERE dominant='platform.bootstrap' AND subordinate='platform.bootstrap')
AND EXISTS (SELECT 1 FROM authority_dominance WHERE dominant='platform.bootstrap' AND subordinate='project.admin')
AND EXISTS (SELECT 1 FROM authority_dominance WHERE dominant='platform.bootstrap' AND subordinate='supervisor.root.establish')
AND EXISTS (SELECT 1 FROM authority_dominance WHERE dominant='supervisor.root' AND subordinate='supervisor.register')
AND EXISTS (SELECT 1 FROM authority_dominance WHERE dominant='supervisor.root' AND subordinate='supervisor.verify')
AND EXISTS (SELECT 1 FROM authority_dominance WHERE dominant='supervisor.root' AND subordinate='supervisor.recover')
) THEN RAISE(ABORT, 'DOMINANCE_SET_MISMATCH') END;
SELECT CASE WHEN NOT (
(SELECT COUNT(*) FROM platform_bootstrap_seed) = 1
AND (SELECT COUNT(*) FROM principals) = 1
AND (SELECT actor_kind FROM principals
WHERE principal_id = (SELECT installer_principal_id FROM platform_bootstrap_seed WHERE seed_id = 1)
) = 'installer'
AND (SELECT issuer_id FROM principals
WHERE principal_id = (SELECT installer_principal_id FROM platform_bootstrap_seed WHERE seed_id = 1)
) IS NOT NULL
AND (SELECT COUNT(*) FROM platform_bootstrap_grants
WHERE granted_by IS NULL AND active = 1
AND grantee_principal_id = (SELECT installer_principal_id FROM platform_bootstrap_seed WHERE seed_id = 1)
) = 1
AND (SELECT COUNT(*) FROM platform_bootstrap_grants) = 1
AND (SELECT active_count FROM platform_active_invariant WHERE id = 1) = 1
) THEN RAISE(ABORT, 'INVALID_BOOTSTRAP_STATE') END;
END;
"""
def default_db_path() -> str:
return os.environ.get(
DB_PATH_ENV,
os.path.expanduser("~/.cache/gitea-tools/arch01/platform.sqlite3"),
)
def _utc_now_iso() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
@dataclass(frozen=True)
class OperationResult:
"""Structured result of a kernel operation (#822 §10)."""
code: str
detail: str = ""
@property
def ok(self) -> bool:
return self.code in (INSTALLED, ALREADY_INSTALLED)
@dataclass
class _ActorContext:
principal: str
kind: str
mode: str
session: Optional[str]
bound_epoch: int
live_epoch: int
expired: bool = False
class PlatformKernel:
"""ARCH-01 authority kernel over a single SQLite connection.
The connection carries the trusted-service actor context: the ``cp_*``
scalar functions read the context this object holds. Only Python code here
can bind or clear it, so no SQL statement can assert an actor identity — the
trusted-service authenticity boundary of #822 §4.
"""
def __init__(self, db_path: Optional[str] = None, *, busy_timeout_ms: int = 5000) -> None:
self.db_path = db_path or default_db_path()
if self.db_path != ":memory:":
parent = os.path.dirname(self.db_path)
if parent:
os.makedirs(parent, exist_ok=True)
self._ctx: Optional[_ActorContext] = None
self._epoch_seq = 0
self._lock = threading.Lock()
# check_same_thread=False is safe: every mutation path is serialized
# by self._lock, so the connection is never used concurrently even when
# callers drive the kernel from different threads (concurrency tests).
self._conn = sqlite3.connect(
self.db_path, isolation_level=None, check_same_thread=False
)
self._conn.execute("PRAGMA foreign_keys = ON")
self._conn.execute(f"PRAGMA busy_timeout = {int(busy_timeout_ms)}")
self._register_actor_functions()
self._migrate()
# -- trusted-service actor functions ---------------------------------- #
def _register_actor_functions(self) -> None:
c = self._conn
c.create_function("cp_actor_principal", 0, lambda: self._ctx.principal if self._ctx else None)
c.create_function("cp_actor_kind", 0, lambda: self._ctx.kind if self._ctx else None)
c.create_function("cp_operation_mode", 0, lambda: self._ctx.mode if self._ctx else None)
c.create_function("cp_service_session", 0, lambda: self._ctx.session if self._ctx else None)
c.create_function("cp_context_epoch", 0, self._fn_context_epoch)
# Trusted-service helper: folds present + non-expired + epoch-consistent
# into the read/re-read epoch equality of #822 §4.
c.create_function("cp_actor_context_valid", 0, self._fn_context_valid)
def _fn_context_epoch(self) -> Optional[int]:
if self._ctx is None or self._ctx.expired:
return None
return self._ctx.live_epoch
def _fn_context_valid(self) -> int:
ctx = self._ctx
if ctx is None or ctx.expired:
return 0
# read/re-read epoch equality: a context whose live epoch has drifted
# from the epoch it was bound to (a stale/replaced connection context)
# is not bound to the active transaction and fails closed.
if ctx.live_epoch != ctx.bound_epoch:
return 0
if ctx.principal is None:
return 0
if ctx.kind not in ACTOR_KINDS or ctx.mode not in OPERATION_MODES:
return 0
return 1
# -- context lifecycle ------------------------------------------------ #
@contextmanager
def actor_context(
self, principal: str, kind: str, mode: str, session: Optional[str] = None
) -> Iterator[None]:
"""Bind a trusted actor context for the duration of the block."""
prev = self._ctx
self._epoch_seq += 1
epoch = self._epoch_seq
self._ctx = _ActorContext(
principal=principal, kind=kind, mode=mode, session=session,
bound_epoch=epoch, live_epoch=epoch,
)
try:
yield
finally:
self._ctx = prev
def _clear_context(self) -> None:
self._ctx = None
# -- migration -------------------------------------------------------- #
def _migrate(self) -> None:
self._conn.executescript(_SCHEMA_SQL)
self._conn.execute(
"INSERT OR IGNORE INTO arch01_meta(key, value) VALUES ('schema_version', ?)",
(str(SCHEMA_VERSION),),
)
self._conn.execute(
"INSERT OR IGNORE INTO arch01_meta(key, value) VALUES "
"('architecture', 'ARCH-01 Slice A: atomic install + authority kernel (#822); "
"disabled by default until readiness checks pass')"
)
# -- introspection ---------------------------------------------------- #
def is_installed(self) -> bool:
row = self._conn.execute("SELECT COUNT(*) FROM install_state").fetchone()
return bool(row[0])
def active_grant_count(self) -> int:
row = self._conn.execute(
"SELECT active_count FROM platform_active_invariant WHERE id = 1"
).fetchone()
return int(row[0]) if row else 0
def audit_events(self) -> list[str]:
return [
r[0]
for r in self._conn.execute(
"SELECT event FROM audit_records ORDER BY audit_id"
).fetchall()
]
def close(self) -> None:
self._conn.close()
# -- operations ------------------------------------------------------- #
def install_platform(
self,
installer_principal_id: str = "platform.installer",
*,
session: Optional[str] = None,
) -> OperationResult:
"""Single atomic install transaction (#822 §4/§7).
``BEGIN IMMEDIATE`` serializes concurrent installs; the loser rechecks
the marker and returns ``ALREADY_INSTALLED``, or — if it never acquires
the write lock — ``CONCURRENT_INSTALLATION_LOST``. On any stage failure
the whole transaction rolls back leaving no partial rows (AC3/AC5).
"""
now = _utc_now_iso()
with self._lock:
try:
self._conn.execute("BEGIN IMMEDIATE")
except sqlite3.OperationalError as exc:
if "locked" in str(exc).lower() or "busy" in str(exc).lower():
return OperationResult(CONCURRENT_INSTALLATION_LOST, str(exc))
raise
try:
if self.is_installed():
self._conn.execute("ROLLBACK")
return OperationResult(ALREADY_INSTALLED, "install marker already present")
with self.actor_context(installer_principal_id, "installer", "install", session):
c = self._conn
# class -> installer principal (temporary NULL issuer)
cur = c.execute(
"INSERT INTO principal_equivalence_classes(created_at) VALUES (?)",
(now,),
)
class_id = cur.lastrowid
c.execute(
"INSERT INTO principals"
"(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
"VALUES (?, 'installer', ?, NULL, ?, ?)",
(installer_principal_id, class_id, installer_principal_id, now),
)
# distinguished operator-key issuer
cur = c.execute(
"INSERT INTO authoritative_issuers(issuer_kind, issuer_ref, created_at) "
"VALUES (?, ?, ?)",
(DISTINGUISHED_ISSUER_KIND, DISTINGUISHED_ISSUER_ID, now),
)
issuer_id = cur.lastrowid
# link installer -> issuer (permitted pre-marker)
c.execute(
"UPDATE principals SET issuer_id = ? WHERE principal_id = ?",
(issuer_id, installer_principal_id),
)
# dominance tuples
c.executemany(
"INSERT INTO authority_dominance(dominant, subordinate) VALUES (?, ?)",
DOMINANCE_TUPLES,
)
# seed
c.execute(
"INSERT INTO platform_bootstrap_seed(seed_id, installer_principal_id, created_at) "
"VALUES (1, ?, ?)",
(installer_principal_id, now),
)
# initial grant (granted_by NULL, active)
c.execute(
"INSERT INTO platform_bootstrap_grants"
"(grantee_principal_id, granted_by, active, created_at) "
"VALUES (?, NULL, 1, ?)",
(installer_principal_id, now),
)
# active invariant
c.execute(
"INSERT INTO platform_active_invariant(id, active_count) VALUES (1, 1)"
)
# audit rows for the security-sensitive operation
c.execute(
"INSERT INTO audit_records(event, principal_id, detail, created_at) "
"VALUES (?, ?, ?, ?)",
(EVT_PRINCIPAL_REGISTERED, installer_principal_id, "installer", now),
)
c.execute(
"INSERT INTO audit_records(event, principal_id, detail, created_at) "
"VALUES (?, ?, ?, ?)",
(EVT_GRANT_CREATED, installer_principal_id, "initial platform.bootstrap grant", now),
)
# install marker LAST -> fires the whole-bootstrap validator
c.execute(
"INSERT INTO install_state(id, marker, installed_at) VALUES (1, 'installed', ?)",
(now,),
)
c.execute(
"INSERT INTO audit_records(event, principal_id, detail, created_at) "
"VALUES (?, ?, ?, ?)",
(EVT_PLATFORM_INSTALLED, installer_principal_id, "platform installed", now),
)
self._conn.execute("COMMIT")
return OperationResult(INSTALLED, "platform installed")
except sqlite3.Error as exc:
self._safe_rollback()
return OperationResult(self._classify(exc), str(exc))
def register_principal(
self,
principal_id: str,
actor_kind: str,
issuer_ref: str,
*,
actor_principal: str,
actor_kind_ctx: str = "operator",
session: Optional[str] = None,
) -> OperationResult:
"""Atomically create an equivalence class and its first principal.
The class is inserted *before* the principal, and ``current_class_id``
is ``NOT NULL`` (#822 AC6): a principal can never exist classless.
The principal references an existing issuer (non-NULL); the temporary
NULL-issuer exception is reserved for the installer during install
(AC7).
"""
if actor_kind not in ACTOR_KINDS:
return OperationResult(INVALID_BOOTSTRAP_STATE, f"bad actor_kind {actor_kind!r}")
now = _utc_now_iso()
with self._lock:
try:
self._conn.execute("BEGIN IMMEDIATE")
except sqlite3.OperationalError as exc:
return OperationResult(AUTHORIZATION_DENIED, str(exc))
try:
if not self.is_installed():
self._conn.execute("ROLLBACK")
return OperationResult(INVALID_BOOTSTRAP_STATE, "platform not installed")
row = self._conn.execute(
"SELECT issuer_id FROM authoritative_issuers WHERE issuer_ref = ?",
(issuer_ref,),
).fetchone()
if row is None:
self._conn.execute("ROLLBACK")
return OperationResult(INVALID_BOOTSTRAP_STATE, f"unknown issuer {issuer_ref!r}")
issuer_id = row[0]
with self.actor_context(actor_principal, actor_kind_ctx, "normal", session):
cur = self._conn.execute(
"INSERT INTO principal_equivalence_classes(created_at) VALUES (?)",
(now,),
)
class_id = cur.lastrowid
self._conn.execute(
"INSERT INTO principals"
"(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
"VALUES (?, ?, ?, ?, ?, ?)",
(principal_id, actor_kind, class_id, issuer_id, actor_principal, now),
)
self._conn.execute(
"INSERT INTO audit_records(event, principal_id, detail, created_at) "
"VALUES (?, ?, ?, ?)",
(EVT_PRINCIPAL_REGISTERED, principal_id, actor_kind, now),
)
self._conn.execute("COMMIT")
return OperationResult(INSTALLED, f"registered {principal_id}")
except sqlite3.Error as exc:
self._safe_rollback()
return OperationResult(self._classify(exc), str(exc))
def grant_platform_bootstrap(
self,
grantee_principal_id: str,
granted_by: str,
*,
actor_kind_ctx: str = "operator",
session: Optional[str] = None,
) -> OperationResult:
"""Create an additional active platform.bootstrap grant.
Serialized on the singleton invariant row via ``BEGIN IMMEDIATE``.
"""
now = _utc_now_iso()
with self._lock:
try:
self._conn.execute("BEGIN IMMEDIATE")
except sqlite3.OperationalError as exc:
return OperationResult(AUTHORIZATION_DENIED, str(exc))
try:
if not self.is_installed():
self._conn.execute("ROLLBACK")
return OperationResult(INVALID_BOOTSTRAP_STATE, "platform not installed")
with self.actor_context(granted_by, actor_kind_ctx, "normal", session):
self._conn.execute(
"INSERT INTO platform_bootstrap_grants"
"(grantee_principal_id, granted_by, active, created_at) "
"VALUES (?, ?, 1, ?)",
(grantee_principal_id, granted_by, now),
)
self._conn.execute(
"UPDATE platform_active_invariant SET active_count = active_count + 1 WHERE id = 1"
)
self._conn.execute(
"INSERT INTO audit_records(event, principal_id, detail, created_at) "
"VALUES (?, ?, ?, ?)",
(EVT_GRANT_CREATED, grantee_principal_id, f"granted_by={granted_by}", now),
)
self._conn.execute("COMMIT")
return OperationResult(INSTALLED, f"granted to {grantee_principal_id}")
except sqlite3.Error as exc:
self._safe_rollback()
return OperationResult(self._classify(exc), str(exc))
def revoke_platform_bootstrap(
self,
grant_id: int,
*,
actor_principal: str,
actor_kind_ctx: str = "operator",
session: Optional[str] = None,
) -> OperationResult:
"""Revoke an active grant, floored so the last one can never drop.
The ``active_count >= 1`` CHECK plus ``BEGIN IMMEDIATE`` serialization
make two concurrent revocations unable to remove the final active grant
(#822 AC11): the decrement that would reach zero fails and rolls back.
"""
now = _utc_now_iso()
with self._lock:
try:
self._conn.execute("BEGIN IMMEDIATE")
except sqlite3.OperationalError as exc:
return OperationResult(AUTHORIZATION_DENIED, str(exc))
try:
if not self.is_installed():
self._conn.execute("ROLLBACK")
return OperationResult(INVALID_BOOTSTRAP_STATE, "platform not installed")
row = self._conn.execute(
"SELECT active, grantee_principal_id FROM platform_bootstrap_grants WHERE grant_id = ?",
(grant_id,),
).fetchone()
if row is None or row[0] != 1:
self._conn.execute("ROLLBACK")
return OperationResult(AUTHORIZATION_DENIED, "grant absent or already inactive")
grantee = row[1]
with self.actor_context(actor_principal, actor_kind_ctx, "normal", session):
# Decrement first: the CHECK floor rejects dropping below 1,
# aborting the whole revoke before the grant flips inactive.
self._conn.execute(
"UPDATE platform_active_invariant SET active_count = active_count - 1 WHERE id = 1"
)
self._conn.execute(
"UPDATE platform_bootstrap_grants SET active = 0, revoked_at = ? WHERE grant_id = ?",
(now, grant_id),
)
self._conn.execute(
"INSERT INTO audit_records(event, principal_id, detail, created_at) "
"VALUES (?, ?, ?, ?)",
(EVT_GRANT_REVOKED, grantee, f"grant_id={grant_id}", now),
)
self._conn.execute("COMMIT")
return OperationResult(INSTALLED, f"revoked grant {grant_id}")
except sqlite3.Error as exc:
self._safe_rollback()
return OperationResult(self._classify(exc), str(exc))
# -- helpers ---------------------------------------------------------- #
def _safe_rollback(self) -> None:
try:
self._conn.execute("ROLLBACK")
except sqlite3.Error:
pass
@staticmethod
def _classify(exc: sqlite3.Error) -> str:
msg = str(exc)
if "INVALID_ACTOR_CONTEXT" in msg:
return INVALID_ACTOR_CONTEXT
if "DOMINANCE_SET_MISMATCH" in msg:
return DOMINANCE_SET_MISMATCH
if "active_count" in msg or "CHECK constraint failed: platform_active_invariant" in msg:
# last-active-grant floor tripped
return AUTHORIZATION_DENIED
if any(tag in msg for tag in (
"INVALID_BOOTSTRAP_STATE", "IMMUTABLE_", "NOT_INSTALLED",
)):
return INVALID_BOOTSTRAP_STATE
return INVALID_BOOTSTRAP_STATE
-295
View File
@@ -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.
+1 -4
View File
@@ -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.
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).
- **Authentication:** none in MVP — protection comes from network placement
- **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 -81
View File
@@ -52,8 +52,7 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
| Path | Description |
|------|-------------|
| `/` | Home / operator overview |
| `/health` | JSON liveness (`status`, `service`, `mode`, `timestamp`, `uptime_seconds`) |
| `/api/v1/system/health` | Structured read-only system health (#634) |
| `/health` | JSON liveness (`status`, `service`, `mode`, `timestamp`) |
| `/queue` | Live PR and issue queue dashboard (#429) |
| `/api/queue` | JSON queue export with pagination metadata |
| `/projects` | Project registry list with status and onboarding progress (#427, #635) |
@@ -79,85 +78,6 @@ 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
@@ -292,6 +212,53 @@ health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
checkout is behind merged safety-gate changes. Restart guidance links to #420;
no tokens or MCP restart actions are exposed.
## Inventory API (#636)
`GET /api/v1/inventory` returns one versioned, read-only snapshot that unifies
what the lease (#433), worktree (#432), and runtime (#430) MVP views each show
separately, so traffic-control and recovery consumers read the same source.
`GET /api/v1/inventory/{section}` returns a single section under the identical
schema (`sessions`, `leases`, `locks`, `worktrees`, `namespaces`); an unknown
section is a `404` with `error: unknown_section`. Both routes are `GET`-only.
Each section carries its own `status` (`ok` / `degraded` / `unavailable`), a
`reason` when not `ok`, and a `scan_ms`. A subsystem that cannot be read
degrades to a reasoned section; it never raises and never emits an empty list
that would read as "nothing is there".
### Field authority
Every section names where its rows came from; authorities are never blended.
| Section | Authority | Source |
|---|---|---|
| `sessions` | `control_plane_db` | #613 control-plane DB (`mode=ro`), authoritative for exclusive ownership (#600/#601) |
| `leases` | `control_plane_db` | #613 control-plane DB; degrades if the `work_items` table is absent |
| `locks` | `filesystem` | durable per-issue lock files (`issue_lock_store`) |
| `worktrees` | `filesystem` | registered git worktrees via the #432 hygiene scanner |
| `namespaces` | `filesystem` | the active profile serving this web process (others are not enumerable) |
The payload restates this map under `field_authority` for machine consumers.
### Ownership safety
`ownership_authority_complete` is true only when every ownership-bearing section
(`sessions`, `leases`, `locks`) read cleanly. While it is false, nothing is
reported as unowned and no collision is asserted from a degraded source —
absence of evidence is reported as absence of evidence, never as free work.
`collisions` surfaces detectable conflicts, each with a `kind` and `severity`:
`lock-without-worktree`, `duplicate-live-lock`, `live-lock-dead-owner` (unexpired
lease, dead pid — a #753 recovery candidate that would read as live to a naive
timestamp check), `stale-lock-dead-owner`, `expired-lock-live-owner` (the
#635/#760 daemon-pid deadlock), `concurrent-active-lease`, `active-lease-past-expiry`,
and `orphan-lease`. Collisions are emitted only from sections that read cleanly.
The control-plane DB is opened through a `mode=ro` URI so a read never creates
or migrates it; paths are collapsed against `$HOME`, URLs lose userinfo and
query strings, and credential-shaped values are redacted at the boundary. Lease
steal/release and worktree deletion are Phase 2+ and have no representation here.
## Tests
```bash
+6 -41
View File
@@ -234,25 +234,12 @@ def _effective_workspace_role() -> str:
def _profile_role_kind(profile: dict) -> str:
"""Resolve a profile's declared role before inferring from permissions.
Declared ``role`` / ``role_kind`` always wins so a controller profile is
never reclassified as reconciler from permission inference (#840).
"""
role = (profile.get("role") or profile.get("role_kind") or "").strip().lower()
"""Resolve a profile's declared role before inferring from permissions."""
role = (profile.get("role") or profile.get("role_kind") or "").strip()
if role:
# Normalize aliases / case.
if "control" in role:
return "controller"
return role
profile_name = (profile.get("profile_name") or "").strip().lower()
for candidate in (
"controller",
"reconciler",
"merger",
"reviewer",
"author",
):
for candidate in ("reconciler", "merger", "reviewer", "author"):
if candidate in profile_name:
return candidate
return _role_kind(
@@ -15551,8 +15538,7 @@ def mcp_get_control_plane_guide(
profile = get_profile()
allowed = profile["allowed_operations"]
forbidden = profile["forbidden_operations"]
# Prefer declared profile role so controller is not mislabeled reconciler (#840).
role = _profile_role_kind(profile)
role = _role_kind(allowed, forbidden)
username = _authenticated_username(h)
identity = {
@@ -15611,16 +15597,6 @@ def mcp_get_control_plane_guide(
"user, and merging requires explicit operator authorization plus the "
"'MERGE PR <n>' confirmation. "
"Review and merge are separate workflow roles. A reviewer approval is not merge authorization.")
elif role == "controller":
guidance.append(
"Controller profile: route work via "
"gitea_route_task_session(task_type='process_work_queue') then "
"gitea_allocate_next_work (allocation_mode=cross_role by default). "
"The allocator returns exactly one authoritative selection with "
"required_role / required_profile / selected_action. Do not "
"implement, review, approve, or merge in this session — schedule "
"the matching role namespace instead. Dashboard output is "
"explanatory only and never replaces allocator selection.")
elif role == "mixed":
guidance.append(
"WARNING: this profile allows both authoring and "
@@ -15830,8 +15806,7 @@ def gitea_whoami(
"environment": profile.get("environment"),
"service": profile.get("service"),
"identity": profile.get("identity"),
"role": profile.get("role") or _profile_role_kind(profile),
"role_kind": _profile_role_kind(profile),
"role": profile.get("role"),
"profile_address": profile.get("profile_path"),
"execution_profile": profile.get("execution_profile"),
"audit_label": profile.get("audit_label"),
@@ -20615,10 +20590,9 @@ def gitea_allocate_next_work(
candidates_json: Any = None,
exclude_issue_numbers: list[int] | None = None,
expected_candidate_set_fingerprint: str | None = None,
allocation_mode: str | None = None,
limit: int = 50,
) -> dict:
"""Controller-owned next-work allocator using the #613 control-plane DB (#600/#840).
"""Controller-owned next-work allocator using the #613 control-plane DB (#600).
Workers must not self-select exclusive work under the standard multi-LLM
workflow. Call this tool instead.
@@ -20628,14 +20602,6 @@ def gitea_allocate_next_work(
``ControlPlaneDB.assign_and_lease`` (never file locks or comment-only
leases as the coordination source).
*allocation_mode* (#840): when the active role is controller (or mode is
``cross_role``), inspect the complete queue and return exactly one
authoritative selection with selected item, action, required_role,
required_profile/namespace, pins, and allocation/lease evidence.
Role-scoped workers pass ``role=author|reviewer|merger|reconciler`` (or
omit for profile role) for single-role filtering. Controller routes only
and does not perform downstream mutations.
Outcomes include: ``assigned_work``, ``preview``, ``wait``,
``blocked_by_terminal_path``, ``no_safe_work``, ``role_ineligible``,
``blocked_by_excluded_own_lease``, ``candidate_set_drift``.
@@ -20770,7 +20736,6 @@ def gitea_allocate_next_work(
controller_instance_id=allocator_service.resolve_controller_instance_id(),
exclude_issue_numbers=exclude_issue_numbers,
expected_candidate_set_fingerprint=expected_candidate_set_fingerprint,
allocation_mode=allocation_mode,
)
except ValueError as exc:
return {
+5 -19
View File
@@ -24,12 +24,7 @@ ROLE_WORKTREE_ENVS: dict[str, str] = {
"reconciler": RECONCILER_WORKTREE_ENV,
}
# Controller has no task worktree env — it routes only (#840).
KNOWN_ROLE_KINDS = frozenset(
{"author", "reviewer", "merger", "reconciler", "controller"}
)
NON_AUTHOR_ROLES = frozenset({"reviewer", "merger", "reconciler", "controller"})
NON_AUTHOR_ROLES = frozenset({"reviewer", "merger", "reconciler"})
def normalize_role_kind(
@@ -42,12 +37,8 @@ def normalize_role_kind(
profile = (profile_name or "").strip().lower()
if role == "reviewer" and "merger" in profile:
return "merger"
if "controller" in profile or role == "controller":
return "controller"
if role in ROLE_WORKTREE_ENVS:
return role
if role in KNOWN_ROLE_KINDS:
return role
return "author"
@@ -89,7 +80,7 @@ def resolve_namespace_workspace(
"""
env_map = env if env is not None else os.environ
role = normalize_role_kind(role_kind, profile_name=profile_name)
role_env_key = ROLE_WORKTREE_ENVS.get(role)
role_env_key = ROLE_WORKTREE_ENVS[role]
# #618: durable author resolution — no silent control/master fallback.
if role == "author" and verify_paths:
@@ -117,17 +108,13 @@ def resolve_namespace_workspace(
)
return workspace, source
role_env_candidate = (
(_env_value(env_map, role_env_key), f"{role_env_key} environment variable", True)
if role_env_key
else (None, "no role worktree env", True)
)
for candidate, source, env_sourced in (
(worktree_path, "worktree_path argument", False),
(worktree, "worktree argument", False),
(_env_value(env_map, ACTIVE_WORKTREE_ENV),
f"{ACTIVE_WORKTREE_ENV} environment variable", True),
role_env_candidate,
(_env_value(env_map, role_env_key),
f"{role_env_key} environment variable", True),
(session_lease_worktree if role in {"reviewer", "merger"} else None,
"reviewer PR lease worktree", False),
# Author lock derivation is handled by the durable path above when
@@ -446,8 +433,7 @@ def assess_namespace_mutation_workspace(
reasons.append(
f"{role} mutation blocked: workspace is the stable control checkout; "
f"create or reconnect to a session-owned worktree under branches/ "
f"or set {ROLE_WORKTREE_ENVS.get(role, ACTIVE_WORKTREE_ENV)} / "
f"{ACTIVE_WORKTREE_ENV}"
f"or set {ROLE_WORKTREE_ENVS[role]} / {ACTIVE_WORKTREE_ENV}"
)
elif (
role in {"reviewer", "merger"}
-35
View File
@@ -81,12 +81,6 @@ AUTHOR_TASKS = frozenset({
"reconcile_landed_pr",
})
CONTROLLER_TASKS = frozenset({
"process_work_queue",
"process-work-queue",
"cross_role_allocate",
})
RECONCILER_TASKS = frozenset({
"cleanup_merged_pr_branch",
# #729: delete_branch is reconciler-owned (gitea.branch.delete is granted
@@ -138,10 +132,6 @@ TASK_REQUIRED_ROLE = {
"reconcile_close_superseded_pr": "reconciler",
"reconcile_close_satisfied_issue": "reconciler",
"reconcile_create_followup_issue": "reconciler",
# #840: controller-owned generic queue allocation / routing.
"process_work_queue": "controller",
"process-work-queue": "controller",
"cross_role_allocate": "controller",
}
WRONG_ROLE_REVIEWER_MSG = (
@@ -157,10 +147,6 @@ WRONG_ROLE_MERGER_MSG = (
"Wrong role/session for merger task. Launch merger MCP namespace."
)
WRONG_ROLE_CONTROLLER_MSG = (
"Wrong role/session for controller task. Launch controller MCP namespace."
)
_session_last_route: dict | None = None
@@ -295,27 +281,6 @@ def route_task_session(
_record_route(result)
return result
if required_role == "controller":
result = {
"task_type": task_type,
"required_role": required_role,
"active_role": active_role_kind,
"active_profile": active_profile,
"route_result": ROUTE_WRONG_ROLE,
"downstream_allowed": False,
"reasons": [
WRONG_ROLE_CONTROLLER_MSG,
"Controller tasks (process_work_queue / cross-role allocate) "
"cannot run in author, reviewer, merger, or reconciler "
"worker sessions.",
],
"message": WRONG_ROLE_CONTROLLER_MSG,
"runtime_switching_supported": runtime_switching_supported,
"profile_switch_blocked": not runtime_switching_supported,
}
_record_route(result)
return result
if required_role == "author":
route = ROUTE_TO_AUTHOR
message = (
+2 -17
View File
@@ -309,10 +309,8 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.pr.create",
"role": "author",
},
# #600: workers and controller may call with gitea.read; role-scoped workers
# pass role=author|reviewer|merger|reconciler. Cross-role routing is the
# controller default (#840). The canonical generic queue *task type* is
# process_work_queue (controller-only below).
# #600: controller-owned allocator — any authenticated profile may call;
# routing enforces role match to selected work. Uses control-plane DB (#613).
"allocate_next_work": {
"permission": "gitea.read",
"role": "author",
@@ -321,19 +319,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.read",
"role": "author",
},
# #840: documented generic queue task — controller routes only.
"process_work_queue": {
"permission": "gitea.read",
"role": "controller",
},
"process-work-queue": {
"permission": "gitea.read",
"role": "controller",
},
"cross_role_allocate": {
"permission": "gitea.read",
"role": "controller",
},
# #601 first-class lease lifecycle — inspect/list need read; mutations gate on
# ownership in the control-plane DB (not a separate Gitea write permission).
-572
View File
@@ -1,572 +0,0 @@
"""Executable acceptance tests for ARCH-01 Slice A (#822).
Each acceptance criterion (#822 §12) and named test (#822 §13) is exercised
against a real SQLite database. The migration runs on a fresh DB in ``setUp``;
the test-run output is the durable evidence the issue requires (§14).
Enforcement being proven:
* ``[TRUSTED-SERVICE]`` the ``cp_*`` actor functions exist only on the
trusted kernel connection; a raw connection cannot satisfy the triggers.
* ``[SCHEMA]`` fail-closed aborts, exact dominance set, NOT-NULL class,
immutability, and the last-active-grant floor are enforced by
CHECK/FK/trigger, verified here including raw-write bypass and concurrency.
"""
from __future__ import annotations
import os
import sqlite3
import tempfile
import threading
import unittest
from concurrent.futures import ThreadPoolExecutor
import arch01_platform as ap
from arch01_platform import (
ALREADY_INSTALLED,
AUTHORIZATION_DENIED,
CONCURRENT_INSTALLATION_LOST,
DISTINGUISHED_ISSUER_ID,
DOMINANCE_SET_MISMATCH,
DOMINANCE_TUPLES,
INSTALLED,
INVALID_ACTOR_CONTEXT,
INVALID_BOOTSTRAP_STATE,
PlatformKernel,
)
INSTALLER = "platform.installer"
_BOOTSTRAP_TABLES = (
"principal_equivalence_classes",
"principals",
"authoritative_issuers",
"authority_dominance",
"platform_bootstrap_seed",
"platform_bootstrap_grants",
"platform_active_invariant",
"install_state",
)
def _count(kernel: PlatformKernel, table: str) -> int:
return kernel._conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
def _count_where(kernel: PlatformKernel, table: str, where: str) -> int:
return kernel._conn.execute(f"SELECT COUNT(*) FROM {table} WHERE {where}").fetchone()[0]
def _all_bootstrap_empty(kernel: PlatformKernel) -> bool:
return all(_count(kernel, t) == 0 for t in _BOOTSTRAP_TABLES)
class Arch01MemoryTest(unittest.TestCase):
"""Single-connection behavior on an in-memory database."""
def setUp(self) -> None:
self.kernel = PlatformKernel(":memory:")
def tearDown(self) -> None:
self.kernel.close()
# -- AC1 -------------------------------------------------------------- #
def test_install_clean(self) -> None: # t_install_clean(+)
res = self.kernel.install_platform(INSTALLER)
self.assertEqual(res.code, INSTALLED)
self.assertTrue(self.kernel.is_installed())
self.assertEqual(_count(self.kernel, "install_state"), 1)
self.assertEqual(self.kernel.active_grant_count(), 1)
self.assertIn(ap.EVT_PLATFORM_INSTALLED, self.kernel.audit_events())
rows = set(
self.kernel._conn.execute(
"SELECT dominant, subordinate FROM authority_dominance"
).fetchall()
)
self.assertEqual(rows, set(DOMINANCE_TUPLES))
issuer_ref = self.kernel._conn.execute(
"SELECT i.issuer_ref FROM principals p JOIN authoritative_issuers i "
"ON p.issuer_id = i.issuer_id WHERE p.principal_id = ?",
(INSTALLER,),
).fetchone()
self.assertEqual(issuer_ref[0], DISTINGUISHED_ISSUER_ID)
# -- AC2 -------------------------------------------------------------- #
def test_install_twice(self) -> None: # t_install_twice(-)
self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
res2 = self.kernel.install_platform(INSTALLER)
self.assertEqual(res2.code, ALREADY_INSTALLED)
self.assertEqual(_count(self.kernel, "principals"), 1)
self.assertEqual(_count(self.kernel, "platform_bootstrap_grants"), 1)
self.assertEqual(_count(self.kernel, "install_state"), 1)
# -- AC3 / AC5 -------------------------------------------------------- #
def test_install_stage_rollback(self) -> None: # t_install_stage_rollback
for stop in range(1, 9):
with self.subTest(stages=stop):
k = PlatformKernel(":memory:")
try:
self._partial_bootstrap_then_rollback(k, stop)
self.assertTrue(
_all_bootstrap_empty(k),
f"partial rows survived rollback at stage {stop}",
)
self.assertFalse(k.is_installed())
finally:
k.close()
def test_no_partial_after_rollback(self) -> None: # t_no_partial_after_rollback
k = PlatformKernel(":memory:")
try:
code = self._seed_bootstrap_and_mark(k, dominance=DOMINANCE_TUPLES[:-1])
self.assertEqual(code, DOMINANCE_SET_MISMATCH)
self.assertTrue(_all_bootstrap_empty(k))
self.assertFalse(k.is_installed())
finally:
k.close()
# -- AC4 -------------------------------------------------------------- #
def test_dominance_missing(self) -> None: # t_dominance_missing(-)
k = PlatformKernel(":memory:")
try:
self.assertEqual(
self._seed_bootstrap_and_mark(k, dominance=DOMINANCE_TUPLES[:-1]),
DOMINANCE_SET_MISMATCH,
)
self.assertFalse(k.is_installed())
finally:
k.close()
def test_dominance_extra(self) -> None: # t_dominance_extra(-)
k = PlatformKernel(":memory:")
try:
extra = DOMINANCE_TUPLES + (("platform.bootstrap", "rogue.extra"),)
self.assertEqual(
self._seed_bootstrap_and_mark(k, dominance=extra),
DOMINANCE_SET_MISMATCH,
)
self.assertFalse(k.is_installed())
finally:
k.close()
def test_dominance_malformed(self) -> None: # t_dominance_malformed(-)
k = PlatformKernel(":memory:")
try:
malformed = DOMINANCE_TUPLES[:-1] + (("supervisor.root", "WRONG.subordinate"),)
self.assertEqual(
self._seed_bootstrap_and_mark(k, dominance=malformed),
DOMINANCE_SET_MISMATCH,
)
self.assertFalse(k.is_installed())
finally:
k.close()
# -- AC6 -------------------------------------------------------------- #
def test_principal_no_class(self) -> None: # t_principal_no_class(-)
with self.kernel.actor_context("op", "operator", "install"):
with self.assertRaises(sqlite3.IntegrityError):
self.kernel._conn.execute(
"INSERT INTO principals"
"(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
"VALUES ('x', 'operator', NULL, NULL, NULL, '2026-01-01T00:00:00Z')"
)
# -- AC7 -------------------------------------------------------------- #
def test_noninstaller_null_issuer(self) -> None: # t_nonobstaller_null_issuer(-)
self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
with self.kernel.actor_context("op", "operator", "normal"):
cur = self.kernel._conn.execute(
"INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
)
class_id = cur.lastrowid
with self.assertRaises(sqlite3.IntegrityError) as ctx:
self.kernel._conn.execute(
"INSERT INTO principals"
"(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
"VALUES ('rogue', 'operator', ?, NULL, NULL, '2026-01-01T00:00:00Z')",
(class_id,),
)
self.assertIn("INVALID_BOOTSTRAP_STATE", str(ctx.exception))
def test_installer_null_issuer_only_during_install(self) -> None:
self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
with self.kernel.actor_context("i2", "installer", "install"):
cur = self.kernel._conn.execute(
"INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
)
class_id = cur.lastrowid
with self.assertRaises(sqlite3.IntegrityError):
self.kernel._conn.execute(
"INSERT INTO principals"
"(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
"VALUES ('i2', 'installer', ?, NULL, NULL, '2026-01-01T00:00:00Z')",
(class_id,),
)
# -- AC8 -------------------------------------------------------------- #
def test_context_missing(self) -> None: # t_context_missing(-)
self.assertIsNone(self.kernel._ctx)
with self.assertRaises(sqlite3.IntegrityError) as ctx:
self.kernel._conn.execute(
"INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
)
self.assertIn("INVALID_ACTOR_CONTEXT", str(ctx.exception))
def test_context_stale(self) -> None: # t_context_stale(-)
with self.kernel.actor_context("op", "operator", "normal"):
self.kernel._ctx.expired = True
with self.assertRaises(sqlite3.IntegrityError) as ctx:
self.kernel._conn.execute(
"INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
)
self.assertIn("INVALID_ACTOR_CONTEXT", str(ctx.exception))
def test_context_epoch_shift(self) -> None: # t_context_epoch_shift(-)
with self.kernel.actor_context("op", "operator", "normal"):
self.kernel._ctx.live_epoch = self.kernel._ctx.bound_epoch + 99
with self.assertRaises(sqlite3.IntegrityError) as ctx:
self.kernel._conn.execute(
"INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
)
self.assertIn("INVALID_ACTOR_CONTEXT", str(ctx.exception))
def test_bad_actor_kind_or_mode_rejected(self) -> None:
for kind, mode in (("intruder", "normal"), ("operator", "sabotage")):
with self.subTest(kind=kind, mode=mode):
with self.kernel.actor_context("op", kind, mode):
with self.assertRaises(sqlite3.IntegrityError):
self.kernel._conn.execute(
"INSERT INTO principal_equivalence_classes(created_at) "
"VALUES ('2026-01-01T00:00:00Z')"
)
# -- AC9 -------------------------------------------------------------- #
def test_bootstrap_immutable_update(self) -> None: # t_bootstrap_immutable_{update}
self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
cases = [
("UPDATE install_state SET installed_at = 'x' WHERE id = 1", "IMMUTABLE_INSTALL_STATE"),
("UPDATE platform_bootstrap_seed SET created_at = 'x' WHERE seed_id = 1", "IMMUTABLE_SEED"),
("UPDATE authority_dominance SET subordinate = 'x' WHERE dominant = 'supervisor.root'", "IMMUTABLE_DOMINANCE"),
(f"UPDATE authoritative_issuers SET issuer_ref = 'x' WHERE issuer_ref = '{DISTINGUISHED_ISSUER_ID}'", "IMMUTABLE_ISSUER"),
(f"UPDATE principals SET actor_kind = 'operator' WHERE principal_id = '{INSTALLER}'", "IMMUTABLE_PRINCIPAL"),
]
for sql, tag in cases:
with self.subTest(sql=sql):
with self.kernel.actor_context("op", "operator", "normal"):
with self.assertRaises(sqlite3.IntegrityError) as ctx:
self.kernel._conn.execute(sql)
self.assertIn(tag, str(ctx.exception))
def test_bootstrap_immutable_delete(self) -> None: # t_bootstrap_immutable_{delete}
self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
cases = [
("DELETE FROM install_state WHERE id = 1", "IMMUTABLE_INSTALL_STATE"),
("DELETE FROM platform_bootstrap_seed WHERE seed_id = 1", "IMMUTABLE_SEED"),
("DELETE FROM authority_dominance", "IMMUTABLE_DOMINANCE"),
("DELETE FROM authoritative_issuers", "IMMUTABLE_ISSUER"),
(f"DELETE FROM principals WHERE principal_id = '{INSTALLER}'", "IMMUTABLE_PRINCIPAL"),
("DELETE FROM platform_bootstrap_grants", "IMMUTABLE_GRANT"),
]
for sql, tag in cases:
with self.subTest(sql=sql):
with self.kernel.actor_context("op", "operator", "normal"):
with self.assertRaises(sqlite3.IntegrityError) as ctx:
self.kernel._conn.execute(sql)
self.assertIn(tag, str(ctx.exception))
def test_grant_reactivation_rejected(self) -> None:
self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
self.kernel.register_principal(
"op1", "operator", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER
)
self.assertEqual(
self.kernel.grant_platform_bootstrap("op1", INSTALLER).code, INSTALLED
)
gid = self.kernel._conn.execute(
"SELECT grant_id FROM platform_bootstrap_grants WHERE grantee_principal_id = 'op1'"
).fetchone()[0]
self.assertEqual(
self.kernel.revoke_platform_bootstrap(gid, actor_principal=INSTALLER).code,
INSTALLED,
)
with self.kernel.actor_context("op", "operator", "normal"):
with self.assertRaises(sqlite3.IntegrityError) as ctx:
self.kernel._conn.execute(
"UPDATE platform_bootstrap_grants SET active = 1 WHERE grant_id = ?",
(gid,),
)
self.assertIn("IMMUTABLE_GRANT", str(ctx.exception))
# -- AC12 ------------------------------------------------------------- #
def test_raw_write_bypass(self) -> None: # t_raw_write_bypass(raw-bypass)
with tempfile.TemporaryDirectory() as tmp:
path = os.path.join(tmp, "p.sqlite3")
k = PlatformKernel(path)
self.assertEqual(k.install_platform(INSTALLER).code, INSTALLED)
k.close()
raw = sqlite3.connect(path)
raw.execute("PRAGMA foreign_keys = ON")
try:
with self.assertRaises(sqlite3.Error):
raw.execute(
"INSERT INTO audit_records(event, created_at) "
"VALUES ('forged', '2026-01-01T00:00:00Z')"
)
raw.commit()
with self.assertRaises(sqlite3.Error):
raw.execute("UPDATE install_state SET installed_at = 'x' WHERE id = 1")
raw.commit()
with self.assertRaises(sqlite3.Error):
raw.execute("DELETE FROM platform_bootstrap_grants")
raw.commit()
finally:
raw.close()
# -- AC13 ------------------------------------------------------------- #
def test_audit_created(self) -> None: # t_audit_created(+)
self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
self.kernel.register_principal(
"op1", "operator", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER
)
self.assertEqual(
self.kernel.grant_platform_bootstrap("op1", INSTALLER).code, INSTALLED
)
gid = self.kernel._conn.execute(
"SELECT grant_id FROM platform_bootstrap_grants WHERE grantee_principal_id = 'op1'"
).fetchone()[0]
self.assertEqual(
self.kernel.revoke_platform_bootstrap(gid, actor_principal=INSTALLER).code,
INSTALLED,
)
events = self.kernel.audit_events()
for evt in (
ap.EVT_PLATFORM_INSTALLED,
ap.EVT_GRANT_CREATED,
ap.EVT_GRANT_REVOKED,
ap.EVT_PRINCIPAL_REGISTERED,
):
self.assertIn(evt, events)
# -- AC14 ------------------------------------------------------------- #
def test_audit_immutable(self) -> None: # t_audit_immutable(raw-bypass)
self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
with self.kernel.actor_context("op", "operator", "normal"):
with self.assertRaises(sqlite3.IntegrityError) as up:
self.kernel._conn.execute("UPDATE audit_records SET event = 'x' WHERE audit_id = 1")
self.assertIn("IMMUTABLE_AUDIT", str(up.exception))
with self.assertRaises(sqlite3.IntegrityError) as dl:
self.kernel._conn.execute("DELETE FROM audit_records WHERE audit_id = 1")
self.assertIn("IMMUTABLE_AUDIT", str(dl.exception))
# -- meta ------------------------------------------------------------- #
def test_schema_meta(self) -> None:
rows = dict(self.kernel._conn.execute("SELECT key, value FROM arch01_meta").fetchall())
self.assertEqual(rows["schema_version"], str(ap.SCHEMA_VERSION))
self.assertIn("disabled by default", rows["architecture"])
def test_register_principal_creates_class_first(self) -> None:
self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
res = self.kernel.register_principal(
"svc1", "service", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER
)
self.assertEqual(res.code, INSTALLED)
row = self.kernel._conn.execute(
"SELECT current_class_id FROM principals WHERE principal_id = 'svc1'"
).fetchone()
self.assertIsNotNone(row[0])
# -- helpers ---------------------------------------------------------- #
def _partial_bootstrap_then_rollback(self, k: PlatformKernel, stop: int) -> None:
"""Execute the first ``stop`` bootstrap statements, then ROLLBACK."""
now = "2026-01-01T00:00:00Z"
k._conn.execute("BEGIN IMMEDIATE")
class_id = None
issuer_id = None
try:
with k.actor_context(INSTALLER, "installer", "install"):
c = k._conn
if stop >= 1:
class_id = c.execute(
"INSERT INTO principal_equivalence_classes(created_at) VALUES (?)", (now,)
).lastrowid
if stop >= 2:
c.execute(
"INSERT INTO principals(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
"VALUES (?, 'installer', ?, NULL, ?, ?)",
(INSTALLER, class_id, INSTALLER, now),
)
if stop >= 3:
issuer_id = c.execute(
"INSERT INTO authoritative_issuers(issuer_kind, issuer_ref, created_at) VALUES ('operator-key', ?, ?)",
(DISTINGUISHED_ISSUER_ID, now),
).lastrowid
if stop >= 4:
c.execute(
"UPDATE principals SET issuer_id = ? WHERE principal_id = ?",
(issuer_id, INSTALLER),
)
if stop >= 5:
c.executemany(
"INSERT INTO authority_dominance(dominant, subordinate) VALUES (?, ?)",
DOMINANCE_TUPLES,
)
if stop >= 6:
c.execute(
"INSERT INTO platform_bootstrap_seed(seed_id, installer_principal_id, created_at) VALUES (1, ?, ?)",
(INSTALLER, now),
)
if stop >= 7:
c.execute(
"INSERT INTO platform_bootstrap_grants(grantee_principal_id, granted_by, active, created_at) VALUES (?, NULL, 1, ?)",
(INSTALLER, now),
)
if stop >= 8:
c.execute("INSERT INTO platform_active_invariant(id, active_count) VALUES (1, 1)")
finally:
k._conn.execute("ROLLBACK")
def _seed_bootstrap_and_mark(self, k: PlatformKernel, dominance) -> str:
"""Seed a full bootstrap with a caller-supplied dominance set, then
attempt the marker insert. Returns the classified failure code (or
INSTALLED). Rolls back on failure so no partial rows remain."""
now = "2026-01-01T00:00:00Z"
k._conn.execute("BEGIN IMMEDIATE")
try:
with k.actor_context(INSTALLER, "installer", "install"):
c = k._conn
class_id = c.execute(
"INSERT INTO principal_equivalence_classes(created_at) VALUES (?)", (now,)
).lastrowid
c.execute(
"INSERT INTO principals(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
"VALUES (?, 'installer', ?, NULL, ?, ?)",
(INSTALLER, class_id, INSTALLER, now),
)
issuer_id = c.execute(
"INSERT INTO authoritative_issuers(issuer_kind, issuer_ref, created_at) VALUES ('operator-key', ?, ?)",
(DISTINGUISHED_ISSUER_ID, now),
).lastrowid
c.execute(
"UPDATE principals SET issuer_id = ? WHERE principal_id = ?",
(issuer_id, INSTALLER),
)
c.executemany(
"INSERT INTO authority_dominance(dominant, subordinate) VALUES (?, ?)",
dominance,
)
c.execute(
"INSERT INTO platform_bootstrap_seed(seed_id, installer_principal_id, created_at) VALUES (1, ?, ?)",
(INSTALLER, now),
)
c.execute(
"INSERT INTO platform_bootstrap_grants(grantee_principal_id, granted_by, active, created_at) VALUES (?, NULL, 1, ?)",
(INSTALLER, now),
)
c.execute("INSERT INTO platform_active_invariant(id, active_count) VALUES (1, 1)")
c.execute(
"INSERT INTO install_state(id, marker, installed_at) VALUES (1, 'installed', ?)",
(now,),
)
k._conn.execute("COMMIT")
return INSTALLED
except sqlite3.Error as exc:
k._safe_rollback()
return PlatformKernel._classify(exc)
class Arch01ConcurrencyTest(unittest.TestCase):
"""Concurrency invariants require file-backed DBs and independent connections."""
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.path = os.path.join(self._tmp.name, "p.sqlite3")
def tearDown(self) -> None:
self._tmp.cleanup()
# -- AC10 ------------------------------------------------------------- #
def test_concurrent_install(self) -> None: # t_concurrent_install(concurrency)
k1 = PlatformKernel(self.path, busy_timeout_ms=0)
k2 = PlatformKernel(self.path, busy_timeout_ms=0)
barrier = threading.Barrier(2)
results = {}
def _install(name, kernel):
barrier.wait()
results[name] = kernel.install_platform(INSTALLER).code
try:
with ThreadPoolExecutor(max_workers=2) as ex:
f1 = ex.submit(_install, "a", k1)
f2 = ex.submit(_install, "b", k2)
f1.result()
f2.result()
codes = sorted(results.values())
self.assertEqual(codes.count(INSTALLED), 1, f"exactly one install expected: {results}")
other = [c for c in results.values() if c != INSTALLED][0]
self.assertIn(other, (ALREADY_INSTALLED, CONCURRENT_INSTALLATION_LOST))
self.assertTrue(k1.is_installed())
self.assertEqual(_count(k1, "install_state"), 1)
self.assertEqual(_count(k1, "principals"), 1)
finally:
k1.close()
k2.close()
# -- AC11 ------------------------------------------------------------- #
def test_concurrent_last_grant_revoke(self) -> None: # t_concurrent_last_grant_revoke
setup = PlatformKernel(self.path)
self.assertEqual(setup.install_platform(INSTALLER).code, INSTALLED)
setup.register_principal("op1", "operator", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER)
self.assertEqual(setup.grant_platform_bootstrap("op1", INSTALLER).code, INSTALLED)
self.assertEqual(setup.active_grant_count(), 2)
gids = [
r[0]
for r in setup._conn.execute(
"SELECT grant_id FROM platform_bootstrap_grants WHERE active = 1 ORDER BY grant_id"
).fetchall()
]
setup.close()
self.assertEqual(len(gids), 2)
k1 = PlatformKernel(self.path, busy_timeout_ms=3000)
k2 = PlatformKernel(self.path, busy_timeout_ms=3000)
barrier = threading.Barrier(2)
results = {}
def _revoke(name, kernel, gid):
barrier.wait()
results[name] = kernel.revoke_platform_bootstrap(gid, actor_principal=INSTALLER).code
try:
with ThreadPoolExecutor(max_workers=2) as ex:
f1 = ex.submit(_revoke, "a", k1, gids[0])
f2 = ex.submit(_revoke, "b", k2, gids[1])
f1.result()
f2.result()
codes = list(results.values())
self.assertEqual(codes.count(INSTALLED), 1, f"exactly one revoke should win: {results}")
self.assertEqual(codes.count(AUTHORIZATION_DENIED), 1, f"one revoke must be denied: {results}")
self.assertEqual(k1.active_grant_count(), 1)
self.assertEqual(_count_where(k1, "platform_bootstrap_grants", "active = 1"), 1)
finally:
k1.close()
k2.close()
def test_revoke_final_grant_denied(self) -> None:
k = PlatformKernel(self.path)
try:
self.assertEqual(k.install_platform(INSTALLER).code, INSTALLED)
gid = k._conn.execute(
"SELECT grant_id FROM platform_bootstrap_grants WHERE active = 1"
).fetchone()[0]
res = k.revoke_platform_bootstrap(gid, actor_principal=INSTALLER)
self.assertEqual(res.code, AUTHORIZATION_DENIED)
self.assertEqual(k.active_grant_count(), 1)
self.assertEqual(_count_where(k, "platform_bootstrap_grants", "active = 1"), 1)
finally:
k.close()
if __name__ == "__main__":
unittest.main()
-581
View File
@@ -1,581 +0,0 @@
"""Authoritative controller cross-role generic queue allocation (#840)."""
from __future__ import annotations
import os
import tempfile
import unittest
from unittest.mock import patch
from allocator_service import (
ALLOCATION_MODE_CROSS_ROLE,
ALLOCATION_MODE_ROLE_SCOPED,
OUTCOME_NO_SAFE,
OUTCOME_PREVIEW,
OUTCOME_WAIT,
ROLE_AUTHOR,
ROLE_CONTROLLER,
ROLE_MERGER,
ROLE_RECONCILER,
ROLE_REVIEWER,
WorkCandidate,
allocate_next_work,
build_selection_dict,
classify_skip,
required_namespace_for_role,
required_profile_for_role,
resolve_allocation_mode,
selected_action_for_candidate,
)
from control_plane_db import ControlPlaneDB
import role_session_router
from role_session_router import (
ROUTE_ALLOWED,
ROUTE_AMBIGUOUS,
ROUTE_WRONG_ROLE,
route_task_session,
)
import namespace_workspace_binding as nwb
import task_capability_map
class CrossRoleAllocationModeTest(unittest.TestCase):
def test_controller_defaults_to_cross_role(self) -> None:
self.assertEqual(
resolve_allocation_mode(ROLE_CONTROLLER),
ALLOCATION_MODE_CROSS_ROLE,
)
def test_worker_defaults_to_role_scoped(self) -> None:
for role in (ROLE_AUTHOR, ROLE_REVIEWER, ROLE_MERGER, ROLE_RECONCILER):
self.assertEqual(
resolve_allocation_mode(role),
ALLOCATION_MODE_ROLE_SCOPED,
)
def test_explicit_modes(self) -> None:
self.assertEqual(
resolve_allocation_mode(ROLE_CONTROLLER, "role_scoped"),
ALLOCATION_MODE_ROLE_SCOPED,
)
self.assertEqual(
resolve_allocation_mode(ROLE_AUTHOR, "cross_role"),
ALLOCATION_MODE_CROSS_ROLE,
)
class CrossRoleSelectionPayloadTest(unittest.TestCase):
def test_selection_contains_required_fields(self) -> None:
c = WorkCandidate(
kind="issue",
number=840,
labels=("status:ready",),
title="cross-role",
priority=20,
)
sel = build_selection_dict(
c,
active_role=ROLE_CONTROLLER,
required_role=ROLE_AUTHOR,
profile_name="prgs-controller",
allocation_mode=ALLOCATION_MODE_CROSS_ROLE,
)
self.assertEqual(sel["number"], 840)
self.assertEqual(sel["kind"], "issue")
self.assertEqual(sel["required_role"], ROLE_AUTHOR)
self.assertEqual(sel["selected_action"], "implement")
self.assertEqual(sel["action"], "implement")
self.assertEqual(sel["required_profile"], "prgs-author")
self.assertEqual(sel["required_namespace"], "gitea-author")
self.assertEqual(sel["pinned"]["number"], 840)
self.assertIsNone(sel["pinned"]["head_sha"])
def test_profile_prefix_preserved(self) -> None:
self.assertEqual(
required_profile_for_role(ROLE_REVIEWER, profile_name="dadeschools-controller"),
"dadeschools-reviewer",
)
self.assertEqual(
required_namespace_for_role(ROLE_MERGER),
"gitea-merger",
)
def test_selected_actions_per_role(self) -> None:
issue = WorkCandidate(kind="issue", number=1, labels=("status:ready",))
pr_review = WorkCandidate(kind="pr", number=2, head_sha="a" * 40)
pr_rc = WorkCandidate(
kind="pr",
number=3,
head_sha="b" * 40,
request_changes_current_head=True,
)
pr_merge = WorkCandidate(
kind="pr",
number=4,
head_sha="c" * 40,
approval_on_current_head=True,
mergeable=True,
)
pr_recon = WorkCandidate(
kind="pr",
number=5,
head_sha="d" * 40,
approval_contaminated=True,
)
self.assertEqual(selected_action_for_candidate(issue, ROLE_AUTHOR), "implement")
self.assertEqual(
selected_action_for_candidate(pr_rc, ROLE_AUTHOR),
"address_pr_change_requests",
)
self.assertEqual(
selected_action_for_candidate(pr_review, ROLE_REVIEWER), "review"
)
self.assertEqual(selected_action_for_candidate(pr_merge, ROLE_MERGER), "merge")
self.assertEqual(
selected_action_for_candidate(pr_recon, ROLE_RECONCILER),
"reconcile_contaminated_approval",
)
class CrossRoleAllocateServiceTest(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
def tearDown(self) -> None:
self._tmp.cleanup()
def _alloc(self, **kwargs):
defaults = dict(
db=self.db,
session_id="ctrl-session",
role=ROLE_CONTROLLER,
remote="prgs",
org="org",
repo="repo",
candidates=[],
apply=False,
profile_name="prgs-controller",
username="controller-bot",
controller_instance_id="ctrl-1",
)
defaults.update(kwargs)
return allocate_next_work(**defaults)
def test_eligible_author_work(self) -> None:
cands = [
WorkCandidate(
kind="issue",
number=100,
labels=("status:ready",),
title="author work",
priority=20,
),
]
res = self._alloc(candidates=cands)
self.assertTrue(res["success"])
self.assertEqual(res["outcome"], OUTCOME_PREVIEW)
self.assertEqual(res["allocation_mode"], ALLOCATION_MODE_CROSS_ROLE)
self.assertIsNotNone(res["selected"])
self.assertEqual(res["selected"]["number"], 100)
self.assertEqual(res["required_role"], ROLE_AUTHOR)
self.assertEqual(res["selected_action"], "implement")
self.assertEqual(res["required_profile"], "prgs-author")
self.assertEqual(res["required_namespace"], "gitea-author")
self.assertIn("allocate", res["controller_allowed_actions"])
self.assertIn("merge", res["controller_forbidden_actions"])
self.assertFalse(res["allocation_evidence"]["lease_created"])
def test_eligible_reviewer_work(self) -> None:
cands = [
WorkCandidate(
kind="pr",
number=200,
head_sha="e" * 40,
title="needs review",
priority=30,
),
]
res = self._alloc(candidates=cands)
self.assertEqual(res["selected"]["number"], 200)
self.assertEqual(res["required_role"], ROLE_REVIEWER)
self.assertEqual(res["selected_action"], "review")
self.assertEqual(res["required_profile"], "prgs-reviewer")
self.assertEqual(res["selected"]["pinned"]["head_sha"], "e" * 40)
def test_eligible_merger_work(self) -> None:
cands = [
WorkCandidate(
kind="pr",
number=300,
head_sha="f" * 40,
approval_on_current_head=True,
mergeable=True,
priority=40,
),
]
res = self._alloc(candidates=cands)
self.assertEqual(res["selected"]["number"], 300)
self.assertEqual(res["required_role"], ROLE_MERGER)
self.assertEqual(res["selected_action"], "merge")
def test_eligible_reconciler_work(self) -> None:
cands = [
WorkCandidate(
kind="pr",
number=400,
head_sha="1" * 40,
approval_contaminated=True,
priority=50,
),
]
res = self._alloc(candidates=cands)
self.assertEqual(res["selected"]["number"], 400)
self.assertEqual(res["required_role"], ROLE_RECONCILER)
self.assertIn("reconcile", res["selected_action"])
def test_no_eligible_work(self) -> None:
cands = [
WorkCandidate(
kind="issue",
number=10,
labels=("status:blocked",),
blocked=True,
priority=99,
),
WorkCandidate(
kind="issue",
number=11,
labels=("status:ready",),
dependency_unmet=True,
dependency_reason="blocked by #10",
priority=98,
),
]
res = self._alloc(candidates=cands)
self.assertTrue(res["success"])
self.assertEqual(res["outcome"], OUTCOME_NO_SAFE)
self.assertIsNone(res["selected"])
self.assertEqual(res["allocation_mode"], ALLOCATION_MODE_CROSS_ROLE)
def test_leased_work_skipped(self) -> None:
cands = [
WorkCandidate(
kind="issue",
number=50,
labels=("status:ready",),
priority=20,
),
WorkCandidate(
kind="issue",
number=51,
labels=("status:ready",),
priority=10,
),
]
# Seed a foreign lease on issue 50 via assign_and_lease under another session.
other = allocate_next_work(
self.db,
session_id="other-worker",
role=ROLE_AUTHOR,
remote="prgs",
org="org",
repo="repo",
candidates=cands[:1],
apply=True,
profile_name="prgs-author",
controller_instance_id="other-ctrl",
)
self.assertEqual(other["outcome"], "assigned_work")
res = self._alloc(candidates=cands)
self.assertIsNotNone(res["selected"])
self.assertEqual(res["selected"]["number"], 51)
self.assertTrue(any(s["number"] == 50 for s in res["skipped"]))
self.assertTrue(res["claims_excluded"])
def test_dependencies_skipped(self) -> None:
cands = [
WorkCandidate(
kind="issue",
number=1,
labels=("status:ready",),
priority=99,
dependency_unmet=True,
dependency_reason="needs #2",
),
WorkCandidate(
kind="issue",
number=2,
labels=("status:ready",),
priority=1,
),
]
res = self._alloc(candidates=cands)
self.assertEqual(res["selected"]["number"], 2)
skipped = {s["number"]: s["reason"] for s in res["skipped"]}
self.assertIn(1, skipped)
self.assertIn("needs #2", skipped[1])
def test_pagination_limit_only_truncates_skip_report(self) -> None:
"""Ranking uses full inventory; reporting limit is MCP-layer only.
Service ranks all candidates; prove higher-priority eligible item
wins even when many skipped precede it.
"""
cands = []
for n in range(1, 30):
cands.append(
WorkCandidate(
kind="issue",
number=n,
labels=("status:ready",),
priority=100 - n,
dependency_unmet=True,
dependency_reason=f"dep {n}",
)
)
cands.append(
WorkCandidate(
kind="issue",
number=999,
labels=("status:ready",),
priority=1,
)
)
res = self._alloc(candidates=cands)
self.assertEqual(res["selected"]["number"], 999)
self.assertGreaterEqual(len(res["skipped"]), 29)
def test_role_scoped_controller_legacy_still_restricts(self) -> None:
"""role_scoped controller only takes reconciler-needed items."""
cands = [
WorkCandidate(
kind="issue",
number=1,
labels=("status:ready",),
priority=50,
),
WorkCandidate(
kind="pr",
number=2,
head_sha="a" * 40,
approval_contaminated=True,
priority=1,
),
]
res = self._alloc(
candidates=cands,
allocation_mode=ALLOCATION_MODE_ROLE_SCOPED,
)
self.assertEqual(res["allocation_mode"], ALLOCATION_MODE_ROLE_SCOPED)
self.assertEqual(res["selected"]["number"], 2)
self.assertEqual(res["required_role"], ROLE_RECONCILER)
def test_cross_role_prefers_highest_priority_across_roles(self) -> None:
cands = [
WorkCandidate(
kind="issue",
number=10,
labels=("status:ready",),
priority=10,
),
WorkCandidate(
kind="pr",
number=20,
head_sha="b" * 40,
priority=50,
),
WorkCandidate(
kind="pr",
number=30,
head_sha="c" * 40,
approval_on_current_head=True,
mergeable=True,
priority=20,
),
]
res = self._alloc(candidates=cands)
# PR #20 highest priority → reviewer
self.assertEqual(res["selected"]["number"], 20)
self.assertEqual(res["required_role"], ROLE_REVIEWER)
def test_apply_creates_lease_evidence_for_required_role(self) -> None:
cands = [
WorkCandidate(
kind="issue",
number=777,
labels=("status:ready",),
priority=20,
),
]
res = self._alloc(candidates=cands, apply=True)
self.assertEqual(res["outcome"], "assigned_work")
self.assertTrue(res["allocation_evidence"]["lease_created"])
self.assertEqual(res["allocation_evidence"]["lease_role"], ROLE_AUTHOR)
proof = res["lease_proof"]
self.assertIsNotNone(proof["lease_id"])
self.assertEqual(proof["lease_role"], ROLE_AUTHOR)
self.assertIn("implement", proof["allowed_actions"])
# Controller isolation: controller still forbids merge/push/create_pr
self.assertIn("merge", res["controller_forbidden_actions"])
self.assertIn("push", res["controller_forbidden_actions"])
def test_metadata_consistency_role_is_controller(self) -> None:
cands = [
WorkCandidate(
kind="issue",
number=1,
labels=("status:ready",),
),
]
res = self._alloc(candidates=cands)
self.assertEqual(res["role"], ROLE_CONTROLLER)
self.assertEqual(res["routing_role"], ROLE_CONTROLLER)
self.assertEqual(res["required_role"], ROLE_AUTHOR)
class ProcessWorkQueueRouterTest(unittest.TestCase):
def tearDown(self) -> None:
role_session_router.clear_route_state()
def test_process_work_queue_allowed_for_controller(self) -> None:
res = route_task_session(
"process_work_queue",
active_profile="prgs-controller",
active_role_kind="controller",
allowed_in_current_session=True,
)
self.assertEqual(res["route_result"], ROUTE_ALLOWED)
self.assertEqual(res["required_role"], "controller")
self.assertTrue(res["downstream_allowed"])
def test_process_work_queue_hyphen_alias(self) -> None:
res = route_task_session(
"process-work-queue",
active_profile="prgs-controller",
active_role_kind="controller",
allowed_in_current_session=True,
)
self.assertEqual(res["route_result"], ROUTE_ALLOWED)
def test_process_work_queue_wrong_role_for_author(self) -> None:
res = route_task_session(
"process_work_queue",
active_profile="prgs-author",
active_role_kind="author",
allowed_in_current_session=False,
)
self.assertEqual(res["route_result"], ROUTE_WRONG_ROLE)
self.assertEqual(res["required_role"], "controller")
self.assertFalse(res["downstream_allowed"])
def test_unknown_still_ambiguous(self) -> None:
res = route_task_session(
"not_a_real_task",
active_profile="prgs-controller",
active_role_kind="controller",
allowed_in_current_session=False,
)
self.assertEqual(res["route_result"], ROUTE_AMBIGUOUS)
def test_capability_map_process_work_queue_is_controller(self) -> None:
self.assertEqual(
task_capability_map.required_role("process_work_queue"),
"controller",
)
self.assertEqual(
task_capability_map.required_permission("process_work_queue"),
"gitea.read",
)
class ControllerRoleMetadataTest(unittest.TestCase):
def test_normalize_role_kind_controller(self) -> None:
self.assertEqual(
nwb.normalize_role_kind("controller"),
"controller",
)
self.assertEqual(
nwb.normalize_role_kind("author", profile_name="prgs-controller"),
"controller",
)
self.assertEqual(
nwb.normalize_role_kind("reconciler", profile_name="prgs-controller"),
"controller",
)
def test_profile_role_kind_prefers_declared_controller(self) -> None:
# Import from worktree package path via sys.path already set by pytest.
import gitea_mcp_server as mcp
profile = {
"profile_name": "prgs-controller",
"role": "controller",
"allowed_operations": [
"gitea.read",
"gitea.issue.comment",
"gitea.pr.close",
],
"forbidden_operations": [
"gitea.pr.approve",
"gitea.pr.merge",
"gitea.pr.create",
"gitea.branch.push",
],
}
# Declared role wins even if permissions look reconciler-like.
self.assertEqual(mcp._profile_role_kind(profile), "controller")
# Name-based fallback.
profile_no_role = dict(profile)
profile_no_role["role"] = None
profile_no_role["role_kind"] = None
self.assertEqual(mcp._profile_role_kind(profile_no_role), "controller")
def test_permission_inference_without_controller_name_stays_reconciler(self) -> None:
import gitea_mcp_server as mcp
# Pure permission inference still may return reconciler when no controller
# declaration exists — that is intentional for reconciler profiles.
role = mcp._role_kind(
["gitea.read", "gitea.pr.close", "gitea.issue.comment"],
["gitea.pr.approve", "gitea.pr.merge", "gitea.pr.create", "gitea.branch.push"],
)
self.assertEqual(role, "reconciler")
class DashboardRemainsExplanatoryTest(unittest.TestCase):
def test_dashboard_prompt_points_at_allocator_not_self_select(self) -> None:
import workflow_dashboard as wd
self.assertIn("gitea_allocate_next_work", wd.PROMPT_CONTROLLER)
self.assertIn("process_work_queue", wd.PROMPT_CONTROLLER)
self.assertIn("never replaces allocator", wd.PROMPT_CONTROLLER.lower())
self.assertNotIn("self-select", wd.PROMPT_CONTROLLER.lower())
class ClassifySkipCrossRoleTest(unittest.TestCase):
def test_controller_cross_role_accepts_author_issue(self) -> None:
c = WorkCandidate(kind="issue", number=1, labels=("status:ready",))
self.assertIsNone(
classify_skip(
c,
role=ROLE_CONTROLLER,
terminal_pr=None,
allocation_mode=ALLOCATION_MODE_CROSS_ROLE,
)
)
def test_legacy_controller_skips_author_issue(self) -> None:
c = WorkCandidate(kind="issue", number=1, labels=("status:ready",))
reason = classify_skip(
c,
role=ROLE_CONTROLLER,
terminal_pr=None,
allocation_mode=ALLOCATION_MODE_ROLE_SCOPED,
)
self.assertIsNotNone(reason)
self.assertIn("does not require controller", reason or "")
if __name__ == "__main__":
unittest.main()
-189
View File
@@ -1,189 +0,0 @@
"""Integration tests for autonomous canonical handoffs and dependency-aware task orchestration (#628).
Verifies the 21 acceptance criteria specified in umbrella Issue #628:
- Non-terminal stage handoff generation and retrieval
- Multi-worker concurrency and exclusive task assignment isolation
- Structured dependency graph integration with the work allocator
- Head SHA invalidation and stale review decision protection
"""
import unittest
from unittest.mock import MagicMock, patch
import os
import json
import tempfile
from canonical_thread_handoff import (
format_cth_body,
parse_cth_comment,
assess_cth_comment,
)
import dependency_graph
from control_plane_db import ControlPlaneDB
from allocator_service import (
WorkCandidate,
classify_skip,
ROLE_AUTHOR,
ROLE_REVIEWER,
ROLE_MERGER,
ROLE_RECONCILER,
OWNERSHIP_OWN,
OWNERSHIP_FOREIGN,
)
class TestIssue628Orchestration(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
self.db = ControlPlaneDB(self.db_path)
def tearDown(self):
self._tmp.cleanup()
def test_canonical_handoff_serialization_and_retrieval(self):
"""AC1 & AC2: Every non-terminal stage stores and retrieves a valid canonical handoff."""
handoff = format_cth_body(
cth_type="Author Handoff",
status="completed",
next_owner="reviewer",
current_blocker="none",
decision="Implementation complete, tests passing",
proof="pytest tests/test_issue_628_orchestration.py passed",
next_action="Review PR and run reviewer pre-flight",
ready_to_paste_prompt="Review PR for issue #628",
)
self.assertIn("CTH: Author Handoff", handoff)
parsed = parse_cth_comment(handoff)
self.assertIsNotNone(parsed)
self.assertEqual(parsed["cth_type"], "Author Handoff")
assessment = assess_cth_comment(handoff)
self.assertFalse(assessment["block"])
def test_exclusive_task_unit_single_owner(self):
"""AC5 & AC6: Concurrency isolation ensures an exclusive task unit has only one active owner."""
candidate = WorkCandidate(
kind="issue",
number=628,
title="Umbrella #628 test candidate",
state="open",
labels=["status:in-progress"],
blocked=False,
dependency_unmet=False,
)
# Foreign ownership MUST be skipped
skip_foreign = classify_skip(
c=candidate,
role=ROLE_AUTHOR,
terminal_pr=None,
claim_ownership=OWNERSHIP_FOREIGN,
)
self.assertIsNotNone(skip_foreign)
self.assertIn("active lease", skip_foreign)
# Own/Self claim remains selectable for session resumption
skip_self = classify_skip(
c=candidate,
role=ROLE_AUTHOR,
terminal_pr=None,
claim_ownership=OWNERSHIP_OWN,
)
self.assertIsNone(skip_self)
def test_durable_dependency_graph_blocking(self):
"""AC8, AC9, AC10: Durable dependency edges exclude blocked tasks from assignment."""
# Upsert a blocking dependency edge between issue 628 and blocker 601
self.db.upsert_dependency_edge(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
source_kind="issue",
source_number=628,
target_kind="issue",
target_number=601,
edge_type=dependency_graph.EDGE_ISSUE_BLOCKED_BY_ISSUE,
state=dependency_graph.STATE_UNMET,
blocking_condition="Target issue #601 is not closed",
completion_condition="Target issue #601 is closed",
evidence={"source": "unit_test"},
)
edges = self.db.list_dependency_edges(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
source_kind="issue",
source_number=628,
)
self.assertEqual(len(edges), 1)
self.assertEqual(edges[0]["state"], "unmet")
self.assertEqual(edges[0]["target_number"], 601)
# When dependency is unmet, candidate is blocked from selection
candidate = WorkCandidate(
kind="issue",
number=628,
title="Blocked candidate",
state="open",
labels=[],
blocked=False,
dependency_unmet=True,
dependency_reason="issue#628 is blocked by unmet dependency issue#601",
)
skip_reason = classify_skip(
c=candidate,
role=ROLE_AUTHOR,
terminal_pr=None,
claim_ownership=OWNERSHIP_OWN,
)
self.assertIsNotNone(skip_reason)
self.assertIn("issue#601", skip_reason)
def test_dependency_completion_reevaluation(self):
"""AC11: Dependency completion updates edge state to MET."""
self.db.upsert_dependency_edge(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
source_kind="issue",
source_number=628,
target_kind="issue",
target_number=601,
edge_type=dependency_graph.EDGE_ISSUE_BLOCKED_BY_ISSUE,
state=dependency_graph.STATE_UNMET,
blocking_condition="Target issue #601 is open",
completion_condition="Target issue #601 is closed",
evidence={"source": "unit_test"},
)
# Mark edge as met upon target issue closure
self.db.upsert_dependency_edge(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
source_kind="issue",
source_number=628,
target_kind="issue",
target_number=601,
edge_type=dependency_graph.EDGE_ISSUE_BLOCKED_BY_ISSUE,
state=dependency_graph.STATE_MET,
blocking_condition="Target issue #601 is open",
completion_condition="Target issue #601 is closed",
evidence={"source": "target_closed_event"},
)
edges = self.db.list_dependency_edges(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
source_kind="issue",
source_number=628,
)
self.assertEqual(len(edges), 1)
self.assertEqual(edges[0]["state"], "met")
if __name__ == "__main__":
unittest.main()
-703
View File
@@ -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()
+468
View File
@@ -0,0 +1,468 @@
"""Tests for the unified web-console inventory API (#636).
Covers the four cases the issue names empty, populated, partial failure, and
the no-false-unowned invariant plus redaction, collision detection, the
resource-split routes, and read-only guarantees against a real control-plane
database and real durable lock files.
"""
import json
import os
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
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 import inventory
def _iso(dt: datetime) -> str:
return dt.astimezone(timezone.utc).isoformat()
def _write_lock(lock_dir: str, name: str, payload: dict) -> str:
path = os.path.join(lock_dir, name)
with open(path, "w", encoding="utf-8") as handle:
json.dump(payload, handle)
return path
def _live_lock_payload(
*,
issue_number: int,
branch: str,
worktree_path: str,
pid: int,
username: str = "jcwalker3",
profile: str = "prgs-author",
) -> dict:
now = datetime.now(timezone.utc)
future = now + timedelta(hours=2)
return {
"branch_name": branch,
"issue_number": issue_number,
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"remote": "prgs",
"pid": pid,
"session_pid": pid,
"lock_generation": 1,
"worktree_path": worktree_path,
"claimant": {"username": username, "profile": profile},
"work_lease": {
"branch": branch,
"issue_number": issue_number,
"operation_type": "author_issue_work",
"created_at": _iso(now),
"expires_at": _iso(future),
"last_heartbeat_at": _iso(now),
"claimant": {"username": username, "profile": profile},
},
}
class _FixtureMixin(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.tmp = self._tmp.name
self.lock_dir = os.path.join(self.tmp, "locks")
os.makedirs(self.lock_dir, mode=0o700)
self.db_path = os.path.join(self.tmp, "control_plane.db")
self.addCleanup(self._tmp.cleanup)
def _seed_db(self) -> control_plane_db.ControlPlaneDB:
db = control_plane_db.ControlPlaneDB(self.db_path)
db.upsert_session(
session_id="prgs-author-1",
role="author",
profile="prgs-author",
namespace="gitea-author",
pid=os.getpid(),
)
db.upsert_work_item(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
kind="issue",
number=636,
)
db.assign_and_lease(
session_id="prgs-author-1",
role="author",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
kind="issue",
number=636,
)
return db
class TestRedaction(unittest.TestCase):
def test_redact_path_collapses_home(self):
home = os.path.expanduser("~")
self.assertEqual(
inventory.redact_path(f"{home}/Development/Gitea-Tools"),
"~/Development/Gitea-Tools",
)
def test_redact_url_strips_userinfo_and_query(self):
self.assertEqual(
inventory.redact_url("https://user:[email protected]/api?token=abc"),
"https://gitea.prgs.cc/api",
)
def test_scrub_drops_credential_keys(self):
scrubbed = inventory.scrub(
{"token": "abc123", "api_key": "k", "profile": "prgs-author"}
)
self.assertEqual(scrubbed["token"], "[redacted]")
self.assertEqual(scrubbed["api_key"], "[redacted]")
self.assertEqual(scrubbed["profile"], "prgs-author")
def test_scrub_is_recursive_and_never_raises(self):
class Weird:
def __repr__(self) -> str:
return "weird-obj"
out = inventory.scrub({"nested": [{"password": "p", "obj": Weird()}]})
self.assertEqual(out["nested"][0]["password"], "[redacted]")
self.assertEqual(out["nested"][0]["obj"], "weird-obj")
class TestEmptyInventory(_FixtureMixin):
def test_empty_db_and_locks_degrade_without_raising(self):
# No DB file, no locks: sessions/leases unavailable, locks ok+empty.
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
load_hygiene=lambda: _StubHygiene(entries=()),
)
sessions = snap.section("sessions")
leases = snap.section("leases")
locks = snap.section("locks")
self.assertEqual(sessions.status, inventory.STATUS_UNAVAILABLE)
self.assertEqual(leases.status, inventory.STATUS_UNAVAILABLE)
self.assertEqual(locks.status, inventory.STATUS_OK)
self.assertEqual(len(locks.items), 0)
# Ownership authority is incomplete because the DB is missing.
self.assertFalse(snap.ownership_authority_complete)
self.assertEqual(snap.collisions, ())
def test_empty_db_present_but_unpopulated(self):
control_plane_db.ControlPlaneDB(self.db_path) # creates schema, no rows
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
load_hygiene=lambda: _StubHygiene(entries=()),
)
self.assertEqual(snap.section("sessions").status, inventory.STATUS_OK)
self.assertEqual(len(snap.section("sessions").items), 0)
self.assertEqual(snap.section("leases").status, inventory.STATUS_OK)
self.assertTrue(snap.ownership_authority_complete)
class TestPopulatedInventory(_FixtureMixin):
def test_sections_populated_and_correlated(self):
self._seed_db()
wt = f"{self.tmp}/branches/issue-636-inventory-api"
_write_lock(
self.lock_dir,
"prgs-Scaled-Tech-Consulting-Gitea-Tools-636.json",
_live_lock_payload(
issue_number=636,
branch="feat/issue-636-inventory-api",
worktree_path=wt,
pid=os.getpid(),
),
)
hygiene = _StubHygiene(
entries=(
_StubEntry(
rel_path="branches/issue-636-inventory-api",
branch="feat/issue-636-inventory-api",
classification="active-issue",
),
)
)
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
load_hygiene=lambda: hygiene,
)
self.assertTrue(snap.ownership_authority_complete)
self.assertEqual(len(snap.section("sessions").items), 1)
self.assertEqual(len(snap.section("leases").items), 1)
self.assertEqual(len(snap.section("locks").items), 1)
self.assertEqual(len(snap.section("worktrees").items), 1)
# The lease, lock, and worktree for #636 correlate onto one row.
row = next(r for r in snap.correlations if r["issue_number"] == 636)
self.assertEqual(row["branch"], "feat/issue-636-inventory-api")
self.assertTrue(row["lock_live"])
self.assertEqual(row["worktree_classification"], "active-issue")
self.assertEqual(len(row["lease_ids"]), 1)
# No collision: live lock, live pid, matching worktree.
self.assertEqual(snap.collisions, ())
def test_serialized_payload_declares_field_authority(self):
self._seed_db()
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
load_hygiene=lambda: _StubHygiene(entries=()),
)
payload = inventory.snapshot_to_dict(snap)
self.assertEqual(payload["api_version"], "v1")
self.assertEqual(payload["schema_version"], 1)
self.assertEqual(payload["field_authority"]["sessions"], "control_plane_db")
self.assertEqual(payload["field_authority"]["locks"], "filesystem")
self.assertIn("sessions", payload["sections"])
class TestPartialFailure(_FixtureMixin):
def test_worktree_scan_failure_degrades_only_that_section(self):
self._seed_db()
def _boom():
raise RuntimeError("git worktree list exploded")
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
load_hygiene=_boom,
)
self.assertEqual(
snap.section("worktrees").status, inventory.STATUS_UNAVAILABLE
)
self.assertIn("exploded", snap.section("worktrees").reason)
# DB-backed sections still healthy.
self.assertEqual(snap.section("sessions").status, inventory.STATUS_OK)
self.assertIn("worktrees", snap.degraded_sections)
def test_degraded_ownership_suppresses_unowned_claim(self):
# DB absent → sessions/leases unavailable → ownership incomplete even
# though a lock exists and could look "unclaimed" by the DB alone.
_write_lock(
self.lock_dir,
"prgs-Scaled-Tech-Consulting-Gitea-Tools-636.json",
_live_lock_payload(
issue_number=636,
branch="feat/issue-636-inventory-api",
worktree_path=f"{self.tmp}/wt",
pid=os.getpid(),
),
)
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
load_hygiene=lambda: _StubHygiene(entries=()),
)
self.assertFalse(snap.ownership_authority_complete)
payload = inventory.snapshot_to_dict(snap)
self.assertIn("may be treated as unowned", payload["ownership_note"])
class TestCollisionDetection(_FixtureMixin):
def test_live_lock_dead_owner_flagged(self):
_write_lock(
self.lock_dir,
"prgs-Scaled-Tech-Consulting-Gitea-Tools-700.json",
_live_lock_payload(
issue_number=700,
branch="feat/issue-700-x",
worktree_path=f"{self.tmp}/wt700",
pid=999_999_999, # not a running pid
),
)
control_plane_db.ControlPlaneDB(self.db_path) # empty but present
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
load_hygiene=lambda: _StubHygiene(entries=()),
)
kinds = {c.kind for c in snap.collisions}
self.assertIn("live-lock-dead-owner", kinds)
# Also lock-without-worktree, since no worktree carries the branch.
self.assertIn("lock-without-worktree", kinds)
def test_duplicate_live_lock_on_same_branch(self):
for issue in (800, 801):
_write_lock(
self.lock_dir,
f"prgs-Scaled-Tech-Consulting-Gitea-Tools-{issue}.json",
_live_lock_payload(
issue_number=issue,
branch="feat/issue-800-shared",
worktree_path=f"{self.tmp}/wt{issue}",
pid=os.getpid(),
),
)
control_plane_db.ControlPlaneDB(self.db_path)
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
load_hygiene=lambda: _StubHygiene(entries=()),
)
self.assertIn(
"duplicate-live-lock", {c.kind for c in snap.collisions}
)
def test_no_collision_when_sections_degraded(self):
# locks ok but worktrees unavailable → lock-without-worktree must NOT
# be asserted (a missing scan is not a missing worktree).
_write_lock(
self.lock_dir,
"prgs-Scaled-Tech-Consulting-Gitea-Tools-636.json",
_live_lock_payload(
issue_number=636,
branch="feat/issue-636-inventory-api",
worktree_path=f"{self.tmp}/wt",
pid=os.getpid(),
),
)
control_plane_db.ControlPlaneDB(self.db_path)
def _boom():
raise RuntimeError("scan down")
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
load_hygiene=_boom,
)
self.assertNotIn(
"lock-without-worktree", {c.kind for c in snap.collisions}
)
class TestSectionInclude(_FixtureMixin):
def test_include_restricts_scanned_sections(self):
self._seed_db()
snap = inventory.load_inventory_snapshot(
db_path=self.db_path,
lock_dir=self.lock_dir,
include=("locks",),
)
self.assertIsNotNone(snap.section("locks"))
self.assertIsNone(snap.section("sessions"))
self.assertIsNone(snap.section("worktrees"))
class TestRoutes(_FixtureMixin):
def setUp(self) -> None:
super().setUp()
# Point the loaders at the fixture DB and lock dir via env, and stub
# the worktree scan so the route does not shell out to git.
self._prev_env = {
"GITEA_CONTROL_PLANE_DB": os.environ.get("GITEA_CONTROL_PLANE_DB"),
"GITEA_ISSUE_LOCK_DIR": os.environ.get("GITEA_ISSUE_LOCK_DIR"),
"WEBUI_TEST_OFFLINE": os.environ.get("WEBUI_TEST_OFFLINE"),
}
os.environ["GITEA_CONTROL_PLANE_DB"] = self.db_path
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir
os.environ["WEBUI_TEST_OFFLINE"] = "1"
self._seed_db()
self.client = TestClient(create_app())
def tearDown(self) -> None:
for key, value in self._prev_env.items():
if value is None:
os.environ.pop(key, None)
else:
os.environ[key] = value
def test_inventory_route_returns_versioned_payload(self):
resp = self.client.get("/api/v1/inventory")
self.assertEqual(resp.status_code, 200)
body = resp.json()
self.assertEqual(body["api_version"], "v1")
self.assertIn("sessions", body["sections"])
self.assertIn("field_authority", body)
def test_section_route_restricts_and_labels(self):
resp = self.client.get("/api/v1/inventory/locks")
self.assertEqual(resp.status_code, 200)
body = resp.json()
self.assertEqual(body["requested_section"], "locks")
self.assertIn("locks", body["sections"])
self.assertNotIn("sessions", body["sections"])
def test_unknown_section_is_404(self):
resp = self.client.get("/api/v1/inventory/bogus")
self.assertEqual(resp.status_code, 404)
self.assertEqual(resp.json()["error"], "unknown_section")
def test_inventory_route_rejects_post(self):
resp = self.client.post("/api/v1/inventory")
self.assertEqual(resp.status_code, 405)
class TestReadOnly(_FixtureMixin):
def test_snapshot_does_not_create_db_file(self):
missing = os.path.join(self.tmp, "does-not-exist.db")
inventory.load_inventory_snapshot(
db_path=missing,
lock_dir=self.lock_dir,
load_hygiene=lambda: _StubHygiene(entries=()),
)
self.assertFalse(os.path.exists(missing))
def test_readonly_connection_refuses_write(self):
self._seed_db()
conn = inventory._open_readonly(self.db_path)
try:
with self.assertRaises(Exception):
conn.execute(
"INSERT INTO sessions(session_id, role, started_at, "
"last_heartbeat_at, status) VALUES ('x','author',"
"'t','t','active')"
)
conn.commit()
finally:
conn.close()
# ── lightweight stand-ins for the #432 hygiene snapshot ──────────────────────
class _StubEntry:
def __init__(
self,
*,
rel_path: str,
branch: str | None = None,
classification: str = "stale-clean",
head_sha: str | None = "abc123",
dirty_tracked: int = 0,
dirty_untracked: bool = False,
detached: bool = False,
registered_worktree: bool = True,
notes: str = "",
) -> None:
self.rel_path = rel_path
self.folder_name = rel_path.split("/", 1)[-1]
self.branch = branch
self.classification = classification
self.head_sha = head_sha
self.dirty_tracked = dirty_tracked
self.dirty_untracked = dirty_untracked
self.detached = detached
self.registered_worktree = registered_worktree
self.notes = notes
class _StubHygiene:
def __init__(self, *, entries=(), scan_error=None) -> None:
self.entries = tuple(entries)
self.scan_error = scan_error
if __name__ == "__main__":
unittest.main()
-499
View File
@@ -1,499 +0,0 @@
"""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()
+29 -108
View File
@@ -2,7 +2,6 @@
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from starlette.applications import Starlette
@@ -32,9 +31,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
@@ -45,11 +41,10 @@ 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,
from webui.inventory import (
SECTION_NAMES as _INVENTORY_SECTIONS,
load_inventory_snapshot,
snapshot_to_dict as inventory_snapshot_to_dict,
)
_READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
@@ -84,43 +79,16 @@ 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)))
@@ -313,49 +281,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)
@@ -365,13 +290,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)
@@ -385,29 +303,32 @@ 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 api_inventory(_request: Request) -> JSONResponse:
"""Unified read-only session/lease/lock/worktree inventory (#636)."""
snapshot = load_inventory_snapshot()
return JSONResponse(inventory_snapshot_to_dict(snapshot))
async def api_inventory_section(request: Request) -> JSONResponse:
"""Resource-split view: one inventory section under the shared schema."""
section = request.path_params["section"]
if section not in _INVENTORY_SECTIONS:
return JSONResponse(
{
"error": "unknown_section",
"detail": f"no inventory section named {section!r}",
"available": sorted(_INVENTORY_SECTIONS),
},
status_code=404,
)
snapshot = load_inventory_snapshot(include=(section,))
payload = inventory_snapshot_to_dict(snapshot)
payload["requested_section"] = section
return JSONResponse(payload)
async def method_not_allowed(request: Request, _exc: Exception) -> Response:
@@ -432,7 +353,6 @@ 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"]),
@@ -467,9 +387,10 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
methods=["POST"],
),
Route("/api/leases", api_leases, methods=["GET"]),
Route("/api/v1/inventory", api_inventory, methods=["GET"]),
Route(
"/api/console/security-model",
api_console_security_model,
"/api/v1/inventory/{section}",
api_inventory_section,
methods=["GET"],
),
],
-281
View File
@@ -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."
),
}
-537
View File
@@ -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,
}
-169
View File
@@ -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",
}
+952
View File
@@ -0,0 +1,952 @@
"""Unified session/lease/lock/worktree inventory for the web console (#636).
Leases (#433), worktrees (#432), and runtime (#430) each ship their own MVP
view, each with its own shape and its own idea of what "owned" means. A
traffic-control or recovery operator has to read all three and correlate them
by hand, which is exactly the step that goes wrong under collision pressure.
This module aggregates them into one versioned, read-only snapshot so the
console, and any worker asking "what is safe to do next", read the same
inventory from the same authority.
Field authority is explicit and never blended. Every section declares where its
rows came from:
* ``control_plane_db`` the #613 substrate: sessions, leases, assignments.
Authoritative for *exclusive ownership* (#600/#601).
* ``filesystem`` durable per-issue lock files (:mod:`issue_lock_store`) and
registered git worktrees. Authoritative for *what exists on this machine*.
* ``gitea`` remote issue/PR state, reached only through existing loaders.
Safety invariants:
* **Read-only.** The control-plane database is opened through a ``mode=ro``
URI. :class:`control_plane_db.ControlPlaneDB` creates directories and runs
migrations in its constructor, which an inventory read must never do, so this
module talks to sqlite directly rather than through that class.
* **Fail-soft, never fail-silent.** A subsystem that cannot be read degrades to
a section carrying ``status`` and ``reason``. It never raises, and it never
produces an empty list that reads like "nothing is there".
* **Never invent active ownership.** This is the invariant that matters most.
A degraded ownership source sets ``ownership_authority_complete`` false, and
while that flag is false no work item is reported unowned and no collision is
asserted. Absence of evidence is reported as absence of evidence.
* **Redaction at the boundary.** Absolute paths are collapsed against the home
directory, URLs lose userinfo and query strings, and no credential-shaped
value is emitted. No session token exists in these sources and none is read.
Phase 1 is read-only. Lease steal/release and worktree deletion are Phase 2+
and deliberately have no representation here, not even a disabled one.
"""
from __future__ import annotations
import os
import re
import sqlite3
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Callable
from urllib.parse import urlparse
import control_plane_db
import issue_lock_store
SCHEMA_VERSION = 1
API_VERSION = "v1"
#: Sections whose absence would make an ownership claim unprovable. If any of
#: these is not ``ok``, the snapshot refuses to describe anything as unowned.
OWNERSHIP_SECTIONS = ("sessions", "leases", "locks")
SECTION_NAMES = ("sessions", "leases", "locks", "worktrees", "namespaces")
STATUS_OK = "ok"
STATUS_DEGRADED = "degraded"
STATUS_UNAVAILABLE = "unavailable"
AUTHORITY_CONTROL_PLANE_DB = "control_plane_db"
AUTHORITY_FILESYSTEM = "filesystem"
AUTHORITY_GITEA = "gitea"
_CREDENTIAL_KEY_RE = re.compile(
r"(token|secret|password|passwd|api[_-]?key|authorization|bearer|credential)",
re.IGNORECASE,
)
_REDACTED = "[redacted]"
@dataclass(frozen=True)
class InventorySection:
"""One subsystem's contribution, with its authority and health."""
name: str
authority: str
status: str
items: tuple[dict[str, Any], ...] = ()
reason: str | None = None
scan_ms: float | None = None
@property
def ok(self) -> bool:
return self.status == STATUS_OK
def to_dict(self) -> dict[str, Any]:
return {
"name": self.name,
"authority": self.authority,
"status": self.status,
"count": len(self.items),
"reason": self.reason,
"scan_ms": self.scan_ms,
"items": [dict(item) for item in self.items],
}
@dataclass(frozen=True)
class CollisionSignal:
"""A detected conflict between two ownership records."""
kind: str
message: str
severity: str = "warning"
issue_number: int | None = None
branch: str | None = None
worktree_path: str | None = None
session_ids: tuple[str, ...] = ()
def to_dict(self) -> dict[str, Any]:
return {
"kind": self.kind,
"severity": self.severity,
"message": self.message,
"issue_number": self.issue_number,
"branch": self.branch,
"worktree_path": self.worktree_path,
"session_ids": list(self.session_ids),
}
@dataclass(frozen=True)
class InventorySnapshot:
"""Versioned aggregate of every inventory section."""
generated_at: str
sections: tuple[InventorySection, ...]
collisions: tuple[CollisionSignal, ...] = ()
correlations: tuple[dict[str, Any], ...] = ()
schema_version: int = SCHEMA_VERSION
api_version: str = API_VERSION
scan_ms: float | None = None
_section_index: dict[str, InventorySection] = field(
default_factory=dict, repr=False, compare=False
)
def section(self, name: str) -> InventorySection | None:
return self._section_index.get(name)
@property
def degraded_sections(self) -> tuple[str, ...]:
return tuple(s.name for s in self.sections if not s.ok)
@property
def ownership_authority_complete(self) -> bool:
"""True only when every ownership-bearing section read cleanly.
While this is false the snapshot must not describe any work item as
unowned: a lease the reader could not load is not an absent lease.
"""
for name in OWNERSHIP_SECTIONS:
section = self._section_index.get(name)
if section is None or not section.ok:
return False
return True
@property
def status(self) -> str:
if all(s.ok for s in self.sections):
return STATUS_OK
return STATUS_DEGRADED
# ── redaction ────────────────────────────────────────────────────────────────
def redact_path(path: str | None) -> str | None:
"""Collapse an absolute path against ``$HOME`` for browser display."""
if not path:
return path
text = str(path)
home = os.path.expanduser("~")
if home and home != "/" and text.startswith(home):
return "~" + text[len(home) :]
return text
def redact_url(value: str | None) -> str | None:
"""Strip userinfo and query string from a URL."""
if not value:
return value
text = str(value)
try:
parsed = urlparse(text)
except ValueError:
return _REDACTED
if not parsed.scheme or not parsed.netloc:
return text
netloc = parsed.hostname or ""
if parsed.port:
netloc = f"{netloc}:{parsed.port}"
rebuilt = f"{parsed.scheme}://{netloc}{parsed.path}"
return rebuilt.rstrip("/") or rebuilt
def scrub(value: Any, *, key: str | None = None) -> Any:
"""Recursively drop credential-shaped values and redact paths/URLs.
Never raises: an unexpected object degrades to its ``repr`` rather than
propagating out of a read-only view.
"""
if key and _CREDENTIAL_KEY_RE.search(key):
return _REDACTED
if isinstance(value, dict):
return {str(k): scrub(v, key=str(k)) for k, v in value.items()}
if isinstance(value, (list, tuple)):
return [scrub(v, key=key) for v in value]
if isinstance(value, str):
if value.startswith(("http://", "https://")):
return redact_url(value)
if value.startswith("/") or value.startswith("~"):
return redact_path(value)
return value
if isinstance(value, (int, float, bool)) or value is None:
return value
return repr(value)
# ── control-plane database (read-only) ───────────────────────────────────────
def _open_readonly(db_path: str) -> sqlite3.Connection:
"""Open the control-plane DB without creating or migrating anything."""
conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=5)
conn.row_factory = sqlite3.Row
return conn
def _table_names(conn: sqlite3.Connection) -> set[str]:
rows = conn.execute(
"SELECT name FROM sqlite_master WHERE type = 'table'"
).fetchall()
return {str(row[0]) for row in rows}
def _load_cp_db_sections(
*,
db_path: str | None = None,
limit: int = 200,
) -> tuple[InventorySection, InventorySection]:
"""Return the ``sessions`` and ``leases`` sections from the #613 DB."""
path = (db_path or control_plane_db.default_db_path()).strip()
def _both_unavailable(reason: str) -> tuple[InventorySection, InventorySection]:
return (
InventorySection(
name="sessions",
authority=AUTHORITY_CONTROL_PLANE_DB,
status=STATUS_UNAVAILABLE,
reason=reason,
),
InventorySection(
name="leases",
authority=AUTHORITY_CONTROL_PLANE_DB,
status=STATUS_UNAVAILABLE,
reason=reason,
),
)
if not path:
return _both_unavailable("control-plane database path is not configured")
if not os.path.exists(path):
return _both_unavailable(
f"control-plane database not present at {redact_path(path)}; "
"no session or lease authority available"
)
started = time.perf_counter()
try:
conn = _open_readonly(path)
except sqlite3.Error as exc:
return _both_unavailable(f"control-plane database could not be opened: {exc}")
try:
tables = _table_names(conn)
if "sessions" not in tables or "leases" not in tables:
missing = sorted({"sessions", "leases"} - tables)
return _both_unavailable(
"control-plane database is missing required tables: "
+ ", ".join(missing)
)
session_rows = [
dict(row)
for row in conn.execute(
"SELECT session_id, role, profile, namespace, pid, started_at,"
" last_heartbeat_at, status FROM sessions"
" ORDER BY last_heartbeat_at DESC LIMIT ?",
(max(1, int(limit)),),
).fetchall()
]
has_work_items = "work_items" in tables
if has_work_items:
lease_sql = (
"SELECT l.lease_id, l.session_id, l.role, l.phase, l.status,"
" l.expires_at, w.remote, w.org, w.repo, w.kind AS work_kind,"
" w.number AS work_number, w.state AS work_state,"
" s.pid AS session_pid, s.profile AS session_profile,"
" s.namespace AS session_namespace, s.status AS session_status"
" FROM leases l"
" JOIN work_items w ON w.work_item_id = l.work_item_id"
" LEFT JOIN sessions s ON s.session_id = l.session_id"
" ORDER BY l.expires_at DESC LIMIT ?"
)
else:
lease_sql = (
"SELECT l.lease_id, l.session_id, l.role, l.phase, l.status,"
" l.expires_at FROM leases l"
" ORDER BY l.expires_at DESC LIMIT ?"
)
lease_rows = [
dict(row)
for row in conn.execute(lease_sql, (max(1, int(limit)),)).fetchall()
]
except sqlite3.Error as exc:
return _both_unavailable(f"control-plane database read failed: {exc}")
finally:
conn.close()
elapsed = (time.perf_counter() - started) * 1000.0
now = datetime.now(timezone.utc)
sessions = tuple(
scrub(
{
"session_id": row.get("session_id"),
"role": row.get("role"),
"profile": row.get("profile"),
"namespace": row.get("namespace"),
"pid": row.get("pid"),
"pid_alive": issue_lock_store.is_process_alive(row.get("pid")),
"started_at": row.get("started_at"),
"last_heartbeat_at": row.get("last_heartbeat_at"),
"status": row.get("status"),
}
)
for row in session_rows
)
leases = tuple(
scrub(
{
"lease_id": row.get("lease_id"),
"session_id": row.get("session_id"),
"role": row.get("role"),
"phase": row.get("phase"),
"status": row.get("status"),
"expires_at": row.get("expires_at"),
"expired": _is_expired(row.get("expires_at"), now=now),
"remote": row.get("remote"),
"org": row.get("org"),
"repo": row.get("repo"),
"work_kind": row.get("work_kind"),
"work_number": row.get("work_number"),
"work_state": row.get("work_state"),
"session_pid": row.get("session_pid"),
"session_profile": row.get("session_profile"),
"session_namespace": row.get("session_namespace"),
"session_status": row.get("session_status"),
}
)
for row in lease_rows
)
degraded_reason = (
None
if has_work_items
else "work_items table absent; lease rows carry no work linkage"
)
lease_status = STATUS_OK if has_work_items else STATUS_DEGRADED
return (
InventorySection(
name="sessions",
authority=AUTHORITY_CONTROL_PLANE_DB,
status=STATUS_OK,
items=sessions,
scan_ms=round(elapsed, 3),
),
InventorySection(
name="leases",
authority=AUTHORITY_CONTROL_PLANE_DB,
status=lease_status,
items=leases,
reason=degraded_reason,
scan_ms=round(elapsed, 3),
),
)
def _is_expired(expires_at: str | None, *, now: datetime) -> bool | None:
if not expires_at:
return None
text = str(expires_at).strip().replace("Z", "+00:00")
try:
parsed = datetime.fromisoformat(text)
except ValueError:
return None
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
return parsed <= now
# ── durable issue locks (filesystem) ─────────────────────────────────────────
def _load_locks_section(*, lock_dir: str | None = None) -> InventorySection:
started = time.perf_counter()
try:
paths = issue_lock_store.iter_lock_files(lock_dir)
except OSError as exc:
return InventorySection(
name="locks",
authority=AUTHORITY_FILESYSTEM,
status=STATUS_UNAVAILABLE,
reason=f"issue lock directory could not be listed: {exc}",
)
items: list[dict[str, Any]] = []
unreadable = 0
for path in paths:
try:
record = issue_lock_store.read_lock_file(path)
except (OSError, ValueError):
unreadable += 1
continue
if not record:
unreadable += 1
continue
try:
freshness = issue_lock_store.assess_lock_freshness(record)
except Exception: # noqa: BLE001 — a read-only view never raises
freshness = {"status": "unknown", "live": False, "stale": False}
claimant = record.get("claimant") or (
(record.get("work_lease") or {}).get("claimant") or {}
)
items.append(
scrub(
{
"issue_number": record.get("issue_number"),
"branch_name": record.get("branch_name"),
"remote": record.get("remote"),
"org": record.get("org"),
"repo": record.get("repo"),
"worktree_path": record.get("worktree_path"),
"pid": record.get("session_pid") or record.get("pid"),
"pid_alive": issue_lock_store.is_process_alive(
record.get("session_pid") or record.get("pid")
),
"claimant_username": (claimant or {}).get("username"),
"claimant_profile": (claimant or {}).get("profile"),
"lock_generation": record.get("lock_generation"),
"freshness_status": freshness.get("status"),
"live": bool(freshness.get("live")),
"stale": bool(freshness.get("stale")),
"freshness_reason": freshness.get("reason"),
"lock_path": record.get("lock_file_path") or path,
}
)
)
elapsed = (time.perf_counter() - started) * 1000.0
reason = (
f"{unreadable} lock file(s) were unreadable and are not represented"
if unreadable
else None
)
return InventorySection(
name="locks",
authority=AUTHORITY_FILESYSTEM,
status=STATUS_DEGRADED if unreadable else STATUS_OK,
items=tuple(items),
reason=reason,
scan_ms=round(elapsed, 3),
)
# ── worktrees (filesystem, via the #432 scanner) ─────────────────────────────
def _load_worktrees_section(
*, load_hygiene: Callable[[], Any] | None = None
) -> InventorySection:
started = time.perf_counter()
try:
loader = load_hygiene
if loader is None:
from webui.worktree_scanner import load_hygiene_snapshot
loader = load_hygiene_snapshot
snapshot = loader()
except Exception as exc: # noqa: BLE001 — fail soft, never fail the request
return InventorySection(
name="worktrees",
authority=AUTHORITY_FILESYSTEM,
status=STATUS_UNAVAILABLE,
reason=f"worktree scan failed: {exc}",
)
items = tuple(
scrub(
{
"rel_path": entry.rel_path,
"folder_name": entry.folder_name,
"classification": entry.classification,
"branch": entry.branch,
"head_sha": entry.head_sha,
"dirty_tracked": entry.dirty_tracked,
"dirty_untracked": entry.dirty_untracked,
"detached": entry.detached,
"registered_worktree": entry.registered_worktree,
"notes": entry.notes,
}
)
for entry in snapshot.entries
)
scan_error = getattr(snapshot, "scan_error", None)
elapsed = (time.perf_counter() - started) * 1000.0
return InventorySection(
name="worktrees",
authority=AUTHORITY_FILESYSTEM,
status=STATUS_DEGRADED if scan_error else STATUS_OK,
items=items,
reason=scan_error,
scan_ms=round(elapsed, 3),
)
# ── namespaces / capability summary ──────────────────────────────────────────
def _load_namespaces_section() -> InventorySection:
started = time.perf_counter()
try:
from gitea_auth import get_profile
profile = get_profile() or {}
except Exception as exc: # noqa: BLE001
return InventorySection(
name="namespaces",
authority=AUTHORITY_FILESYSTEM,
status=STATUS_UNAVAILABLE,
reason=f"active profile could not be resolved: {exc}",
)
allowed = list(profile.get("allowed_operations") or [])
forbidden = list(profile.get("forbidden_operations") or [])
profile_name = str(profile.get("profile_name") or "")
namespace = None
try:
import role_namespace_gate
namespace = role_namespace_gate.infer_mcp_namespace(profile_name)
except Exception: # noqa: BLE001 — namespace inference is advisory
namespace = None
item = scrub(
{
"profile_name": profile_name,
"role": profile.get("role"),
"mcp_namespace": namespace,
"allowed_operations": sorted(allowed),
"forbidden_operations": sorted(forbidden),
"capability_summary": {
"can_author": "gitea.pr.create" in allowed,
"can_review": "gitea.pr.approve" in allowed,
"can_merge": "gitea.pr.merge" in allowed,
"can_close_pr": "gitea.pr.close" in allowed,
},
"active": True,
}
)
elapsed = (time.perf_counter() - started) * 1000.0
return InventorySection(
name="namespaces",
authority=AUTHORITY_FILESYSTEM,
status=STATUS_OK,
items=(item,),
reason=(
"only the profile serving this web process is observable; other "
"namespaces are not enumerable from here"
),
scan_ms=round(elapsed, 3),
)
# ── correlation and collision detection ──────────────────────────────────────
def _issue_from_branch(branch: str | None) -> int | None:
match = re.search(r"issue-(\d+)", str(branch or ""), re.IGNORECASE)
return int(match.group(1)) if match else None
def correlate(
*,
leases: InventorySection,
locks: InventorySection,
worktrees: InventorySection,
sessions: InventorySection,
) -> tuple[tuple[dict[str, Any], ...], tuple[CollisionSignal, ...]]:
"""Join lease owner ↔ lock ↔ worktree ↔ namespace where evidence allows.
Correlation rows are emitted from whatever sections did load. Collision
signals are only emitted from sections that are ``ok``: a conflict inferred
from a partially-read source would be a false accusation.
"""
correlations: list[dict[str, Any]] = []
collisions: list[CollisionSignal] = []
worktree_by_branch: dict[str, dict[str, Any]] = {}
for entry in worktrees.items:
branch = (entry.get("branch") or "").strip()
if branch:
worktree_by_branch.setdefault(branch, entry)
session_by_id = {
str(s.get("session_id")): s for s in sessions.items if s.get("session_id")
}
# Lock-centred rows: a durable lock names an issue, a branch, and a worktree.
for lock in locks.items:
branch = (lock.get("branch_name") or "").strip()
worktree = worktree_by_branch.get(branch)
matching_leases = [
lease
for lease in leases.items
if lease.get("work_kind") == "issue"
and lease.get("work_number") == lock.get("issue_number")
]
correlations.append(
{
"issue_number": lock.get("issue_number"),
"branch": branch or None,
"lock_live": bool(lock.get("live")),
"lock_claimant": lock.get("claimant_profile"),
"lock_pid": lock.get("pid"),
"lock_pid_alive": lock.get("pid_alive"),
"worktree_rel_path": (worktree or {}).get("rel_path"),
"worktree_classification": (worktree or {}).get("classification"),
"worktree_registered": (worktree or {}).get("registered_worktree"),
"lease_ids": [
lease.get("lease_id")
for lease in matching_leases
if lease.get("lease_id")
],
"lease_sessions": [
lease.get("session_id")
for lease in matching_leases
if lease.get("session_id")
],
}
)
if locks.ok and worktrees.ok:
# A claim whose lease window is still open but has no registered
# worktree is an anomaly regardless of whether its pid is alive; a
# fully time-expired lease is on its way out and is not flagged.
if (
lock.get("freshness_status") != "expired"
and branch
and worktree is None
):
collisions.append(
CollisionSignal(
kind="lock-without-worktree",
severity="warning",
issue_number=lock.get("issue_number"),
branch=branch,
worktree_path=lock.get("worktree_path"),
message=(
f"Live lock on issue #{lock.get('issue_number')} names "
f"branch {branch!r} but no registered worktree carries "
"that branch (#404)"
),
)
)
if locks.ok:
# A lock whose recorded pid is gone is held by nobody: a clean #753
# dead-session recovery candidate. Subclassify by the lease window,
# because the two cases need different operator urgency. When the
# window is still open the lock would read as live to a naive
# timestamp check even though the owner is dead — the more dangerous
# case — so it is flagged distinctly from a fully time-expired lease.
if lock.get("pid_alive") is False and lock.get("stale"):
if lock.get("freshness_status") == "expired":
collisions.append(
CollisionSignal(
kind="stale-lock-dead-owner",
severity="warning",
issue_number=lock.get("issue_number"),
branch=branch or None,
message=(
f"Lock on issue #{lock.get('issue_number')} is stale "
f"and its recorded pid {lock.get('pid')} is not running "
"(#753 dead-session recovery candidate)"
),
)
)
else:
collisions.append(
CollisionSignal(
kind="live-lock-dead-owner",
severity="warning",
issue_number=lock.get("issue_number"),
branch=branch or None,
message=(
f"Lock on issue #{lock.get('issue_number')} has an "
"unexpired lease but its recorded pid "
f"{lock.get('pid')} is not running; it would read as "
"live to a timestamp check (#753 dead-session recovery "
"candidate)"
),
)
)
# The #635 trap: the lease has expired but the recorded pid is a
# still-running daemon, so neither dead-pid reclaim nor exact-owner
# renewal applies. This is the collision an operator must see.
elif (
lock.get("freshness_status") == "expired"
and lock.get("pid_alive") is True
):
collisions.append(
CollisionSignal(
kind="expired-lock-live-owner",
severity="error",
issue_number=lock.get("issue_number"),
branch=branch or None,
message=(
f"Lock on issue #{lock.get('issue_number')} has an expired "
f"lease but its recorded pid {lock.get('pid')} is still "
"running (daemon-pid deadlock; needs an operator decision, "
"#635/#760)"
),
)
)
# Two live locks on one branch, or two active leases on one work item.
if locks.ok:
by_branch: dict[str, list[dict[str, Any]]] = {}
for lock in locks.items:
if not lock.get("live"):
continue
branch = (lock.get("branch_name") or "").strip()
if branch:
by_branch.setdefault(branch, []).append(lock)
for branch, entries in sorted(by_branch.items()):
if len(entries) > 1:
collisions.append(
CollisionSignal(
kind="duplicate-live-lock",
severity="error",
branch=branch,
message=(
f"{len(entries)} live locks name branch {branch!r}: "
"issues "
+ ", ".join(
f"#{e.get('issue_number')}" for e in entries
)
),
)
)
if leases.ok:
by_work: dict[tuple[str, int], list[dict[str, Any]]] = {}
for lease in leases.items:
if str(lease.get("status") or "").lower() != "active":
continue
kind = str(lease.get("work_kind") or "").strip().lower()
number = lease.get("work_number")
if not kind or number is None:
continue
by_work.setdefault((kind, int(number)), []).append(lease)
for (kind, number), entries in sorted(by_work.items()):
sessions_held = {
str(e.get("session_id")) for e in entries if e.get("session_id")
}
if len(sessions_held) > 1:
collisions.append(
CollisionSignal(
kind="concurrent-active-lease",
severity="error",
issue_number=number if kind == "issue" else None,
session_ids=tuple(sorted(sessions_held)),
message=(
f"{len(sessions_held)} sessions hold an active lease on "
f"{kind} #{number}"
),
)
)
for entry in entries:
if entry.get("expired") is True:
collisions.append(
CollisionSignal(
kind="active-lease-past-expiry",
severity="warning",
issue_number=number if kind == "issue" else None,
session_ids=(
(str(entry.get("session_id")),)
if entry.get("session_id")
else ()
),
message=(
f"Lease {entry.get('lease_id')} on {kind} #{number} "
"is still marked active past its expiry"
),
)
)
# A lease whose owning session is gone is an orphan, not free work.
if leases.ok and sessions.ok:
for lease in leases.items:
if str(lease.get("status") or "").lower() != "active":
continue
session_id = str(lease.get("session_id") or "")
if session_id and session_id not in session_by_id:
collisions.append(
CollisionSignal(
kind="orphan-lease",
severity="error",
session_ids=(session_id,),
message=(
f"Active lease {lease.get('lease_id')} names session "
f"{session_id}, which has no session record"
),
)
)
return tuple(correlations), tuple(collisions)
# ── snapshot assembly ────────────────────────────────────────────────────────
def load_inventory_snapshot(
*,
db_path: str | None = None,
lock_dir: str | None = None,
load_hygiene: Callable[[], Any] | None = None,
include: tuple[str, ...] | None = None,
) -> InventorySnapshot:
"""Build the unified inventory snapshot.
Every section is loaded independently and fails soft. *include* restricts
the sections that are scanned; omitted sections are simply absent rather
than reported as empty, so a resource-split request cannot be mistaken for
a whole-inventory answer.
"""
started = time.perf_counter()
wanted = tuple(include) if include else SECTION_NAMES
sections: list[InventorySection] = []
sessions_section: InventorySection | None = None
leases_section: InventorySection | None = None
if "sessions" in wanted or "leases" in wanted:
sessions_section, leases_section = _load_cp_db_sections(db_path=db_path)
if "sessions" in wanted:
sections.append(sessions_section)
if "leases" in wanted:
sections.append(leases_section)
locks_section = (
_load_locks_section(lock_dir=lock_dir)
if "locks" in wanted
else _empty_section("locks", AUTHORITY_FILESYSTEM)
)
if "locks" in wanted:
sections.append(locks_section)
worktrees_section = (
_load_worktrees_section(load_hygiene=load_hygiene)
if "worktrees" in wanted
else _empty_section("worktrees", AUTHORITY_FILESYSTEM)
)
if "worktrees" in wanted:
sections.append(worktrees_section)
if "namespaces" in wanted:
sections.append(_load_namespaces_section())
correlations, collisions = correlate(
leases=leases_section or _empty_section("leases", AUTHORITY_CONTROL_PLANE_DB),
locks=locks_section,
worktrees=worktrees_section,
sessions=sessions_section
or _empty_section("sessions", AUTHORITY_CONTROL_PLANE_DB),
)
elapsed = (time.perf_counter() - started) * 1000.0
index = {section.name: section for section in sections}
return InventorySnapshot(
generated_at=datetime.now(timezone.utc).isoformat(),
sections=tuple(sections),
collisions=collisions,
correlations=correlations,
scan_ms=round(elapsed, 3),
_section_index=index,
)
def _empty_section(name: str, authority: str) -> InventorySection:
"""A section that was not requested — never a claim that it is empty."""
return InventorySection(
name=name,
authority=authority,
status=STATUS_UNAVAILABLE,
reason="section not requested in this scan",
)
def snapshot_to_dict(snapshot: InventorySnapshot) -> dict[str, Any]:
"""Serialize the snapshot for the versioned API."""
return {
"api_version": snapshot.api_version,
"schema_version": snapshot.schema_version,
"generated_at": snapshot.generated_at,
"status": snapshot.status,
"scan_ms": snapshot.scan_ms,
"ownership_authority_complete": snapshot.ownership_authority_complete,
"ownership_note": (
"Every ownership source read cleanly; an item absent from leases "
"and locks is genuinely unclaimed."
if snapshot.ownership_authority_complete
else "One or more ownership sources are degraded; nothing in this "
"snapshot may be treated as unowned. Collisions are reported only "
"from sections that read cleanly."
),
"degraded_sections": list(snapshot.degraded_sections),
"field_authority": {
"sessions": AUTHORITY_CONTROL_PLANE_DB,
"leases": AUTHORITY_CONTROL_PLANE_DB,
"locks": AUTHORITY_FILESYSTEM,
"worktrees": AUTHORITY_FILESYSTEM,
"namespaces": AUTHORITY_FILESYSTEM,
},
"sections": {section.name: section.to_dict() for section in snapshot.sections},
"correlations": [dict(row) for row in snapshot.correlations],
"collisions": [signal.to_dict() for signal in snapshot.collisions],
}
-682
View File
@@ -1,682 +0,0 @@
"""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),
}
+3 -5
View File
@@ -65,11 +65,9 @@ PROMPT_RECONCILER = (
"reconciliation (already-landed / post-merge cleanup). Do not approve or merge."
)
PROMPT_CONTROLLER = (
"CONTROLLER session: call gitea_route_task_session(task_type='process_work_queue') "
"then gitea_allocate_next_work (cross_role default) for {remote}/{org}/{repo}; "
"use the returned required_role/profile/action to schedule exactly one downstream "
"role cycle. Dashboard is explanatory only and never replaces allocator selection. "
"Do not implement, review, approve, or merge in-band."
"CONTROLLER session: inspect gitea_workflow_dashboard + control-plane leases, "
"diagnose blocked/terminal-locked items for {remote}/{org}/{repo}, and schedule "
"exactly one fresh role-scoped cycle. Do not implement, review, or merge in-band."
)
PROMPT_IDLE = (
"IDLE: no safe assignable work for role '{role}' on {remote}/{org}/{repo}. "