Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 sysadmin
|
22e0a41bd5 |
@@ -1,94 +0,0 @@
|
||||
# MCP scoped recovery playbook (#669)
|
||||
|
||||
**Parent:** [#655](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/655)
|
||||
**Vision / roadmap:** [#652](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/652) · [#653](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/653)
|
||||
**Class matrix:** [#663](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/663) · `docs/mcp-restart-classes.md`
|
||||
**Coordinator:** [#658](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/658) · `restart_coordinator.py`
|
||||
**Audit lineage:** [#665](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/665)
|
||||
|
||||
## Decision
|
||||
|
||||
Full-server MCP reset is a **last resort**. Prefer the narrowest recovery that
|
||||
can clear the symptom. The coordinator **refuses** `rolling_mcp_restart`,
|
||||
`full_mcp_restart`, and `host_restart` unless:
|
||||
|
||||
1. The inventory carries a prior **attempt log** of at least one *insufficient*
|
||||
narrower recovery, **or**
|
||||
2. **Break-glass** is authorized
|
||||
(`request_break_glass` + `GITEA_BREAKGLASS_RESTART_AUTHORIZATION`).
|
||||
|
||||
Break-glass still never bypasses the #663 class matrix (role/permission).
|
||||
|
||||
## Ladder (narrow → broad)
|
||||
|
||||
| Rank | Action | Self-service | Implementation / delegation |
|
||||
|---:|---|---|---|
|
||||
| 0 | `client_reconnect` | yes | Host auto-reconnect / client reconnect · [#584](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/584) · `docs/mcp-namespace-eof-recovery.md` |
|
||||
| 1 | `capability_refresh` | yes | `gitea_resolve_task_capability` + `gitea_whoami` · [#610](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/610) · [#685](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/685) |
|
||||
| 2 | `session_reconnect` | yes | Runtime rebind + explicit `worktree_path` · [#543](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/543) · [#618](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/618) |
|
||||
| 3 | `configuration_reload` | no | Class `configuration_reload` · console reload · [#642](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/642) |
|
||||
| 4 | `lease_recovery` | no | Lock/lease recovery paths · [#702](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/702) · [#753](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/753) · [#790](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/790) |
|
||||
| 5 | `worker_restart` | no | Class `worker_restart` · #663 |
|
||||
| 6 | `role_runtime_restart` | no | Class `role_runtime_restart` · console restart · #642/#663 |
|
||||
| 7 | `connector_restart` | no | Class `connector_restart` · #663 |
|
||||
| 8 | `rolling_mcp_restart` | no | Class `rolling_mcp_restart` · design [#668](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/668) · **attempt log required** |
|
||||
| 9 | `full_mcp_restart` | no | Class `full_mcp_restart` · **attempt log required** |
|
||||
| 10 | `host_restart` | no | Class `host_restart` · **attempt log required** |
|
||||
|
||||
Machine-readable source of truth: `recovery_playbook.RECOVERY_LADDER` and
|
||||
`recovery_playbook.ladder_document()`.
|
||||
|
||||
## Attempt log shape
|
||||
|
||||
Each prior attempt is a mapping:
|
||||
|
||||
```json
|
||||
{
|
||||
"action": "client_reconnect",
|
||||
"outcome": "insufficient",
|
||||
"reason": "transport still closed after IDE reconnect",
|
||||
"actor": "prgs-controller-12345",
|
||||
"recorded_at": "2026-07-25T21:00:00+00:00"
|
||||
}
|
||||
```
|
||||
|
||||
Outcomes that count toward escalation: `failed`, `insufficient`, `denied`,
|
||||
`unresolved`, `timeout`, `error`.
|
||||
|
||||
Pass attempts into the coordinator via inventory
|
||||
`prior_recovery_attempts` or the MCP tool argument
|
||||
`prior_recovery_attempts_json` on `gitea_request_mcp_restart`.
|
||||
|
||||
Helper: `recovery_playbook.build_attempt_record(...)`.
|
||||
|
||||
## Symptom → first rung
|
||||
|
||||
`recovery_playbook.recommend_actions(symptoms=[...])` maps symptoms such as
|
||||
`transport_eof`, `stale_capability`, `stale_lease`, `daemon_corrupt` to the
|
||||
narrowest recommended action, then walks the ladder. Soft recommendations
|
||||
never replace the hard gate on broad restarts.
|
||||
|
||||
## Enforcement points
|
||||
|
||||
1. **`recovery_playbook.assess_escalation`** — pure gate.
|
||||
2. **`restart_coordinator.evaluate_restart_impact`** — when `restart_class` is
|
||||
set (policy-enforced path), broad classes require the gate; report fields
|
||||
`attempt_log_satisfied`, `playbook_escalation`, `break_glass`.
|
||||
3. **`gitea_request_mcp_restart`** — accepts attempt JSON and env-authorized
|
||||
break-glass; never restarts a process.
|
||||
|
||||
## Metrics
|
||||
|
||||
`recovery_playbook.recovery_metrics(attempts)` reports the fraction of
|
||||
successful recoveries that avoided full/host restart
|
||||
(`fraction_avoided_full_restart`).
|
||||
|
||||
## Non-goals
|
||||
|
||||
* HA multi-instance execution (#668 design only here).
|
||||
* Normalizing `pkill` (#630 contamination stays forbidden).
|
||||
* Silent mutation of leases or processes from the playbook itself.
|
||||
|
||||
## Manual process kills
|
||||
|
||||
Remain forbidden and contaminating (#630). The playbook never recommends them.
|
||||
@@ -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.
|
||||
@@ -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
|
||||
|
||||
@@ -95,18 +96,12 @@ gitea_request_mcp_restart(remote, host, org, repo,
|
||||
target_session_id=None, target_role=None,
|
||||
target_connector=None,
|
||||
drain_proof_json=None,
|
||||
request_break_glass=False,
|
||||
prior_recovery_attempts_json=None)
|
||||
request_break_glass=False)
|
||||
```
|
||||
|
||||
It **never restarts anything**: `apply_supported` is always `false` and
|
||||
`restart_performed` is always `false`.
|
||||
|
||||
`prior_recovery_attempts_json` (#669) is an optional JSON array of prior
|
||||
narrow recovery attempts. Rolling / full / host classes require at least one
|
||||
*insufficient* narrower attempt (or authorized break-glass). See
|
||||
`docs/mcp-recovery-playbook.md`.
|
||||
|
||||
### Dry-run versus apply
|
||||
|
||||
| Call | Behavior |
|
||||
@@ -118,10 +113,9 @@ narrow recovery attempts. Rolling / full / host classes require at least one
|
||||
|
||||
An apply requires **both** authorizations, and they are independent:
|
||||
|
||||
1. **Restart-class authorization** (#663 / #669) — the requester's role and
|
||||
permissions must allow the requested class, the class's approval requirement
|
||||
must be satisfied, any target-scoped class must name its target, and broad
|
||||
classes must satisfy the recovery-playbook attempt-log gate. Failing any of
|
||||
1. **Restart-class authorization** (#663) — the requester's role and permissions
|
||||
must allow the requested class, the class's approval requirement must be
|
||||
satisfied, and any target-scoped class must name its target. Failing any of
|
||||
these makes `allow_restart` `false`.
|
||||
2. **Drain-proof gate** (#661) — a valid, unexpired, clean proof bound to the
|
||||
current impact fingerprint, or an authorized break-glass.
|
||||
@@ -134,10 +128,7 @@ the authorization that produced it.
|
||||
|
||||
### Break-glass
|
||||
|
||||
Break-glass bypasses the **drain proof only** — never the restart-class matrix
|
||||
(role/permission). Separately, authorized break-glass also satisfies the #669
|
||||
attempt-log requirement for broad restarts (rolling/full/host), because that
|
||||
gate is not a class-matrix permission check.
|
||||
Break-glass bypasses the **drain proof only** — never the restart-class matrix.
|
||||
It is honoured solely when `request_break_glass` is set *and* the environment
|
||||
carries `GITEA_BREAKGLASS_RESTART_AUTHORIZATION`; like operator override, the
|
||||
tool argument expresses caller intent and cannot be self-asserted by a worker
|
||||
|
||||
@@ -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
|
||||
|
||||
+122
-35
@@ -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)
|
||||
@@ -22575,9 +22576,8 @@ def gitea_request_mcp_restart(
|
||||
target_connector: str | None = None,
|
||||
drain_proof_json: str | None = None,
|
||||
request_break_glass: bool = False,
|
||||
prior_recovery_attempts_json: str | None = None,
|
||||
) -> dict:
|
||||
"""Evaluate a proposed MCP restart and return an impact preview (#658/#669).
|
||||
"""Evaluate a proposed MCP restart and return an impact preview (#658).
|
||||
|
||||
Central restart coordinator: resolves the requested restart class, gathers
|
||||
live control-plane state (sessions,
|
||||
@@ -22601,16 +22601,10 @@ def gitea_request_mcp_restart(
|
||||
independent — the drain gate proves the blast radius was drained and knows
|
||||
nothing about whether this requester may request this class — so a class the
|
||||
matrix denied never reports an authorized apply. Break-glass bypasses the
|
||||
drain proof and, when env-authorized, the #669 attempt-log requirement for
|
||||
broad restarts; it never bypasses the class matrix. ``apply_gate`` carries
|
||||
drain proof only; it never bypasses the class matrix. ``apply_gate`` carries
|
||||
``drain_gate_allow`` and ``restart_class_authorized`` so a denial is
|
||||
attributable to the authorization that produced it.
|
||||
|
||||
``prior_recovery_attempts_json`` (#669) is an optional JSON array of prior
|
||||
narrow recovery attempts ``{action, outcome, reason, ...}``. Rolling / full
|
||||
/ host restart classes require at least one *insufficient* narrower attempt
|
||||
unless break-glass is authorized.
|
||||
|
||||
Operator override authority is read from the process environment
|
||||
(``GITEA_OPERATOR_RESTART_OVERRIDE_AUTHORIZATION``), never self-asserted by
|
||||
the requesting session: ``request_override`` only expresses caller intent
|
||||
@@ -22716,37 +22710,12 @@ def gitea_request_mcp_restart(
|
||||
requester_role
|
||||
)
|
||||
|
||||
prior_recovery_attempts: list[dict] = []
|
||||
if prior_recovery_attempts_json:
|
||||
try:
|
||||
parsed_attempts = json.loads(prior_recovery_attempts_json)
|
||||
if isinstance(parsed_attempts, list):
|
||||
prior_recovery_attempts = [
|
||||
dict(a) for a in parsed_attempts if isinstance(a, dict)
|
||||
]
|
||||
else:
|
||||
incomplete_reasons.append(
|
||||
"prior_recovery_attempts_json must be a JSON array (#669)"
|
||||
)
|
||||
inventory_complete = False
|
||||
except (ValueError, TypeError) as exc:
|
||||
incomplete_reasons.append(
|
||||
f"invalid prior_recovery_attempts_json: {_redact(str(exc))}"
|
||||
)
|
||||
inventory_complete = False
|
||||
|
||||
break_glass_authorized = bool(
|
||||
(os.environ.get("GITEA_BREAKGLASS_RESTART_AUTHORIZATION") or "").strip()
|
||||
)
|
||||
break_glass = bool(request_break_glass and break_glass_authorized)
|
||||
|
||||
inventory = {
|
||||
"sessions": sessions,
|
||||
"leases": leases,
|
||||
"terminal_lock": terminal_lock,
|
||||
"inventory_complete": inventory_complete,
|
||||
"incomplete_reasons": incomplete_reasons,
|
||||
"prior_recovery_attempts": prior_recovery_attempts,
|
||||
}
|
||||
|
||||
report = restart_coordinator.evaluate_restart_impact(
|
||||
@@ -22762,7 +22731,6 @@ def gitea_request_mcp_restart(
|
||||
target_session_id=target_session_id,
|
||||
target_role=target_role,
|
||||
target_connector=target_connector,
|
||||
break_glass=break_glass,
|
||||
)
|
||||
|
||||
payload = report.as_dict()
|
||||
@@ -22783,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
|
||||
@@ -22795,6 +22795,13 @@ def gitea_request_mcp_restart(
|
||||
except (ValueError, TypeError) as exc:
|
||||
proof_parse_error = f"invalid drain_proof_json: {_redact(str(exc))}"
|
||||
|
||||
break_glass_authorized = bool(
|
||||
(
|
||||
os.environ.get("GITEA_BREAKGLASS_RESTART_AUTHORIZATION") or ""
|
||||
).strip()
|
||||
)
|
||||
break_glass = bool(request_break_glass and break_glass_authorized)
|
||||
|
||||
expected_fp = drain_proof.impact_fingerprint(report.as_dict())
|
||||
gate = drain_proof.gate_apply_restart(
|
||||
proof=proof_obj,
|
||||
@@ -22835,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
|
||||
|
||||
|
||||
|
||||
@@ -1,583 +0,0 @@
|
||||
"""Scoped MCP recovery playbook (#669).
|
||||
|
||||
Operational recovery must prefer the *narrowest* action that can fix the
|
||||
symptom. Full MCP / host restarts are last-resort rungs on a documented
|
||||
ladder; the coordinator refuses those rungs unless a prior attempt log
|
||||
shows narrower recoveries already failed (or break-glass is authorized).
|
||||
|
||||
This module is pure classification and recommendation:
|
||||
|
||||
* No network, filesystem, or process I/O.
|
||||
* Never restarts anything.
|
||||
* Narrow recovery *execution* is delegated to existing tools/docs (linked
|
||||
per rung) — the playbook records which rung to try next and whether
|
||||
escalation to a broad restart is allowed.
|
||||
|
||||
Design lineage: umbrella #655, class matrix #663, coordinator #658,
|
||||
auto-reconnect #584, stale-runtime #610, contamination #630, audit #665.
|
||||
Vision #652 / roadmap #653.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from enum import Enum
|
||||
from typing import Any, Mapping, Sequence
|
||||
|
||||
PLAYBOOK_VERSION = "1.0.0-issue-669"
|
||||
|
||||
# Attempt outcomes that count as "tried and insufficient" for escalation.
|
||||
INSUFFICIENT_OUTCOMES = frozenset(
|
||||
{
|
||||
"failed",
|
||||
"insufficient",
|
||||
"denied",
|
||||
"unresolved",
|
||||
"timeout",
|
||||
"error",
|
||||
}
|
||||
)
|
||||
|
||||
# Break-glass / operator override still records that the ladder was skipped.
|
||||
OUTCOME_BREAK_GLASS = "break_glass"
|
||||
OUTCOME_SUCCESS = "success"
|
||||
OUTCOME_SKIPPED = "skipped"
|
||||
|
||||
|
||||
class RecoveryAction(str, Enum):
|
||||
"""Ordered recovery ladder (narrow → broad)."""
|
||||
|
||||
CLIENT_RECONNECT = "client_reconnect"
|
||||
CAPABILITY_REFRESH = "capability_refresh"
|
||||
SESSION_RECONNECT = "session_reconnect"
|
||||
CONFIGURATION_RELOAD = "configuration_reload"
|
||||
LEASE_RECOVERY = "lease_recovery"
|
||||
WORKER_RESTART = "worker_restart"
|
||||
ROLE_RUNTIME_RESTART = "role_runtime_restart"
|
||||
CONNECTOR_RESTART = "connector_restart"
|
||||
ROLLING_MCP_RESTART = "rolling_mcp_restart"
|
||||
FULL_MCP_RESTART = "full_mcp_restart"
|
||||
HOST_RESTART = "host_restart"
|
||||
|
||||
|
||||
# Classes that require a prior narrow-attempt log (unless break-glass).
|
||||
BROAD_RESTART_ACTIONS: frozenset[RecoveryAction] = frozenset(
|
||||
{
|
||||
RecoveryAction.ROLLING_MCP_RESTART,
|
||||
RecoveryAction.FULL_MCP_RESTART,
|
||||
RecoveryAction.HOST_RESTART,
|
||||
}
|
||||
)
|
||||
|
||||
# Map #663 restart_class strings onto playbook actions.
|
||||
RESTART_CLASS_TO_ACTION: dict[str, RecoveryAction] = {
|
||||
"client_reconnect": RecoveryAction.CLIENT_RECONNECT,
|
||||
"session_reconnect": RecoveryAction.SESSION_RECONNECT,
|
||||
"configuration_reload": RecoveryAction.CONFIGURATION_RELOAD,
|
||||
"worker_restart": RecoveryAction.WORKER_RESTART,
|
||||
"role_runtime_restart": RecoveryAction.ROLE_RUNTIME_RESTART,
|
||||
"connector_restart": RecoveryAction.CONNECTOR_RESTART,
|
||||
"rolling_mcp_restart": RecoveryAction.ROLLING_MCP_RESTART,
|
||||
"full_mcp_restart": RecoveryAction.FULL_MCP_RESTART,
|
||||
"host_restart": RecoveryAction.HOST_RESTART,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RecoveryRung:
|
||||
"""One rung on the recovery ladder."""
|
||||
|
||||
action: RecoveryAction
|
||||
rank: int
|
||||
summary: str
|
||||
# Existing implementation or explicit delegation target.
|
||||
implementation: str
|
||||
issue_links: tuple[str, ...]
|
||||
self_service: bool
|
||||
# Restart-class permission when this rung is requested via coordinator.
|
||||
restart_class: str | None = None
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"action": self.action.value,
|
||||
"rank": self.rank,
|
||||
"summary": self.summary,
|
||||
"implementation": self.implementation,
|
||||
"issue_links": list(self.issue_links),
|
||||
"self_service": self.self_service,
|
||||
"restart_class": self.restart_class,
|
||||
}
|
||||
|
||||
|
||||
# Canonical ladder. Rank 0 is narrowest.
|
||||
RECOVERY_LADDER: tuple[RecoveryRung, ...] = (
|
||||
RecoveryRung(
|
||||
RecoveryAction.CLIENT_RECONNECT,
|
||||
0,
|
||||
"Reconnect the IDE/client MCP transport (EOF / transport flap).",
|
||||
"Host auto-reconnect or explicit client reconnect; "
|
||||
"docs/mcp-namespace-eof-recovery.md",
|
||||
("#584", "#655"),
|
||||
True,
|
||||
"client_reconnect",
|
||||
),
|
||||
RecoveryRung(
|
||||
RecoveryAction.CAPABILITY_REFRESH,
|
||||
1,
|
||||
"Re-resolve task capability and clear stale permission context.",
|
||||
"Delegated: gitea_resolve_task_capability + gitea_whoami "
|
||||
"(no process change).",
|
||||
("#610", "#685", "#655"),
|
||||
True,
|
||||
None,
|
||||
),
|
||||
RecoveryRung(
|
||||
RecoveryAction.SESSION_RECONNECT,
|
||||
2,
|
||||
"Rebind identity, workspace, and namespace for one session.",
|
||||
"Delegated: gitea_get_runtime_context + explicit worktree_path "
|
||||
"rebind (#618); docs/mcp-namespace-health.md",
|
||||
("#543", "#618", "#655"),
|
||||
True,
|
||||
"session_reconnect",
|
||||
),
|
||||
RecoveryRung(
|
||||
RecoveryAction.CONFIGURATION_RELOAD,
|
||||
3,
|
||||
"Gracefully reload configuration without replacing the daemon.",
|
||||
"restart_coordinator class configuration_reload; console "
|
||||
"system.reload_namespace (#642).",
|
||||
("#642", "#663", "#655"),
|
||||
False,
|
||||
"configuration_reload",
|
||||
),
|
||||
RecoveryRung(
|
||||
RecoveryAction.LEASE_RECOVERY,
|
||||
4,
|
||||
"Recover or rebind stale leases/locks without a process restart.",
|
||||
"Delegated: issue lock recovery / lease lifecycle paths "
|
||||
"(#702, #753, #790).",
|
||||
("#702", "#753", "#790", "#655"),
|
||||
False,
|
||||
None,
|
||||
),
|
||||
RecoveryRung(
|
||||
RecoveryAction.WORKER_RESTART,
|
||||
5,
|
||||
"Restart one worker after its own lease and mutation scope drains.",
|
||||
"restart_coordinator class worker_restart (#663).",
|
||||
("#663", "#655"),
|
||||
False,
|
||||
"worker_restart",
|
||||
),
|
||||
RecoveryRung(
|
||||
RecoveryAction.ROLE_RUNTIME_RESTART,
|
||||
6,
|
||||
"Restart one role runtime and re-probe that namespace only.",
|
||||
"restart_coordinator class role_runtime_restart; console "
|
||||
"system.restart_namespace (#642).",
|
||||
("#642", "#663", "#655"),
|
||||
False,
|
||||
"role_runtime_restart",
|
||||
),
|
||||
RecoveryRung(
|
||||
RecoveryAction.CONNECTOR_RESTART,
|
||||
7,
|
||||
"Restart one connector while unrelated runtimes stay available.",
|
||||
"restart_coordinator class connector_restart (#663).",
|
||||
("#663", "#655"),
|
||||
False,
|
||||
"connector_restart",
|
||||
),
|
||||
RecoveryRung(
|
||||
RecoveryAction.ROLLING_MCP_RESTART,
|
||||
8,
|
||||
"Drain/restart/verify one instance at a time (HA path).",
|
||||
"restart_coordinator class rolling_mcp_restart; design #668.",
|
||||
("#668", "#663", "#655"),
|
||||
False,
|
||||
"rolling_mcp_restart",
|
||||
),
|
||||
RecoveryRung(
|
||||
RecoveryAction.FULL_MCP_RESTART,
|
||||
9,
|
||||
"Full stable-control MCP process restart after verified full drain.",
|
||||
"restart_coordinator class full_mcp_restart; requires attempt log "
|
||||
"unless break-glass (#669).",
|
||||
("#658", "#661", "#663", "#669", "#655"),
|
||||
False,
|
||||
"full_mcp_restart",
|
||||
),
|
||||
RecoveryRung(
|
||||
RecoveryAction.HOST_RESTART,
|
||||
10,
|
||||
"Host/infrastructure restart — broadest last-resort action.",
|
||||
"restart_coordinator class host_restart; operator-owned.",
|
||||
("#663", "#669", "#655"),
|
||||
False,
|
||||
"host_restart",
|
||||
),
|
||||
)
|
||||
|
||||
_LADDER_BY_ACTION: dict[RecoveryAction, RecoveryRung] = {
|
||||
rung.action: rung for rung in RECOVERY_LADDER
|
||||
}
|
||||
|
||||
# Symptom tokens → preferred first rung (decision tree, #663 lineage).
|
||||
SYMPTOM_TO_FIRST_ACTION: dict[str, RecoveryAction] = {
|
||||
"transport_eof": RecoveryAction.CLIENT_RECONNECT,
|
||||
"client_closing_eof": RecoveryAction.CLIENT_RECONNECT,
|
||||
"transport_flap": RecoveryAction.CLIENT_RECONNECT,
|
||||
"namespace_disconnected": RecoveryAction.CLIENT_RECONNECT,
|
||||
"stale_capability": RecoveryAction.CAPABILITY_REFRESH,
|
||||
"permission_stale": RecoveryAction.CAPABILITY_REFRESH,
|
||||
"runtime_reconnect_required": RecoveryAction.CAPABILITY_REFRESH,
|
||||
"stale_runtime": RecoveryAction.SESSION_RECONNECT,
|
||||
"worktree_unbound": RecoveryAction.SESSION_RECONNECT,
|
||||
"namespace_unhealthy": RecoveryAction.SESSION_RECONNECT,
|
||||
"config_drift": RecoveryAction.CONFIGURATION_RELOAD,
|
||||
"profile_misbound": RecoveryAction.CONFIGURATION_RELOAD,
|
||||
"stale_lease": RecoveryAction.LEASE_RECOVERY,
|
||||
"dead_pid_lock": RecoveryAction.LEASE_RECOVERY,
|
||||
"orphan_worktree": RecoveryAction.LEASE_RECOVERY,
|
||||
"single_worker_stuck": RecoveryAction.WORKER_RESTART,
|
||||
"role_runtime_dead": RecoveryAction.ROLE_RUNTIME_RESTART,
|
||||
"connector_dead": RecoveryAction.CONNECTOR_RESTART,
|
||||
"ha_instance_unhealthy": RecoveryAction.ROLLING_MCP_RESTART,
|
||||
"daemon_corrupt": RecoveryAction.FULL_MCP_RESTART,
|
||||
"full_process_deadlock": RecoveryAction.FULL_MCP_RESTART,
|
||||
"host_unresponsive": RecoveryAction.HOST_RESTART,
|
||||
}
|
||||
|
||||
|
||||
def _utc_now() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def resolve_action(value: RecoveryAction | str) -> RecoveryAction:
|
||||
"""Resolve a recovery action or fail closed for unknown values."""
|
||||
if isinstance(value, RecoveryAction):
|
||||
return value
|
||||
text = str(value or "").strip()
|
||||
# Accept #663 restart_class aliases.
|
||||
if text in RESTART_CLASS_TO_ACTION:
|
||||
return RESTART_CLASS_TO_ACTION[text]
|
||||
try:
|
||||
return RecoveryAction(text)
|
||||
except ValueError as exc:
|
||||
raise ValueError(
|
||||
f"unknown recovery action {value!r}; deny (fail closed, #669)"
|
||||
) from exc
|
||||
|
||||
|
||||
def ladder_rank(action: RecoveryAction | str) -> int:
|
||||
resolved = resolve_action(action)
|
||||
return _LADDER_BY_ACTION[resolved].rank
|
||||
|
||||
|
||||
def rung_for(action: RecoveryAction | str) -> RecoveryRung:
|
||||
return _LADDER_BY_ACTION[resolve_action(action)]
|
||||
|
||||
|
||||
def normalize_attempt(raw: Mapping[str, Any]) -> dict[str, Any] | None:
|
||||
"""Normalize one prior-recovery attempt record; return None if unusable."""
|
||||
if not isinstance(raw, Mapping):
|
||||
return None
|
||||
action_raw = raw.get("action") or raw.get("recovery_action") or raw.get(
|
||||
"restart_class"
|
||||
)
|
||||
if not action_raw:
|
||||
return None
|
||||
try:
|
||||
action = resolve_action(str(action_raw))
|
||||
except ValueError:
|
||||
return None
|
||||
outcome = str(
|
||||
raw.get("outcome") or raw.get("status") or raw.get("result") or ""
|
||||
).strip().lower()
|
||||
if not outcome:
|
||||
return None
|
||||
recorded_at = raw.get("recorded_at") or raw.get("at") or raw.get("timestamp")
|
||||
reason = str(raw.get("reason") or raw.get("detail") or "").strip()
|
||||
actor = str(raw.get("actor") or raw.get("session_id") or "").strip()
|
||||
return {
|
||||
"action": action.value,
|
||||
"outcome": outcome,
|
||||
"reason": reason,
|
||||
"actor": actor,
|
||||
"recorded_at": recorded_at,
|
||||
"rank": ladder_rank(action),
|
||||
"raw": dict(raw),
|
||||
}
|
||||
|
||||
|
||||
def normalize_attempt_log(
|
||||
attempts: Sequence[Mapping[str, Any]] | None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Return usable attempt records in ladder order."""
|
||||
out: list[dict[str, Any]] = []
|
||||
for raw in attempts or ():
|
||||
norm = normalize_attempt(raw)
|
||||
if norm is not None:
|
||||
out.append(norm)
|
||||
out.sort(key=lambda a: (a["rank"], str(a.get("recorded_at") or "")))
|
||||
return out
|
||||
|
||||
|
||||
def narrower_insufficient_attempts(
|
||||
attempts: Sequence[Mapping[str, Any]] | None,
|
||||
*,
|
||||
requested: RecoveryAction | str,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Return prior attempts narrower than *requested* that were insufficient."""
|
||||
target_rank = ladder_rank(requested)
|
||||
usable = []
|
||||
for attempt in normalize_attempt_log(attempts):
|
||||
if attempt["rank"] >= target_rank:
|
||||
continue
|
||||
if attempt["outcome"] in INSUFFICIENT_OUTCOMES:
|
||||
usable.append(attempt)
|
||||
return usable
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class EscalationAssessment:
|
||||
"""Whether a requested broad recovery may proceed given the attempt log."""
|
||||
|
||||
requested_action: str
|
||||
allowed: bool
|
||||
require_attempt_log: bool
|
||||
break_glass: bool
|
||||
reasons: list[str] = field(default_factory=list)
|
||||
qualifying_attempts: list[dict[str, Any]] = field(default_factory=list)
|
||||
recommended_next: list[dict[str, Any]] = field(default_factory=list)
|
||||
playbook_version: str = PLAYBOOK_VERSION
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"playbook_version": self.playbook_version,
|
||||
"requested_action": self.requested_action,
|
||||
"allowed": self.allowed,
|
||||
"require_attempt_log": self.require_attempt_log,
|
||||
"break_glass": self.break_glass,
|
||||
"reasons": list(self.reasons),
|
||||
"qualifying_attempts": list(self.qualifying_attempts),
|
||||
"recommended_next": list(self.recommended_next),
|
||||
}
|
||||
|
||||
|
||||
def assess_escalation(
|
||||
requested: RecoveryAction | str,
|
||||
*,
|
||||
prior_recovery_attempts: Sequence[Mapping[str, Any]] | None = None,
|
||||
break_glass: bool = False,
|
||||
) -> EscalationAssessment:
|
||||
"""Gate broad restarts on a prior narrow-attempt log (#669 AC3).
|
||||
|
||||
Narrow / mid-ladder actions do not require a prior attempt log.
|
||||
``full_mcp_restart``, ``host_restart``, and ``rolling_mcp_restart``
|
||||
require at least one *insufficient* narrower attempt unless
|
||||
``break_glass`` is true.
|
||||
"""
|
||||
action = resolve_action(requested)
|
||||
require_log = action in BROAD_RESTART_ACTIONS
|
||||
reasons: list[str] = []
|
||||
qualifying = narrower_insufficient_attempts(
|
||||
prior_recovery_attempts, requested=action
|
||||
)
|
||||
|
||||
if not require_log:
|
||||
return EscalationAssessment(
|
||||
requested_action=action.value,
|
||||
allowed=True,
|
||||
require_attempt_log=False,
|
||||
break_glass=bool(break_glass),
|
||||
reasons=["narrow recovery; attempt log not required"],
|
||||
qualifying_attempts=qualifying,
|
||||
recommended_next=[],
|
||||
)
|
||||
|
||||
if break_glass:
|
||||
return EscalationAssessment(
|
||||
requested_action=action.value,
|
||||
allowed=True,
|
||||
require_attempt_log=True,
|
||||
break_glass=True,
|
||||
reasons=[
|
||||
"break-glass authorized; broad restart permitted without "
|
||||
"narrow-attempt log (#669)"
|
||||
],
|
||||
qualifying_attempts=qualifying,
|
||||
recommended_next=[],
|
||||
)
|
||||
|
||||
if qualifying:
|
||||
return EscalationAssessment(
|
||||
requested_action=action.value,
|
||||
allowed=True,
|
||||
require_attempt_log=True,
|
||||
break_glass=False,
|
||||
reasons=[
|
||||
f"{len(qualifying)} narrower recovery attempt(s) recorded as "
|
||||
"insufficient; escalation permitted"
|
||||
],
|
||||
qualifying_attempts=qualifying,
|
||||
recommended_next=[],
|
||||
)
|
||||
|
||||
# Deny: recommend the next untried narrow rung(s).
|
||||
recommended = recommend_actions(
|
||||
symptoms=(),
|
||||
prior_recovery_attempts=prior_recovery_attempts,
|
||||
max_actions=3,
|
||||
)
|
||||
reasons.append(
|
||||
f"{action.value} requires a prior attempt log of insufficient "
|
||||
"narrower recoveries (or break-glass); none found — deny (fail "
|
||||
"closed, #669)"
|
||||
)
|
||||
return EscalationAssessment(
|
||||
requested_action=action.value,
|
||||
allowed=False,
|
||||
require_attempt_log=True,
|
||||
break_glass=False,
|
||||
reasons=reasons,
|
||||
qualifying_attempts=[],
|
||||
recommended_next=recommended.get("recommended_actions") or [],
|
||||
)
|
||||
|
||||
|
||||
def recommend_actions(
|
||||
*,
|
||||
symptoms: Sequence[str] = (),
|
||||
prior_recovery_attempts: Sequence[Mapping[str, Any]] | None = None,
|
||||
max_actions: int = 5,
|
||||
) -> dict[str, Any]:
|
||||
"""Return ordered recommended recovery actions for the given symptoms.
|
||||
|
||||
Soft mode (rollout): recommendations only — callers decide whether to
|
||||
hard-gate. Hard mode for broad restarts is :func:`assess_escalation`.
|
||||
"""
|
||||
attempted_success = {
|
||||
a["action"]
|
||||
for a in normalize_attempt_log(prior_recovery_attempts)
|
||||
if a["outcome"] == OUTCOME_SUCCESS
|
||||
}
|
||||
attempted_any = {
|
||||
a["action"] for a in normalize_attempt_log(prior_recovery_attempts)
|
||||
}
|
||||
|
||||
first_actions: list[RecoveryAction] = []
|
||||
for symptom in symptoms:
|
||||
key = str(symptom or "").strip().lower().replace(" ", "_").replace("-", "_")
|
||||
mapped = SYMPTOM_TO_FIRST_ACTION.get(key)
|
||||
if mapped is not None and mapped not in first_actions:
|
||||
first_actions.append(mapped)
|
||||
|
||||
# Default entry: client reconnect then walk the ladder.
|
||||
if not first_actions:
|
||||
first_actions = [RecoveryAction.CLIENT_RECONNECT]
|
||||
|
||||
recommended: list[dict[str, Any]] = []
|
||||
seen: set[str] = set()
|
||||
min_rank = min(ladder_rank(a) for a in first_actions)
|
||||
|
||||
for rung in RECOVERY_LADDER:
|
||||
if rung.rank < min_rank:
|
||||
continue
|
||||
if rung.action.value in attempted_success:
|
||||
continue
|
||||
if rung.action.value in seen:
|
||||
continue
|
||||
# Prefer rungs not yet attempted; still list previously-failed ones
|
||||
# only if nothing else remains.
|
||||
entry = rung.as_dict()
|
||||
entry["already_attempted"] = rung.action.value in attempted_any
|
||||
recommended.append(entry)
|
||||
seen.add(rung.action.value)
|
||||
if len(recommended) >= max(1, int(max_actions)):
|
||||
break
|
||||
|
||||
return {
|
||||
"playbook_version": PLAYBOOK_VERSION,
|
||||
"symptoms": [str(s) for s in symptoms],
|
||||
"recommended_actions": recommended,
|
||||
"ladder": [r.as_dict() for r in RECOVERY_LADDER],
|
||||
"read_only": True,
|
||||
"hard_gate_note": (
|
||||
"Broad restarts (rolling/full/host) still require "
|
||||
"assess_escalation / coordinator attempt-log enforcement."
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def build_attempt_record(
|
||||
action: RecoveryAction | str,
|
||||
*,
|
||||
outcome: str,
|
||||
reason: str = "",
|
||||
actor: str = "",
|
||||
recorded_at: str | None = None,
|
||||
extra: Mapping[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Build a durable-shaped attempt log entry for inventory/audit (#665)."""
|
||||
resolved = resolve_action(action)
|
||||
record = {
|
||||
"action": resolved.value,
|
||||
"outcome": str(outcome or "").strip().lower(),
|
||||
"reason": str(reason or "").strip(),
|
||||
"actor": str(actor or "").strip(),
|
||||
"recorded_at": recorded_at or _utc_now().isoformat(),
|
||||
"rank": ladder_rank(resolved),
|
||||
"playbook_version": PLAYBOOK_VERSION,
|
||||
}
|
||||
if extra:
|
||||
record["extra"] = dict(extra)
|
||||
return record
|
||||
|
||||
|
||||
def recovery_metrics(
|
||||
attempts: Sequence[Mapping[str, Any]] | None,
|
||||
) -> dict[str, Any]:
|
||||
"""Compute the fraction of recoveries that avoided full/host restart.
|
||||
|
||||
A recovery *episode* is approximated as one attempt with
|
||||
``outcome=success``. Successes on non-broad rungs count as avoided full
|
||||
restart; successes on full/host count as full-restart recoveries.
|
||||
"""
|
||||
norms = normalize_attempt_log(attempts)
|
||||
successes = [a for a in norms if a["outcome"] == OUTCOME_SUCCESS]
|
||||
broad_success = [
|
||||
a
|
||||
for a in successes
|
||||
if resolve_action(a["action"])
|
||||
in {RecoveryAction.FULL_MCP_RESTART, RecoveryAction.HOST_RESTART}
|
||||
]
|
||||
avoided = [a for a in successes if a not in broad_success]
|
||||
total = len(successes)
|
||||
fraction_avoided = (len(avoided) / total) if total else None
|
||||
return {
|
||||
"playbook_version": PLAYBOOK_VERSION,
|
||||
"attempts_total": len(norms),
|
||||
"successes_total": total,
|
||||
"successes_avoided_full_restart": len(avoided),
|
||||
"successes_full_or_host_restart": len(broad_success),
|
||||
"fraction_avoided_full_restart": fraction_avoided,
|
||||
"insufficient_attempts": sum(
|
||||
1 for a in norms if a["outcome"] in INSUFFICIENT_OUTCOMES
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def ladder_document() -> dict[str, Any]:
|
||||
"""Machine-readable ladder for docs/tools inventory."""
|
||||
return {
|
||||
"playbook_version": PLAYBOOK_VERSION,
|
||||
"parent_issues": ["#655", "#652", "#653"],
|
||||
"enforcement_issue": "#669",
|
||||
"ladder": [r.as_dict() for r in RECOVERY_LADDER],
|
||||
"broad_restart_actions": [a.value for a in sorted(BROAD_RESTART_ACTIONS, key=lambda x: x.value)],
|
||||
"insufficient_outcomes": sorted(INSUFFICIENT_OUTCOMES),
|
||||
"symptom_map": {k: v.value for k, v in sorted(SYMPTOM_TO_FIRST_ACTION.items())},
|
||||
}
|
||||
@@ -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
|
||||
+2
-46
@@ -1,4 +1,4 @@
|
||||
"""MCP restart coordinator and impact analysis (#658 / #669).
|
||||
"""MCP restart coordinator and impact analysis (#658).
|
||||
|
||||
Before any sanctioned MCP restart, a central coordinator must evaluate the
|
||||
live control-plane state — active sessions, leases/locks, in-flight issue/PR
|
||||
@@ -16,9 +16,6 @@ Design rules (mirrors the read-only posture of ``workflow_dashboard`` /
|
||||
a mutative apply path is a later child gated by a drain proof (non-goal here).
|
||||
* **Fail closed.** If the inventory is not explicitly complete, the verdict is
|
||||
``unsafe`` / deny — an incomplete evaluation must never green-light a restart.
|
||||
* **Narrow-first (#669).** Broad classes (rolling / full / host) require a
|
||||
prior attempt log of insufficient narrower recoveries unless break-glass is
|
||||
authorized. See :mod:`recovery_playbook`.
|
||||
* **No secrets.** Session ids, pids, and profiles are operational metadata, not
|
||||
credentials; nothing secret flows through this module.
|
||||
|
||||
@@ -35,9 +32,8 @@ from enum import Enum
|
||||
from typing import Any, Mapping, Sequence
|
||||
|
||||
import lease_lifecycle
|
||||
import recovery_playbook
|
||||
|
||||
COORDINATOR_VERSION = "1.2.0-issue-669"
|
||||
COORDINATOR_VERSION = "1.1.0-issue-663"
|
||||
|
||||
# Restart verdicts. Exactly the three the acceptance criteria name.
|
||||
VERDICT_SAFE = "safe"
|
||||
@@ -353,10 +349,6 @@ class RestartImpactReport:
|
||||
counts: dict[str, int]
|
||||
audit_record: dict[str, Any]
|
||||
incomplete_reasons: list[str] = field(default_factory=list)
|
||||
# #669 playbook escalation gate (attempt-log enforcement).
|
||||
playbook_escalation: dict[str, Any] = field(default_factory=dict)
|
||||
attempt_log_satisfied: bool = True
|
||||
break_glass: bool = False
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
@@ -390,9 +382,6 @@ class RestartImpactReport:
|
||||
"prior_recovery_attempts": list(self.prior_recovery_attempts),
|
||||
"counts": dict(self.counts),
|
||||
"audit_record": dict(self.audit_record),
|
||||
"playbook_escalation": dict(self.playbook_escalation),
|
||||
"attempt_log_satisfied": self.attempt_log_satisfied,
|
||||
"break_glass": self.break_glass,
|
||||
}
|
||||
|
||||
|
||||
@@ -507,7 +496,6 @@ def evaluate_restart_impact(
|
||||
target_session_id: str | None = None,
|
||||
target_role: str | None = None,
|
||||
target_connector: str | None = None,
|
||||
break_glass: bool = False,
|
||||
) -> RestartImpactReport:
|
||||
"""Evaluate a proposed MCP restart and return an impact preview.
|
||||
|
||||
@@ -596,30 +584,6 @@ def evaluate_restart_impact(
|
||||
dict(a) for a in (inventory.get("prior_recovery_attempts") or [])
|
||||
]
|
||||
|
||||
# #669: broad restarts require a prior narrow-attempt log unless break-glass.
|
||||
playbook_escalation: dict[str, Any] = {}
|
||||
attempt_log_satisfied = True
|
||||
if policy_enforced and resolved_class is not None:
|
||||
try:
|
||||
escalation = recovery_playbook.assess_escalation(
|
||||
resolved_class.value,
|
||||
prior_recovery_attempts=prior_recovery_attempts,
|
||||
break_glass=bool(break_glass),
|
||||
)
|
||||
playbook_escalation = escalation.as_dict()
|
||||
attempt_log_satisfied = bool(escalation.allowed)
|
||||
if not attempt_log_satisfied:
|
||||
authorization_reasons.extend(list(escalation.reasons))
|
||||
except ValueError as exc:
|
||||
# Unknown mapping should never happen for enum values; fail closed.
|
||||
attempt_log_satisfied = False
|
||||
playbook_escalation = {
|
||||
"allowed": False,
|
||||
"reasons": [str(exc)],
|
||||
"playbook_version": recovery_playbook.PLAYBOOK_VERSION,
|
||||
}
|
||||
authorization_reasons.append(str(exc))
|
||||
|
||||
session_impacts = [
|
||||
_classify_session(
|
||||
s,
|
||||
@@ -718,7 +682,6 @@ def evaluate_restart_impact(
|
||||
and role_authorized
|
||||
and approval_satisfied
|
||||
and target_complete
|
||||
and attempt_log_satisfied
|
||||
)
|
||||
|
||||
if policy_enforced and not authorization_ok:
|
||||
@@ -782,7 +745,6 @@ def evaluate_restart_impact(
|
||||
"affected_issues": len(affected_issues),
|
||||
"affected_prs": len(affected_prs),
|
||||
"prior_recovery_attempts": len(prior_recovery_attempts),
|
||||
"attempt_log_satisfied": attempt_log_satisfied,
|
||||
}
|
||||
|
||||
audit_record = {
|
||||
@@ -803,9 +765,6 @@ def evaluate_restart_impact(
|
||||
"allow_restart": allow_restart,
|
||||
"blast_radius": blast_radius,
|
||||
"counts": counts,
|
||||
"attempt_log_satisfied": attempt_log_satisfied,
|
||||
"break_glass": bool(break_glass),
|
||||
"playbook_version": recovery_playbook.PLAYBOOK_VERSION,
|
||||
}
|
||||
|
||||
return RestartImpactReport(
|
||||
@@ -845,7 +804,4 @@ def evaluate_restart_impact(
|
||||
counts=counts,
|
||||
audit_record=audit_record,
|
||||
incomplete_reasons=incomplete_reasons,
|
||||
playbook_escalation=playbook_escalation,
|
||||
attempt_log_satisfied=attempt_log_satisfied,
|
||||
break_glass=bool(break_glass),
|
||||
)
|
||||
|
||||
@@ -35,17 +35,6 @@ BREAK_GLASS_ENV = "GITEA_BREAKGLASS_RESTART_AUTHORIZATION"
|
||||
QUIET_SESSIONS: list[dict] = []
|
||||
QUIET_LEASES: list[dict] = []
|
||||
|
||||
# #669: broad restarts need a prior narrow-attempt log (unless break-glass).
|
||||
PRIOR_NARROW_ATTEMPTS_JSON = json.dumps(
|
||||
[
|
||||
{
|
||||
"action": "client_reconnect",
|
||||
"outcome": "insufficient",
|
||||
"reason": "still flapping after reconnect",
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
class _FakeDB:
|
||||
"""Minimal control-plane DB stand-in for the restart inventory."""
|
||||
@@ -139,7 +128,6 @@ class TestConjunction(_RestartToolHarness):
|
||||
preview = self._call(
|
||||
role="operator",
|
||||
restart_class="full_mcp_restart",
|
||||
prior_recovery_attempts_json=PRIOR_NARROW_ATTEMPTS_JSON,
|
||||
env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
|
||||
)
|
||||
self.assertTrue(preview["allow_restart"],
|
||||
@@ -148,7 +136,6 @@ class TestConjunction(_RestartToolHarness):
|
||||
result = self._call(
|
||||
role="operator",
|
||||
restart_class="full_mcp_restart",
|
||||
prior_recovery_attempts_json=PRIOR_NARROW_ATTEMPTS_JSON,
|
||||
dry_run=False,
|
||||
drain_proof_json=self._clean_proof_for(preview),
|
||||
env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
|
||||
@@ -355,7 +342,6 @@ class TestExistingPathsStillWork(_RestartToolHarness):
|
||||
result = self._call(
|
||||
role="operator",
|
||||
restart_class="full_mcp_restart",
|
||||
prior_recovery_attempts_json=PRIOR_NARROW_ATTEMPTS_JSON,
|
||||
dry_run=False,
|
||||
env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
|
||||
)
|
||||
|
||||
@@ -1,217 +0,0 @@
|
||||
"""Unit tests for the scoped recovery playbook (#669)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import recovery_playbook as rp
|
||||
import restart_coordinator as rc
|
||||
|
||||
|
||||
def test_ladder_covers_eleven_ordered_rungs():
|
||||
ranks = [r.rank for r in rp.RECOVERY_LADDER]
|
||||
assert ranks == list(range(len(rp.RECOVERY_LADDER)))
|
||||
assert len(rp.RECOVERY_LADDER) == 11
|
||||
assert rp.RECOVERY_LADDER[0].action is rp.RecoveryAction.CLIENT_RECONNECT
|
||||
assert rp.RECOVERY_LADDER[-1].action is rp.RecoveryAction.HOST_RESTART
|
||||
|
||||
|
||||
def test_ladder_document_links_parent_issues():
|
||||
doc = rp.ladder_document()
|
||||
assert "#655" in doc["parent_issues"]
|
||||
assert "#652" in doc["parent_issues"]
|
||||
assert "#653" in doc["parent_issues"]
|
||||
assert doc["enforcement_issue"] == "#669"
|
||||
assert "full_mcp_restart" in doc["broad_restart_actions"]
|
||||
|
||||
|
||||
def test_recommend_transport_eof_starts_at_client_reconnect():
|
||||
plan = rp.recommend_actions(symptoms=["transport_eof"])
|
||||
assert plan["recommended_actions"][0]["action"] == "client_reconnect"
|
||||
assert plan["recommended_actions"][0]["issue_links"]
|
||||
|
||||
|
||||
def test_recommend_skips_successful_prior_attempts():
|
||||
attempts = [
|
||||
rp.build_attempt_record(
|
||||
"client_reconnect", outcome="success", reason="reconnected"
|
||||
)
|
||||
]
|
||||
plan = rp.recommend_actions(
|
||||
symptoms=["transport_eof"], prior_recovery_attempts=attempts
|
||||
)
|
||||
actions = [a["action"] for a in plan["recommended_actions"]]
|
||||
assert "client_reconnect" not in actions
|
||||
assert actions[0] == "capability_refresh"
|
||||
|
||||
|
||||
def test_escalation_denied_without_attempt_log():
|
||||
result = rp.assess_escalation("full_mcp_restart", prior_recovery_attempts=[])
|
||||
assert result.allowed is False
|
||||
assert result.require_attempt_log is True
|
||||
assert any("#669" in r for r in result.reasons)
|
||||
assert result.recommended_next # soft recommendations still provided
|
||||
|
||||
|
||||
def test_escalation_allowed_after_insufficient_narrower():
|
||||
attempts = [
|
||||
rp.build_attempt_record(
|
||||
"client_reconnect",
|
||||
outcome="insufficient",
|
||||
reason="still flapping",
|
||||
),
|
||||
rp.build_attempt_record(
|
||||
"session_reconnect",
|
||||
outcome="failed",
|
||||
reason="namespace still dead",
|
||||
),
|
||||
]
|
||||
result = rp.assess_escalation(
|
||||
"full_mcp_restart", prior_recovery_attempts=attempts
|
||||
)
|
||||
assert result.allowed is True
|
||||
assert len(result.qualifying_attempts) == 2
|
||||
|
||||
|
||||
def test_escalation_break_glass_bypasses_attempt_log():
|
||||
result = rp.assess_escalation(
|
||||
"host_restart", prior_recovery_attempts=[], break_glass=True
|
||||
)
|
||||
assert result.allowed is True
|
||||
assert result.break_glass is True
|
||||
|
||||
|
||||
def test_narrow_action_does_not_require_attempt_log():
|
||||
result = rp.assess_escalation(
|
||||
"client_reconnect", prior_recovery_attempts=[]
|
||||
)
|
||||
assert result.allowed is True
|
||||
assert result.require_attempt_log is False
|
||||
|
||||
|
||||
def test_same_rank_attempt_does_not_qualify_for_escalation():
|
||||
attempts = [
|
||||
rp.build_attempt_record(
|
||||
"full_mcp_restart", outcome="failed", reason="already failed full"
|
||||
)
|
||||
]
|
||||
result = rp.assess_escalation(
|
||||
"full_mcp_restart", prior_recovery_attempts=attempts
|
||||
)
|
||||
assert result.allowed is False
|
||||
|
||||
|
||||
def test_success_outcome_does_not_qualify_for_escalation():
|
||||
attempts = [
|
||||
rp.build_attempt_record(
|
||||
"client_reconnect", outcome="success", reason="fixed"
|
||||
)
|
||||
]
|
||||
result = rp.assess_escalation(
|
||||
"full_mcp_restart", prior_recovery_attempts=attempts
|
||||
)
|
||||
assert result.allowed is False
|
||||
|
||||
|
||||
def test_recovery_metrics_fraction_avoided():
|
||||
attempts = [
|
||||
rp.build_attempt_record("client_reconnect", outcome="success"),
|
||||
rp.build_attempt_record("session_reconnect", outcome="success"),
|
||||
rp.build_attempt_record("full_mcp_restart", outcome="success"),
|
||||
]
|
||||
metrics = rp.recovery_metrics(attempts)
|
||||
assert metrics["successes_total"] == 3
|
||||
assert metrics["successes_avoided_full_restart"] == 2
|
||||
assert metrics["successes_full_or_host_restart"] == 1
|
||||
assert abs(metrics["fraction_avoided_full_restart"] - (2 / 3)) < 1e-9
|
||||
|
||||
|
||||
def test_coordinator_denies_full_restart_without_attempt_log():
|
||||
inv = {
|
||||
"inventory_complete": True,
|
||||
"sessions": [],
|
||||
"leases": [],
|
||||
"prior_recovery_attempts": [],
|
||||
}
|
||||
report = rc.evaluate_restart_impact(
|
||||
inv,
|
||||
restart_class=rc.RestartClass.FULL_MCP_RESTART,
|
||||
requester_role="controller",
|
||||
requester_permissions=rc.permissions_for_role("controller"),
|
||||
controller_approved=True,
|
||||
operator_authorized=True,
|
||||
)
|
||||
assert report.allow_restart is False
|
||||
assert report.attempt_log_satisfied is False
|
||||
assert report.verdict == rc.VERDICT_UNSAFE
|
||||
blob = " ".join(report.reasons + report.authorization_reasons)
|
||||
assert "#669" in blob or "attempt log" in blob
|
||||
|
||||
|
||||
def test_coordinator_allows_full_restart_with_attempt_log():
|
||||
inv = {
|
||||
"inventory_complete": True,
|
||||
"sessions": [],
|
||||
"leases": [],
|
||||
"prior_recovery_attempts": [
|
||||
{
|
||||
"action": "client_reconnect",
|
||||
"outcome": "insufficient",
|
||||
"reason": "still broken",
|
||||
}
|
||||
],
|
||||
}
|
||||
report = rc.evaluate_restart_impact(
|
||||
inv,
|
||||
restart_class=rc.RestartClass.FULL_MCP_RESTART,
|
||||
requester_role="controller",
|
||||
requester_permissions=rc.permissions_for_role("controller"),
|
||||
controller_approved=True,
|
||||
operator_authorized=True,
|
||||
)
|
||||
assert report.attempt_log_satisfied is True
|
||||
assert report.allow_restart is True
|
||||
assert report.verdict == rc.VERDICT_SAFE
|
||||
|
||||
|
||||
def test_coordinator_break_glass_allows_without_log():
|
||||
inv = {
|
||||
"inventory_complete": True,
|
||||
"sessions": [],
|
||||
"leases": [],
|
||||
"prior_recovery_attempts": [],
|
||||
}
|
||||
report = rc.evaluate_restart_impact(
|
||||
inv,
|
||||
restart_class=rc.RestartClass.FULL_MCP_RESTART,
|
||||
requester_role="controller",
|
||||
requester_permissions=rc.permissions_for_role("controller"),
|
||||
controller_approved=True,
|
||||
operator_authorized=True,
|
||||
break_glass=True,
|
||||
)
|
||||
assert report.break_glass is True
|
||||
assert report.attempt_log_satisfied is True
|
||||
assert report.allow_restart is True
|
||||
|
||||
|
||||
def test_coordinator_client_reconnect_unaffected():
|
||||
inv = {
|
||||
"inventory_complete": True,
|
||||
"sessions": [],
|
||||
"leases": [],
|
||||
"prior_recovery_attempts": [],
|
||||
}
|
||||
report = rc.evaluate_restart_impact(
|
||||
inv,
|
||||
restart_class=rc.RestartClass.CLIENT_RECONNECT,
|
||||
requester_role="author",
|
||||
requester_permissions=rc.permissions_for_role("author"),
|
||||
)
|
||||
assert report.attempt_log_satisfied is True
|
||||
assert report.allow_restart is True
|
||||
|
||||
|
||||
def test_restart_class_alias_accepted():
|
||||
assert (
|
||||
rp.resolve_action("full_mcp_restart")
|
||||
is rp.RecoveryAction.FULL_MCP_RESTART
|
||||
)
|
||||
@@ -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()
|
||||
@@ -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("<script>", 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()
|
||||
@@ -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,
|
||||
@@ -347,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
|
||||
@@ -825,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"]),
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
@@ -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
-5
@@ -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.
|
||||
"""
|
||||
@@ -61,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"),
|
||||
|
||||
+2
-27
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user