Compare commits

..
Author SHA1 Message Date
sysadmin 22e0a41bd5 feat(restart): audit lifecycle events and durable incidents (#665)
Add restart_audit with mcp.restart.* event schema, redacted emission via
gitea_audit, correlation ids, and incident materialization for failed drain
and break-glass. Wire gitea_request_mcp_restart to always audit impact
previews and fail closed on privileged apply when the audit sink is enabled
but write fails.

Closes #665
2026-07-25 17:10:16 -04:00
25 changed files with 1019 additions and 6259 deletions
+24 -98
View File
@@ -23,7 +23,6 @@ import json
import os
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Mapping, Sequence
from control_plane_db import (
@@ -739,46 +738,6 @@ def normalize_exclude_issue_numbers(
return sorted(out)
def _claim_expires_at(claim: Any) -> datetime | None:
"""Parse a claim's ``expires_at``, or ``None`` when it is absent/malformed."""
if not isinstance(claim, Mapping):
return None
text = str(claim.get("expires_at") or "").strip()
if not text:
return None
if text.endswith("Z"):
text = text[:-1] + "+00:00"
try:
parsed = datetime.fromisoformat(text)
except ValueError:
return None
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
return parsed.astimezone(timezone.utc)
def _drop_expired_claims(
claims: Mapping[tuple[str, int], dict[str, Any]],
*,
now: datetime | None = None,
) -> dict[tuple[str, int], dict[str, Any]]:
"""Claims minus those whose lease has already expired (#643).
The read-only mirror of ``expire_stale_leases``: the sweep marks such rows
``expired`` so they stop being returned as claims, and this reaches the same
view without writing. A claim with no parseable ``expires_at`` is **kept** —
an unreadable expiry is not evidence that work is free.
"""
moment = now or datetime.now(timezone.utc)
kept: dict[tuple[str, int], dict[str, Any]] = {}
for key, claim in (claims or {}).items():
expires_at = _claim_expires_at(claim)
if expires_at is not None and expires_at <= moment:
continue
kept[key] = claim
return kept
def candidate_set_fingerprint(
candidates: Sequence[WorkCandidate],
*,
@@ -867,22 +826,12 @@ def allocate_next_work(
exclude_issue_numbers: Sequence[int] | None = None,
expected_candidate_set_fingerprint: str | None = None,
allocation_mode: str | None = None,
side_effect_free: bool = False,
) -> 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.
*side_effect_free* (#643): a dry run that writes **nothing** to the
control-plane DB. A plain ``apply=False`` still registered a session row and
swept stale leases globally, so a caller advertising a read-only preview was
mutating on every call. Under this flag both writes are suppressed and stale
leases are instead filtered out of the claim map in memory, which yields the
same selection the sweep would have produced without persisting anything.
Incompatible with *apply* — the combination fails closed rather than
silently reserving.
*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
@@ -936,57 +885,40 @@ def allocate_next_work(
"allocation_mode": (allocation_mode or "").strip() or None,
}
# A side-effect-free run may never reserve: reserving is a write, and the
# flag is the caller's assertion that this call writes nothing (#643).
if side_effect_free and apply:
session_id = (session_id or "").strip() or f"alloc-{uuid.uuid4().hex[:12]}"
try:
db.upsert_session(
session_id=session_id,
role=role_norm,
profile=profile_name,
pid=os.getpid(),
controller_instance_id=controller_instance_id,
)
except Exception as exc: # noqa: BLE001 — surface structured
return {
"success": False,
"outcome": OUTCOME_NO_SAFE,
"apply": True,
"reasons": [
"side_effect_free is incompatible with apply=True; an "
"assignment is a write (fail closed, #643)"
f"failed to register session in control-plane DB: {exc} "
"(fail closed, #613)"
],
"skipped": [],
"assignment": None,
"substrate": "control_plane_db",
}
session_id = (session_id or "").strip() or f"alloc-{uuid.uuid4().hex[:12]}"
if not side_effect_free:
try:
db.upsert_session(
session_id=session_id,
role=role_norm,
profile=profile_name,
pid=os.getpid(),
controller_instance_id=controller_instance_id,
)
except Exception as exc: # noqa: BLE001 — surface structured
return {
"success": False,
"outcome": OUTCOME_NO_SAFE,
"reasons": [
f"failed to register session in control-plane DB: {exc} "
"(fail closed, #613)"
],
"skipped": [],
"assignment": None,
"substrate": "control_plane_db",
}
# Expire stale leases globally before selection.
try:
db.expire_stale_leases()
except Exception as exc: # noqa: BLE001
return {
"success": False,
"outcome": OUTCOME_NO_SAFE,
"reasons": [f"lease expiry failed: {exc} (fail closed)"],
"skipped": [],
"assignment": None,
"substrate": "control_plane_db",
}
# Expire stale leases globally before selection.
try:
db.expire_stale_leases()
except Exception as exc: # noqa: BLE001
return {
"success": False,
"outcome": OUTCOME_NO_SAFE,
"reasons": [f"lease expiry failed: {exc} (fail closed)"],
"skipped": [],
"assignment": None,
"substrate": "control_plane_db",
}
terminal = None
try:
@@ -1021,12 +953,6 @@ def allocate_next_work(
"assignment": None,
"substrate": "control_plane_db",
}
if side_effect_free:
# ``list_active_claims`` filters on status alone, so without the
# global sweep an already-expired lease would still read as a live
# claim and the preview would report work as taken that is free.
# Drop those in memory: same view the sweep produces, no write.
claims = _drop_expired_claims(claims)
try:
exclude_nums = normalize_exclude_issue_numbers(exclude_issue_numbers)
+93
View File
@@ -0,0 +1,93 @@
# MCP restart audit events and incidents (#665)
Restarts and recovery attempts leave a forensic trail. Failed drains and
break-glass paths also raise durable Gitea incident issues so unsafe restarts
cannot be silently repeated.
Parent umbrella: **#655**. Related: impact coordinator **#658**, drain proof
**#661**, restart classes **#663**, break-glass **#664**, post-restart reconcile
**#662**, vision **#652**, roadmap **#653**, console recovery **#642**.
## Components
| Piece | Where | Responsibility |
|-------|-------|----------------|
| Event schema + emission | `restart_audit.py` | `mcp.restart.*` vocabulary, redacted payload builder, append-only sink via `gitea_audit` |
| Fail-closed privileged gate | `restart_audit.require_audit_or_deny` | When `GITEA_AUDIT_LOG` is set and the write fails, privileged apply is denied |
| Incident descriptors | `restart_audit.build_incident_descriptor` | Durable follow-up issues (failed drain, break-glass, reconcile unresolved, unguarded) |
| Materializer | `restart_audit.materialize_incident` | Injected `create_issue_fn` (network kept out of pure tests) |
| Wiring | `gitea_request_mcp_restart` | Correlation id, impact-preview audit, apply-gate / break-glass audit + incident creation |
## Event vocabulary
| Event type | When |
|------------|------|
| `mcp.restart.impact_preview` | Every `gitea_request_mcp_restart` evaluation |
| `mcp.restart.drain_enter` | Drain window starts (schema reserved; emit from drain path) |
| `mcp.restart.drain_exit` | Drain window ends |
| `mcp.restart.drain_proof` | Drain-proof verification result |
| `mcp.restart.apply_gate` | Apply hard gate (`dry_run=False`) |
| `mcp.restart.break_glass` | Authorized break-glass bypass |
| `mcp.restart.post_restart_reconcile` | Post-restart reconcile outcome |
| `mcp.restart.narrower_recovery` | Narrower recovery attempt recorded |
| `mcp.restart.unguarded_detected` | Unguarded restart path detected |
All free text is redacted before sink write or issue body assembly. Emission
never raises; callers decide fail-closed policy.
## Correlation
Each restart lifecycle mints a short `correlation_id` (`rst-` + 16 hex) shared
across impact preview → apply gate → incident descriptors so operators can join
the trail.
## Privileged deny-on-audit-fail
Rollout policy (issue #665):
1. Configure `GITEA_AUDIT_LOG` so writes land.
2. Only then enforce deny when a privileged restart path cannot audit.
When audit is **not** configured, privileged apply still proceeds (no false
denials during rollout). When audit **is** configured and the write fails,
`apply_authorized` is cleared.
## Incidents
| Kind | Trigger |
|------|---------|
| `restart_failed_drain` | Apply denied by drain hard gate / failed proof |
| `restart_break_glass` | Any authorized break-glass apply |
| `restart_reconcile_unresolved` | Post-restart reconcile left work unresolved |
| `restart_unguarded_detected` | Unguarded restart attempt detected |
Break-glass **always** creates an incident descriptor (and a Gitea issue when
the create path is available). Failed drain does the same. Incident bodies
include correlation id, session, class, scope, proof id, and redacted reasons.
Default labels: `mcp-health`, `safety`, `observability`, `status:ready`,
`type:bug`, `workflow-hardening`.
## Tool payload surface
`gitea_request_mcp_restart` returns:
* `correlation_id` — lifecycle join key
* `restart_audit.impact_preview_written` — sink success for the preview event
* `restart_audit.apply_gate_written` — sink success for apply/break-glass (apply only)
* `restart_audit.incident_result` — materialization outcome when an incident was required
* `incident` — durable descriptor (when gate requires follow-up)
## Security
* No secrets in audit payloads or issue bodies.
* This module never restarts a process.
* Drain proof verification remains #661; audit only records the decision.
* Incident creation failures are recorded in `incident_result.reasons` and never
crash the restart evaluation path (audit write failure still fails closed for
privileged apply when the sink is enabled).
## Tests
See `tests/test_restart_audit.py`: schema, redaction, emission, deny policy,
incident materialization mocks, break-glass / failed-drain selection.
+1
View File
@@ -33,6 +33,7 @@ recovery behavior for all nine classes.
| `ControlPlaneDB.list_sessions` | `control_plane_db.py` | Read-only session inventory (the process-level unit a restart kills). |
| `gitea_request_mcp_restart` | `gitea_mcp_server.py` | MCP tool: gathers inventory from the #613 DB, calls the coordinator, returns the report, and on `dry_run=False` runs the #661 drain-proof hard gate. Never restarts a process. |
| `drain_proof.gate_apply_restart` | `drain_proof.py` | The #661 hard gate: verifies a drain proof against the current impact fingerprint, or records an authorized break-glass bypass. |
| `restart_audit` | `restart_audit.py` | #665 forensic trail: `mcp.restart.*` events via `gitea_audit`, correlation ids, durable incidents for failed drain / break-glass. See [`mcp-restart-audit.md`](./mcp-restart-audit.md). |
## Dimensions evaluated
+9 -41
View File
@@ -94,7 +94,6 @@ already define, and a regression test asserts each mapping matches.
| `record_analytics_usage` | operator | gated_write | `runtime.record_analytics_usage` | Yes | No | No | 2 |
| `system.reload_namespace` | controller | privileged | `runtime.reload_namespace` | Yes | No | No | 2 |
| `system.restart_namespace` | admin | destructive | `runtime.restart_namespace` | Yes | **Yes** | **Yes** | 2 |
| `initiate_workflow` | operator | gated_write | `gitea.read` | Yes | No | No | 2 |
**Dual control** means the acting principal may not be the sole authority: a
second distinct principal must confirm. **Break-glass** means the action is
@@ -113,12 +112,6 @@ by the console — both hand off to a host supervisor, and neither exposes a raw
process kill. See
[`sanctioned-restart-controls.md`](sanctioned-restart-controls.md) (#642).
`initiate_workflow` (#643) is operator-class because its outcome is a *claim*,
not a Gitea verdict. Requesting reviewer or merger work reserves that work
through the allocator; it does not grant the right to approve or merge, which
stays with the MCP role profile and its own capability gates. See
[`webui-requests.md`](webui-requests.md).
### Authorization decision
`authorize(action_id, principal, for_execution=False)` returns a decision
@@ -133,24 +126,9 @@ record and **denies by default**. The deny reasons are closed and enumerated:
| `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.
`execution_enabled` on the decision reports whether the action has a live
execution path at all, and is computed by `execution_wired(action)`. There are
exactly two ways to be wired:
1. the action's `phase` is at or below `ACTIVE_PHASE`; or
2. the action declares an `execution_env_flag` **and** that variable is set.
Every action that declares no flag therefore reports `execution_enabled: false`
while the console is in Phase 1, so no caller can read an allow as permission
to mutate. The per-action flag exists because raising `ACTIVE_PHASE` would
enable execution for every action of that phase at once, including ones whose
execution path is not implemented. One implemented action goes live on its own
flag instead of dragging its unimplemented phase-mates with it.
`initiate_workflow` is the only action that currently declares a flag
(`WEBUI_REQUESTS_EXECUTION`), and it stays denied until an operator sets it.
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
@@ -257,22 +235,13 @@ second one. The integration points are already wired and observable:
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.
- **`POST /api/v1/requests/preview` and `.../apply`** (#643) are the first
actions to use this model for a real execution path. Preview always returns a
decision and an audited `previewed` record; apply requires `confirm=true`,
emits `succeeded` or `denied`, and reserves work only through the allocator.
See [`webui-requests.md`](webui-requests.md).
A Phase 2 action must: use `execution_wired` rather than a private enable flag,
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.
Raising `ACTIVE_PHASE` remains the way to open a whole phase at once, and is
deliberately *not* what #643 did: an action-scoped opt-in cannot enable an
action whose execution path nobody wrote.
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
@@ -325,7 +294,6 @@ Until Phase 2 wires it, probe protection rests on network placement alone, as
| `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 |
| `WEBUI_REQUESTS_EXECUTION` | unset | Opt in to `initiate_workflow` execution (#643) |
All are read server-side only. None is ever rendered into a page or returned by
an API.
-64
View File
@@ -80,8 +80,6 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
| `/sessions` | Runtime and session view (#641) — health + inventory sessions/namespaces/worktrees |
| `/api/sessions` | JSON export for the runtime/session view |
| `/api/v1/sessions` | Versioned alias of `/api/sessions` |
| `/gitea` | Gitea issue↔PR linkage console (#645) — both directions, with the evidence for each edge |
| `/api/v1/gitea/linkage` | JSON linkage export; `502` when the read could not be answered |
| `/inventory` | Phase 1 shell stub — unified inventory (backed by #636) |
| `/timeline` | Phase 1 shell stub — workflow event timeline |
| `/policy` | Phase 1 shell stub — capability/role policy placeholder |
@@ -329,68 +327,6 @@ Honesty rules specific to this view:
The write-time redactor is a narrow denylist and is not relied on. The field
itself is kept — it is the `#630` evidence naming which daemon was killed.
## Gitea issue/PR linkage (#645)
`/gitea` is the Phase 3 read-only linkage console: which PR carries which issue,
which issues are claimed by more than one PR, and what the latest Canonical
Thread Handoff on a thread said. Gitea remains the source of truth — this
surface reads it and never writes to it. There is no issue/PR editor, no review,
and no merge control.
Query parameters (all optional):
| Parameter | Meaning |
|-----------|---------|
| `project` | Registry project id to scope the read (default: first registry entry) |
| `state` | `open` (default) or `all`; `all` widens the window to merged/closed items, where a landed edge lives |
| `issue=N` / `pr=N` | Focus one thread and load *its* latest canonical handoff |
`GET /api/v1/gitea/linkage` returns the same model as JSON
(`schema_version: 1`). It answers `502` when the read could not be answered, so
an automated consumer cannot mistake a fail-closed payload for "no links exist".
The HTML page always answers `200` and renders the reason instead — an operator
view must show why a read failed rather than withhold the page.
### How an edge is found
Each edge carries the evidence that produced it, strongest first:
| Evidence | Meaning |
|----------|---------|
| `closes_keyword` | The PR title or body declares `closes/fixes/resolves #N`. Gitea itself acts on this keyword. |
| `branch_marker` | The PR head branch carries the canonical `(fix\|feat\|docs\|chore)/issue-N-…` marker minted by the issue lock. |
| `body_reference` | The PR body mentions `#N` with no closing keyword. A mention is not a claim to close. |
Only closing and branch-marker edges populate the **issue → PR** direction: a
bare mention is a cross-link, and counting it as ownership would invent
contested issues out of ordinary references. The mention stays visible on the
**PR → issue** side, labelled as such. A PR whose two strongest edges tie is
flagged `ambiguous`; an issue claimed by two PRs is flagged `contested`.
### Honesty rules specific to this view
* **A partial read never reads as an absence.** Linkage is a claim about the
loaded window only. When pagination did not complete, every empty edge cell
renders `none found (partial inventory)` rather than `none`, and the JSON
carries `inventory_complete: false` plus per-row `links_authoritative: false`.
* **A failed read renders no table at all.** Missing credentials, an unknown
project, or a fetch error produce `ok: false` with a reason. An empty linkage
table would assert that no issue is linked to any PR, which such a read is not
in a position to claim.
* **Handoffs are loaded, never assumed.** CTH comments are thread-scoped, so
only the focused issue or PR has its comments fetched. Every other row reports
`not_loaded` with the reason; a thread whose comments *were* loaded and carried
no CTH says exactly that. A comment-source failure degrades the handoff alone —
the linkage tables still render.
* **Unrecognised handoff headings are reported, not republished.** A `## CTH:`
heading outside `CTH_TYPES` renders as `unrecognized`.
* **Redaction precedes display.** Titles, labels, handoff fields, and error
reasons pass through `webui.console_redaction` before serialization, and the
page HTML-escapes everything it renders.
* **Deep links are opt-in.** A link out to the Gitea web UI appears only when
`GITEA_MCP_REVEAL_ENDPOINTS=1` is set server-side, matching how the MCP tools
gate URL exposure. Item numbers stay usable without it.
## System-health dashboard (#639)
`/system-health` renders the same snapshot the `/api/v1/system/health` API
-81
View File
@@ -1,81 +0,0 @@
# Web Console: Notifications & Human-Attention Routing (#648)
- **Status:** Phase 3 Live
- **Tracking Issue:** [#648](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/648)
- **Parent Epic:** [#631](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/631)
- **Attention Boundary Reference:** [#628](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/628)
---
## 1. Overview
The **Notifications & Human-Attention Console** (`/notifications`, `/api/v1/notifications`) provides intelligent event classification and human-attention routing for autonomous workflow operations.
To prevent alert fatigue while ensuring critical escalation boundaries are never missed, events are classified into three distinct **Attention Classes**:
1. **`human-required`** (Urgent Escalation Boundary):
- Items requiring immediate human intervention or business decisions.
- Triggers: Auth failures, hard stops, irrecoverable state, decision locks, failed report validations, critical probe errors.
- Display: Highlighted in red (`badge-blocked`) with a `HUMAN REQUIRED` badge.
2. **`operator`** (Operational Inbox):
- Items requiring controller or operator review/triage during routine execution.
- Triggers: Blocked PRs (merge conflicts), stale leases, duplicate PRs on issues, unassigned ready work.
- Display: Displayed in orange/yellow (`badge-claimed`).
3. **`routine`** (Background Workflow Transitions):
- Normal, healthy workflow transitions and state progressions.
- Triggers: Active PRs/issues in standard state, clean branch creation, routine heartbeats.
- Display: Filtered out of default inbox views to eliminate notification spam; viewable on demand via the "Routine" or "All" tab.
---
## 2. API Endpoints
### `GET /api/v1/notifications`
*Compatibility Alias:* `GET /api/notifications`
#### Query Parameters:
- `project_id` (optional): Filter notifications by project ID.
- `attention_class` (optional): `inbox` (default: human-required + operator), `human-required`, `operator`, `routine`, `all`.
#### Example JSON Response:
```json
{
"project_id": "gitea-tools",
"repo_label": "Scaled-Tech-Consulting/Gitea-Tools",
"human_required_count": 0,
"operator_count": 2,
"routine_count": 5,
"total_count": 7,
"fetch_error": null,
"inbox_items": [
{
"id": "notif-pr-block-742",
"attention_class": "operator",
"category": "blocker",
"title": "Blocked PR #742",
"summary": "PR #742 requires merge conflict resolution.",
"work_kind": "pr",
"work_number": 742,
"project_id": "gitea-tools",
"repo_label": "Scaled-Tech-Consulting/Gitea-Tools",
"created_at": "2026-07-25T16:39:47Z",
"deep_link": "/traffic",
"requires_human": false,
"extra": {}
}
],
"all_items": [...]
}
```
---
## 3. UI Navigation
- Access via the **Traffic** navigation menu: **Traffic → Notifications**.
- The main view displays:
- **Metrics Summary Bar**: Highlighting counts for Human Required, Operator Inbox, and Routine items.
- **Attention Filter Tabs**: Toggle between Inbox (Human + Operator), Human Required, Operator, Routine, and All.
- **Structured Event Table**: Displays category, title, summary, work item links, and timestamps.
-160
View File
@@ -1,160 +0,0 @@
# Web console requests: intent preview and workflow initiation (#643)
**Phase 2. Preview is always live and always read-only. Initiation is wired but
denied until an operator opts in.**
Before this surface, starting role work meant pasting a prompt into a terminal
and trusting the operator to have checked the allocator first. Nothing enforced
that check, so two sessions could reach for the same issue and each believe it
was theirs. This page replaces the paste with a *request*: a desired role, an
issue or PR, and a stated intent, answered by an authorization decision and —
on confirmation — an exclusive assignment from the allocator.
| Concern | Module |
|---------|--------|
| Request model, preview, initiation | `webui/request_service.py` |
| Form and preview rendering | `webui/request_views.py` |
| Authorization | `webui/console_authz.py` (`initiate_workflow`) |
| Audit | `webui/console_audit.py` |
| Ownership substrate | `allocator_service.py` + `control_plane_db.py` |
## Surfaces
| Path | Method | Purpose |
|------|--------|---------|
| `/requests` | GET | Request form |
| `/requests` | POST | Render an intent preview. **Never assigns.** |
| `/api/v1/requests/preview` | POST | Intent preview as JSON |
| `/api/v1/requests/apply` | POST | Initiate — confirmed, audited, allocator-owned |
The HTML form has no initiate button on purpose. Initiating requires a
confirmed POST to `/api/v1/requests/apply`, so a stray form submission cannot
reserve work as a side effect.
## The request
```json
{
"desired_role": "author",
"work_kind": "issue",
"work_number": 643,
"intent_summary": "implement request preview and initiation",
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"expected_head_sha": null
}
```
`desired_role` is one of `author`, `reviewer`, `merger`, `reconciler`,
`controller`. `work_kind` is `issue` or `pr`. `remote`/`org`/`repo` default to
the first project in the registry when omitted; when neither the request nor
the registry resolves them, the request is rejected rather than pointed at some
other repository. `intent_summary` is required — it is what the audit record
states as the reason — and is truncated to 500 characters.
Parsing rejects rather than corrects. An unknown role, an unknown work kind, a
non-positive number, or a missing intent each return `400` with a `reason_code`
and the offending `field`.
## Preview
Five checks, each with its own verdict, reason code, and detail:
| Check | Passes when |
|-------|-------------|
| `authorization` | The console principal holds `operator` or above |
| `capability` | The desired role maps to a declared profile and MCP namespace |
| `lease_availability` | No active claim holds the work unit |
| `next_safe_action` | The allocator would independently select this exact work unit |
| `head_pin` | PR work resolves to a head SHA, and a supplied SHA still matches |
A preview also returns the role's `allowed_actions` and `prohibited_actions`
(from `allocator_service.ROLE_ACTIONS`), the `required_profile` and
`required_namespace` the work must run under, and a `correlation_id` that ties
the preview to its audit record and to any assignment that follows.
Preview is read-only in the strict sense: it calls the allocator with
`apply=false` and writes nothing but an audit line. An unauthorized principal
never reaches the allocator or the control-plane DB at all, so a denial cannot
be used to enumerate the queue.
## Initiation
`POST /api/v1/requests/apply` refuses in this order, and every refusal returns
before any assignment is attempted:
| Condition | Outcome | Status |
|-----------|---------|--------|
| Unparseable request | `invalid_request` | 400 |
| Not authorized, or execution not wired | `denied` | 403 |
| `confirm` not set | `denied` / `confirmation_required` | 409 |
| Work unit already claimed | `blocked` / `duplicate_assignment` | 409 |
| Allocator would select other work | `wait` / `not_next_safe_work` | 409 |
| Allocator declines on apply | `blocked` or `wait` | 409 |
| Evidence unavailable | `wait` / `evidence_unavailable` | 503 |
| Assigned | `assigned_work` | 201 |
A success returns the assignment plus a `handoff` block naming the profile, the
namespace, and the actions that stay forbidden — enough for the operator to
continue in the right MCP namespace without guessing.
### Why apply runs the allocator twice
The allocator is the only source of exclusive ownership (#600 / #613), and it
selects work; it does not take orders. So `apply` runs a dry-run first and
proceeds only when the allocator would independently pick the requested work
unit. If it would not, the request reports `wait` and mutates nothing.
A request is therefore a *confirmation* of the allocator's decision, never an
override of it. The apply call carries the dry-run's
`candidate_set_fingerprint` as a CAS pin (#776), so a queue that changed
between the two calls fails closed rather than assigning against a stale view.
The result is checked again on the way out: an assignment naming a different
work unit is not read as success.
### Fail-closed defaults
- An unreadable control-plane DB denies. It is never treated as "nothing holds
this work unit".
- An incomplete queue inventory denies (#758). Ranking a partial candidate set
can select the wrong work.
- An allocator that raises denies.
- PR work with no resolvable head SHA denies; a supplied SHA that no longer
matches denies with `head_moved`.
## Enabling initiation
Execution is wired off. Set `WEBUI_REQUESTS_EXECUTION=1` to enable it for the
`initiate_workflow` action only — see
[`webui-authz-audit.md`](webui-authz-audit.md) for why this is an
action-scoped flag rather than a phase bump. With the variable unset, `apply`
returns `403` with `reason_code: unauthorized` no matter who asks.
Enabling execution does **not** enable approvals or merges. Those are phase 3
console actions and remain forbidden in every path here; the console reserves
work and hands off, and the MCP role profile enforces what that role may then
do.
## Audit
Every preview and every apply emits a console audit record (schema in
[`webui-authz-audit.md`](webui-authz-audit.md)):
| Event | `result` |
|-------|----------|
| Preview | `previewed` |
| Refusal at any stage | `denied` |
| Assignment created | `succeeded` |
`correlation.request_id` carries the request's `correlation_id`, and a
successful record's `metadata` carries `assignment_id` and `lease_id`, so an
assignment can be traced back to the intent that produced it. The operator's
`intent_summary` travels in `metadata` and passes through the standard
redaction pass before persistence like every other field.
## Non-goals
- No browser-initiated approve or merge, in this phase or any other.
- No bypass of allocator exclusive ownership; no self-selection of work.
- No auto-start from raw monitoring incidents (#612 stays downstream).
+113
View File
@@ -2069,6 +2069,7 @@ import lease_policy # noqa: E402
import workflow_dashboard # noqa: E402 # #605 live queue/lease dashboard
import restart_coordinator # noqa: E402 # #658 MCP restart coordinator/impact
import drain_proof # noqa: E402 # #661 pre-restart drain proof and hard gate
import restart_audit # noqa: E402 # #665 restart audit events + incidents
import incident_bridge # noqa: E402
import sentry_observability # noqa: E402 (#606 optional Sentry observability)
import sentry_incident_bridge # noqa: E402 (#607 Sentry→Gitea incident bridge)
@@ -22750,6 +22751,38 @@ def gitea_request_mcp_restart(
# and a durable incident is raised. Break-glass is the only bypass and its
# authorization is read from the environment, never self-asserted.
payload["apply_supported"] = False
# #665: correlation id threads impact preview → apply gate → incidents.
correlation_id = restart_audit.new_correlation_id()
payload["correlation_id"] = correlation_id
auth_user = None
try:
# remote-only: host override is for operator diagnostics, not required here
auth_user = (gitea_whoami(remote=remote) or {}).get("username")
except Exception: # noqa: BLE001 — identity is best-effort for audit
auth_user = None
preview_audit = restart_audit.record_restart_lifecycle(
event_type=restart_audit.EVENT_IMPACT_PREVIEW,
outcome=str(report.verdict or "unknown"),
correlation_id=correlation_id,
remote=remote,
org=o,
repo=r,
requesting_session_id=sid,
restart_class=restart_class,
profile_name=profile_name,
authenticated_username=auth_user,
reasons=list(report.reasons or []),
details={
"dry_run": True,
"allow_restart": bool(report.allow_restart),
"inventory_complete": inventory_complete,
},
privileged=False,
)
payload["restart_audit"] = {
"correlation_id": correlation_id,
"impact_preview_written": preview_audit["audit_written"],
}
if not dry_run:
proof_obj: dict | None = None
proof_parse_error: str | None = None
@@ -22809,6 +22842,86 @@ def gitea_request_mcp_restart(
)
if not gate.allow and gate.incident is not None:
payload["incident"] = gate.incident
# #665: audit apply-gate + materialize durable incidents for failed
# drain and break-glass. Privileged apply denies if audit is enabled
# and the sink write fails.
incident_desc = restart_audit.incident_from_apply_gate(
gate_payload={
**gate_payload,
"incident": gate.incident,
"allow": gate.allow,
},
break_glass=break_glass,
correlation_id=correlation_id,
requesting_session_id=sid,
restart_class=restart_class,
remote=remote,
org=o,
repo=r,
)
if incident_desc is not None:
payload["incident"] = incident_desc
def _create_restart_incident_issue(
*, title, body, labels, org=None, repo=None, **_kw
):
return gitea_create_issue(
title=title,
body=body,
labels=labels,
remote=remote,
host=h,
org=org or o,
repo=repo or r,
)
apply_audit = restart_audit.record_restart_lifecycle(
event_type=(
restart_audit.EVENT_BREAK_GLASS
if break_glass
else restart_audit.EVENT_APPLY_GATE
),
outcome=(
"break_glass"
if break_glass
else ("allow" if payload["apply_authorized"] else "deny")
),
correlation_id=correlation_id,
remote=remote,
org=o,
repo=r,
requesting_session_id=sid,
restart_class=restart_class,
profile_name=profile_name,
authenticated_username=auth_user,
reasons=list(gate_payload.get("reasons") or []),
details={
"apply_authorized": payload["apply_authorized"],
"drain_gate_allow": gate_payload.get("drain_gate_allow"),
"restart_class_authorized": restart_class_authorized,
"break_glass": break_glass,
"proof_id": gate_payload.get("proof_id"),
},
privileged=True,
create_incident=incident_desc,
create_issue_fn=_create_restart_incident_issue
if incident_desc is not None
else None,
dry_run_incident=False,
)
payload["restart_audit"] = {
"correlation_id": correlation_id,
"impact_preview_written": preview_audit["audit_written"],
"apply_gate_written": apply_audit["audit_written"],
"incident_result": apply_audit.get("incident_result"),
}
if apply_audit["deny_reasons"]:
payload["apply_authorized"] = False
payload["reasons"] = list(payload.get("reasons") or []) + list(
apply_audit["deny_reasons"]
)
payload["success"] = True
return payload
+426
View File
@@ -0,0 +1,426 @@
"""MCP restart lifecycle audit events and incident materialization (#665).
Restarts and recovery attempts must leave a forensic trail: impact previews,
drain enter/exit, drain-proof results, apply gate verdicts, break-glass, and
post-restart reconcile outcomes. Failed drains and break-glass must also raise
durable Gitea incident issues so they cannot be silently repeated.
This module is the pure + sink layer for that trail:
* **Schema** — ``mcp.restart.*`` event names and a redacted payload builder.
* **Emission** — append-only via :mod:`gitea_audit` (off when ``GITEA_AUDIT_LOG``
is unset; privileged apply can still *require* a successful write).
* **Incidents** — descriptors for failed drain / break-glass / unguarded restart,
plus an optional materializer that creates a Gitea issue through an injected
``create_issue_fn`` (keeps this module free of network I/O in tests).
Design rules:
* **No secrets.** All free text is redacted before write or issue body assembly.
* **Never raises from emission.** ``emit_restart_event`` returns False on sink
failure so callers can decide fail-closed policy for privileged restarts.
* **Does not restart.** Audit never executes a process restart.
* **Drain proof stays #661.** This module records what the gate decided; it
does not re-verify proofs.
"""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from typing import Any, Callable, Mapping, Sequence
import gitea_audit
# ── Event vocabulary (stable identifiers for operators + tests) ───────────────
EVENT_IMPACT_PREVIEW = "mcp.restart.impact_preview"
EVENT_DRAIN_ENTER = "mcp.restart.drain_enter"
EVENT_DRAIN_EXIT = "mcp.restart.drain_exit"
EVENT_DRAIN_PROOF = "mcp.restart.drain_proof"
EVENT_APPLY_GATE = "mcp.restart.apply_gate"
EVENT_BREAK_GLASS = "mcp.restart.break_glass"
EVENT_POST_RESTART_RECONCILE = "mcp.restart.post_restart_reconcile"
EVENT_NARROWER_RECOVERY = "mcp.restart.narrower_recovery"
EVENT_UNGUARDED_DETECTED = "mcp.restart.unguarded_detected"
RESTART_EVENT_TYPES: frozenset[str] = frozenset(
{
EVENT_IMPACT_PREVIEW,
EVENT_DRAIN_ENTER,
EVENT_DRAIN_EXIT,
EVENT_DRAIN_PROOF,
EVENT_APPLY_GATE,
EVENT_BREAK_GLASS,
EVENT_POST_RESTART_RECONCILE,
EVENT_NARROWER_RECOVERY,
EVENT_UNGUARDED_DETECTED,
}
)
# Incident kinds (durable Gitea issues).
INCIDENT_FAILED_DRAIN = "restart_failed_drain"
INCIDENT_BREAK_GLASS = "restart_break_glass"
INCIDENT_RECONCILE_UNRESOLVED = "restart_reconcile_unresolved"
INCIDENT_UNGUARDED = "restart_unguarded_detected"
DEFAULT_INCIDENT_LABELS: tuple[str, ...] = (
"mcp-health",
"safety",
"observability",
"status:ready",
"type:bug",
"workflow-hardening",
)
CreateIssueFn = Callable[..., dict[str, Any]]
def _utc_now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def new_correlation_id() -> str:
"""Mint a short correlation id shared across a restart lifecycle."""
return f"rst-{uuid.uuid4().hex[:16]}"
def build_restart_event(
*,
event_type: str,
outcome: str,
correlation_id: str | None = None,
remote: str | None = None,
org: str | None = None,
repo: str | None = None,
requesting_session_id: str | None = None,
restart_class: str | None = None,
profile_name: str | None = None,
authenticated_username: str | None = None,
reasons: Sequence[str] | None = None,
details: Mapping[str, Any] | None = None,
now: str | None = None,
) -> dict[str, Any]:
"""Build a redacted ``mcp.restart.*`` audit event.
Raises ``ValueError`` on unknown event types so a typo cannot silently land
under a free-form action name.
"""
name = str(event_type or "").strip()
if name not in RESTART_EVENT_TYPES:
raise ValueError(
f"unknown restart audit event_type {name!r}; expected one of "
f"{sorted(RESTART_EVENT_TYPES)}"
)
redacted_reasons = [
gitea_audit.redact(str(r)) for r in (reasons or []) if str(r).strip()
]
redacted_details = gitea_audit.redact(dict(details or {}))
if not isinstance(redacted_details, dict):
redacted_details = {"value": redacted_details}
event = gitea_audit.build_event(
action=name,
result=str(outcome or "unknown"),
remote=remote,
repository=f"{org}/{repo}" if org and repo else None,
profile_name=profile_name,
authenticated_username=authenticated_username,
reason="; ".join(redacted_reasons) if redacted_reasons else None,
request_metadata={
"event_family": "mcp.restart",
"correlation_id": correlation_id or new_correlation_id(),
"restart_class": restart_class,
"requesting_session_id": requesting_session_id,
"org": org,
"repo": repo,
"details": redacted_details,
"reasons": redacted_reasons,
},
now=now or _utc_now_iso(),
operation=name,
)
event["action_type"] = "restart_lifecycle"
event["event_type"] = name
event["correlation_id"] = (event.get("request_metadata") or {}).get(
"correlation_id"
)
return event
def emit_restart_event(event: Mapping[str, Any], *, path: str | None = None) -> bool:
"""Append *event* to the audit sink. Never raises. Returns write success."""
try:
return bool(gitea_audit.write_event(dict(event), path=path))
except Exception:
return False
def require_audit_or_deny(
*,
privileged: bool,
written: bool,
audit_enabled: bool | None = None,
) -> list[str]:
"""Return deny reasons when a privileged restart path fails to audit.
When audit is not configured (``GITEA_AUDIT_LOG`` unset), privileged apply
still proceeds under the rollout policy "enable audit before enforcing
deny-on-audit-fail" — but *if* audit is enabled and the write fails,
privileged apply is denied (fail closed).
"""
enabled = (
gitea_audit.audit_enabled() if audit_enabled is None else bool(audit_enabled)
)
if not privileged:
return []
if not enabled:
return []
if written:
return []
return [
"privileged restart path requires a successful audit write; "
"audit sink failed (fail closed, #665)"
]
# ── Incident descriptors ──────────────────────────────────────────────────────
def build_incident_descriptor(
*,
kind: str,
reasons: Sequence[str],
correlation_id: str | None = None,
requesting_session_id: str | None = None,
restart_class: str | None = None,
remote: str | None = None,
org: str | None = None,
repo: str | None = None,
proof_id: str | None = None,
at: str | None = None,
) -> dict[str, Any]:
"""Build a durable incident descriptor (no network)."""
titles = {
INCIDENT_FAILED_DRAIN: "Restart denied: drain proof failed the hard gate",
INCIDENT_BREAK_GLASS: "Break-glass MCP restart authorized",
INCIDENT_RECONCILE_UNRESOLVED: "Post-restart reconcile left unresolved work",
INCIDENT_UNGUARDED: "Unguarded MCP restart attempt detected",
}
title = titles.get(kind, f"MCP restart incident ({kind})")
redacted_reasons = [
gitea_audit.redact(str(r)) for r in reasons if str(r).strip()
]
return {
"kind": kind,
"title": title,
"labels": list(DEFAULT_INCIDENT_LABELS),
"reasons": redacted_reasons,
"correlation_id": correlation_id,
"requesting_session_id": requesting_session_id,
"restart_class": restart_class,
"remote": remote,
"org": org,
"repo": repo,
"proof_id": proof_id,
"at": at or _utc_now_iso(),
"source": "restart_audit#665",
}
def incident_body(descriptor: Mapping[str, Any]) -> str:
"""Render a redacted markdown body for a Gitea incident issue."""
reasons = descriptor.get("reasons") or []
reason_lines = "\n".join(f"- {gitea_audit.redact(str(r))}" for r in reasons) or (
"- (no reasons recorded)"
)
return "\n".join(
[
"<!-- mcp-restart-incident:v1 -->",
f"## MCP restart incident (`{descriptor.get('kind')}`)",
"",
f"**Correlation:** `{descriptor.get('correlation_id') or 'none'}`",
f"**Session:** `{descriptor.get('requesting_session_id') or 'none'}`",
f"**Class:** `{descriptor.get('restart_class') or 'none'}`",
f"**Scope:** `{descriptor.get('remote')}/{descriptor.get('org')}/"
f"{descriptor.get('repo')}`",
f"**At:** `{descriptor.get('at')}`",
f"**Proof id:** `{descriptor.get('proof_id') or 'none'}`",
"",
"### Reasons",
reason_lines,
"",
"### Operator next steps",
"- Treat this as durable follow-up work under the restart-governance umbrella (#655).",
"- Do not invent a second restart path; use sanctioned coordinator tools only.",
"- Raw secrets must never appear in this issue (already redacted).",
"",
f"_Source: {descriptor.get('source')}_",
]
)
def materialize_incident(
descriptor: Mapping[str, Any],
*,
create_issue_fn: CreateIssueFn | None,
dry_run: bool = False,
) -> dict[str, Any]:
"""Create a Gitea issue from *descriptor* when *create_issue_fn* is provided.
Returns a result dict with ``created`` / ``issue_number`` / ``dry_run`` /
``reasons``. Never raises.
"""
base: dict[str, Any] = {
"created": False,
"dry_run": bool(dry_run),
"issue_number": None,
"kind": descriptor.get("kind"),
"reasons": [],
"descriptor": dict(descriptor),
}
if dry_run:
base["reasons"] = ["dry-run only; no Gitea issue created"]
return base
if create_issue_fn is None:
base["reasons"] = [
"create_issue_fn not provided; incident descriptor retained only"
]
return base
try:
result = create_issue_fn(
title=str(descriptor.get("title") or "MCP restart incident"),
body=incident_body(descriptor),
labels=list(descriptor.get("labels") or DEFAULT_INCIDENT_LABELS),
org=descriptor.get("org"),
repo=descriptor.get("repo"),
)
number = None
if isinstance(result, dict):
number = result.get("number") or result.get("issue_number")
if number is not None:
base["created"] = True
base["issue_number"] = int(number)
base["reasons"] = [f"created incident issue #{int(number)}"]
else:
base["reasons"] = ["create_issue_fn returned no issue number"]
except Exception as exc: # noqa: BLE001 — never break restart path here
base["reasons"] = [
f"incident issue creation failed: {gitea_audit.redact(str(exc))}"
]
return base
def record_restart_lifecycle(
*,
event_type: str,
outcome: str,
correlation_id: str,
remote: str | None = None,
org: str | None = None,
repo: str | None = None,
requesting_session_id: str | None = None,
restart_class: str | None = None,
profile_name: str | None = None,
authenticated_username: str | None = None,
reasons: Sequence[str] | None = None,
details: Mapping[str, Any] | None = None,
privileged: bool = False,
create_incident: Mapping[str, Any] | None = None,
create_issue_fn: CreateIssueFn | None = None,
dry_run_incident: bool = False,
audit_path: str | None = None,
) -> dict[str, Any]:
"""Emit one restart audit event and optionally materialize an incident.
Returns ``{event, audit_written, deny_reasons, incident_result}``.
"""
event = build_restart_event(
event_type=event_type,
outcome=outcome,
correlation_id=correlation_id,
remote=remote,
org=org,
repo=repo,
requesting_session_id=requesting_session_id,
restart_class=restart_class,
profile_name=profile_name,
authenticated_username=authenticated_username,
reasons=reasons,
details=details,
)
written = emit_restart_event(event, path=audit_path)
deny = require_audit_or_deny(privileged=privileged, written=written)
incident_result = None
if create_incident is not None:
incident_result = materialize_incident(
create_incident,
create_issue_fn=create_issue_fn,
dry_run=dry_run_incident,
)
return {
"event": event,
"audit_written": written,
"deny_reasons": deny,
"incident_result": incident_result,
"correlation_id": correlation_id,
}
def incident_from_apply_gate(
*,
gate_payload: Mapping[str, Any],
break_glass: bool,
correlation_id: str,
requesting_session_id: str | None,
restart_class: str | None,
remote: str | None,
org: str | None,
repo: str | None,
) -> dict[str, Any] | None:
"""Choose an incident descriptor from an apply-gate payload, if required."""
reasons = list(gate_payload.get("reasons") or [])
proof_id = gate_payload.get("proof_id")
if break_glass:
return build_incident_descriptor(
kind=INCIDENT_BREAK_GLASS,
reasons=reasons
or ["break-glass restart path used; durable incident required (#665)"],
correlation_id=correlation_id,
requesting_session_id=requesting_session_id,
restart_class=restart_class,
remote=remote,
org=org,
repo=repo,
proof_id=proof_id if isinstance(proof_id, str) else None,
)
# Failed drain / deny path.
incident = gate_payload.get("incident")
if isinstance(incident, Mapping) and incident:
# Normalize gate-provided descriptor into our schema.
return build_incident_descriptor(
kind=INCIDENT_FAILED_DRAIN,
reasons=list(incident.get("reasons") or reasons),
correlation_id=correlation_id,
requesting_session_id=requesting_session_id
or incident.get("requesting_session_id"),
restart_class=restart_class,
remote=remote,
org=org,
repo=repo,
proof_id=incident.get("proof_id") or proof_id,
at=incident.get("at"),
)
if not gate_payload.get("allow") and not gate_payload.get("drain_gate_allow", True):
return build_incident_descriptor(
kind=INCIDENT_FAILED_DRAIN,
reasons=reasons or ["restart apply denied"],
correlation_id=correlation_id,
requesting_session_id=requesting_session_id,
restart_class=restart_class,
remote=remote,
org=org,
repo=repo,
proof_id=proof_id if isinstance(proof_id, str) else None,
)
return None
-158
View File
@@ -7,7 +7,6 @@ import tempfile
import threading
import unittest
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
from allocator_service import (
OUTCOME_ASSIGNED,
@@ -16,7 +15,6 @@ from allocator_service import (
OUTCOME_PREVIEW,
OUTCOME_WAIT,
WorkCandidate,
_drop_expired_claims,
allocate_next_work,
candidate_from_dict,
classify_skip,
@@ -364,161 +362,5 @@ class AllocatorServiceTest(unittest.TestCase):
self.assertIn("unavailable", res["reasons"][0].lower())
class SideEffectFreeAllocationTest(unittest.TestCase):
"""``side_effect_free`` dry runs write nothing to the control plane (#643).
A plain ``apply=False`` still called ``upsert_session`` and
``expire_stale_leases`` before the apply branch was consulted, so a caller
advertising a read-only preview mutated on every call — one unreferenced
session row per preview, plus a global lease sweep.
"""
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="s-preview",
role="author",
remote="prgs",
org="org",
repo="repo",
candidates=[
WorkCandidate(kind="issue", number=643, labels=("status:ready",))
],
apply=False,
profile_name="prgs-author",
username="jcwalker3",
)
defaults.update(kwargs)
return allocate_next_work(**defaults)
def _session_ids(self) -> set[str]:
return {str(r.get("session_id")) for r in self.db.list_sessions()}
def test_side_effect_free_preview_writes_no_session_row(self):
before = self._session_ids()
result = self._alloc(side_effect_free=True)
self.assertEqual(result["outcome"], OUTCOME_PREVIEW)
self.assertEqual(self._session_ids(), before)
self.assertNotIn("s-preview", self._session_ids())
def test_plain_dry_run_still_registers_a_session(self):
# The default is unchanged for every existing caller.
self._alloc()
self.assertIn("s-preview", self._session_ids())
def test_repeated_previews_do_not_accumulate_rows(self):
for index in range(5):
self._alloc(side_effect_free=True, session_id=f"s-{index}")
self.assertEqual(self._session_ids(), set())
def test_side_effect_free_does_not_sweep_stale_leases(self):
self.db.upsert_session(session_id="owner", role="author", pid=1)
assigned = self.db.assign_and_lease(
session_id="owner",
role="author",
remote="prgs",
org="org",
repo="repo",
kind="issue",
number=999,
lease_ttl_seconds=-60, # already expired
)
self.assertEqual(assigned.outcome, "assigned")
self._alloc(side_effect_free=True)
# The expired row is still 'active' in the DB: nothing swept it.
statuses = {
r["lease_id"]: r["status"]
for r in self.db.list_leases(
remote="prgs", org="org", repo="repo",
statuses=("active", "expired"),
)
}
self.assertEqual(statuses.get(assigned.lease_id), "active")
def test_expired_claims_are_filtered_in_memory_so_work_stays_selectable(self):
"""The read-only mirror of the sweep: expired claims must not block."""
self.db.upsert_session(session_id="owner", role="author", pid=1)
self.db.assign_and_lease(
session_id="owner",
role="author",
remote="prgs",
org="org",
repo="repo",
kind="issue",
number=643,
lease_ttl_seconds=-60, # expired: must not withhold #643
)
result = self._alloc(side_effect_free=True)
self.assertEqual(result["outcome"], OUTCOME_PREVIEW)
self.assertEqual(result["selected"]["number"], 643)
def test_a_live_claim_still_withholds_the_work(self):
self.db.upsert_session(session_id="owner", role="author", pid=1)
self.db.assign_and_lease(
session_id="owner",
role="author",
remote="prgs",
org="org",
repo="repo",
kind="issue",
number=643,
lease_ttl_seconds=3600,
)
result = self._alloc(side_effect_free=True)
self.assertNotEqual(result["outcome"], OUTCOME_ASSIGNED)
self.assertNotEqual((result.get("selected") or {}).get("number"), 643)
def test_side_effect_free_with_apply_fails_closed(self):
result = self._alloc(side_effect_free=True, apply=True)
self.assertFalse(result["success"])
self.assertEqual(result["outcome"], OUTCOME_NO_SAFE)
self.assertIsNone(result["assignment"])
self.assertIn("incompatible with apply", result["reasons"][0])
# And it reserved nothing.
self.assertEqual(
self.db.list_leases(remote="prgs", org="org", repo="repo"), []
)
class DropExpiredClaimsTest(unittest.TestCase):
"""The in-memory expiry filter behind side-effect-free previews (#643)."""
def test_unparseable_expiry_is_kept_rather_than_assumed_free(self):
claims = {
("issue", 1): {"lease_id": "l1", "expires_at": "not-a-date"},
("issue", 2): {"lease_id": "l2"},
("issue", 3): {"lease_id": "l3", "expires_at": None},
}
self.assertEqual(_drop_expired_claims(claims), claims)
def test_expired_dropped_and_future_kept(self):
now = datetime(2026, 7, 25, 12, 0, tzinfo=timezone.utc)
claims = {
("issue", 1): {"expires_at": "2026-07-25T11:59:59+00:00"},
("issue", 2): {"expires_at": "2026-07-25T12:00:01+00:00"},
("issue", 3): {"expires_at": "2026-07-25T12:00:00+00:00"}, # boundary
}
kept = _drop_expired_claims(claims, now=now)
self.assertEqual(set(kept), {("issue", 2)})
def test_naive_and_zulu_timestamps_are_treated_as_utc(self):
now = datetime(2026, 7, 25, 12, 0, tzinfo=timezone.utc)
claims = {
("issue", 1): {"expires_at": "2026-07-25T11:00:00"}, # naive, past
("issue", 2): {"expires_at": "2026-07-25T13:00:00Z"}, # zulu, future
}
kept = _drop_expired_claims(claims, now=now)
self.assertEqual(set(kept), {("issue", 2)})
if __name__ == "__main__":
unittest.main()
+342
View File
@@ -0,0 +1,342 @@
"""Tests for MCP restart lifecycle audit events and incidents (#665)."""
from __future__ import annotations
import json
import os
import tempfile
import unittest
from unittest.mock import patch
import gitea_audit
import restart_audit as ra
class TestEventSchema(unittest.TestCase):
def test_all_lifecycle_event_types_are_named(self):
expected = {
"mcp.restart.impact_preview",
"mcp.restart.drain_enter",
"mcp.restart.drain_exit",
"mcp.restart.drain_proof",
"mcp.restart.apply_gate",
"mcp.restart.break_glass",
"mcp.restart.post_restart_reconcile",
"mcp.restart.narrower_recovery",
"mcp.restart.unguarded_detected",
}
self.assertEqual(set(ra.RESTART_EVENT_TYPES), expected)
def test_build_restart_event_core_fields(self):
event = ra.build_restart_event(
event_type=ra.EVENT_IMPACT_PREVIEW,
outcome="safe",
correlation_id="rst-abc123",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
requesting_session_id="sess-1",
restart_class="full_mcp_restart",
profile_name="prgs-author",
authenticated_username="bot",
reasons=["inventory complete"],
details={"allow_restart": True},
now="2026-07-25T12:00:00+00:00",
)
self.assertEqual(event["event_type"], ra.EVENT_IMPACT_PREVIEW)
self.assertEqual(event["action"], ra.EVENT_IMPACT_PREVIEW)
self.assertEqual(event["action_type"], "restart_lifecycle")
self.assertEqual(event["result"], "safe")
self.assertEqual(event["correlation_id"], "rst-abc123")
self.assertEqual(event["profile_name"], "prgs-author")
self.assertEqual(event["authenticated_username"], "bot")
meta = event["request_metadata"]
self.assertEqual(meta["event_family"], "mcp.restart")
self.assertEqual(meta["correlation_id"], "rst-abc123")
self.assertEqual(meta["restart_class"], "full_mcp_restart")
self.assertEqual(meta["details"]["allow_restart"], True)
def test_unknown_event_type_raises(self):
with self.assertRaises(ValueError) as ctx:
ra.build_restart_event(
event_type="mcp.restart.not_a_real_event",
outcome="x",
correlation_id="rst-1",
)
self.assertIn("unknown restart audit event_type", str(ctx.exception))
def test_reasons_and_details_are_redacted(self):
event = ra.build_restart_event(
event_type=ra.EVENT_APPLY_GATE,
outcome="deny",
correlation_id="rst-sec",
reasons=["token secret-xyz rejected", "ok"],
details={"token": "leak-token", "status": "denied"},
)
self.assertNotIn("secret-xyz", event.get("reason") or "")
meta = event["request_metadata"]
self.assertEqual(meta["details"]["token"], gitea_audit.REDACTED)
self.assertEqual(meta["details"]["status"], "denied")
for reason in meta["reasons"]:
self.assertNotIn("secret-xyz", reason)
def test_new_correlation_id_shape(self):
cid = ra.new_correlation_id()
self.assertTrue(cid.startswith("rst-"))
self.assertEqual(len(cid), len("rst-") + 16)
class TestEmitAndRequire(unittest.TestCase):
def test_emit_appends_json_line(self):
with tempfile.TemporaryDirectory() as d:
path = os.path.join(d, "audit.log")
event = ra.build_restart_event(
event_type=ra.EVENT_DRAIN_PROOF,
outcome="pass",
correlation_id="rst-write",
)
self.assertTrue(ra.emit_restart_event(event, path=path))
with open(path, encoding="utf-8") as fh:
lines = fh.read().splitlines()
self.assertEqual(len(lines), 1)
loaded = json.loads(lines[0])
self.assertEqual(loaded["event_type"], ra.EVENT_DRAIN_PROOF)
self.assertEqual(loaded["correlation_id"], "rst-write")
def test_emit_never_raises(self):
self.assertFalse(
ra.emit_restart_event({"action": "x"}, path="/no/such/dir/audit.log")
)
def test_require_audit_denies_privileged_when_write_fails_and_enabled(self):
deny = ra.require_audit_or_deny(
privileged=True, written=False, audit_enabled=True
)
self.assertEqual(len(deny), 1)
self.assertIn("fail closed", deny[0])
def test_require_audit_allows_when_audit_disabled(self):
# Rollout policy: enable audit before enforcing deny-on-audit-fail.
deny = ra.require_audit_or_deny(
privileged=True, written=False, audit_enabled=False
)
self.assertEqual(deny, [])
def test_require_audit_noop_for_non_privileged(self):
deny = ra.require_audit_or_deny(
privileged=False, written=False, audit_enabled=True
)
self.assertEqual(deny, [])
def test_require_audit_allows_when_written(self):
deny = ra.require_audit_or_deny(
privileged=True, written=True, audit_enabled=True
)
self.assertEqual(deny, [])
class TestIncidents(unittest.TestCase):
def test_break_glass_descriptor(self):
desc = ra.build_incident_descriptor(
kind=ra.INCIDENT_BREAK_GLASS,
reasons=["break-glass authorized"],
correlation_id="rst-bg",
requesting_session_id="s1",
restart_class="full_mcp_restart",
remote="prgs",
org="O",
repo="R",
)
self.assertEqual(desc["kind"], ra.INCIDENT_BREAK_GLASS)
self.assertIn("Break-glass", desc["title"])
self.assertIn("mcp-health", desc["labels"])
self.assertEqual(desc["source"], "restart_audit#665")
def test_incident_body_redacts_and_includes_correlation(self):
desc = ra.build_incident_descriptor(
kind=ra.INCIDENT_FAILED_DRAIN,
reasons=["token secret-xyz failed proof"],
correlation_id="rst-body",
remote="prgs",
org="O",
repo="R",
proof_id="proof-1",
)
body = ra.incident_body(desc)
self.assertIn("rst-body", body)
self.assertIn("proof-1", body)
self.assertIn("mcp-restart-incident:v1", body)
self.assertNotIn("secret-xyz", body)
def test_materialize_dry_run(self):
desc = ra.build_incident_descriptor(
kind=ra.INCIDENT_FAILED_DRAIN,
reasons=["denied"],
correlation_id="rst-dr",
)
result = ra.materialize_incident(desc, create_issue_fn=lambda **k: {}, dry_run=True)
self.assertFalse(result["created"])
self.assertTrue(result["dry_run"])
self.assertIn("dry-run", result["reasons"][0])
def test_materialize_without_create_fn(self):
desc = ra.build_incident_descriptor(
kind=ra.INCIDENT_FAILED_DRAIN,
reasons=["denied"],
correlation_id="rst-nfn",
)
result = ra.materialize_incident(desc, create_issue_fn=None)
self.assertFalse(result["created"])
self.assertIn("create_issue_fn not provided", result["reasons"][0])
def test_materialize_creates_issue(self):
created = {}
def _create(*, title, body, labels, org=None, repo=None, **_kw):
created["title"] = title
created["body"] = body
created["labels"] = labels
created["org"] = org
created["repo"] = repo
return {"number": 999}
desc = ra.build_incident_descriptor(
kind=ra.INCIDENT_BREAK_GLASS,
reasons=["break-glass"],
correlation_id="rst-create",
org="O",
repo="R",
)
result = ra.materialize_incident(desc, create_issue_fn=_create)
self.assertTrue(result["created"])
self.assertEqual(result["issue_number"], 999)
self.assertIn("Break-glass", created["title"])
self.assertIn("rst-create", created["body"])
self.assertEqual(created["org"], "O")
def test_materialize_never_raises_on_create_failure(self):
def _boom(**_kw):
raise RuntimeError("token secret-xyz network")
desc = ra.build_incident_descriptor(
kind=ra.INCIDENT_FAILED_DRAIN,
reasons=["x"],
correlation_id="rst-boom",
)
result = ra.materialize_incident(desc, create_issue_fn=_boom)
self.assertFalse(result["created"])
self.assertIn("failed", result["reasons"][0])
self.assertNotIn("secret-xyz", result["reasons"][0])
class TestIncidentFromApplyGate(unittest.TestCase):
def test_break_glass_always_incident(self):
desc = ra.incident_from_apply_gate(
gate_payload={"allow": True, "reasons": [], "proof_id": None},
break_glass=True,
correlation_id="rst-bg2",
requesting_session_id="s",
restart_class="full_mcp_restart",
remote="prgs",
org="O",
repo="R",
)
self.assertIsNotNone(desc)
self.assertEqual(desc["kind"], ra.INCIDENT_BREAK_GLASS)
def test_failed_drain_from_gate_incident(self):
desc = ra.incident_from_apply_gate(
gate_payload={
"allow": False,
"drain_gate_allow": False,
"reasons": ["proof expired"],
"incident": {
"reasons": ["proof expired"],
"proof_id": "p1",
},
},
break_glass=False,
correlation_id="rst-fd",
requesting_session_id="s",
restart_class="full_mcp_restart",
remote="prgs",
org="O",
repo="R",
)
self.assertIsNotNone(desc)
self.assertEqual(desc["kind"], ra.INCIDENT_FAILED_DRAIN)
self.assertEqual(desc["proof_id"], "p1")
def test_allow_without_break_glass_no_incident(self):
desc = ra.incident_from_apply_gate(
gate_payload={
"allow": True,
"drain_gate_allow": True,
"reasons": [],
},
break_glass=False,
correlation_id="rst-ok",
requesting_session_id="s",
restart_class="full_mcp_restart",
remote="prgs",
org="O",
repo="R",
)
self.assertIsNone(desc)
class TestRecordLifecycle(unittest.TestCase):
def test_record_emits_and_materializes(self):
created = []
def _create(**kwargs):
created.append(kwargs)
return {"number": 42}
with tempfile.TemporaryDirectory() as d:
path = os.path.join(d, "audit.log")
with patch.dict(os.environ, {"GITEA_AUDIT_LOG": path}, clear=False):
incident = ra.build_incident_descriptor(
kind=ra.INCIDENT_BREAK_GLASS,
reasons=["bg"],
correlation_id="rst-lc",
org="O",
repo="R",
)
out = ra.record_restart_lifecycle(
event_type=ra.EVENT_BREAK_GLASS,
outcome="break_glass",
correlation_id="rst-lc",
remote="prgs",
org="O",
repo="R",
privileged=True,
create_incident=incident,
create_issue_fn=_create,
audit_path=path,
)
self.assertTrue(out["audit_written"])
self.assertEqual(out["deny_reasons"], [])
self.assertTrue(out["incident_result"]["created"])
self.assertEqual(out["incident_result"]["issue_number"], 42)
self.assertEqual(len(created), 1)
def test_privileged_deny_when_audit_write_fails(self):
with patch.dict(
os.environ, {"GITEA_AUDIT_LOG": "/no/such/dir/a.log"}, clear=False
):
with patch("restart_audit.emit_restart_event", return_value=False):
with patch("gitea_audit.audit_enabled", return_value=True):
out = ra.record_restart_lifecycle(
event_type=ra.EVENT_APPLY_GATE,
outcome="deny",
correlation_id="rst-deny",
privileged=True,
audit_path="/no/such/dir/a.log",
)
self.assertFalse(out["audit_written"])
self.assertEqual(len(out["deny_reasons"]), 1)
if __name__ == "__main__":
unittest.main()
-511
View File
@@ -1,511 +0,0 @@
"""Tests for the Gitea issue↔PR linkage console (#645, Phase 3).
Covers the acceptance criteria of the issue:
* AC1 — issue↔PR linkage is visible for the selected project/repo, in both
directions, with the evidence that produced each edge.
* AC2 — the latest canonical handoff (CTH) is summarized for a focused thread.
* AC3 — an external Gitea link appears only under the admin reveal opt-in.
* AC4 — every case is driven by mocked Gitea payloads; no network.
Plus the invariants this console must not violate: a partial or failed read is
never rendered as "no link exists", an unfetched thread is never rendered as
"no handoff", redaction happens before display, and the surface stays read-only.
"""
from __future__ import annotations
import json
import os
import sys
import unittest
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from tests.webui_testclient import TestClient
from canonical_thread_handoff import format_cth_body
from webui.app import create_app
from webui.linkage_loader import (
EVIDENCE_BRANCH,
EVIDENCE_CLOSES,
EVIDENCE_REFERENCE,
HANDOFF_LOADED,
HANDOFF_NOT_LOADED,
HANDOFF_UNAVAILABLE,
LinkageSnapshot,
load_linkage_snapshot,
resolve_linkage,
resolve_pr_links,
snapshot_to_dict,
summarize_handoff,
)
from webui.linkage_views import render_linkage_page
from webui.nav import nav_hrefs
from webui.queue_loader import PaginationMeta
def _pagination(*, complete: bool = True, count: int = 0) -> PaginationMeta:
return PaginationMeta(
page=1,
per_page=50,
returned_count=count,
has_more=not complete,
is_final_page=complete,
inventory_complete=complete,
pages_fetched=1,
)
def _pr(
number: int,
*,
title: str = "",
body: str = "",
head: str = "",
state: str = "open",
labels: tuple[str, ...] = (),
) -> dict:
return {
"number": number,
"title": title or f"pr {number}",
"body": body,
"state": state,
"head": {"ref": head},
"labels": [{"name": name} for name in labels],
}
def _issue(
number: int,
*,
title: str = "",
state: str = "open",
labels: tuple[str, ...] = (),
) -> dict:
return {
"number": number,
"title": title or f"issue {number}",
"state": state,
"labels": [{"name": name} for name in labels],
}
def _fetcher(items: list[dict], *, complete: bool = True):
def _fetch(*_args, **_kwargs):
return items, _pagination(complete=complete, count=len(items))
return _fetch
def _load(
issues: list[dict],
prs: list[dict],
*,
complete: bool = True,
**kwargs,
) -> LinkageSnapshot:
return load_linkage_snapshot(
fetch_prs=_fetcher(prs, complete=complete),
fetch_issues=_fetcher(issues, complete=complete),
**kwargs,
)
def _cth(comment_id: int, *, created_at: str, status: str, next_owner: str) -> dict:
return {
"id": comment_id,
"created_at": created_at,
"user": {"login": "jcwalker3"},
"body": format_cth_body(
cth_type="Author Handoff",
status=status,
next_owner=next_owner,
current_blocker="none",
decision="implemented",
proof="full suite green",
next_action="review PR",
ready_to_paste_prompt="Review PR #902 now.",
),
}
class TestLinkageEvidence(unittest.TestCase):
"""AC1 — every edge records how it was found, and keeps all candidates."""
def test_closes_keyword_in_body_is_strongest_evidence(self):
links = resolve_pr_links(_pr(902, body="Closes #643"))
self.assertEqual([link.issue_number for link in links], [643])
self.assertEqual(links[0].evidence, (EVIDENCE_CLOSES,))
self.assertTrue(links[0].closes)
def test_closes_keyword_in_title_counts(self):
links = resolve_pr_links(_pr(902, title="feat(webui): preview (Closes #643)"))
self.assertEqual(links[0].evidence, (EVIDENCE_CLOSES,))
def test_canonical_branch_marker_links_without_a_keyword(self):
links = resolve_pr_links(_pr(902, head="feat/issue-643-request-preview"))
self.assertEqual([link.issue_number for link in links], [643])
self.assertEqual(links[0].evidence, (EVIDENCE_BRANCH,))
self.assertFalse(links[0].closes)
def test_non_canonical_branch_is_not_treated_as_a_marker(self):
self.assertEqual(resolve_pr_links(_pr(902, head="issue-643-preview")), ())
def test_bare_mention_is_recorded_as_the_weakest_evidence(self):
links = resolve_pr_links(_pr(902, body="context in #643"))
self.assertEqual(links[0].evidence, (EVIDENCE_REFERENCE,))
self.assertFalse(links[0].closes)
def test_several_evidence_kinds_merge_onto_one_edge(self):
links = resolve_pr_links(
_pr(902, body="Closes #643 — see #643", head="feat/issue-643-preview")
)
self.assertEqual(len(links), 1)
self.assertEqual(
links[0].evidence,
(EVIDENCE_CLOSES, EVIDENCE_BRANCH, EVIDENCE_REFERENCE),
)
def test_stronger_evidence_sorts_first(self):
links = resolve_pr_links(_pr(902, body="Closes #643, related #700"))
self.assertEqual([link.issue_number for link in links], [643, 700])
def test_self_reference_is_not_linkage(self):
links = resolve_pr_links(_pr(902, body="supersedes #902"))
self.assertEqual(links, ())
def test_every_candidate_is_kept_never_collapsed_to_a_guess(self):
links = resolve_pr_links(_pr(902, body="Closes #643\nCloses #644"))
self.assertEqual([link.issue_number for link in links], [643, 644])
class TestLinkageIndex(unittest.TestCase):
def test_issue_direction_ignores_mention_only_edges(self):
index = resolve_linkage([_pr(902, body="context in #643")])
self.assertIsNone(index.issue_prs.get(643))
self.assertEqual(index.pr_links[902][0].evidence, (EVIDENCE_REFERENCE,))
def test_contested_issue_is_reported_when_two_prs_claim_it(self):
index = resolve_linkage(
[_pr(902, body="Closes #643"), _pr(903, head="feat/issue-643-again")]
)
self.assertEqual(index.contested_issues(), (643,))
self.assertEqual(index.issue_prs[643], (902, 903))
def test_single_claim_is_not_contested(self):
index = resolve_linkage([_pr(902, body="Closes #643")])
self.assertEqual(index.contested_issues(), ())
def test_ambiguous_when_two_issues_tie_at_the_strongest_evidence(self):
index = resolve_linkage([_pr(902, body="Closes #643\nCloses #644")])
self.assertTrue(index.ambiguous(902))
def test_weaker_candidate_alongside_a_stronger_one_is_not_ambiguous(self):
index = resolve_linkage([_pr(902, body="Closes #643, see #700")])
self.assertFalse(index.ambiguous(902))
self.assertEqual(index.primary_issue(902).issue_number, 643)
def test_malformed_pr_row_is_skipped_not_raised_on(self):
index = resolve_linkage([{"title": "no number"}, _pr(902, body="Closes #643")])
self.assertEqual(sorted(index.pr_links), [902])
class TestLinkageSnapshot(unittest.TestCase):
"""AC1 — linkage is visible per project/repo, in both directions."""
def test_both_directions_are_populated(self):
snapshot = _load([_issue(643)], [_pr(902, body="Closes #643")])
self.assertTrue(snapshot.ok)
self.assertEqual([node.number for node in snapshot.issues], [643])
self.assertEqual(snapshot.issues[0].linked_prs, (902,))
self.assertEqual(snapshot.prs[0].links[0].issue_number, 643)
def test_repo_scope_comes_from_the_registry_project(self):
snapshot = _load([], [])
self.assertIn("/", snapshot.repo_label)
self.assertTrue(snapshot.project_id)
def test_unknown_project_fails_closed_with_a_reason(self):
snapshot = _load([_issue(643)], [], project_id="no-such-project")
self.assertFalse(snapshot.ok)
self.assertIn("not found in registry", snapshot.fetch_error)
self.assertEqual(snapshot.issues, ())
def test_orphan_pr_is_identifiable(self):
snapshot = _load([], [_pr(902), _pr(903, body="Closes #643")])
self.assertEqual([node.number for node in snapshot.orphan_prs], [902])
def test_state_scope_defaults_to_open_and_is_reported(self):
self.assertEqual(_load([], []).state_scope, "open")
self.assertEqual(_load([], [], state="all").state_scope, "all")
def test_unsupported_state_falls_back_to_open(self):
self.assertEqual(_load([], [], state="../etc").state_scope, "open")
def test_state_is_passed_through_to_the_fetchers(self):
seen: list[str] = []
def _fetch(*_args, **kwargs):
seen.append(kwargs.get("state", ""))
return [], _pagination()
load_linkage_snapshot(state="all", fetch_prs=_fetch, fetch_issues=_fetch)
self.assertEqual(seen, ["all", "all"])
class TestPartialInventoryIsNotAnAbsenceClaim(unittest.TestCase):
"""An empty edge list from a partial read must never read as 'no link'."""
def test_incomplete_pagination_marks_links_non_authoritative(self):
snapshot = _load([_issue(643)], [], complete=False)
self.assertFalse(snapshot.inventory_complete)
self.assertFalse(snapshot.issues[0].links_authoritative)
def test_complete_pagination_marks_links_authoritative(self):
snapshot = _load([_issue(643)], [], complete=True)
self.assertTrue(snapshot.inventory_complete)
self.assertTrue(snapshot.issues[0].links_authoritative)
def test_partial_window_renders_a_qualified_empty_cell(self):
html = render_linkage_page(_load([_issue(643)], [], complete=False))
self.assertIn("none found (partial inventory)", html)
def test_complete_window_renders_a_plain_none(self):
html = render_linkage_page(_load([_issue(643)], [], complete=True))
self.assertNotIn("partial inventory", html)
self.assertIn(">none<", html)
def test_missing_credentials_fail_closed_without_a_table(self):
with mock.patch(
"webui.linkage_loader._offline_test_mode", return_value=False
), mock.patch("webui.linkage_loader.get_auth_header", return_value=""):
snapshot = load_linkage_snapshot()
self.assertFalse(snapshot.ok)
self.assertIn("credentials unavailable", snapshot.fetch_error)
html = render_linkage_page(snapshot)
self.assertIn("Linkage unavailable", html)
self.assertNotIn("Issues → pull requests", html)
def test_fetch_failure_is_reported_not_raised(self):
def _boom(*_args, **_kwargs):
raise RuntimeError("gitea 502")
snapshot = load_linkage_snapshot(fetch_prs=_boom, fetch_issues=_boom)
self.assertFalse(snapshot.ok)
self.assertIn("Gitea fetch failed", snapshot.fetch_error)
class TestHandoffSummary(unittest.TestCase):
"""AC2 — the latest canonical handoff is summarized for a focused thread."""
def test_latest_cth_wins(self):
summary = summarize_handoff([
_cth(1, created_at="2026-07-24T10:00:00Z", status="in progress",
next_owner="author"),
_cth(2, created_at="2026-07-25T10:00:00Z", status="PR-open",
next_owner="reviewer"),
])
self.assertEqual(summary.comment_id, 2)
self.assertEqual(summary.status, "PR-open")
self.assertEqual(summary.next_owner, "reviewer")
self.assertTrue(summary.cth_type_known)
def test_thread_without_a_cth_summarizes_to_none(self):
self.assertIsNone(summarize_handoff([{"id": 1, "body": "ordinary comment"}]))
def test_unknown_heading_is_reported_not_republished(self):
summary = summarize_handoff([
{
"id": 5,
"created_at": "2026-07-25T10:00:00Z",
"user": {"login": "someone"},
"body": "<!-- cth:v1 -->\n## CTH: Totally Made Up\n\nStatus: odd\n",
}
])
self.assertFalse(summary.cth_type_known)
self.assertEqual(summary.cth_type, "unrecognized")
self.assertNotIn("Totally Made Up", json.dumps(summary.to_dict()))
def test_focused_pr_loads_its_handoff(self):
snapshot = _load(
[_issue(643)],
[_pr(902, body="Closes #643")],
pr=902,
comment_source=lambda kind, number: [
_cth(2, created_at="2026-07-25T10:00:00Z", status="PR-open",
next_owner="reviewer")
],
)
self.assertEqual(snapshot.handoff_status.state, HANDOFF_LOADED)
self.assertEqual(snapshot.focus, ("pr", 902))
self.assertEqual(snapshot.prs[0].handoff.status, "PR-open")
def test_unfocused_rows_report_not_loaded_never_none(self):
snapshot = _load(
[_issue(643)],
[_pr(902, body="Closes #643"), _pr(903)],
pr=902,
comment_source=lambda kind, number: [],
)
other = next(node for node in snapshot.prs if node.number == 903)
self.assertIsNone(other.handoff)
self.assertEqual(other.handoff_status.state, HANDOFF_NOT_LOADED)
self.assertIn("not loaded", render_linkage_page(snapshot))
def test_no_focus_means_no_thread_is_claimed_handoff_free(self):
snapshot = _load([_issue(643)], [])
self.assertEqual(snapshot.handoff_status.state, HANDOFF_NOT_LOADED)
self.assertIn("thread-scoped", snapshot.handoff_status.reason)
def test_comment_source_failure_degrades_only_the_handoff(self):
def _boom(_kind, _number):
raise RuntimeError("comments 500")
snapshot = _load(
[_issue(643)], [_pr(902, body="Closes #643")], pr=902, comment_source=_boom
)
self.assertTrue(snapshot.ok)
self.assertEqual(snapshot.handoff_status.state, HANDOFF_UNAVAILABLE)
self.assertEqual(snapshot.issues[0].linked_prs, (902,))
self.assertIn("unavailable", render_linkage_page(snapshot))
def test_loaded_thread_with_no_cth_says_so_explicitly(self):
snapshot = _load(
[_issue(643)],
[_pr(902, body="Closes #643")],
pr=902,
comment_source=lambda kind, number: [{"id": 1, "body": "hi"}],
)
self.assertIn(
"no Canonical Thread Handoff comment found", render_linkage_page(snapshot)
)
class TestDeepLinks(unittest.TestCase):
"""AC3 — an external Gitea link is emitted only when permitted."""
def test_deep_links_are_withheld_by_default(self):
with mock.patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": ""}):
snapshot = _load([_issue(643)], [])
html = render_linkage_page(snapshot)
self.assertFalse(snapshot.deep_links_enabled)
self.assertIsNone(snapshot.issues[0].deep_link)
self.assertIn("Gitea deep links are withheld", html)
def test_reveal_opt_in_emits_the_link(self):
with mock.patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}):
snapshot = _load([_issue(643)], [_pr(902, body="Closes #643")])
html = render_linkage_page(snapshot)
self.assertTrue(snapshot.deep_links_enabled)
self.assertIn("/issues/643", snapshot.issues[0].deep_link)
self.assertIn("/pulls/902", snapshot.prs[0].deep_link)
self.assertIn(f'href="{snapshot.issues[0].deep_link}"', html)
class TestRedactionBoundary(unittest.TestCase):
def test_secret_shaped_title_is_redacted_before_display(self):
snapshot = _load(
[_issue(643, title="token=ghp_thisisnotarealsecretvalue0001")], []
)
payload = json.dumps(snapshot_to_dict(snapshot))
self.assertNotIn("ghp_thisisnotarealsecretvalue0001", payload)
self.assertNotIn(
"ghp_thisisnotarealsecretvalue0001", render_linkage_page(snapshot)
)
def test_handoff_fields_are_redacted(self):
comment = _cth(
2, created_at="2026-07-25T10:00:00Z", status="ok", next_owner="reviewer"
)
comment["body"] += "\nDecision: password=hunter2hunter2\n"
snapshot = _load(
[_issue(643)],
[_pr(902, body="Closes #643")],
pr=902,
comment_source=lambda kind, number: [comment],
)
self.assertNotIn("hunter2hunter2", json.dumps(snapshot_to_dict(snapshot)))
self.assertNotIn("hunter2hunter2", render_linkage_page(snapshot))
def test_html_escapes_markup_in_a_title(self):
snapshot = _load([_issue(643, title="<script>alert(1)</script>")], [])
html = render_linkage_page(snapshot)
self.assertNotIn("<script>alert(1)</script>", html)
self.assertIn("&lt;script&gt;", html)
class TestLinkageRoutes(unittest.TestCase):
def setUp(self):
self.snapshot = _load(
[_issue(643, labels=("status:ready",))],
[_pr(902, body="Closes #643", labels=("status:pr-open",))],
)
self.client = TestClient(create_app())
def test_page_renders_both_tables(self):
with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
response = self.client.get("/gitea")
self.assertEqual(response.status_code, 200)
self.assertIn("Issues → pull requests", response.text)
self.assertIn("Pull requests → issues", response.text)
self.assertIn("#643", response.text)
def test_api_exports_the_same_model(self):
with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
response = self.client.get("/api/v1/gitea/linkage")
self.assertEqual(response.status_code, 200)
payload = response.json()
self.assertTrue(payload["ok"])
self.assertEqual(payload["issues"][0]["linked_prs"], [902])
self.assertEqual(payload["prs"][0]["links"][0]["issue_number"], 643)
self.assertEqual(payload["schema_version"], 1)
def test_api_declares_the_evidence_vocabulary(self):
with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
payload = self.client.get("/api/v1/gitea/linkage").json()
names = {entry["name"] for entry in payload["evidence_kinds"]}
self.assertEqual(names, {EVIDENCE_CLOSES, EVIDENCE_BRANCH, EVIDENCE_REFERENCE})
def test_api_fails_closed_with_a_non_200_when_the_read_failed(self):
failed = _load([], [], project_id="no-such-project")
with mock.patch("webui.app.load_linkage_snapshot", return_value=failed):
response = self.client.get("/api/v1/gitea/linkage")
self.assertEqual(response.status_code, 502)
self.assertFalse(response.json()["ok"])
def test_page_still_renders_when_the_read_failed(self):
failed = _load([], [], project_id="no-such-project")
with mock.patch("webui.app.load_linkage_snapshot", return_value=failed):
response = self.client.get("/gitea")
self.assertEqual(response.status_code, 200)
self.assertIn("Linkage unavailable", response.text)
def test_query_parameters_reach_the_loader(self):
with mock.patch(
"webui.app.load_linkage_snapshot", return_value=self.snapshot
) as loader:
self.client.get("/gitea?project=gitea-tools&state=all&pr=902")
loader.assert_called_once()
args, kwargs = loader.call_args
self.assertEqual(args[0], "gitea-tools")
self.assertEqual(kwargs["state"], "all")
self.assertEqual(kwargs["pr"], 902)
self.assertIsNone(kwargs["issue"])
def test_surface_stays_read_only(self):
for path in ("/gitea", "/api/v1/gitea/linkage"):
with self.subTest(path=path):
self.assertEqual(self.client.post(path).status_code, 405)
def test_nav_exposes_the_linkage_page_as_live(self):
self.assertIn("/gitea", nav_hrefs())
home = self.client.get("/").text
self.assertIn('href="/gitea"', home)
self.assertIn(">Gitea<", home)
if __name__ == "__main__":
unittest.main()
-465
View File
@@ -1,465 +0,0 @@
"""Unit tests for Phase 3 Notifications and Human-Attention Console (#648)."""
from __future__ import annotations
import pytest
from starlette.testclient import TestClient
from webui.app import create_app
from webui.notifications import (
ATTENTION_HUMAN_REQUIRED,
ATTENTION_OPERATOR,
ATTENTION_ROUTINE,
CATEGORY_AUTH,
CATEGORY_BLOCKER,
CATEGORY_LEASE,
CATEGORY_SYSTEM,
CATEGORY_VALIDATION,
CATEGORY_WORKFLOW,
NotificationItem,
NotificationSnapshot,
classify_attention_event,
load_notifications_snapshot,
snapshot_to_dict,
)
from webui.notification_views import render_notifications_page
from webui.project_registry import load_registry
from webui.queue_loader import QueueItem, QueueSnapshot
from webui.lease_loader import CollisionWarning, LeaseSnapshot
from webui.system_health import DependencyProbe, SystemHealthSnapshot, VersionInfo, StaleRuntime
def test_classify_attention_event_rules():
# 1. Critical escalation boundaries -> human-required
att_cls, req_human = classify_attention_event(
CATEGORY_AUTH, "Auth error", "Unauthorized access attempt", is_auth_failure=True
)
assert att_cls == ATTENTION_HUMAN_REQUIRED
assert req_human is True
att_cls, req_human = classify_attention_event(
CATEGORY_SYSTEM, "Hard stop", "Hard stop triggered", is_hard_stop=True
)
assert att_cls == ATTENTION_HUMAN_REQUIRED
assert req_human is True
att_cls, req_human = classify_attention_event(
CATEGORY_VALIDATION, "Validation Error", "Report validation failed", is_validation_failure=True
)
assert att_cls == ATTENTION_HUMAN_REQUIRED
assert req_human is True
# 2. Operational issues -> operator
att_cls, req_human = classify_attention_event(
CATEGORY_BLOCKER, "PR Blocked", "Merge conflict detected", is_blocker=True
)
assert att_cls == ATTENTION_OPERATOR
assert req_human is False
att_cls, req_human = classify_attention_event(
CATEGORY_LEASE, "Lease Expired", "Session lease expired", is_stale=True
)
assert att_cls == ATTENTION_OPERATOR
assert req_human is False
# 3. Routine workflow transitions -> routine
att_cls, req_human = classify_attention_event(
CATEGORY_WORKFLOW, "PR Active", "PR in review"
)
assert att_cls == ATTENTION_ROUTINE
assert req_human is False
def test_notification_snapshot_aggregation():
reg = load_registry()
proj_id = reg.projects[0].id if reg.projects else "gitea-tools"
mock_queue = QueueSnapshot(
project_id=proj_id,
repo_label="org/repo",
prs=(
QueueItem(
number=101,
title="Blocked PR",
badges=("blocked",),
extra={},
),
QueueItem(
number=102,
title="Normal PR",
badges=("in-review",),
extra={},
),
),
issues=(),
pr_pagination=None,
issue_pagination=None,
)
mock_leases = LeaseSnapshot(
project_id=proj_id,
repo_label="org/repo",
issue_lock=None,
claim_inventory={},
reviewer_leases=(
{
"pr_number": 101,
"status": "expired",
"is_expired": True,
},
),
duplicate_prs=(
CollisionWarning(
kind="duplicate_pr",
message="Multiple open PRs for issue #101",
issue_number=101,
pr_numbers=(101, 103),
),
),
duplicate_branches=(),
collision_history=(),
fetch_error=None,
)
mock_version = VersionInfo(
git_sha="abc1234",
git_describe="v1.0.0",
control_plane_schema_version=1,
python_version="3.11",
known=True,
)
mock_stale = StaleRuntime(
daemon_head="abc1234",
checkout_head="abc1234",
remote_head="abc1234",
stale=False,
determinable=True,
mutation_safe=True,
reasons=(),
)
mock_health = SystemHealthSnapshot(
status="degraded",
ready=False,
readiness_complete=True,
readiness_reasons=("Auth failure",),
service="webui",
mode="test",
version=mock_version,
started_at="2026-07-25T00:00:00Z",
uptime_seconds=100.0,
timestamp="2026-07-25T00:00:00Z",
deep_probes_requested=True,
dependencies=(
DependencyProbe(
name="auth_service",
kind="auth",
status="unauthorized",
detail="Token expired",
required=True,
),
),
mcp_namespaces=(),
stale_runtime=mock_stale,
probe_errors=(),
)
snapshot = load_notifications_snapshot(
proj_id,
load_queue=lambda _id: mock_queue,
load_leases=lambda **_kwargs: mock_leases,
load_health=lambda **_kwargs: mock_health,
)
assert snapshot.project_id == proj_id
assert snapshot.total_count == 5
assert snapshot.human_required_count >= 1 # auth probe failure
assert snapshot.operator_count >= 3 # blocked PR + expired lease + duplicate PR collision
assert snapshot.routine_count >= 1 # normal PR
# Inbox items should include operator and human-required items only
inbox_classes = {item.attention_class for item in snapshot.inbox_items}
assert ATTENTION_ROUTINE not in inbox_classes
assert ATTENTION_OPERATOR in inbox_classes
assert ATTENTION_HUMAN_REQUIRED in inbox_classes
def test_snapshot_to_dict_and_redaction():
item = NotificationItem(
id="notif-1",
attention_class=ATTENTION_HUMAN_REQUIRED,
category=CATEGORY_AUTH,
title="Auth Error",
summary="Failed auth header: Bearer secret_token_12345",
work_kind="system",
work_number=None,
project_id="test-proj",
repo_label="org/repo",
created_at="2026-07-25T16:00:00Z",
requires_human=True,
)
snap = NotificationSnapshot(
project_id="test-proj",
repo_label="org/repo",
items=(item,),
human_required_count=1,
operator_count=0,
routine_count=0,
total_count=1,
)
data = snapshot_to_dict(snap)
assert data["project_id"] == "test-proj"
assert data["human_required_count"] == 1
assert len(data["inbox_items"]) == 1
# Redaction test
summary = data["inbox_items"][0]["summary"]
assert "secret_token_12345" not in summary
assert "<redacted>" in summary or "Bearer" in summary
def test_notifications_html_views():
item = NotificationItem(
id="notif-1",
attention_class=ATTENTION_HUMAN_REQUIRED,
category=CATEGORY_AUTH,
title="Critical Auth Failure",
summary="Auth failure details",
work_kind="issue",
work_number=42,
project_id="test-proj",
repo_label="org/repo",
created_at="2026-07-25T16:00:00Z",
requires_human=True,
)
snap = NotificationSnapshot(
project_id="test-proj",
repo_label="org/repo",
items=(item,),
human_required_count=1,
operator_count=0,
routine_count=0,
total_count=1,
)
html = render_notifications_page(snap, filter_class="inbox")
assert "Notifications &amp; Attention Inbox" in html or "Notifications & Attention Inbox" in html
assert "Critical Auth Failure" in html
assert "HUMAN REQUIRED" in html
assert "Human Required" in html
def test_notifications_app_routes():
app = create_app()
client = TestClient(app)
# 1. HTML Route
res = client.get("/notifications")
assert res.status_code == 200
assert "Notifications" in res.text
assert "Attention Inbox" in res.text
# 2. API Route /api/v1/notifications
res_api = client.get("/api/v1/notifications")
assert res_api.status_code == 200
json_data = res_api.json()
assert "human_required_count" in json_data
assert "operator_count" in json_data
assert "routine_count" in json_data
assert "inbox_items" in json_data
# 3. Compatibility Alias /api/notifications
res_alias = client.get("/api/notifications")
assert res_alias.status_code == 200
assert res_alias.json()["project_id"] == json_data["project_id"]
def test_classify_ignores_human_authored_title_and_summary_keywords():
"""B1: keywords in human-authored titles must not escalate routine work (#905)."""
# Routine transition whose title/summary mention critical-boundary words
att_cls, req_human = classify_attention_event(
CATEGORY_WORKFLOW,
"record irrecoverable decision lock provenance",
"PR #999 'record irrecoverable decision lock provenance' is in routine state in-review.",
)
assert att_cls == ATTENTION_ROUTINE
assert req_human is False
att_cls, req_human = classify_attention_event(
CATEGORY_WORKFLOW,
"fix unauthorized token path",
"Issue #1 'fix unauthorized token path' state: claimed. hard stop docs only.",
)
assert att_cls == ATTENTION_ROUTINE
assert req_human is False
# Structured flags still escalate (machine-driven)
att_cls, req_human = classify_attention_event(
CATEGORY_SYSTEM,
"anything",
"anything with hard stop in text",
is_hard_stop=True,
)
assert att_cls == ATTENTION_HUMAN_REQUIRED
assert req_human is True
def test_notification_ids_are_unique_across_probe_errors_and_collisions():
"""B2: published notification ids must be unique within a snapshot (#905)."""
reg = load_registry()
proj_id = reg.projects[0].id if reg.projects else "gitea-tools"
mock_queue = QueueSnapshot(
project_id=proj_id,
repo_label="org/repo",
prs=(),
issues=(),
pr_pagination=None,
issue_pagination=None,
)
mock_leases = LeaseSnapshot(
project_id=proj_id,
repo_label="org/repo",
issue_lock=None,
claim_inventory={},
reviewer_leases=(),
duplicate_prs=(
CollisionWarning(
kind="duplicate_pr",
message="Multiple open PRs for issue #10",
issue_number=10,
pr_numbers=(10, 11),
),
CollisionWarning(
kind="duplicate_branch",
message="Another collision without issue",
issue_number=None,
pr_numbers=(12, 13),
),
CollisionWarning(
kind="duplicate_pr",
message="Second issue collision",
issue_number=10,
pr_numbers=(14, 15),
),
),
duplicate_branches=(),
collision_history=(),
fetch_error=None,
)
mock_version = VersionInfo(
git_sha="abc1234",
git_describe="v1.0.0",
control_plane_schema_version=1,
python_version="3.11",
known=True,
)
mock_stale = StaleRuntime(
daemon_head="abc1234",
checkout_head="abc1234",
remote_head="abc1234",
stale=False,
determinable=True,
mutation_safe=True,
reasons=(),
)
mock_health = SystemHealthSnapshot(
status="degraded",
ready=False,
readiness_complete=True,
readiness_reasons=(),
service="webui",
mode="test",
version=mock_version,
started_at="2026-07-25T00:00:00Z",
uptime_seconds=100.0,
timestamp="2026-07-25T00:00:00Z",
deep_probes_requested=True,
dependencies=(),
mcp_namespaces=(),
stale_runtime=mock_stale,
probe_errors=("error alpha", "error beta"),
)
snapshot = load_notifications_snapshot(
proj_id,
load_queue=lambda _id: mock_queue,
load_leases=lambda **_kwargs: mock_leases,
load_health=lambda **_kwargs: mock_health,
)
ids = [item.id for item in snapshot.items]
assert len(ids) == len(set(ids)), f"duplicate notification ids: {ids}"
assert any(i.startswith(f"notif-sys-err-{proj_id}-") for i in ids)
assert any(i.startswith("notif-collision-") for i in ids)
def test_probe_errors_do_not_set_fetch_error():
"""B3: probe_errors must not be reported as fetch_error (#905)."""
reg = load_registry()
proj_id = reg.projects[0].id if reg.projects else "gitea-tools"
mock_queue = QueueSnapshot(
project_id=proj_id,
repo_label="org/repo",
prs=(),
issues=(),
pr_pagination=None,
issue_pagination=None,
fetch_error=None,
)
mock_leases = LeaseSnapshot(
project_id=proj_id,
repo_label="org/repo",
issue_lock=None,
claim_inventory={},
reviewer_leases=(),
duplicate_prs=(),
duplicate_branches=(),
collision_history=(),
fetch_error=None,
)
mock_version = VersionInfo(
git_sha="abc1234",
git_describe="v1.0.0",
control_plane_schema_version=1,
python_version="3.11",
known=True,
)
mock_stale = StaleRuntime(
daemon_head="abc1234",
checkout_head="abc1234",
remote_head="abc1234",
stale=False,
determinable=True,
mutation_safe=True,
reasons=(),
)
mock_health = SystemHealthSnapshot(
status="degraded",
ready=False,
readiness_complete=True,
readiness_reasons=(),
service="webui",
mode="test",
version=mock_version,
started_at="2026-07-25T00:00:00Z",
uptime_seconds=100.0,
timestamp="2026-07-25T00:00:00Z",
deep_probes_requested=True,
dependencies=(),
mcp_namespaces=(),
stale_runtime=mock_stale,
probe_errors=("probe blew up",),
)
snapshot = load_notifications_snapshot(
proj_id,
load_queue=lambda _id: mock_queue,
load_leases=lambda **_kwargs: mock_leases,
load_health=lambda **_kwargs: mock_health,
)
assert snapshot.fetch_error is None
# probe errors still appear as items
assert any("probe blew up" in item.summary for item in snapshot.items)
File diff suppressed because it is too large Load Diff
-182
View File
@@ -53,11 +53,6 @@ from webui.session_loader import (
snapshot_to_dict as session_view_snapshot_to_dict,
)
from webui.session_views import render_sessions_page
from webui.linkage_loader import (
load_linkage_snapshot,
snapshot_to_dict as linkage_snapshot_to_dict,
)
from webui.linkage_views import render_linkage_page
from webui.inventory import (
SECTION_NAMES as _INVENTORY_SECTIONS,
load_inventory_snapshot,
@@ -77,13 +72,6 @@ from webui.system_health import (
snapshot_to_dict as system_health_to_dict,
)
from webui.system_health_views import render_system_health_page
from webui.notifications import (
load_notifications_snapshot,
snapshot_to_dict as notifications_snapshot_to_dict,
)
from webui.notification_views import render_notifications_page
from webui import request_service
from webui.request_views import render_requests_page
_READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
_AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
@@ -354,41 +342,6 @@ async def api_sessions(_request: Request) -> JSONResponse:
return JSONResponse(session_view_snapshot_to_dict(load_session_view_snapshot()))
def _linkage_snapshot(request: Request):
"""Load one linkage snapshot from the request's scope and focus parameters."""
return load_linkage_snapshot(
request.query_params.get("project") or None,
state=request.query_params.get("state"),
issue=_query_int(request, "issue"),
pr=_query_int(request, "pr"),
)
async def gitea_linkage(request: Request) -> HTMLResponse:
"""Gitea issue↔PR linkage console (#645) — read-only.
Always 200, including on a failed read: this is an operator view, and it
must render *why* linkage could not be loaded rather than withhold the page.
The snapshot itself carries ``ok=False`` and the page refuses to draw a
linkage table it cannot stand behind.
"""
return HTMLResponse(render_linkage_page(_linkage_snapshot(request)))
async def api_v1_gitea_linkage(request: Request) -> JSONResponse:
"""JSON export of the issue↔PR linkage model (#645).
Unlike the HTML view, the API answers with 502 when the snapshot could not
be loaded, so an automated consumer cannot read a fail-closed payload as a
successful "no links exist" result.
"""
snapshot = _linkage_snapshot(request)
return JSONResponse(
linkage_snapshot_to_dict(snapshot),
status_code=200 if snapshot.ok else 502,
)
async def _parse_audit_form(request: Request) -> tuple[str, str | None]:
if request.method == "GET":
return "", None
@@ -786,125 +739,6 @@ async def api_v1_analytics_ingest(request: Request) -> JSONResponse:
)
async def notifications_route(request: Request) -> HTMLResponse:
project_id = request.query_params.get("project_id")
attention_class = request.query_params.get("attention_class") or "inbox"
snap = load_notifications_snapshot(project_id)
html = render_notifications_page(
snap, filter_class=attention_class, filter_project=project_id
)
return HTMLResponse(html)
async def api_notifications(request: Request) -> JSONResponse:
project_id = request.query_params.get("project_id")
snap = load_notifications_snapshot(project_id)
data = notifications_snapshot_to_dict(snap)
return JSONResponse(data)
def _default_request_scope() -> dict[str, str]:
"""Resolve remote/org/repo from the project registry for request forms.
Returns an empty mapping when the registry cannot be read, which makes
``parse_request`` reject a request that did not name its own scope rather
than letting it default to some other repository.
"""
from webui.queue_loader import _host_from_url # host normalisation helper
registry, error = _load_project_registry()
if error is not None or not registry.projects:
return {}
project = registry.projects[0]
host = _host_from_url(project.remote_host)
return {
"remote": _derive_remote(host),
"org": project.gitea_owner or "",
"repo": project.repo_name or "",
}
async def _request_payload(request: Request) -> dict[str, object]:
"""Read a request body as JSON or form-encoded. Never raises."""
content_type = (request.headers.get("content-type") or "").lower()
if "application/json" in content_type:
try:
body = await request.json()
except Exception:
return {}
return dict(body) if isinstance(body, dict) else {}
try:
form = await request.form()
except Exception:
return {}
return {key: form[key] for key in form}
async def requests_page(request: Request) -> HTMLResponse:
"""Operator request form and intent preview (#643).
POST here only ever *previews*. Initiation is a separate confirmed call to
``/api/v1/requests/apply`` so that submitting this form cannot reserve
work as a side effect.
"""
submitted: dict[str, object] = {}
preview = None
error = None
if request.method == "POST":
submitted = await _request_payload(request)
work_request, error = request_service.parse_request(
submitted, default_scope=_default_request_scope()
)
if work_request is not None:
preview = request_service.preview_request(
work_request,
principal=resolve_principal(headers=dict(request.headers)),
)
return HTMLResponse(
render_requests_page(
preview=preview, error=error, submitted=submitted
)
)
async def api_v1_request_preview(request: Request) -> JSONResponse:
"""Dry-run authorization and intent preview for a work request (#643)."""
payload = await _request_payload(request)
work_request, error = request_service.parse_request(
payload, default_scope=_default_request_scope()
)
if work_request is None:
return JSONResponse(error.to_dict(), status_code=400)
preview = request_service.preview_request(
work_request,
principal=resolve_principal(headers=dict(request.headers)),
)
return JSONResponse(
preview.to_dict(), status_code=200 if preview.authorized else 403
)
async def api_v1_request_apply(request: Request) -> JSONResponse:
"""Initiate a previewed work request through the allocator (#643).
Fail-closed at every step: unauthorized, unconfirmed, not-next-safe, and
already-claimed all return without attempting an assignment.
"""
payload = await _request_payload(request)
work_request, error = request_service.parse_request(
payload, default_scope=_default_request_scope()
)
if work_request is None:
return JSONResponse(error.to_dict(), status_code=400)
confirm = _truthy_flag(str(payload.get("confirm") or ""))
result = request_service.apply_request(
work_request,
principal=resolve_principal(headers=dict(request.headers)),
confirm=confirm,
)
status = int(result.pop("status_code", 403))
return JSONResponse(result, status_code=status)
async def method_not_allowed(request: Request, _exc: Exception) -> Response:
path = request.url.path
if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
@@ -933,9 +767,6 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
Route("/api/queue", api_queue, methods=["GET"]),
Route("/traffic", traffic, methods=["GET"]),
Route("/api/traffic", api_traffic, methods=["GET"]),
Route("/notifications", notifications_route, methods=["GET"]),
Route("/api/notifications", api_notifications, methods=["GET"]),
Route("/api/v1/notifications", api_notifications, methods=["GET"]),
Route("/projects", projects, methods=["GET"]),
Route("/projects/{project_id}", project_detail, methods=["GET"]),
Route("/api/projects", api_projects, methods=["GET"]),
@@ -954,8 +785,6 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
Route("/api/sessions", api_sessions, methods=["GET"]),
Route("/api/v1/sessions", api_sessions, methods=["GET"]),
Route("/api/v1/timeline", api_v1_timeline, methods=["GET"]),
Route("/gitea", gitea_linkage, methods=["GET"]),
Route("/api/v1/gitea/linkage", api_v1_gitea_linkage, methods=["GET"]),
Route("/analytics", analytics, methods=["GET"]),
Route("/api/analytics", api_v1_analytics, methods=["GET"]),
Route("/api/v1/analytics", api_v1_analytics, methods=["GET"]),
@@ -977,17 +806,6 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
api_action_attempt,
methods=["POST"],
),
Route("/requests", requests_page, methods=["GET", "POST"]),
Route(
"/api/v1/requests/preview",
api_v1_request_preview,
methods=["POST"],
),
Route(
"/api/v1/requests/apply",
api_v1_request_apply,
methods=["POST"],
),
Route("/api/leases", api_leases, methods=["GET"]),
Route("/api/v1/inventory", api_inventory, methods=["GET"]),
Route(
+8 -71
View File
@@ -115,12 +115,6 @@ class ConsoleAction:
break_glass: bool
phase: int
summary: str
# Opt-in switch for an action whose execution path is genuinely wired
# ahead of its phase becoming globally active (#643). Naming a variable
# here enables nothing on its own: the variable must also be set in the
# environment. An action that leaves this ``None`` can only execute once
# ACTIVE_PHASE reaches its phase, exactly as before.
execution_env_flag: str | None = None
@property
def mcp_permission(self) -> str:
@@ -283,27 +277,6 @@ _ACTION_SPECS: tuple[ConsoleAction, ...] = (
phase=2,
summary="Restart one MCP namespace via the host supervisor.",
),
# #643: submit a work request — desired role, issue/PR, intent — and let
# the allocator reserve it. This is the one Phase 2 action whose execution
# path is actually implemented (``webui.request_service``), so it carries
# the opt-in flag; it stays denied until an operator sets that variable.
# Authority is operator-class because the outcome is a claim, not a Gitea
# verdict: initiating reviewer or merger *work* does not grant the right
# to approve or merge, which stays with the MCP role profile.
ConsoleAction(
action_id="initiate_workflow",
task_key="allocate_next_work",
action_class=CLASS_WRITE,
minimum_role=OPERATOR,
requires_confirmation=True,
dual_control=False,
break_glass=False,
phase=2,
summary=(
"Preview and initiate allocator-owned workflow work for a role."
),
execution_env_flag="WEBUI_REQUESTS_EXECUTION",
),
)
ACTIONS: dict[str, ConsoleAction] = {a.action_id: a for a in _ACTION_SPECS}
@@ -457,33 +430,6 @@ ALLOW_PREVIEW = "allowed_preview_only"
# gated on this model landing; nothing here enables it.
ACTIVE_PHASE = 1
_TRUTHY = frozenset({"1", "true", "yes", "on"})
def execution_wired(
action: ConsoleAction | None, env: dict[str, str] | None = None
) -> bool:
"""Whether *action* has a live execution path right now.
Two ways to be wired, and only two. The action's phase is active, or the
action declares an opt-in environment variable *and* that variable is set.
Everything else — including every action that never declares a flag — is
unwired, so the default across the registry stays deny.
Bumping ``ACTIVE_PHASE`` would enable execution for every action of that
phase at once. The per-action flag exists so a single implemented action
can go live without dragging its unimplemented phase-mates with it.
"""
if action is None:
return False
if action.phase <= ACTIVE_PHASE:
return True
flag = (action.execution_env_flag or "").strip()
if not flag:
return False
source = env if env is not None else os.environ
return (source.get(flag) or "").strip().lower() in _TRUTHY
@dataclass(frozen=True)
class AuthorizationDecision:
@@ -523,19 +469,16 @@ def authorize(
principal: Principal | None = None,
*,
for_execution: bool = False,
env: dict[str, str] | None = None,
) -> AuthorizationDecision:
"""Decide whether *principal* may invoke *action_id*. Deny by default.
``for_execution`` distinguishes a read-only preview from a real invocation.
``execution_enabled`` reports whether the action has a live execution path
at all (:func:`execution_wired`) — for every action without an explicit
opt-in flag that stays ``False`` while the console is in Phase 1, so no
caller can read an allow as permission to mutate.
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)
wired = execution_wired(action, env)
if action is None:
return AuthorizationDecision(
@@ -554,7 +497,7 @@ def authorize(
"requires_confirmation": action.requires_confirmation,
"dual_control": action.dual_control,
"break_glass": action.break_glass,
"execution_enabled": wired,
"execution_enabled": False,
}
if not who.authenticated:
@@ -587,19 +530,13 @@ def authorize(
**base,
)
if for_execution and not wired:
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}"
+ (
f" and {action.execution_env_flag} is not set"
if action.execution_env_flag
else ""
)
+ ". Execution is not wired."
f"console is in phase {ACTIVE_PHASE}. Execution is not wired."
),
**base,
)
@@ -608,8 +545,8 @@ def authorize(
allowed=True,
reason_code=ALLOW_PREVIEW,
detail=(
"Principal holds the required role. Execution proceeds only for an "
"action with a wired execution path; everything else is preview."
"Principal holds the required role. Preview only — execution "
"remains disabled until the Phase 2 action framework ships."
),
**base,
)
-771
View File
@@ -1,771 +0,0 @@
"""Gitea issue↔PR linkage model for the console (#645, Phase 3).
Operators lose context between an issue and the PR that closes it: which PR
carries which issue, whether two PRs claim the same issue, and what the latest
canonical handoff on that thread said. The evidence exists in Gitea, but only
as free text scattered across PR titles, bodies, and branch names.
This module resolves that linkage into one read-only model:
* :func:`resolve_linkage` is a pure function from raw Gitea issue/PR payloads to
a :class:`LinkageIndex`. It records *how* each edge was found (a ``Closes #N``
keyword, the canonical ``feat/issue-N-…`` branch marker, or a bare ``#N``
body reference) and never collapses several candidates into one silent guess.
* :func:`load_linkage_snapshot` scopes that index to a registry project and
optionally attaches the latest Canonical Thread Handoff (CTH) summary for one
focused issue or PR.
Design rules, matching the rest of the console:
- **Read-only.** Gitea is read through the shared authenticated helpers. No
endpoint here mutates anything, and no write action is registered.
- **Qualified absence.** Linkage is a claim about a *loaded* window of Gitea.
When pagination did not complete, when credentials were unavailable, or when
only open items were fetched, the snapshot says so and every "no linked PR"
is marked non-authoritative. An empty edge list from a partial read is not
evidence that no link exists.
- **Handoff is loaded, never assumed.** CTH comments are thread-scoped, so they
are fetched only for an explicitly focused issue or PR. Every other row
reports ``not_loaded`` rather than rendering as "no handoff".
- **Redaction at the boundary.** Titles, labels, handoff fields, and error
reasons are free text from Gitea and cross :mod:`webui.console_redaction`
before they leave this module.
- **Deep links are opt-in.** A link to the Gitea web UI is emitted only under
the ``GITEA_MCP_REVEAL_ENDPOINTS`` admin opt-in, exactly as the MCP tools
gate their own URL exposure.
Non-goals (from the issue): no issue/PR editor, no browser review or merge, no
reimplementation of Gitea search.
"""
from __future__ import annotations
import os
import re
from dataclasses import dataclass
from typing import Any, Callable, Iterable, Sequence
from gitea_auth import api_fetch_page, get_auth_header, gitea_url, repo_api_url
from webui import console_redaction
from webui.project_registry import ProjectRecord, load_registry
from webui.queue_loader import (
PaginationMeta,
_fetch_issues,
_fetch_prs,
_host_from_url,
)
#: Version of the serialized linkage contract. Bump on any breaking change.
LINKAGE_SCHEMA_VERSION = 1
# --- Linkage evidence -------------------------------------------------------
# Ordered strongest to weakest. The strength ordering is what makes an
# ambiguous PR detectable: two candidates at the same strength are a genuine
# ambiguity, while a weaker candidate alongside a stronger one is not.
EVIDENCE_CLOSES = "closes_keyword"
EVIDENCE_BRANCH = "branch_marker"
EVIDENCE_REFERENCE = "body_reference"
EVIDENCE_ORDER: tuple[str, ...] = (
EVIDENCE_CLOSES,
EVIDENCE_BRANCH,
EVIDENCE_REFERENCE,
)
_EVIDENCE_RANK = {name: rank for rank, name in enumerate(EVIDENCE_ORDER)}
EVIDENCE_DESCRIPTIONS: dict[str, str] = {
EVIDENCE_CLOSES: (
"the PR title or body declares 'closes/fixes/resolves #N' — Gitea itself "
"acts on this keyword, so it is the strongest available evidence"
),
EVIDENCE_BRANCH: (
"the PR head branch carries the canonical issue marker "
"'(fix|feat|docs|chore)/issue-N-…' minted by the issue lock"
),
EVIDENCE_REFERENCE: (
"the PR body mentions '#N' without a closing keyword; a mention is not "
"a claim that the PR closes that issue"
),
}
_CLOSES_RE = re.compile(r"(?:closes|fixes|resolves)\s+#(\d+)", re.IGNORECASE)
_REFERENCE_RE = re.compile(r"#(\d+)")
_BRANCH_MARKER_RE = re.compile(
r"^(?:fix|feat|docs|chore)/issue-(\d+)(?:[-/]|$)", re.IGNORECASE
)
# Handoff-source states. ``not_loaded`` is deliberately distinct from "none
# found": a row whose comments were never fetched proves nothing about whether
# a handoff exists on that thread.
HANDOFF_NOT_LOADED = "not_loaded"
HANDOFF_LOADED = "loaded"
HANDOFF_UNAVAILABLE = "unavailable"
# Which item states were fetched. Linkage claims are scoped to this window.
STATE_OPEN = "open"
STATE_ALL = "all"
_SUPPORTED_STATES = (STATE_OPEN, STATE_ALL)
def _redact(value: Any) -> Any:
"""Redact one free-text field, failing closed to the placeholder."""
if value is None:
return None
return console_redaction.redact_text(str(value))
def deep_links_enabled(env: dict[str, str] | None = None) -> bool:
"""Whether Gitea web-UI deep links may be emitted (admin/debug opt-in)."""
source = env if env is not None else os.environ
return (source.get("GITEA_MCP_REVEAL_ENDPOINTS") or "").strip().lower() in {
"1",
"true",
"yes",
"on",
}
def _deep_link(host: str, org: str, repo: str, kind: str, number: int) -> str | None:
"""Build a Gitea web link for one item, or None when reveal is not enabled."""
if not deep_links_enabled() or not (host and org and repo):
return None
segment = "pulls" if kind == "pr" else "issues"
try:
return gitea_url(host, f"/{org}/{repo}/{segment}/{int(number)}")
except Exception:
return None
# --- Pure linkage resolution -------------------------------------------------
@dataclass(frozen=True)
class IssueLink:
"""One resolved edge from a PR to an issue, with the evidence that found it."""
issue_number: int
evidence: tuple[str, ...]
@property
def strength(self) -> int:
"""Rank of the strongest evidence backing this edge (lower is stronger)."""
return min(
(_EVIDENCE_RANK.get(name, len(EVIDENCE_ORDER)) for name in self.evidence),
default=len(EVIDENCE_ORDER),
)
@property
def closes(self) -> bool:
"""True only when the PR *declares* it closes the issue."""
return EVIDENCE_CLOSES in self.evidence
def to_dict(self) -> dict[str, Any]:
return {
"issue_number": self.issue_number,
"evidence": list(self.evidence),
"closes": self.closes,
}
def _sorted_links(links: Iterable[IssueLink]) -> tuple[IssueLink, ...]:
return tuple(sorted(links, key=lambda link: (link.strength, link.issue_number)))
def resolve_pr_links(pr: dict[str, Any]) -> tuple[IssueLink, ...]:
"""Resolve every issue a PR points at, strongest evidence first.
Every candidate is kept. Collapsing to a single "linked issue" is what makes
a mislinked or double-claimed PR invisible, so the caller decides what to do
with several candidates rather than being handed one guess.
"""
found: dict[int, set[str]] = {}
def _add(number: Any, evidence: str) -> None:
try:
issue_number = int(number)
except (TypeError, ValueError):
return
if issue_number <= 0:
return
found.setdefault(issue_number, set()).add(evidence)
title = str(pr.get("title") or "")
body = str(pr.get("body") or "")
for text in (title, body):
for match in _CLOSES_RE.finditer(text):
_add(match.group(1), EVIDENCE_CLOSES)
head_ref = str((pr.get("head") or {}).get("ref") or "")
branch_match = _BRANCH_MARKER_RE.match(head_ref.strip())
if branch_match:
_add(branch_match.group(1), EVIDENCE_BRANCH)
# The ``#N`` inside "Closes #N" is the *same* textual occurrence as the
# closing keyword, not a second, independent mention. Blanking the closing
# phrases first keeps "mention" meaning what the legend says it means: a
# reference the PR made without claiming to close anything.
for match in _REFERENCE_RE.finditer(_CLOSES_RE.sub(" ", body)):
_add(match.group(1), EVIDENCE_REFERENCE)
# A PR's own number appearing in its body is self-reference, not linkage.
try:
found.pop(int(pr.get("number")), None)
except (TypeError, ValueError):
pass
return _sorted_links(
IssueLink(
issue_number=number,
evidence=tuple(name for name in EVIDENCE_ORDER if name in evidence),
)
for number, evidence in found.items()
)
@dataclass(frozen=True)
class LinkageIndex:
"""Resolved linkage over one loaded window of issues and PRs."""
pr_links: dict[int, tuple[IssueLink, ...]]
issue_prs: dict[int, tuple[int, ...]]
def primary_issue(self, pr_number: int) -> IssueLink | None:
"""The strongest edge for a PR, or None when it points at no issue."""
links = self.pr_links.get(int(pr_number)) or ()
return links[0] if links else None
def ambiguous(self, pr_number: int) -> bool:
"""True when two or more issues tie at the PR's strongest evidence."""
links = self.pr_links.get(int(pr_number)) or ()
if len(links) < 2:
return False
best = links[0].strength
return sum(1 for link in links if link.strength == best) > 1
def contested_issues(self) -> tuple[int, ...]:
"""Issues claimed by more than one PR in the loaded window."""
return tuple(
number for number, prs in sorted(self.issue_prs.items()) if len(prs) > 1
)
def resolve_linkage(prs: Sequence[dict[str, Any]]) -> LinkageIndex:
"""Build the bidirectional linkage index for a loaded window of PRs.
Only *closing* and *branch-marker* edges populate the issue→PR direction: a
bare ``#N`` mention is a reference, and treating it as "this PR is the work
for issue N" would invent contested issues out of ordinary cross-links. The
weaker edge stays visible on the PR→issue side, where it is labelled.
"""
pr_links: dict[int, tuple[IssueLink, ...]] = {}
issue_prs: dict[int, list[int]] = {}
for pr in prs or []:
try:
pr_number = int(pr["number"])
except (KeyError, TypeError, ValueError):
continue
links = resolve_pr_links(pr)
pr_links[pr_number] = links
for link in links:
if link.evidence == (EVIDENCE_REFERENCE,):
continue
bucket = issue_prs.setdefault(link.issue_number, [])
if pr_number not in bucket:
bucket.append(pr_number)
return LinkageIndex(
pr_links=pr_links,
issue_prs={number: tuple(sorted(items)) for number, items in issue_prs.items()},
)
# --- Canonical handoff summary ----------------------------------------------
@dataclass(frozen=True)
class HandoffSummary:
"""The latest CTH comment on one thread, redacted for display."""
comment_id: int | None
created_at: str | None
author: str | None
cth_type: str
cth_type_known: bool
status: str | None
next_owner: str | None
current_blocker: str | None
decision: str | None
next_action: str | None
def to_dict(self) -> dict[str, Any]:
return {
"comment_id": self.comment_id,
"created_at": self.created_at,
"author": self.author,
"cth_type": self.cth_type,
"cth_type_known": self.cth_type_known,
"status": self.status,
"next_owner": self.next_owner,
"current_blocker": self.current_blocker,
"decision": self.decision,
"next_action": self.next_action,
}
@dataclass(frozen=True)
class HandoffStatus:
"""Why a thread's handoff summary is present, absent, or unknown."""
state: str
reason: str | None = None
target: str | None = None
@property
def loaded(self) -> bool:
return self.state == HANDOFF_LOADED
def to_dict(self) -> dict[str, Any]:
return {"state": self.state, "reason": self.reason, "target": self.target}
def summarize_handoff(comments: Sequence[dict[str, Any]]) -> HandoffSummary | None:
"""Summarize the newest CTH comment in *comments*, or None when there is none.
Every field is redacted before it is returned: a handoff body is operator
free text that regularly quotes commands, and it is rendered verbatim on the
page this feeds.
"""
from canonical_thread_handoff import find_latest_cth, is_known_cth_type
try:
latest = find_latest_cth(list(comments or []))
except Exception:
return None
if not latest:
return None
fields = latest.get("fields") or {}
cth_type = str(latest.get("cth_type") or "").strip()
known = is_known_cth_type(cth_type)
try:
comment_id: int | None = int(latest.get("comment_id"))
except (TypeError, ValueError):
comment_id = None
return HandoffSummary(
comment_id=comment_id,
created_at=_redact(latest.get("created_at")),
author=_redact(latest.get("author")),
# An unrecognised heading is reported as such rather than republished:
# the heading is free text, and CTH_TYPES is the only authority for what
# a handoff type may be.
cth_type=cth_type if known else "unrecognized",
cth_type_known=known,
status=_redact(fields.get("status")),
next_owner=_redact(fields.get("next owner")),
current_blocker=_redact(fields.get("current blocker")),
decision=_redact(fields.get("decision")),
next_action=_redact(fields.get("next action")),
)
CommentSource = Callable[[str, int], list[dict[str, Any]]]
def build_comment_source(host: str, org: str, repo: str) -> CommentSource | None:
"""Build an authenticated ``(kind, number) -> comments`` fetcher, or None.
Returns None when the console is running in offline test mode or when no
credential is available for *host*, so the caller reports the handoff source
as unavailable instead of as an empty thread.
"""
if _offline_test_mode() or not (host and org and repo):
return None
auth = get_auth_header(host)
if not auth:
return None
def _fetch(kind: str, number: int) -> list[dict[str, Any]]:
segment = "pulls" if kind == "pr" else "issues"
url = f"{repo_api_url(host, org, repo)}/{segment}/{int(number)}/comments"
comments: list[dict[str, Any]] = []
page = 1
while page <= 20:
raw, meta = api_fetch_page(url, auth, page=page, limit=50)
comments.extend(raw)
if bool(meta["is_final_page"]):
break
page += 1
return comments
return _fetch
# --- Snapshot ----------------------------------------------------------------
@dataclass(frozen=True)
class LinkageNode:
"""One issue or PR row with its resolved links and display metadata."""
kind: str
number: int
title: str
state: str
labels: tuple[str, ...] = ()
links: tuple[IssueLink, ...] = ()
linked_prs: tuple[int, ...] = ()
ambiguous: bool = False
contested: bool = False
deep_link: str | None = None
handoff: HandoffSummary | None = None
handoff_status: HandoffStatus = HandoffStatus(HANDOFF_NOT_LOADED)
links_authoritative: bool = True
def to_dict(self) -> dict[str, Any]:
return {
"kind": self.kind,
"number": self.number,
"title": self.title,
"state": self.state,
"labels": list(self.labels),
"links": [link.to_dict() for link in self.links],
"linked_prs": list(self.linked_prs),
"ambiguous": self.ambiguous,
"contested": self.contested,
"deep_link": self.deep_link,
"links_authoritative": self.links_authoritative,
"handoff": self.handoff.to_dict() if self.handoff else None,
"handoff_status": self.handoff_status.to_dict(),
}
@dataclass(frozen=True)
class LinkageSnapshot:
"""One answered linkage query over a scoped window of a Gitea repo."""
ok: bool
project_id: str
repo_label: str
host: str
state_scope: str
issues: tuple[LinkageNode, ...] = ()
prs: tuple[LinkageNode, ...] = ()
contested_issues: tuple[int, ...] = ()
focus: tuple[str, int] | None = None
inventory_complete: bool = False
deep_links_enabled: bool = False
handoff_status: HandoffStatus = HandoffStatus(HANDOFF_NOT_LOADED)
fetch_error: str | None = None
@property
def orphan_prs(self) -> tuple[LinkageNode, ...]:
"""PRs in the loaded window that point at no issue at all."""
return tuple(node for node in self.prs if not node.links)
def to_dict(self) -> dict[str, Any]:
return {
"ok": self.ok,
"schema_version": LINKAGE_SCHEMA_VERSION,
"project_id": self.project_id,
"repo": self.repo_label,
"state_scope": self.state_scope,
"inventory_complete": self.inventory_complete,
"deep_links_enabled": self.deep_links_enabled,
"focus": (
None
if self.focus is None
else {"kind": self.focus[0], "number": self.focus[1]}
),
"handoff_source": self.handoff_status.to_dict(),
"fetch_error": self.fetch_error,
"contested_issues": list(self.contested_issues),
"issues": [node.to_dict() for node in self.issues],
"prs": [node.to_dict() for node in self.prs],
"evidence_kinds": [
{"name": name, "description": EVIDENCE_DESCRIPTIONS[name]}
for name in EVIDENCE_ORDER
],
}
def snapshot_to_dict(snapshot: LinkageSnapshot) -> dict[str, Any]:
"""JSON-serializable export for ``/api/v1/gitea/linkage``."""
return snapshot.to_dict()
def _offline_test_mode() -> bool:
return (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip().lower() in {
"1",
"true",
"yes",
}
def _labels_of(item: dict[str, Any]) -> tuple[str, ...]:
return tuple(
str(_redact(label.get("name")))
for label in (item.get("labels") or [])
if label.get("name")
)
def _failed_snapshot(
*,
project_id: str,
repo_label: str,
host: str,
state_scope: str,
reason: str,
) -> LinkageSnapshot:
"""A read that could not be answered. Never an empty-and-healthy snapshot."""
return LinkageSnapshot(
ok=False,
project_id=project_id,
repo_label=repo_label,
host=host,
state_scope=state_scope,
inventory_complete=False,
deep_links_enabled=deep_links_enabled(),
handoff_status=HandoffStatus(
HANDOFF_UNAVAILABLE, reason="linkage inventory could not be loaded"
),
fetch_error=str(_redact(reason)),
)
def _resolve_project(project_id: str | None) -> ProjectRecord | None:
registry = load_registry()
if project_id:
for entry in registry.projects:
if entry.id == project_id:
return entry
return None
return registry.projects[0] if registry.projects else None
def _normalize_state(state: str | None) -> str:
text = (state or STATE_OPEN).strip().lower()
return text if text in _SUPPORTED_STATES else STATE_OPEN
def load_linkage_snapshot(
project_id: str | None = None,
*,
state: str | None = None,
issue: int | None = None,
pr: int | None = None,
fetch_prs: Callable[..., tuple[list[dict], PaginationMeta]] | None = None,
fetch_issues: Callable[..., tuple[list[dict], PaginationMeta]] | None = None,
comment_source: CommentSource | None = None,
) -> LinkageSnapshot:
"""Load issue↔PR linkage for a registry project.
``issue``/``pr`` focus one thread: the focused row is the only one whose
Canonical Thread Handoff comments are fetched, because handoff comments are
thread-scoped and loading them for a whole queue would be one request per
row. Every unfocused row reports its handoff as ``not_loaded``.
"""
state_scope = _normalize_state(state)
try:
project = _resolve_project(project_id)
except Exception as exc: # registry invalid — fail closed with the reason
return _failed_snapshot(
project_id=project_id or "",
repo_label="",
host="",
state_scope=state_scope,
reason=f"project registry unavailable: {exc}",
)
if project is None:
return _failed_snapshot(
project_id=project_id or "",
repo_label="",
host="",
state_scope=state_scope,
reason=(
f"project {project_id!r} not found in registry"
if project_id
else "no projects registered"
),
)
host = _host_from_url(project.remote_host)
repo_label = f"{project.gitea_owner}/{project.repo_name}"
offline_test = _offline_test_mode()
def _empty_fetch(*_args, **_kwargs):
return [], PaginationMeta(
page=1,
per_page=50,
returned_count=0,
has_more=False,
is_final_page=True,
# An offline stub loaded nothing; claiming a complete inventory here
# would let the page assert that no issue has a linked PR.
inventory_complete=False,
pages_fetched=0,
)
pr_fetch = fetch_prs or (_empty_fetch if offline_test else _fetch_prs)
issue_fetch = fetch_issues or (_empty_fetch if offline_test else _fetch_issues)
using_live_fetch = not offline_test and (fetch_prs is None or fetch_issues is None)
auth = get_auth_header(host) if using_live_fetch else "test-auth"
if using_live_fetch and not auth:
return _failed_snapshot(
project_id=project.id,
repo_label=repo_label,
host=host,
state_scope=state_scope,
reason=(
f"Gitea credentials unavailable for {host}; linkage cannot be "
"loaded (fail closed — not rendering an empty linkage table)"
),
)
try:
raw_prs, pr_pagination = pr_fetch(
host, project.gitea_owner, project.repo_name, auth, state=state_scope
)
raw_issues, issue_pagination = issue_fetch(
host, project.gitea_owner, project.repo_name, auth, state=state_scope
)
except Exception as exc: # noqa: BLE001 — operator-visible fetch failure
return _failed_snapshot(
project_id=project.id,
repo_label=repo_label,
host=host,
state_scope=state_scope,
reason=f"Gitea fetch failed: {exc}",
)
inventory_complete = bool(
getattr(pr_pagination, "inventory_complete", False)
and getattr(issue_pagination, "inventory_complete", False)
)
index = resolve_linkage(raw_prs)
contested = index.contested_issues()
focus: tuple[str, int] | None = None
if pr is not None:
focus = ("pr", int(pr))
elif issue is not None:
focus = ("issue", int(issue))
unfocused_reason = (
"canonical handoff comments are thread-scoped; focus one issue or PR "
"to load its latest handoff"
)
handoff_status = HandoffStatus(HANDOFF_NOT_LOADED, reason=unfocused_reason)
focus_handoff: HandoffSummary | None = None
if focus is not None:
source = comment_source
if source is None and not offline_test:
source = build_comment_source(host, project.gitea_owner, project.repo_name)
target = f"{focus[0]}#{focus[1]}"
if source is None:
handoff_status = HandoffStatus(
HANDOFF_UNAVAILABLE,
reason="no authenticated comment source available for this read",
target=target,
)
else:
try:
focus_handoff = summarize_handoff(source(focus[0], focus[1]) or [])
handoff_status = HandoffStatus(HANDOFF_LOADED, target=target)
except Exception as exc: # fail soft: degrade this source only
handoff_status = HandoffStatus(
HANDOFF_UNAVAILABLE,
reason=str(_redact(f"handoff fetch failed: {exc}")),
target=target,
)
def _node_handoff(
kind: str, number: int
) -> tuple[HandoffSummary | None, HandoffStatus]:
"""Attach the handoff only to the focused row; qualify every other row."""
if focus == (kind, number):
return (focus_handoff, handoff_status)
return (
None,
HandoffStatus(
HANDOFF_NOT_LOADED,
reason=unfocused_reason if focus is None else "not the focused thread",
),
)
def _number_of(raw: dict[str, Any]) -> int | None:
try:
return int(raw["number"])
except (KeyError, TypeError, ValueError):
return None
def _sort_key(raw: dict[str, Any]) -> int:
number = _number_of(raw)
return -1 if number is None else number
issue_nodes: list[LinkageNode] = []
for raw in sorted(raw_issues or [], key=_sort_key, reverse=True):
number = _number_of(raw)
if number is None:
continue
node_handoff, node_status = _node_handoff("issue", number)
linked_prs = index.issue_prs.get(number, ())
issue_nodes.append(
LinkageNode(
kind="issue",
number=number,
title=str(_redact(raw.get("title")) or ""),
state=str(raw.get("state") or ""),
labels=_labels_of(raw),
linked_prs=linked_prs,
contested=len(linked_prs) > 1,
deep_link=_deep_link(
host, project.gitea_owner, project.repo_name, "issue", number
),
handoff=node_handoff,
handoff_status=node_status,
links_authoritative=inventory_complete,
)
)
pr_nodes: list[LinkageNode] = []
for raw in sorted(raw_prs or [], key=_sort_key, reverse=True):
number = _number_of(raw)
if number is None:
continue
node_handoff, node_status = _node_handoff("pr", number)
links = index.pr_links.get(number, ())
pr_nodes.append(
LinkageNode(
kind="pr",
number=number,
title=str(_redact(raw.get("title")) or ""),
state=str(raw.get("state") or ""),
labels=_labels_of(raw),
links=links,
ambiguous=index.ambiguous(number),
contested=any(link.issue_number in contested for link in links),
deep_link=_deep_link(
host, project.gitea_owner, project.repo_name, "pr", number
),
handoff=node_handoff,
handoff_status=node_status,
links_authoritative=inventory_complete,
)
)
return LinkageSnapshot(
ok=True,
project_id=project.id,
repo_label=repo_label,
host=host,
state_scope=state_scope,
issues=tuple(issue_nodes),
prs=tuple(pr_nodes),
contested_issues=contested,
focus=focus,
inventory_complete=inventory_complete,
deep_links_enabled=deep_links_enabled(),
handoff_status=handoff_status,
)
-364
View File
@@ -1,364 +0,0 @@
"""HTML views for the Gitea issue↔PR linkage console (#645, Phase 3).
Read-only renderer over :mod:`webui.linkage_loader`. The page's job is to make
three things impossible to misread:
* **why** an edge exists — every link carries its evidence badge, so a bare
``#N`` mention never looks like a closing claim;
* **what was not loaded** — a partial inventory, an unfocused thread, or an
unavailable handoff source renders as an explicit qualifier, never as an
affirmative "none";
* **that nothing here mutates** — there is no review, merge, or edit control,
and the deep link out to Gitea appears only under the admin reveal opt-in.
"""
from __future__ import annotations
from html import escape
from typing import Sequence
from webui.layout import render_page
from webui.linkage_loader import (
EVIDENCE_BRANCH,
EVIDENCE_CLOSES,
EVIDENCE_DESCRIPTIONS,
EVIDENCE_ORDER,
EVIDENCE_REFERENCE,
HANDOFF_LOADED,
HANDOFF_NOT_LOADED,
HandoffSummary,
LinkageNode,
LinkageSnapshot,
)
_EVIDENCE_CSS = {
EVIDENCE_CLOSES: "badge-health-ok",
EVIDENCE_BRANCH: "badge-health-skipped",
EVIDENCE_REFERENCE: "badge-health-unproven",
}
_EVIDENCE_LABEL = {
EVIDENCE_CLOSES: "closes",
EVIDENCE_BRANCH: "branch",
EVIDENCE_REFERENCE: "mention",
}
def _badge(text: str, css: str) -> str:
return f'<span class="badge {css}">{escape(text)}</span>'
def _labels(names: Sequence[str]) -> str:
if not names:
return '<span class="muted">—</span>'
return " ".join(_badge(name, "badge-health-skipped") for name in names)
def _ref(node: LinkageNode) -> str:
"""Render an item reference, hyperlinked only when deep links are revealed."""
label = f"#{node.number}"
if node.deep_link:
return f'<a href="{escape(node.deep_link)}"><code>{escape(label)}</code></a>'
return f"<code>{escape(label)}</code>"
def _scope_card(snapshot: LinkageSnapshot) -> str:
focus = (
"none"
if snapshot.focus is None
else f"{snapshot.focus[0]}#{snapshot.focus[1]}"
)
completeness = (
_badge("complete", "badge-health-ok")
if snapshot.inventory_complete
else _badge("partial", "badge-health-degraded")
)
links_note = (
"Every linkage edge below is a claim about this loaded window only."
if snapshot.inventory_complete
else (
"Pagination did not complete for this window, so an empty link list "
"means <em>none found in what was loaded</em> — not that no link exists."
)
)
deep_links = (
_badge("enabled", "badge-health-ok")
if snapshot.deep_links_enabled
else _badge("hidden", "badge-health-skipped")
)
return f"""<div class="health-card">
<h3>Scope</h3>
<table class="detail">
<tr><th>Project</th><td><code>{escape(snapshot.project_id or "")}</code></td></tr>
<tr><th>Repository</th><td><code>{escape(snapshot.repo_label or "")}</code></td></tr>
<tr><th>Item state</th><td><code>{escape(snapshot.state_scope)}</code></td></tr>
<tr><th>Focused thread</th><td><code>{escape(focus)}</code></td></tr>
<tr><th>Inventory</th><td>{completeness}</td></tr>
<tr><th>Gitea deep links</th><td>{deep_links}</td></tr>
</table>
<p class="muted">{links_note}</p>
</div>"""
def _error_card(snapshot: LinkageSnapshot) -> str:
if snapshot.ok and not snapshot.fetch_error:
return ""
return (
'<div class="health-card health-stale"><strong>Linkage unavailable:</strong> '
f"{escape(snapshot.fetch_error or 'the linkage read did not complete')}. "
"No linkage table is rendered: an empty table would read as "
"<em>no issue is linked to any PR</em>, which this read cannot claim."
"</div>"
)
def _contested_card(snapshot: LinkageSnapshot) -> str:
if not snapshot.contested_issues:
return ""
refs = ", ".join(f"<code>#{number}</code>" for number in snapshot.contested_issues)
return (
'<div class="health-card health-stale">'
f"<strong>Contested issues:</strong> {refs}. More than one PR in this "
"window claims each of these — duplicate work or a superseded PR. "
"Resolution stays in Gitea and the workflow; this console only reports it."
"</div>"
)
def _evidence_badges(evidence: Sequence[str]) -> str:
return " ".join(
_badge(
_EVIDENCE_LABEL.get(name, name),
_EVIDENCE_CSS.get(name, "badge-health-skipped"),
)
for name in EVIDENCE_ORDER
if name in evidence
)
def _handoff_inline(handoff: HandoffSummary) -> str:
type_css = "badge-health-ok" if handoff.cth_type_known else "badge-health-degraded"
return (
f'{_badge(handoff.cth_type or "", type_css)}'
f'<div class="muted" style="font-size:0.82rem;">'
f'{escape(handoff.status or "")}{escape(handoff.next_owner or "")}</div>'
)
def _handoff_cell(node: LinkageNode) -> str:
"""Render the handoff column, distinguishing 'none found' from 'not loaded'."""
status = node.handoff_status
if status.state == HANDOFF_LOADED:
if node.handoff is None:
return '<span class="muted">no canonical handoff on this thread</span>'
return _handoff_inline(node.handoff)
if status.state == HANDOFF_NOT_LOADED:
return (
f'{_badge("not loaded", "badge-health-skipped")}'
f'<div class="muted" style="font-size:0.82rem;">'
f'{escape(status.reason or "")}</div>'
)
return (
f'{_badge("unavailable", "badge-health-degraded")}'
f'<div class="muted" style="font-size:0.82rem;">'
f'{escape(status.reason or "")}</div>'
)
def _issue_rows(snapshot: LinkageSnapshot) -> str:
rows = []
for node in snapshot.issues:
if node.linked_prs:
linked = ", ".join(f"<code>#{number}</code>" for number in node.linked_prs)
if node.contested:
linked += " " + _badge("contested", "badge-blocked")
elif node.links_authoritative:
linked = '<span class="muted">none</span>'
else:
# The distinction an operator needs: nothing found in a window that
# was not fully loaded is not the same as nothing existing.
linked = '<span class="muted">none found (partial inventory)</span>'
rows.append(
"<tr>"
f"<td>{_ref(node)}</td>"
f"<td>{escape(node.title)}</td>"
f"<td>{escape(node.state or '')}</td>"
f"<td>{_labels(node.labels)}</td>"
f"<td>{linked}</td>"
f"<td>{_handoff_cell(node)}</td>"
"</tr>"
)
if not rows:
return '<tr><td colspan="6" class="muted">No issues in the loaded window.</td></tr>'
return "".join(rows)
def _pr_rows(snapshot: LinkageSnapshot) -> str:
rows = []
for node in snapshot.prs:
if node.links:
linked = "".join(
f"<div><code>#{link.issue_number}</code> "
f"{_evidence_badges(link.evidence)}</div>"
for link in node.links
)
if node.ambiguous:
linked += _badge("ambiguous", "badge-blocked")
if node.contested:
linked += " " + _badge("contested", "badge-blocked")
elif node.links_authoritative:
linked = '<span class="muted">no issue reference</span>'
else:
linked = '<span class="muted">none found (partial inventory)</span>'
rows.append(
"<tr>"
f"<td>{_ref(node)}</td>"
f"<td>{escape(node.title)}</td>"
f"<td>{escape(node.state or '')}</td>"
f"<td>{_labels(node.labels)}</td>"
f"<td>{linked}</td>"
f"<td>{_handoff_cell(node)}</td>"
"</tr>"
)
if not rows:
return (
'<tr><td colspan="6" class="muted">No pull requests in the loaded '
"window.</td></tr>"
)
return "".join(rows)
def _focus_card(snapshot: LinkageSnapshot) -> str:
"""Render the focused thread's latest canonical handoff, when one was loaded."""
if snapshot.focus is None:
return f"""<div class="prompt-card">
<h3>Canonical handoff</h3>
<p class="muted">{escape(snapshot.handoff_status.reason or "")}
Add <code>?issue=N</code> or <code>?pr=N</code> to load the latest
Canonical Thread Handoff for one thread.</p>
</div>"""
kind, number = snapshot.focus
target = f"{kind} #{number}"
if not snapshot.handoff_status.loaded:
return f"""<div class="prompt-card">
<h3>Canonical handoff — {escape(target)}</h3>
<p class="muted">{_badge("unavailable", "badge-health-degraded")}
{escape(snapshot.handoff_status.reason or "handoff source did not run")}.
This is not evidence that the thread carries no handoff.</p>
</div>"""
handoff = next(
(
node.handoff
for node in (snapshot.issues + snapshot.prs)
if node.kind == kind and node.number == number and node.handoff
),
None,
)
if handoff is None:
return f"""<div class="prompt-card">
<h3>Canonical handoff — {escape(target)}</h3>
<p class="muted">Comments loaded; no Canonical Thread Handoff comment found on
this thread.</p>
</div>"""
type_css = "badge-health-ok" if handoff.cth_type_known else "badge-health-degraded"
unknown_note = (
""
if handoff.cth_type_known
else (
'<p class="muted">The comment\'s heading is not a declared CTH type, '
"so it is reported as unrecognized rather than republished.</p>"
)
)
return f"""<div class="prompt-card">
<h3>Canonical handoff — {escape(target)}</h3>
<p class="meta">{_badge(handoff.cth_type or "", type_css)}
by <code>{escape(handoff.author or "unknown")}</code>
at <code>{escape(handoff.created_at or "unknown")}</code></p>
{unknown_note}
<table class="detail">
<tr><th>Status</th><td>{escape(handoff.status or "")}</td></tr>
<tr><th>Next owner</th><td>{escape(handoff.next_owner or "")}</td></tr>
<tr><th>Current blocker</th><td>{escape(handoff.current_blocker or "")}</td></tr>
<tr><th>Decision</th><td>{escape(handoff.decision or "")}</td></tr>
<tr><th>Next action</th><td>{escape(handoff.next_action or "")}</td></tr>
</table>
<p class="muted">Full event history:
<a href="/api/v1/timeline?{escape(kind)}={number}"><code>/api/v1/timeline</code></a></p>
</div>"""
def _legend_card(snapshot: LinkageSnapshot) -> str:
items = "".join(
f"<li>{_badge(_EVIDENCE_LABEL[name], _EVIDENCE_CSS[name])}"
f"{escape(EVIDENCE_DESCRIPTIONS[name])}</li>"
for name in EVIDENCE_ORDER
)
reveal_note = (
"Gitea deep links are shown because the "
"<code>GITEA_MCP_REVEAL_ENDPOINTS</code> admin opt-in is set."
if snapshot.deep_links_enabled
else (
"Gitea deep links are withheld. Set "
"<code>GITEA_MCP_REVEAL_ENDPOINTS=1</code> server-side to reveal "
"them; item numbers stay usable without them."
)
)
return f"""<div class="prompt-card">
<h3>How an edge was found</h3>
<ul class="reasons">{items}</ul>
<p class="muted">{reveal_note}</p>
<p class="muted">Read-only surface: no issue or PR editing, no review, and no merge.
JSON export: <a href="/api/v1/gitea/linkage"><code>/api/v1/gitea/linkage</code></a></p>
</div>"""
def render_linkage_page(snapshot: LinkageSnapshot) -> str:
"""Render the full HTML page for the Gitea linkage console."""
if not snapshot.ok:
return render_page(
title="Gitea linkage",
body_html=f"""<h2>Gitea issue and PR linkage</h2>
<p class="meta">Phase 3 read-only linkage console (#645).</p>
{_error_card(snapshot)}
{_scope_card(snapshot)}""",
)
body = f"""<h2>Gitea issue and PR linkage</h2>
<p class="meta">Phase 3 read-only linkage console (#645). Gitea remains the source of
truth; this page reads it and never writes to it.</p>
{_scope_card(snapshot)}
{_contested_card(snapshot)}
<div class="prompt-card">
<h3>Issues → pull requests</h3>
<table class="registry">
<thead>
<tr>
<th>Issue</th><th>Title</th><th>State</th><th>Labels</th>
<th>Linked PRs</th><th>Latest handoff</th>
</tr>
</thead>
<tbody>{_issue_rows(snapshot)}</tbody>
</table>
</div>
<div class="prompt-card">
<h3>Pull requests → issues</h3>
<table class="registry">
<thead>
<tr>
<th>PR</th><th>Title</th><th>State</th><th>Labels</th>
<th>Linked issues</th><th>Latest handoff</th>
</tr>
</thead>
<tbody>{_pr_rows(snapshot)}</tbody>
</table>
</div>
{_focus_card(snapshot)}
{_legend_card(snapshot)}
"""
return render_page(title="Gitea linkage", body_html=body)
+1 -7
View File
@@ -6,8 +6,7 @@ destination is a GET view or a Phase 1 placeholder. No mutation links.
Nav groups follow the #631 Phase 1 information architecture: Health, Traffic,
Runtime/Sessions, Projects, Inventory, Timeline, Policy (placeholder), and
Insights (placeholder), joined by the Phase 3 Gitea linkage group (#645).
Later-phase surfaces are declared as ``stub`` items and
Insights (placeholder). Later-phase surfaces are declared as ``stub`` items and
backed by ``STUB_PAGES`` so their nav links resolve to a graceful placeholder
instead of a 404.
"""
@@ -46,8 +45,6 @@ NAV_GROUPS: tuple[NavGroup, ...] = (
NavItem("/queue", "Queue"),
NavItem("/leases", "Leases"),
NavItem("/actions", "Actions"),
NavItem("/notifications", "Notifications"),
NavItem("/requests", "Requests"),
)),
NavGroup("Runtime/Sessions", (
NavItem("/runtime", "Runtime health"),
@@ -63,9 +60,6 @@ NAV_GROUPS: tuple[NavGroup, ...] = (
NavGroup("Timeline", (
NavItem("/timeline", "Timeline", "stub"),
)),
NavGroup("Gitea", (
NavItem("/gitea", "Issue/PR linkage"),
)),
NavGroup("Policy", (
NavItem("/policy", "Policy", "stub"),
NavItem("/prompts", "Prompts"),
-158
View File
@@ -1,158 +0,0 @@
"""HTML rendering for Phase 3 Notifications and Human-Attention Console (#648)."""
from __future__ import annotations
from html import escape
from typing import Sequence
from webui.layout import render_page
from webui.notifications import (
ATTENTION_HUMAN_REQUIRED,
ATTENTION_OPERATOR,
ATTENTION_ROUTINE,
NotificationItem,
NotificationSnapshot,
)
def _render_attention_badge(attention_class: str) -> str:
cls = "badge"
if attention_class == ATTENTION_HUMAN_REQUIRED:
cls += " badge-blocked"
elif attention_class == ATTENTION_OPERATOR:
cls += " badge-claimed"
else:
cls += " muted"
return f'<span class="{cls}">{escape(attention_class)}</span>'
def _render_notification_row(item: NotificationItem) -> str:
category_label = escape(item.category.upper())
id_str = escape(item.id)
title_str = escape(item.title)
summary_str = escape(item.summary)
att_badge = _render_attention_badge(item.attention_class)
work_item_html = ""
if item.work_number and item.work_kind:
kind_label = escape(item.work_kind.upper())
num_str = f"#{item.work_number}"
link = item.deep_link or "#"
work_item_html = f'<a href="{escape(link)}"><code>{kind_label} {num_str}</code></a>'
requires_human_label = (
'<span class="badge badge-blocked" style="font-size:0.75rem;">HUMAN REQUIRED</span>'
if item.requires_human
else ""
)
return f"""<tr>
<td><code>{category_label}</code><br><span class="muted" style="font-size:0.75rem;">{id_str}</span></td>
<td>
<div><strong>{title_str}</strong> {att_badge} {requires_human_label}</div>
<div class="muted" style="font-size:0.85rem; margin-top:0.25rem;">{summary_str}</div>
</td>
<td>{work_item_html}</td>
<td><span class="muted" style="font-size:0.8rem;">{escape(item.created_at[:19])}</span></td>
</tr>"""
def _render_notifications_table(items: Sequence[NotificationItem], empty_message: str) -> str:
if not items:
return f'<p class="muted" style="padding:1rem 0;">{escape(empty_message)}</p>'
rows = "".join(_render_notification_row(item) for item in items)
return f"""<table class="registry">
<thead>
<tr>
<th style="width: 18%;">Category & ID</th>
<th style="width: 52%;">Title & Attention Summary</th>
<th style="width: 15%;">Work Item</th>
<th style="width: 15%;">Time</th>
</tr>
</thead>
<tbody>
{rows}
</tbody>
</table>"""
def render_notifications_page(
snapshot: NotificationSnapshot,
*,
filter_class: str = "inbox",
filter_project: str | None = None,
) -> str:
"""Render the notifications and attention inbox page."""
title = "Notifications & Attention Inbox"
err_html = ""
if snapshot.fetch_error:
err_html = f'<div class="stub" style="border-color:#e53e3e; background:#fff5f5; color:#c53030; margin-bottom:1rem;"><p><strong>Fetch Warning:</strong> {escape(snapshot.fetch_error)}</p></div>'
# Determine items to render based on filter_class
if filter_class == ATTENTION_HUMAN_REQUIRED:
display_items = snapshot.human_required_items
active_tab_title = "Human-Required Escalations"
elif filter_class == ATTENTION_OPERATOR:
display_items = snapshot.operator_items
active_tab_title = "Operator Inbox Items"
elif filter_class == ATTENTION_ROUTINE:
display_items = snapshot.routine_items
active_tab_title = "Routine Workflow Transitions"
elif filter_class == "all":
display_items = snapshot.items
active_tab_title = "All Events (including Routine)"
else: # "inbox" default
display_items = snapshot.inbox_items
active_tab_title = "Attention Inbox (Human + Operator)"
hr_cls = "badge-blocked" if snapshot.human_required_count > 0 else "muted"
op_cls = "badge-claimed" if snapshot.operator_count > 0 else "muted"
metrics_html = f"""<div style="display:flex; gap:1rem; margin-bottom:1.5rem;">
<div class="health-card" style="flex:1;">
<span class="muted" style="font-size:0.85rem;">Human Required</span>
<h2 style="margin:0.2rem 0;"><span class="badge {hr_cls}" style="font-size:1.4rem;">{snapshot.human_required_count}</span></h2>
<p class="muted" style="font-size:0.8rem; margin:0;">Critical escalation boundary</p>
</div>
<div class="health-card" style="flex:1;">
<span class="muted" style="font-size:0.85rem;">Operator Inbox</span>
<h2 style="margin:0.2rem 0;"><span class="badge {op_cls}" style="font-size:1.4rem;">{snapshot.operator_count}</span></h2>
<p class="muted" style="font-size:0.8rem; margin:0;">Operational items needing review</p>
</div>
<div class="health-card" style="flex:1;">
<span class="muted" style="font-size:0.85rem;">Routine Transitions</span>
<h2 style="margin:0.2rem 0;"><span class="badge muted" style="font-size:1.4rem;">{snapshot.routine_count}</span></h2>
<p class="muted" style="font-size:0.8rem; margin:0;">Background transitions (filtered)</p>
</div>
</div>"""
# Filter navigation links
def _tab_link(target_class: str, label: str) -> str:
is_active = (filter_class == target_class)
style = "font-weight:bold; border-bottom:2px solid currentColor;" if is_active else "color:#4a5568;"
return f'<a href="/notifications?attention_class={target_class}" style="margin-right:1.25rem; text-decoration:none; padding-bottom:0.25rem; {style}">{label}</a>'
tabs_html = f"""<div style="margin-bottom:1.25rem; border-bottom:1px solid #e2e8f0; padding-bottom:0.5rem;">
{_tab_link("inbox", f"Attention Inbox ({snapshot.human_required_count + snapshot.operator_count})")}
{_tab_link("human-required", f"Human Required ({snapshot.human_required_count})")}
{_tab_link("operator", f"Operator ({snapshot.operator_count})")}
{_tab_link("routine", f"Routine ({snapshot.routine_count})")}
{_tab_link("all", f"All Events ({snapshot.total_count})")}
</div>"""
table_html = _render_notifications_table(
display_items,
f"No items match attention filter '{filter_class}'.",
)
body = f"""<h2>{escape(title)}</h2>
<p class="muted">Phase 3 console surface for human-attention routing (#648). Routine workflow transitions are filtered by default to eliminate notification fatigue.</p>
{err_html}
{metrics_html}
{tabs_html}
<h3>{escape(active_tab_title)}</h3>
{table_html}"""
return render_page(title=title, body_html=body)
-486
View File
@@ -1,486 +0,0 @@
"""Notifications and human-attention routing module for Phase 3 web console (#648).
Defines attention classes, event classification rules, and inbox aggregation so
operators receive direct alerts only for human-required escalation boundaries
(#628) while routine workflow transitions remain available for pull-based review.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Callable
from webui import console_redaction
from webui.project_registry import load_registry
from webui.queue_loader import QueueSnapshot, load_queue_snapshot
from webui.lease_loader import LeaseSnapshot, load_lease_snapshot
from webui.system_health import SystemHealthSnapshot, load_system_health
# Attention class definitions (#628, #648)
ATTENTION_ROUTINE = "routine"
ATTENTION_OPERATOR = "operator"
ATTENTION_HUMAN_REQUIRED = "human-required"
ATTENTION_CLASSES = (
ATTENTION_ROUTINE,
ATTENTION_OPERATOR,
ATTENTION_HUMAN_REQUIRED,
)
# Notification categories
CATEGORY_AUTH = "auth"
CATEGORY_BLOCKER = "blocker"
CATEGORY_LEASE = "lease"
CATEGORY_VALIDATION = "validation"
CATEGORY_WORKFLOW = "workflow"
CATEGORY_SYSTEM = "system"
CATEGORIES = (
CATEGORY_AUTH,
CATEGORY_BLOCKER,
CATEGORY_LEASE,
CATEGORY_VALIDATION,
CATEGORY_WORKFLOW,
CATEGORY_SYSTEM,
)
@dataclass(frozen=True)
class NotificationItem:
"""A single notification or inbox event."""
id: str
attention_class: str # "routine", "operator", "human-required"
category: str # "auth", "blocker", "lease", "validation", etc.
title: str
summary: str
work_kind: str | None # "issue", "pr", "session", "system"
work_number: int | None
project_id: str
repo_label: str
created_at: str
deep_link: str | None = None
requires_human: bool = False
extra: dict[str, Any] = field(default_factory=dict)
def as_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"attention_class": self.attention_class,
"category": self.category,
"title": self.title,
"summary": console_redaction.redact_text(self.summary),
"work_kind": self.work_kind,
"work_number": self.work_number,
"project_id": self.project_id,
"repo_label": self.repo_label,
"created_at": self.created_at,
"deep_link": self.deep_link,
"requires_human": self.requires_human,
"extra": self.extra,
}
@dataclass(frozen=True)
class NotificationSnapshot:
"""Snapshot of notifications and attention inbox state."""
project_id: str
repo_label: str
items: tuple[NotificationItem, ...]
human_required_count: int
operator_count: int
routine_count: int
total_count: int
fetch_error: str | None = None
@property
def inbox_items(self) -> tuple[NotificationItem, ...]:
"""Items requiring operator or human attention (excluding routine)."""
return tuple(
item
for item in self.items
if item.attention_class in {ATTENTION_OPERATOR, ATTENTION_HUMAN_REQUIRED}
)
@property
def human_required_items(self) -> tuple[NotificationItem, ...]:
return tuple(
item for item in self.items if item.attention_class == ATTENTION_HUMAN_REQUIRED
)
@property
def operator_items(self) -> tuple[NotificationItem, ...]:
return tuple(
item for item in self.items if item.attention_class == ATTENTION_OPERATOR
)
@property
def routine_items(self) -> tuple[NotificationItem, ...]:
return tuple(
item for item in self.items if item.attention_class == ATTENTION_ROUTINE
)
def as_dict(self) -> dict[str, Any]:
return {
"project_id": self.project_id,
"repo_label": self.repo_label,
"human_required_count": self.human_required_count,
"operator_count": self.operator_count,
"routine_count": self.routine_count,
"total_count": self.total_count,
"fetch_error": self.fetch_error,
"inbox_items": [item.as_dict() for item in self.inbox_items],
"all_items": [item.as_dict() for item in self.items],
}
def classify_attention_event(
category: str,
title: str,
summary: str,
*,
is_hard_stop: bool = False,
is_auth_failure: bool = False,
is_irrecoverable: bool = False,
is_decision_lock: bool = False,
is_validation_failure: bool = False,
is_stale: bool = False,
is_blocker: bool = False,
) -> tuple[str, bool]:
"""Classify an event into an attention class and human requirement flag.
Rules (#628, #648):
1. Critical boundaries (hard stop, auth failure, irrecoverable state,
decision lock, validation failure) -> ATTENTION_HUMAN_REQUIRED (requires_human=True).
2. Operational queues (blocker, stale lease, unassigned ready work, queue collision)
-> ATTENTION_OPERATOR (requires_human=False).
3. Routine state transitions (clean progression, healthy heartbeats) -> ATTENTION_ROUTINE (requires_human=False).
Classification uses structured flags and category only. Human-authored
``title`` / ``summary`` text is never substring-matched for escalation
(PR #905 review B1) — callers that need text signals must set flags from
machine-generated status/detail fields before calling this function.
"""
del title, summary # kept for API stability; never used for classification
if (
is_hard_stop
or is_auth_failure
or is_irrecoverable
or is_decision_lock
or is_validation_failure
or category in {CATEGORY_AUTH, CATEGORY_VALIDATION}
):
return ATTENTION_HUMAN_REQUIRED, True
if is_stale or is_blocker or category in {CATEGORY_BLOCKER, CATEGORY_LEASE}:
return ATTENTION_OPERATOR, False
return ATTENTION_ROUTINE, False
def load_notifications_snapshot(
project_id: str | None = None,
*,
load_queue: Callable[..., QueueSnapshot] | None = None,
load_leases: Callable[..., LeaseSnapshot] | None = None,
load_health: Callable[..., SystemHealthSnapshot] | None = None,
) -> NotificationSnapshot:
"""Load and classify attention notifications across queue, leases, and system health."""
registry = load_registry()
project = None
if project_id:
for entry in registry.projects:
if entry.id == project_id:
project = entry
break
else:
project = registry.projects[0] if registry.projects else None
if project is None:
return NotificationSnapshot(
project_id=project_id or "",
repo_label="",
items=(),
human_required_count=0,
operator_count=0,
routine_count=0,
total_count=0,
fetch_error="project not found in registry",
)
queue_loader_fn = load_queue or load_queue_snapshot
lease_loader_fn = load_leases or load_lease_snapshot
health_loader_fn = load_health or load_system_health
try:
queue_snap = queue_loader_fn(project.id)
except TypeError:
queue_snap = queue_loader_fn(project_id=project.id)
try:
lease_snap = lease_loader_fn(project_id=project.id)
except TypeError:
lease_snap = lease_loader_fn(project.id)
try:
health_snap = health_loader_fn(project_id=project.id)
except TypeError:
try:
health_snap = health_loader_fn(project.id)
except TypeError:
health_snap = health_loader_fn()
items: list[NotificationItem] = []
now_iso = datetime.now(timezone.utc).isoformat()
# 1. System health alerts (highest priority)
for err_idx, probe_err in enumerate(getattr(health_snap, "probe_errors", ())):
att_cls, req_human = classify_attention_event(
CATEGORY_SYSTEM,
"System Health Probe Error",
probe_err,
is_blocker=True,
)
items.append(
NotificationItem(
id=f"notif-sys-err-{project.id}-{err_idx}",
attention_class=att_cls,
category=CATEGORY_SYSTEM,
title="System Health Error",
summary=f"System health error: {probe_err}",
work_kind="system",
work_number=None,
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
created_at=now_iso,
deep_link="/system",
requires_human=req_human,
)
)
for probe in getattr(health_snap, "dependencies", ()):
if probe.status not in ("ok", "healthy"):
att_cls, req_human = classify_attention_event(
CATEGORY_SYSTEM,
f"Probe Failure: {probe.name}",
probe.detail or probe.status,
is_hard_stop=("stop" in probe.status or "fatal" in probe.status),
is_auth_failure=("auth" in probe.name.lower() or "unauthorized" in probe.status.lower()),
is_blocker=True,
)
items.append(
NotificationItem(
id=f"notif-probe-{probe.name}",
attention_class=att_cls,
category=CATEGORY_AUTH if "auth" in probe.name.lower() else CATEGORY_SYSTEM,
title=f"Health Probe Alert: {probe.name}",
summary=f"Probe '{probe.name}' reported status '{probe.status}': {probe.detail}",
work_kind="system",
work_number=None,
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
created_at=now_iso,
deep_link="/system",
requires_human=req_human,
)
)
# 2. Queue items (PRs and Issues)
for pr in queue_snap.prs:
if "blocked" in pr.badges:
att_cls, req_human = classify_attention_event(
CATEGORY_BLOCKER,
f"PR #{pr.number} Blocked",
f"PR #{pr.number} '{pr.title}' is blocked or has merge conflicts.",
is_blocker=True,
)
items.append(
NotificationItem(
id=f"notif-pr-block-{pr.number}",
attention_class=att_cls,
category=CATEGORY_BLOCKER,
title=f"Blocked PR #{pr.number}",
summary=f"PR #{pr.number} ({pr.title}) requires merge conflict resolution.",
work_kind="pr",
work_number=pr.number,
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
created_at=now_iso,
deep_link=f"/traffic",
requires_human=req_human,
)
)
elif "stale" in pr.badges:
att_cls, req_human = classify_attention_event(
CATEGORY_WORKFLOW,
f"PR #{pr.number} Stale",
f"PR #{pr.number} '{pr.title}' has had no activity for over 14 days.",
is_stale=True,
)
items.append(
NotificationItem(
id=f"notif-pr-stale-{pr.number}",
attention_class=att_cls,
category=CATEGORY_WORKFLOW,
title=f"Stale PR #{pr.number}",
summary=f"PR #{pr.number} ({pr.title}) is stale.",
work_kind="pr",
work_number=pr.number,
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
created_at=now_iso,
deep_link=f"/queue",
requires_human=req_human,
)
)
else:
# Routine PR transition
att_cls, req_human = classify_attention_event(
CATEGORY_WORKFLOW,
f"PR #{pr.number} Active",
f"PR #{pr.number} '{pr.title}' is in routine state {', '.join(pr.badges)}.",
)
items.append(
NotificationItem(
id=f"notif-pr-routine-{pr.number}",
attention_class=att_cls,
category=CATEGORY_WORKFLOW,
title=f"Routine PR #{pr.number}",
summary=f"PR #{pr.number} ({pr.title}) state: {', '.join(pr.badges)}.",
work_kind="pr",
work_number=pr.number,
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
created_at=now_iso,
deep_link=f"/queue",
requires_human=req_human,
)
)
for issue in queue_snap.issues:
if "duplicate" in issue.badges:
att_cls, req_human = classify_attention_event(
CATEGORY_BLOCKER,
f"Issue #{issue.number} Duplicate PRs",
f"Issue #{issue.number} has multiple linked PRs.",
is_blocker=True,
)
items.append(
NotificationItem(
id=f"notif-issue-dup-{issue.number}",
attention_class=att_cls,
category=CATEGORY_BLOCKER,
title=f"Duplicate PRs on Issue #{issue.number}",
summary=f"Issue #{issue.number} ({issue.title}) linked to multiple PRs.",
work_kind="issue",
work_number=issue.number,
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
created_at=now_iso,
deep_link=f"/traffic",
requires_human=req_human,
)
)
elif "claimed" in issue.badges or "in-review" in issue.badges:
att_cls, req_human = classify_attention_event(
CATEGORY_WORKFLOW,
f"Issue #{issue.number} Active",
f"Issue #{issue.number} '{issue.title}' in state {', '.join(issue.badges)}.",
)
items.append(
NotificationItem(
id=f"notif-issue-routine-{issue.number}",
attention_class=att_cls,
category=CATEGORY_WORKFLOW,
title=f"Routine Issue #{issue.number}",
summary=f"Issue #{issue.number} ({issue.title}) state: {', '.join(issue.badges)}.",
work_kind="issue",
work_number=issue.number,
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
created_at=now_iso,
deep_link=f"/queue",
requires_human=req_human,
)
)
# 3. Leases / Collisions
for lease in lease_snap.reviewer_leases:
if lease.get("is_expired") or lease.get("status") == "expired":
pr_num = lease.get("pr_number") or lease.get("work_item_number")
att_cls, req_human = classify_attention_event(
CATEGORY_LEASE,
f"Reviewer Lease Expired for PR #{pr_num}",
f"Reviewer lease for PR #{pr_num} has expired.",
is_stale=True,
)
items.append(
NotificationItem(
id=f"notif-lease-exp-pr-{pr_num}",
attention_class=att_cls,
category=CATEGORY_LEASE,
title=f"Expired Reviewer Lease (PR #{pr_num})",
summary=f"Reviewer lease for PR #{pr_num} expired.",
work_kind="pr",
work_number=pr_num,
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
created_at=now_iso,
deep_link="/leases",
requires_human=req_human,
)
)
for col_idx, collision in enumerate(lease_snap.duplicate_prs):
att_cls, req_human = classify_attention_event(
CATEGORY_BLOCKER,
f"Duplicate PR Collision ({collision.kind})",
collision.message,
is_blocker=True,
)
issue_part = collision.issue_number if collision.issue_number is not None else "none"
kind_part = (collision.kind or "unknown").replace(" ", "-")
items.append(
NotificationItem(
id=f"notif-collision-{kind_part}-{issue_part}-{col_idx}",
attention_class=att_cls,
category=CATEGORY_BLOCKER,
title=f"Collision Alert ({collision.kind})",
summary=collision.message,
work_kind="issue" if collision.issue_number else "pr",
work_number=collision.issue_number,
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
created_at=now_iso,
deep_link="/leases",
requires_human=req_human,
)
)
human_req_count = sum(1 for i in items if i.attention_class == ATTENTION_HUMAN_REQUIRED)
operator_count = sum(1 for i in items if i.attention_class == ATTENTION_OPERATOR)
routine_count = sum(1 for i in items if i.attention_class == ATTENTION_ROUTINE)
# Fetch errors are transport/load failures only — not probe results that
# already surface as first-class notification items (PR #905 review B3).
fetch_err = queue_snap.fetch_error or lease_snap.fetch_error
if isinstance(fetch_err, (tuple, list)):
fetch_err = "; ".join(fetch_err) if fetch_err else None
return NotificationSnapshot(
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
items=tuple(items),
human_required_count=human_req_count,
operator_count=operator_count,
routine_count=routine_count,
total_count=len(items),
fetch_error=fetch_err,
)
def snapshot_to_dict(snapshot: NotificationSnapshot) -> dict[str, Any]:
"""JSON-serializable export for /api/v1/notifications."""
return snapshot.as_dict()
+2 -27
View File
@@ -211,19 +211,6 @@ def _pagination_from_pages(
)
_SUPPORTED_FETCH_STATES = ("open", "closed", "all")
def _safe_state(state: str | None) -> str:
"""Constrain a caller-supplied item state before it reaches a query string.
The value is interpolated into the Gitea URL, so an unrecognised state falls
back to ``open`` rather than being passed through.
"""
text = (state or "open").strip().lower()
return text if text in _SUPPORTED_FETCH_STATES else "open"
def _fetch_prs(
host: str,
org: str,
@@ -231,15 +218,8 @@ def _fetch_prs(
auth: str,
*,
per_page: int = 50,
state: str = "open",
) -> tuple[list[dict], PaginationMeta]:
"""Fetch PRs in *state* (``open``, ``closed``, or ``all``).
The queue dashboard only ever wants the open window, so ``open`` stays the
default. The linkage console (#645) widens it, because a landed issue↔PR
edge lives on a merged PR.
"""
url = f"{repo_api_url(host, org, repo)}/pulls?state={_safe_state(state)}"
url = f"{repo_api_url(host, org, repo)}/pulls?state=open"
all_raw: list[dict] = []
pages_fetched = 0
is_final = False
@@ -268,13 +248,8 @@ def _fetch_issues(
auth: str,
*,
per_page: int = 50,
state: str = "open",
) -> tuple[list[dict], PaginationMeta]:
"""Fetch issues in *state* (``open``, ``closed``, or ``all``); see :func:`_fetch_prs`."""
url = (
f"{repo_api_url(host, org, repo)}/issues"
f"?state={_safe_state(state)}&type=issues"
)
url = f"{repo_api_url(host, org, repo)}/issues?state=open&type=issues"
all_raw: list[dict] = []
page = 1
pages_fetched = 0
File diff suppressed because it is too large Load Diff
-164
View File
@@ -1,164 +0,0 @@
"""HTML views for the operator request surface (#643).
The form is deliberately a *preview* form. It has no initiate button, because
initiating requires a confirmed POST to ``/api/v1/requests/apply`` and a stray
form submission must not be able to produce one by accident.
Nothing rendered here is trusted input: every interpolated value is escaped,
and the page renders only values the service already produced rather than
echoing a raw request body back.
"""
from __future__ import annotations
import html
import json
from typing import Any
from webui.layout import render_page
from webui.request_service import (
REQUESTABLE_ROLES,
WORK_KINDS,
RequestError,
RequestPreview,
)
REQUESTS_PATH = "/requests"
PREVIEW_API_PATH = "/api/v1/requests/preview"
APPLY_API_PATH = "/api/v1/requests/apply"
def _escape(text: Any) -> str:
return html.escape(str(text if text is not None else ""), quote=True)
REQUEST_PAGE_STYLES = """
<style>
.request-form { display: grid; gap: 0.75rem; max-width: 44rem; }
.request-form label { display: grid; gap: 0.25rem; font-size: 0.9rem; }
.request-check { margin: 0.35rem 0; }
.request-check .verdict-ok { color: var(--accent); }
.request-check .verdict-fail { color: #d14; }
.request-prohibited code { margin-right: 0.4rem; }
</style>
"""
def _options(values: tuple[str, ...], selected: Any) -> str:
return "".join(
f"<option value='{_escape(value)}'"
+ (" selected" if selected == value else "")
+ f">{_escape(value)}</option>"
for value in values
)
def _form(values: dict[str, Any] | None = None) -> str:
current = dict(values or {})
number = current.get("work_number")
return (
f"<form class='request-form' method='post' action='{REQUESTS_PATH}'>"
"<label>Desired role<select name='desired_role'>"
f"{_options(REQUESTABLE_ROLES, current.get('desired_role'))}"
"</select></label>"
"<label>Work kind<select name='work_kind'>"
f"{_options(WORK_KINDS, current.get('work_kind'))}"
"</select></label>"
"<label>Issue or PR number"
"<input type='number' name='work_number' min='1' required "
f"value='{_escape(number) if number else ''}'></label>"
"<label>Intent summary"
"<input type='text' name='intent_summary' maxlength='500' required "
f"value='{_escape(current.get('intent_summary'))}'></label>"
"<label>Expected head SHA <span class='muted'>(PR work only)</span>"
"<input type='text' name='expected_head_sha' "
f"value='{_escape(current.get('expected_head_sha'))}'></label>"
"<button type='submit' class='copy-btn'>Preview request</button>"
"<p class='muted meta'>Preview is read-only and creates no assignment. "
f"Initiating requires a confirmed POST to <code>{APPLY_API_PATH}</code>."
"</p>"
"</form>"
)
def _checks_block(preview: RequestPreview) -> str:
rows = []
for check in preview.checks:
verdict = "PASS" if check.ok else "FAIL"
css = "verdict-ok" if check.ok else "verdict-fail"
rows.append(
"<li class='request-check'>"
f"<span class='{css}'><strong>{verdict}</strong></span> "
f"<code>{_escape(check.name)}</code> — {_escape(check.detail)} "
f"<span class='muted meta'>({_escape(check.reason_code)})</span>"
"</li>"
)
return "<ul>" + "".join(rows) + "</ul>"
def _preview_block(preview: RequestPreview) -> str:
verdict = "AUTHORIZED" if preview.authorized else "DENIED"
prohibited = "".join(
f"<code>{_escape(action)}</code>" for action in preview.prohibited_actions
)
request = preview.request
evidence = json.dumps(preview.allocator_evidence, indent=2, default=str)
return (
"<h3>Intent preview</h3>"
f"<p><strong>{verdict}</strong> — {_escape(preview.detail)}</p>"
"<p class='meta'>"
f"Role <code>{_escape(request.desired_role)}</code> · "
f"{_escape(request.work_kind)} <code>{_escape(request.display_ref)}</code>"
f" · profile <code>{_escape(preview.required_profile)}</code> · "
f"namespace <code>{_escape(preview.required_namespace)}</code> · "
f"permission <code>{_escape(preview.required_permission)}</code>"
"</p>"
f"<p>Intent: {_escape(request.intent_summary)}</p>"
f"{_checks_block(preview)}"
f"<p><strong>Next safe action:</strong> "
f"{_escape(preview.next_safe_action)}</p>"
"<p class='request-prohibited'><strong>Prohibited for this role:</strong> "
+ (prohibited or "<span class='muted'>none declared</span>")
+ "</p>"
"<p class='muted meta'>Correlation id "
f"<code>{_escape(preview.correlation_id)}</code></p>"
"<details><summary>Allocator evidence</summary>"
f"<pre class='prompt-text'>{_escape(evidence)}</pre>"
"</details>"
)
def _error_block(error: RequestError) -> str:
field = (
f"<p class='meta'>Field: <code>{_escape(error.field_name)}</code></p>"
if error.field_name
else ""
)
return (
"<h3>Request rejected</h3>"
f"<p><strong>{_escape(error.reason_code)}</strong> — "
f"{_escape(error.detail)}</p>{field}"
)
def render_requests_page(
*,
preview: RequestPreview | None = None,
error: RequestError | None = None,
submitted: dict[str, Any] | None = None,
) -> str:
"""Render the request form, plus a preview or rejection when one exists."""
body = (
"<h2>Requests</h2>"
"<p>Submit a work request — desired role, issue or PR, and intent — "
"and see whether it would be authorized before anything is reserved. "
"Initiation goes through the allocator (#600/#613); this console never "
"self-selects work, never approves, and never merges.</p>"
+ _form(submitted)
+ (_error_block(error) if error is not None else "")
+ (_preview_block(preview) if preview is not None else "")
+ f"<p class='meta'><a href='{PREVIEW_API_PATH}'>Preview API</a> · "
"<a href='/api/console/security-model'>RBAC model</a></p>"
+ REQUEST_PAGE_STYLES
)
return render_page(title="Requests", body_html=body)
-10
View File
@@ -201,16 +201,6 @@ def _candidates_from_queue_snapshot(q_snap: QueueSnapshot) -> list[WorkCandidate
return candidates
def candidates_from_queue_snapshot(q_snap: QueueSnapshot) -> list[WorkCandidate]:
"""Public alias for :func:`_candidates_from_queue_snapshot` (#643).
The request-initiation service ranks the same candidate set this view
renders, so both must agree on how a queue row becomes a candidate. One
construction, two callers — not two that can drift apart.
"""
return _candidates_from_queue_snapshot(q_snap)
def _claim_lease_records(inventory: dict[str, Any] | None) -> list[dict[str, Any]]:
"""Normalize ``build_claim_inventory`` entries into lease records.