feat(mcp): enforce scoped recovery playbook before full restart (Closes #669 )

Add recovery_playbook.py with the narrow-to-broad recovery ladder, symptom routing, attempt-log helpers, and escalation metrics. Wire the attempt-log gate into restart_coordinator so rolling/full/host restarts require prior insufficient narrower attempts (or break-glass). Document the ladder and update gitea_request_mcp_restart for prior_recovery_attempts_json. Co-Authored-By: Grok 4.5 <[email protected]>
Merge pull request 'feat(webui): Gitea issue and PR linkage console (Closes #645 )' (#904 ) from feat/issue-645-linkage-console into master
2026-07-25 17:35:11 -04:00 · 2026-07-25 16:32:21 -05:00 · 2026-07-25 16:40:07 -04:00
17 changed files with 2788 additions and 994 deletions
@@ -0,0 +1,94 @@
 # 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.
@@ -1,93 +0,0 @@
 # 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,7 +33,6 @@ 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
@@ -96,12 +95,18 @@ 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)
+                          request_break_glass=False,
                          prior_recovery_attempts_json=None)
 ```
 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 |
@@ -113,9 +118,10 @@ It **never restarts anything**: `apply_supported` is always `false` and
 An apply requires **both** authorizations, and they are independent:
-1. **Restart-class authorization** (#663) — the requester's role and permissions
+1. **Restart-class authorization** (#663 / #669) — the requester's role and
-   must allow the requested class, the class's approval requirement must be
+   permissions must allow the requested class, the class's approval requirement
-   satisfied, and any target-scoped class must name its target. Failing any of
+   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
   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.
@@ -128,7 +134,10 @@ the authorization that produced it.
 ### Break-glass
-Break-glass bypasses the **drain proof only** — never the restart-class matrix.
+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.
 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,6 +80,8 @@ 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 |
@@ -327,6 +329,68 @@ 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
@@ -2069,7 +2069,6 @@ 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)
@@ -22576,8 +22575,9 @@ 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).
+    """Evaluate a proposed MCP restart and return an impact preview (#658/#669).
    Central restart coordinator: resolves the requested restart class, gathers
    live control-plane state (sessions,
@@ -22601,10 +22601,16 @@ 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 only; it never bypasses the class matrix. ``apply_gate`` carries
+    drain proof and, when env-authorized, the #669 attempt-log requirement for
    broad restarts; 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
@@ -22710,12 +22716,37 @@ 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(
@@ -22731,6 +22762,7 @@ 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()
@@ -22751,38 +22783,6 @@ 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,13 +22795,6 @@ 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,
@@ -22842,86 +22835,6 @@ 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
@@ -0,0 +1,583 @@
 """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())},
    }
@@ -1,426 +0,0 @@
 """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
@@ -1,4 +1,4 @@
-"""MCP restart coordinator and impact analysis (#658).
+"""MCP restart coordinator and impact analysis (#658 / #669).
 Before any sanctioned MCP restart, a central coordinator must evaluate the
 live control-plane state — active sessions, leases/locks, in-flight issue/PR
@@ -16,6 +16,9 @@ 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.
@@ -32,8 +35,9 @@ from enum import Enum
 from typing import Any, Mapping, Sequence
 import lease_lifecycle
 import recovery_playbook
-COORDINATOR_VERSION = "1.1.0-issue-663"
+COORDINATOR_VERSION = "1.2.0-issue-669"
 # Restart verdicts. Exactly the three the acceptance criteria name.
 VERDICT_SAFE = "safe"
@@ -349,6 +353,10 @@ 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 {
@@ -382,6 +390,9 @@ 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,
        }
@@ -496,6 +507,7 @@ 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.
@@ -584,6 +596,30 @@ 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,
@@ -682,6 +718,7 @@ def evaluate_restart_impact(
        and role_authorized
        and approval_satisfied
        and target_complete
        and attempt_log_satisfied
    )
    if policy_enforced and not authorization_ok:
@@ -745,6 +782,7 @@ 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 = {
@@ -765,6 +803,9 @@ 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(
@@ -804,4 +845,7 @@ 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,6 +35,17 @@ 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."""
@@ -128,6 +139,7 @@ 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"],
@@ -136,6 +148,7 @@ 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"},
@@ -342,6 +355,7 @@ 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"},
        )
@@ -0,0 +1,217 @@
 """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
    )
@@ -1,342 +0,0 @@
 """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()
@@ -0,0 +1,511 @@
 """Tests for the Gitea issue↔PR linkage console (#645, Phase 3).
 Covers the acceptance criteria of the issue:
 * AC1 — issue↔PR linkage is visible for the selected project/repo, in both
  directions, with the evidence that produced each edge.
 * AC2 — the latest canonical handoff (CTH) is summarized for a focused thread.
 * AC3 — an external Gitea link appears only under the admin reveal opt-in.
 * AC4 — every case is driven by mocked Gitea payloads; no network.
 Plus the invariants this console must not violate: a partial or failed read is
 never rendered as "no link exists", an unfetched thread is never rendered as
 "no handoff", redaction happens before display, and the surface stays read-only.
 """
 from __future__ import annotations
 import json
 import os
 import sys
 import unittest
 from pathlib import Path
 from unittest import mock
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 from tests.webui_testclient import TestClient
 from canonical_thread_handoff import format_cth_body
 from webui.app import create_app
 from webui.linkage_loader import (
    EVIDENCE_BRANCH,
    EVIDENCE_CLOSES,
    EVIDENCE_REFERENCE,
    HANDOFF_LOADED,
    HANDOFF_NOT_LOADED,
    HANDOFF_UNAVAILABLE,
    LinkageSnapshot,
    load_linkage_snapshot,
    resolve_linkage,
    resolve_pr_links,
    snapshot_to_dict,
    summarize_handoff,
 )
 from webui.linkage_views import render_linkage_page
 from webui.nav import nav_hrefs
 from webui.queue_loader import PaginationMeta
 def _pagination(*, complete: bool = True, count: int = 0) -> PaginationMeta:
    return PaginationMeta(
        page=1,
        per_page=50,
        returned_count=count,
        has_more=not complete,
        is_final_page=complete,
        inventory_complete=complete,
        pages_fetched=1,
    )
 def _pr(
    number: int,
    *,
    title: str = "",
    body: str = "",
    head: str = "",
    state: str = "open",
    labels: tuple[str, ...] = (),
 ) -> dict:
    return {
        "number": number,
        "title": title or f"pr {number}",
        "body": body,
        "state": state,
        "head": {"ref": head},
        "labels": [{"name": name} for name in labels],
    }
 def _issue(
    number: int,
    *,
    title: str = "",
    state: str = "open",
    labels: tuple[str, ...] = (),
 ) -> dict:
    return {
        "number": number,
        "title": title or f"issue {number}",
        "state": state,
        "labels": [{"name": name} for name in labels],
    }
 def _fetcher(items: list[dict], *, complete: bool = True):
    def _fetch(*_args, **_kwargs):
        return items, _pagination(complete=complete, count=len(items))
    return _fetch
 def _load(
    issues: list[dict],
    prs: list[dict],
    *,
    complete: bool = True,
    **kwargs,
 ) -> LinkageSnapshot:
    return load_linkage_snapshot(
        fetch_prs=_fetcher(prs, complete=complete),
        fetch_issues=_fetcher(issues, complete=complete),
        **kwargs,
    )
 def _cth(comment_id: int, *, created_at: str, status: str, next_owner: str) -> dict:
    return {
        "id": comment_id,
        "created_at": created_at,
        "user": {"login": "jcwalker3"},
        "body": format_cth_body(
            cth_type="Author Handoff",
            status=status,
            next_owner=next_owner,
            current_blocker="none",
            decision="implemented",
            proof="full suite green",
            next_action="review PR",
            ready_to_paste_prompt="Review PR #902 now.",
        ),
    }
 class TestLinkageEvidence(unittest.TestCase):
    """AC1 — every edge records how it was found, and keeps all candidates."""
    def test_closes_keyword_in_body_is_strongest_evidence(self):
        links = resolve_pr_links(_pr(902, body="Closes #643"))
        self.assertEqual([link.issue_number for link in links], [643])
        self.assertEqual(links[0].evidence, (EVIDENCE_CLOSES,))
        self.assertTrue(links[0].closes)
    def test_closes_keyword_in_title_counts(self):
        links = resolve_pr_links(_pr(902, title="feat(webui): preview (Closes #643)"))
        self.assertEqual(links[0].evidence, (EVIDENCE_CLOSES,))
    def test_canonical_branch_marker_links_without_a_keyword(self):
        links = resolve_pr_links(_pr(902, head="feat/issue-643-request-preview"))
        self.assertEqual([link.issue_number for link in links], [643])
        self.assertEqual(links[0].evidence, (EVIDENCE_BRANCH,))
        self.assertFalse(links[0].closes)
    def test_non_canonical_branch_is_not_treated_as_a_marker(self):
        self.assertEqual(resolve_pr_links(_pr(902, head="issue-643-preview")), ())
    def test_bare_mention_is_recorded_as_the_weakest_evidence(self):
        links = resolve_pr_links(_pr(902, body="context in #643"))
        self.assertEqual(links[0].evidence, (EVIDENCE_REFERENCE,))
        self.assertFalse(links[0].closes)
    def test_several_evidence_kinds_merge_onto_one_edge(self):
        links = resolve_pr_links(
            _pr(902, body="Closes #643 — see #643", head="feat/issue-643-preview")
        )
        self.assertEqual(len(links), 1)
        self.assertEqual(
            links[0].evidence,
            (EVIDENCE_CLOSES, EVIDENCE_BRANCH, EVIDENCE_REFERENCE),
        )
    def test_stronger_evidence_sorts_first(self):
        links = resolve_pr_links(_pr(902, body="Closes #643, related #700"))
        self.assertEqual([link.issue_number for link in links], [643, 700])
    def test_self_reference_is_not_linkage(self):
        links = resolve_pr_links(_pr(902, body="supersedes #902"))
        self.assertEqual(links, ())
    def test_every_candidate_is_kept_never_collapsed_to_a_guess(self):
        links = resolve_pr_links(_pr(902, body="Closes #643\nCloses #644"))
        self.assertEqual([link.issue_number for link in links], [643, 644])
 class TestLinkageIndex(unittest.TestCase):
    def test_issue_direction_ignores_mention_only_edges(self):
        index = resolve_linkage([_pr(902, body="context in #643")])
        self.assertIsNone(index.issue_prs.get(643))
        self.assertEqual(index.pr_links[902][0].evidence, (EVIDENCE_REFERENCE,))
    def test_contested_issue_is_reported_when_two_prs_claim_it(self):
        index = resolve_linkage(
            [_pr(902, body="Closes #643"), _pr(903, head="feat/issue-643-again")]
        )
        self.assertEqual(index.contested_issues(), (643,))
        self.assertEqual(index.issue_prs[643], (902, 903))
    def test_single_claim_is_not_contested(self):
        index = resolve_linkage([_pr(902, body="Closes #643")])
        self.assertEqual(index.contested_issues(), ())
    def test_ambiguous_when_two_issues_tie_at_the_strongest_evidence(self):
        index = resolve_linkage([_pr(902, body="Closes #643\nCloses #644")])
        self.assertTrue(index.ambiguous(902))
    def test_weaker_candidate_alongside_a_stronger_one_is_not_ambiguous(self):
        index = resolve_linkage([_pr(902, body="Closes #643, see #700")])
        self.assertFalse(index.ambiguous(902))
        self.assertEqual(index.primary_issue(902).issue_number, 643)
    def test_malformed_pr_row_is_skipped_not_raised_on(self):
        index = resolve_linkage([{"title": "no number"}, _pr(902, body="Closes #643")])
        self.assertEqual(sorted(index.pr_links), [902])
 class TestLinkageSnapshot(unittest.TestCase):
    """AC1 — linkage is visible per project/repo, in both directions."""
    def test_both_directions_are_populated(self):
        snapshot = _load([_issue(643)], [_pr(902, body="Closes #643")])
        self.assertTrue(snapshot.ok)
        self.assertEqual([node.number for node in snapshot.issues], [643])
        self.assertEqual(snapshot.issues[0].linked_prs, (902,))
        self.assertEqual(snapshot.prs[0].links[0].issue_number, 643)
    def test_repo_scope_comes_from_the_registry_project(self):
        snapshot = _load([], [])
        self.assertIn("/", snapshot.repo_label)
        self.assertTrue(snapshot.project_id)
    def test_unknown_project_fails_closed_with_a_reason(self):
        snapshot = _load([_issue(643)], [], project_id="no-such-project")
        self.assertFalse(snapshot.ok)
        self.assertIn("not found in registry", snapshot.fetch_error)
        self.assertEqual(snapshot.issues, ())
    def test_orphan_pr_is_identifiable(self):
        snapshot = _load([], [_pr(902), _pr(903, body="Closes #643")])
        self.assertEqual([node.number for node in snapshot.orphan_prs], [902])
    def test_state_scope_defaults_to_open_and_is_reported(self):
        self.assertEqual(_load([], []).state_scope, "open")
        self.assertEqual(_load([], [], state="all").state_scope, "all")
    def test_unsupported_state_falls_back_to_open(self):
        self.assertEqual(_load([], [], state="../etc").state_scope, "open")
    def test_state_is_passed_through_to_the_fetchers(self):
        seen: list[str] = []
        def _fetch(*_args, **kwargs):
            seen.append(kwargs.get("state", ""))
            return [], _pagination()
        load_linkage_snapshot(state="all", fetch_prs=_fetch, fetch_issues=_fetch)
        self.assertEqual(seen, ["all", "all"])
 class TestPartialInventoryIsNotAnAbsenceClaim(unittest.TestCase):
    """An empty edge list from a partial read must never read as 'no link'."""
    def test_incomplete_pagination_marks_links_non_authoritative(self):
        snapshot = _load([_issue(643)], [], complete=False)
        self.assertFalse(snapshot.inventory_complete)
        self.assertFalse(snapshot.issues[0].links_authoritative)
    def test_complete_pagination_marks_links_authoritative(self):
        snapshot = _load([_issue(643)], [], complete=True)
        self.assertTrue(snapshot.inventory_complete)
        self.assertTrue(snapshot.issues[0].links_authoritative)
    def test_partial_window_renders_a_qualified_empty_cell(self):
        html = render_linkage_page(_load([_issue(643)], [], complete=False))
        self.assertIn("none found (partial inventory)", html)
    def test_complete_window_renders_a_plain_none(self):
        html = render_linkage_page(_load([_issue(643)], [], complete=True))
        self.assertNotIn("partial inventory", html)
        self.assertIn(">none<", html)
    def test_missing_credentials_fail_closed_without_a_table(self):
        with mock.patch(
            "webui.linkage_loader._offline_test_mode", return_value=False
        ), mock.patch("webui.linkage_loader.get_auth_header", return_value=""):
            snapshot = load_linkage_snapshot()
        self.assertFalse(snapshot.ok)
        self.assertIn("credentials unavailable", snapshot.fetch_error)
        html = render_linkage_page(snapshot)
        self.assertIn("Linkage unavailable", html)
        self.assertNotIn("Issues → pull requests", html)
    def test_fetch_failure_is_reported_not_raised(self):
        def _boom(*_args, **_kwargs):
            raise RuntimeError("gitea 502")
        snapshot = load_linkage_snapshot(fetch_prs=_boom, fetch_issues=_boom)
        self.assertFalse(snapshot.ok)
        self.assertIn("Gitea fetch failed", snapshot.fetch_error)
 class TestHandoffSummary(unittest.TestCase):
    """AC2 — the latest canonical handoff is summarized for a focused thread."""
    def test_latest_cth_wins(self):
        summary = summarize_handoff([
            _cth(1, created_at="2026-07-24T10:00:00Z", status="in progress",
                 next_owner="author"),
            _cth(2, created_at="2026-07-25T10:00:00Z", status="PR-open",
                 next_owner="reviewer"),
        ])
        self.assertEqual(summary.comment_id, 2)
        self.assertEqual(summary.status, "PR-open")
        self.assertEqual(summary.next_owner, "reviewer")
        self.assertTrue(summary.cth_type_known)
    def test_thread_without_a_cth_summarizes_to_none(self):
        self.assertIsNone(summarize_handoff([{"id": 1, "body": "ordinary comment"}]))
    def test_unknown_heading_is_reported_not_republished(self):
        summary = summarize_handoff([
            {
                "id": 5,
                "created_at": "2026-07-25T10:00:00Z",
                "user": {"login": "someone"},
                "body": "<!-- cth:v1 -->\n## CTH: Totally Made Up\n\nStatus: odd\n",
            }
        ])
        self.assertFalse(summary.cth_type_known)
        self.assertEqual(summary.cth_type, "unrecognized")
        self.assertNotIn("Totally Made Up", json.dumps(summary.to_dict()))
    def test_focused_pr_loads_its_handoff(self):
        snapshot = _load(
            [_issue(643)],
            [_pr(902, body="Closes #643")],
            pr=902,
            comment_source=lambda kind, number: [
                _cth(2, created_at="2026-07-25T10:00:00Z", status="PR-open",
                     next_owner="reviewer")
            ],
        )
        self.assertEqual(snapshot.handoff_status.state, HANDOFF_LOADED)
        self.assertEqual(snapshot.focus, ("pr", 902))
        self.assertEqual(snapshot.prs[0].handoff.status, "PR-open")
    def test_unfocused_rows_report_not_loaded_never_none(self):
        snapshot = _load(
            [_issue(643)],
            [_pr(902, body="Closes #643"), _pr(903)],
            pr=902,
            comment_source=lambda kind, number: [],
        )
        other = next(node for node in snapshot.prs if node.number == 903)
        self.assertIsNone(other.handoff)
        self.assertEqual(other.handoff_status.state, HANDOFF_NOT_LOADED)
        self.assertIn("not loaded", render_linkage_page(snapshot))
    def test_no_focus_means_no_thread_is_claimed_handoff_free(self):
        snapshot = _load([_issue(643)], [])
        self.assertEqual(snapshot.handoff_status.state, HANDOFF_NOT_LOADED)
        self.assertIn("thread-scoped", snapshot.handoff_status.reason)
    def test_comment_source_failure_degrades_only_the_handoff(self):
        def _boom(_kind, _number):
            raise RuntimeError("comments 500")
        snapshot = _load(
            [_issue(643)], [_pr(902, body="Closes #643")], pr=902, comment_source=_boom
        )
        self.assertTrue(snapshot.ok)
        self.assertEqual(snapshot.handoff_status.state, HANDOFF_UNAVAILABLE)
        self.assertEqual(snapshot.issues[0].linked_prs, (902,))
        self.assertIn("unavailable", render_linkage_page(snapshot))
    def test_loaded_thread_with_no_cth_says_so_explicitly(self):
        snapshot = _load(
            [_issue(643)],
            [_pr(902, body="Closes #643")],
            pr=902,
            comment_source=lambda kind, number: [{"id": 1, "body": "hi"}],
        )
        self.assertIn(
            "no Canonical Thread Handoff comment found", render_linkage_page(snapshot)
        )
 class TestDeepLinks(unittest.TestCase):
    """AC3 — an external Gitea link is emitted only when permitted."""
    def test_deep_links_are_withheld_by_default(self):
        with mock.patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": ""}):
            snapshot = _load([_issue(643)], [])
            html = render_linkage_page(snapshot)
        self.assertFalse(snapshot.deep_links_enabled)
        self.assertIsNone(snapshot.issues[0].deep_link)
        self.assertIn("Gitea deep links are withheld", html)
    def test_reveal_opt_in_emits_the_link(self):
        with mock.patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}):
            snapshot = _load([_issue(643)], [_pr(902, body="Closes #643")])
            html = render_linkage_page(snapshot)
        self.assertTrue(snapshot.deep_links_enabled)
        self.assertIn("/issues/643", snapshot.issues[0].deep_link)
        self.assertIn("/pulls/902", snapshot.prs[0].deep_link)
        self.assertIn(f'href="{snapshot.issues[0].deep_link}"', html)
 class TestRedactionBoundary(unittest.TestCase):
    def test_secret_shaped_title_is_redacted_before_display(self):
        snapshot = _load(
            [_issue(643, title="token=ghp_thisisnotarealsecretvalue0001")], []
        )
        payload = json.dumps(snapshot_to_dict(snapshot))
        self.assertNotIn("ghp_thisisnotarealsecretvalue0001", payload)
        self.assertNotIn(
            "ghp_thisisnotarealsecretvalue0001", render_linkage_page(snapshot)
        )
    def test_handoff_fields_are_redacted(self):
        comment = _cth(
            2, created_at="2026-07-25T10:00:00Z", status="ok", next_owner="reviewer"
        )
        comment["body"] += "\nDecision: password=hunter2hunter2\n"
        snapshot = _load(
            [_issue(643)],
            [_pr(902, body="Closes #643")],
            pr=902,
            comment_source=lambda kind, number: [comment],
        )
        self.assertNotIn("hunter2hunter2", json.dumps(snapshot_to_dict(snapshot)))
        self.assertNotIn("hunter2hunter2", render_linkage_page(snapshot))
    def test_html_escapes_markup_in_a_title(self):
        snapshot = _load([_issue(643, title="<script>alert(1)</script>")], [])
        html = render_linkage_page(snapshot)
        self.assertNotIn("<script>alert(1)</script>", html)
        self.assertIn("&lt;script&gt;", html)
 class TestLinkageRoutes(unittest.TestCase):
    def setUp(self):
        self.snapshot = _load(
            [_issue(643, labels=("status:ready",))],
            [_pr(902, body="Closes #643", labels=("status:pr-open",))],
        )
        self.client = TestClient(create_app())
    def test_page_renders_both_tables(self):
        with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
            response = self.client.get("/gitea")
        self.assertEqual(response.status_code, 200)
        self.assertIn("Issues → pull requests", response.text)
        self.assertIn("Pull requests → issues", response.text)
        self.assertIn("#643", response.text)
    def test_api_exports_the_same_model(self):
        with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
            response = self.client.get("/api/v1/gitea/linkage")
        self.assertEqual(response.status_code, 200)
        payload = response.json()
        self.assertTrue(payload["ok"])
        self.assertEqual(payload["issues"][0]["linked_prs"], [902])
        self.assertEqual(payload["prs"][0]["links"][0]["issue_number"], 643)
        self.assertEqual(payload["schema_version"], 1)
    def test_api_declares_the_evidence_vocabulary(self):
        with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
            payload = self.client.get("/api/v1/gitea/linkage").json()
        names = {entry["name"] for entry in payload["evidence_kinds"]}
        self.assertEqual(names, {EVIDENCE_CLOSES, EVIDENCE_BRANCH, EVIDENCE_REFERENCE})
    def test_api_fails_closed_with_a_non_200_when_the_read_failed(self):
        failed = _load([], [], project_id="no-such-project")
        with mock.patch("webui.app.load_linkage_snapshot", return_value=failed):
            response = self.client.get("/api/v1/gitea/linkage")
        self.assertEqual(response.status_code, 502)
        self.assertFalse(response.json()["ok"])
    def test_page_still_renders_when_the_read_failed(self):
        failed = _load([], [], project_id="no-such-project")
        with mock.patch("webui.app.load_linkage_snapshot", return_value=failed):
            response = self.client.get("/gitea")
        self.assertEqual(response.status_code, 200)
        self.assertIn("Linkage unavailable", response.text)
    def test_query_parameters_reach_the_loader(self):
        with mock.patch(
            "webui.app.load_linkage_snapshot", return_value=self.snapshot
        ) as loader:
            self.client.get("/gitea?project=gitea-tools&state=all&pr=902")
        loader.assert_called_once()
        args, kwargs = loader.call_args
        self.assertEqual(args[0], "gitea-tools")
        self.assertEqual(kwargs["state"], "all")
        self.assertEqual(kwargs["pr"], 902)
        self.assertIsNone(kwargs["issue"])
    def test_surface_stays_read_only(self):
        for path in ("/gitea", "/api/v1/gitea/linkage"):
            with self.subTest(path=path):
                self.assertEqual(self.client.post(path).status_code, 405)
    def test_nav_exposes_the_linkage_page_as_live(self):
        self.assertIn("/gitea", nav_hrefs())
        home = self.client.get("/").text
        self.assertIn('href="/gitea"', home)
        self.assertIn(">Gitea<", home)
 if __name__ == "__main__":
    unittest.main()
@@ -53,6 +53,11 @@ 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,
@@ -342,6 +347,41 @@ 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
@@ -785,6 +825,8 @@ 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"]),
@@ -0,0 +1,771 @@
 """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,
    )
@@ -0,0 +1,364 @@
 """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)
@@ -6,7 +6,8 @@ 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). Later-phase surfaces are declared as ``stub`` items and
+Insights (placeholder), joined by the Phase 3 Gitea linkage group (#645).
 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.
 """
@@ -60,6 +61,9 @@ 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"),
@@ -211,6 +211,19 @@ 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,
@@ -218,8 +231,15 @@ def _fetch_prs(
    auth: str,
    *,
    per_page: int = 50,
    state: str = "open",
 ) -> tuple[list[dict], PaginationMeta]:
-    url = f"{repo_api_url(host, org, repo)}/pulls?state=open"
+    """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)}"
    all_raw: list[dict] = []
    pages_fetched = 0
    is_final = False
@@ -248,8 +268,13 @@ def _fetch_issues(
    auth: str,
    *,
    per_page: int = 50,
    state: str = "open",
 ) -> tuple[list[dict], PaginationMeta]:
-    url = f"{repo_api_url(host, org, repo)}/issues?state=open&type=issues"
+    """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"
    )
    all_raw: list[dict] = []
    page = 1
    pages_fetched = 0