Merge branch 'master' into feat/issue-643-request-preview-initiate

Addresses both blockers from the PR #902 review (review 589) for issue #643.

B1 — an allocator-created assignment could be orphaned and reported as no
mutation.

apply_request re-previews, then re-runs the allocator with apply=True. The CAS
fingerprint hashes only {kind, number} plus exclusions, so a competing lease
taken on the requested unit inside the window leaves the fingerprint identical:
the pin passes, the selection loop skips the now-claimed unit and commits an
assignment on the *next* one, and _selection_matches then fails on egress. The
old code returned mutation_performed False with that lease still committed and
owned by a synthetic session nothing heartbeats. The window contains a second
full load_queue_snapshot(), so it is seconds wide, and foreign sessions acting
on this repo concurrently are an observed condition.

The egress mismatch now releases the assignment the allocator created before
refusing. When the release succeeds the refusal reports mutation_performed
False and a compensation record; when it fails the response carries
mutation_performed True, an explicit orphaned_assignment, and a
gitea_release_workflow_lease reclaim action, because claiming nothing changed
while a lease is live is the defect rather than a report of it.

The same state was reachable through _run_allocator's bare except Exception:
allocate_next_work only catches InvalidWorkKindError, LeaseRequiredError and
ControlPlaneError, so anything raised after assign_and_lease committed arrived
as "no result" with a durable lease. A None result on the apply path now sweeps
and releases whatever this flow's session owns. That sweep is only possible
because of the B2 fix below — the session id is now stable across the flow, so
the lease is findable.

B2 — the "read-only" preview wrote to the control-plane DB.

allocate_next_work called db.upsert_session and db.expire_stale_leases
unconditionally, before the apply branch was consulted, and default_allocator
minted a fresh webui-request-<hex> per call. Every preview therefore appended a
never-reused session row and mutated global lease state while the payload said
dry_run True / mutation_performed False, driven by an operator refreshing a
form. One apply wrote two rows and bound the lease to the second, which is why
B1's orphan had no reclaimable owner.

allocate_next_work gains a keyword-only side_effect_free flag, default False so
every existing caller is byte-for-byte unchanged. Under the flag both writes are
suppressed and expired leases are instead filtered out of the claim map in
memory, which reaches the same selection the sweep would have produced without
persisting anything; a claim whose expiry cannot be parsed is kept, since an
unreadable expiry is not evidence that work is free. side_effect_free with
apply=True fails closed rather than silently reserving. request_service mints
one session id per request flow and threads it through both the dry-run and the
apply, and the dry-run now routes through the side-effect-free path.

Coverage.

test_allocator_drift_on_apply_is_not_read_as_an_assignment asserted the defect —
it built the orphan state and then required mutation_performed to be False, which
a leak satisfies. It now requires the compensating release. Added: release-failure
surfacing a reclaim action, an assignment with no lease id, the post-commit
exception route, and session-id identity across the flow. The two areas the review
named as having zero coverage now have it: default_allocator past its two
fail-closed early returns (side_effect_free routing, session-id pass-through and
minting, scope and fingerprint propagation) and default_claims_source (scoped
read, and an unreadable substrate denying rather than reading as "nothing
claimed"). allocator_service gains side-effect-free tests against a real
temp-file DB plus the expiry-filter unit tests.

Every new guard was mutation-tested by reverting it one at a time: B1
compensation removed → 3 failures; post-commit sweep removed → 1; session id
re-minted per call → 2; side_effect_free ignored → 3; in-memory expiry filter
removed → 1.

Verification: WEBUI_TEST_OFFLINE=1 python -m pytest tests/ -q from this
branches/ worktree gives 23 failed, 5263 passed, 6 skipped, 899 subtests. The
sorted FAILED set is identical to the reviewer's clean-master baseline at
2f4dec83 (23 failed, 5190 passed) — no new, changed, or disappeared failure —
and 21 tests were added over the reviewed head's 5242. Zero conflict markers;
py_compile passes; git diff --check clean.

Refs #643, PR #902

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01V6xFqovhbArPv61j9KCGkL

allocator_service.py

@@ -23,6 +23,7 @@ import json
 import os
 import uuid
 from dataclasses import dataclass, field
+from datetime import datetime, timezone
 from typing import Any, Mapping, Sequence
 
 from control_plane_db import (
@@ -738,6 +739,46 @@ def normalize_exclude_issue_numbers(
     return sorted(out)
 
 
+def _claim_expires_at(claim: Any) -> datetime | None:
+    """Parse a claim's ``expires_at``, or ``None`` when it is absent/malformed."""
+    if not isinstance(claim, Mapping):
+        return None
+    text = str(claim.get("expires_at") or "").strip()
+    if not text:
+        return None
+    if text.endswith("Z"):
+        text = text[:-1] + "+00:00"
+    try:
+        parsed = datetime.fromisoformat(text)
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        parsed = parsed.replace(tzinfo=timezone.utc)
+    return parsed.astimezone(timezone.utc)
+
+
+def _drop_expired_claims(
+    claims: Mapping[tuple[str, int], dict[str, Any]],
+    *,
+    now: datetime | None = None,
+) -> dict[tuple[str, int], dict[str, Any]]:
+    """Claims minus those whose lease has already expired (#643).
+
+    The read-only mirror of ``expire_stale_leases``: the sweep marks such rows
+    ``expired`` so they stop being returned as claims, and this reaches the same
+    view without writing. A claim with no parseable ``expires_at`` is **kept** —
+    an unreadable expiry is not evidence that work is free.
+    """
+    moment = now or datetime.now(timezone.utc)
+    kept: dict[tuple[str, int], dict[str, Any]] = {}
+    for key, claim in (claims or {}).items():
+        expires_at = _claim_expires_at(claim)
+        if expires_at is not None and expires_at <= moment:
+            continue
+        kept[key] = claim
+    return kept
+
+
 def candidate_set_fingerprint(
     candidates: Sequence[WorkCandidate],
     *,
@@ -826,12 +867,22 @@ def allocate_next_work(
     exclude_issue_numbers: Sequence[int] | None = None,
     expected_candidate_set_fingerprint: str | None = None,
     allocation_mode: str | None = None,
+    side_effect_free: bool = False,
 ) -> dict[str, Any]:
     """Select and optionally reserve the next work unit via control-plane DB.
 
     *apply=False* (default): dry-run selection only — no lease/assignment.
     *apply=True*: atomic ``assign_and_lease`` for the selected candidate.
 
+    *side_effect_free* (#643): a dry run that writes **nothing** to the
+    control-plane DB. A plain ``apply=False`` still registered a session row and
+    swept stale leases globally, so a caller advertising a read-only preview was
+    mutating on every call. Under this flag both writes are suppressed and stale
+    leases are instead filtered out of the claim map in memory, which yields the
+    same selection the sweep would have produced without persisting anything.
+    Incompatible with *apply* — the combination fails closed rather than
+    silently reserving.
+
     *allocation_mode* (#840): ``cross_role`` (default for controller) inspects
     the complete queue and returns one authoritative selection naming the
     required downstream role/profile/action. ``role_scoped`` keeps prior
@@ -885,40 +936,57 @@ def allocate_next_work(
             "allocation_mode": (allocation_mode or "").strip() or None,
         }
 
-    session_id = (session_id or "").strip() or f"alloc-{uuid.uuid4().hex[:12]}"
+    # A side-effect-free run may never reserve: reserving is a write, and the
-    try:
+    # flag is the caller's assertion that this call writes nothing (#643).
-        db.upsert_session(
+    if side_effect_free and apply:
-            session_id=session_id,
-            role=role_norm,
-            profile=profile_name,
-            pid=os.getpid(),
-            controller_instance_id=controller_instance_id,
-        )
-    except Exception as exc:  # noqa: BLE001 — surface structured
         return {
             "success": False,
             "outcome": OUTCOME_NO_SAFE,
+            "apply": True,
             "reasons": [
-                f"failed to register session in control-plane DB: {exc} "
+                "side_effect_free is incompatible with apply=True; an "
-                "(fail closed, #613)"
+                "assignment is a write (fail closed, #643)"
             ],
             "skipped": [],
             "assignment": None,
             "substrate": "control_plane_db",
         }
 
-    # Expire stale leases globally before selection.
+    session_id = (session_id or "").strip() or f"alloc-{uuid.uuid4().hex[:12]}"
-    try:
+    if not side_effect_free:
-        db.expire_stale_leases()
+        try:
-    except Exception as exc:  # noqa: BLE001
+            db.upsert_session(
-        return {
+                session_id=session_id,
-            "success": False,
+                role=role_norm,
-            "outcome": OUTCOME_NO_SAFE,
+                profile=profile_name,
-            "reasons": [f"lease expiry failed: {exc} (fail closed)"],
+                pid=os.getpid(),
-            "skipped": [],
+                controller_instance_id=controller_instance_id,
-            "assignment": None,
+            )
-            "substrate": "control_plane_db",
+        except Exception as exc:  # noqa: BLE001 — surface structured
-        }
+            return {
+                "success": False,
+                "outcome": OUTCOME_NO_SAFE,
+                "reasons": [
+                    f"failed to register session in control-plane DB: {exc} "
+                    "(fail closed, #613)"
+                ],
+                "skipped": [],
+                "assignment": None,
+                "substrate": "control_plane_db",
+            }
+
+        # Expire stale leases globally before selection.
+        try:
+            db.expire_stale_leases()
+        except Exception as exc:  # noqa: BLE001
+            return {
+                "success": False,
+                "outcome": OUTCOME_NO_SAFE,
+                "reasons": [f"lease expiry failed: {exc} (fail closed)"],
+                "skipped": [],
+                "assignment": None,
+                "substrate": "control_plane_db",
+            }
 
     terminal = None
     try:
@@ -953,6 +1021,12 @@ def allocate_next_work(
                 "assignment": None,
                 "substrate": "control_plane_db",
             }
+        if side_effect_free:
+            # ``list_active_claims`` filters on status alone, so without the
+            # global sweep an already-expired lease would still read as a live
+            # claim and the preview would report work as taken that is free.
+            # Drop those in memory: same view the sweep produces, no write.
+            claims = _drop_expired_claims(claims)
 
     try:
         exclude_nums = normalize_exclude_issue_numbers(exclude_issue_numbers)

docs/webui-authz-audit.md

@@ -94,6 +94,7 @@ already define, and a regression test asserts each mapping matches.
 | `record_analytics_usage` | operator | gated_write | `runtime.record_analytics_usage` | Yes | No | No | 2 |
 | `system.reload_namespace` | controller | privileged | `runtime.reload_namespace` | Yes | No | No | 2 |
 | `system.restart_namespace` | admin | destructive | `runtime.restart_namespace` | Yes | **Yes** | **Yes** | 2 |
+| `initiate_workflow` | operator | gated_write | `gitea.read` | Yes | No | No | 2 |
 
 **Dual control** means the acting principal may not be the sole authority: a
 second distinct principal must confirm. **Break-glass** means the action is
@@ -112,6 +113,12 @@ by the console — both hand off to a host supervisor, and neither exposes a raw
 process kill. See
 [`sanctioned-restart-controls.md`](sanctioned-restart-controls.md) (#642).
 
+`initiate_workflow` (#643) is operator-class because its outcome is a *claim*,
+not a Gitea verdict. Requesting reviewer or merger work reserves that work
+through the allocator; it does not grant the right to approve or merge, which
+stays with the MCP role profile and its own capability gates. See
+[`webui-requests.md`](webui-requests.md).
+
 ### Authorization decision
 
 `authorize(action_id, principal, for_execution=False)` returns a decision
@@ -126,9 +133,24 @@ record and **denies by default**. The deny reasons are closed and enumerated:
 | `phase_not_active` | Execution requested for an action whose phase is not open. |
 | `allowed_preview_only` | Authorized — preview only, execution still disabled. |
 
-There is no implicit allow branch. Even the allow result reports
+There is no implicit allow branch.
-`execution_enabled: false` while the console is in Phase 1, so no caller can
+
-read an allow as permission to mutate.
+`execution_enabled` on the decision reports whether the action has a live
+execution path at all, and is computed by `execution_wired(action)`. There are
+exactly two ways to be wired:
+
+1. the action's `phase` is at or below `ACTIVE_PHASE`; or
+2. the action declares an `execution_env_flag` **and** that variable is set.
+
+Every action that declares no flag therefore reports `execution_enabled: false`
+while the console is in Phase 1, so no caller can read an allow as permission
+to mutate. The per-action flag exists because raising `ACTIVE_PHASE` would
+enable execution for every action of that phase at once, including ones whose
+execution path is not implemented. One implemented action goes live on its own
+flag instead of dragging its unimplemented phase-mates with it.
+
+`initiate_workflow` is the only action that currently declares a flag
+(`WEBUI_REQUESTS_EXECUTION`), and it stays denied until an operator sets it.
 
 ## Secret redaction
 
@@ -235,13 +257,22 @@ second one. The integration points are already wired and observable:
   instead of adding a parallel check.
 - **`GET /api/console/security-model`** publishes the RBAC matrix, redaction
   policy, and audit policy as JSON for operators and tests.
+- **`POST /api/v1/requests/preview` and `.../apply`** (#643) are the first
+  actions to use this model for a real execution path. Preview always returns a
+  decision and an audited `previewed` record; apply requires `confirm=true`,
+  emits `succeeded` or `denied`, and reserves work only through the allocator.
+  See [`webui-requests.md`](webui-requests.md).
 
-To open Phase 2, a child issue must: raise `ACTIVE_PHASE`, implement the
+A Phase 2 action must: use `execution_wired` rather than a private enable flag,
-confirmation and dual-control flow the matrix already declares, emit a
+implement the confirmation and dual-control flow the matrix already declares,
-`succeeded` or `failed` record alongside the `gitea_audit` mutation record, and
+emit a `succeeded` or `failed` record alongside the `gitea_audit` mutation
-keep `viewer` unable to reach any of it. Turning on execution without the
+record, and keep `viewer` unable to reach any of it. Turning on execution
-confirmation flow contradicts a declared requirement and is a review failure,
+without the confirmation flow contradicts a declared requirement and is a
-not a shortcut.
+review failure, not a shortcut.
+
+Raising `ACTIVE_PHASE` remains the way to open a whole phase at once, and is
+deliberately *not* what #643 did: an action-scoped opt-in cannot enable an
+action whose execution path nobody wrote.
 
 ## Local-dev mode
 
@@ -294,6 +325,7 @@ Until Phase 2 wires it, probe protection rests on network placement alone, as
 | `WEBUI_ROLE_MAP` | unset | JSON subject → role map |
 | `WEBUI_REQUIRE_PROBE_AUTH` | unset | Require auth for non-public probes |
 | `WEBUI_CONSOLE_AUDIT_LOG` | unset | Append-only audit sink path |
+| `WEBUI_REQUESTS_EXECUTION` | unset | Opt in to `initiate_workflow` execution (#643) |
 
 All are read server-side only. None is ever rendered into a page or returned by
 an API.

docs/webui-local-dev.md

@@ -80,15 +80,10 @@ 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 |
-| `/providers` | AI-provider connection status (#650) — declared registry only, no secrets |
+| `/insights` | Phase 1 shell stub — operational insights placeholder |
-| `/api/v1/providers` | JSON provider connections; `502` when the registry cannot be loaded |
-| `/insights` | Evidence-backed operational insights (#650) — advisory only |
-| `/api/v1/insights` | JSON insights export with evidence refs and source availability |
 
 Most routes are GET-only. POST/PUT/PATCH/DELETE return `405` with
 `read-only-mvp`, except `/audit` and `/api/audit` which accept POST for
@@ -332,118 +327,6 @@ Honesty rules specific to this view:
   The write-time redactor is a narrow denylist and is not relied on. The field
   itself is kept — it is the `#630` evidence naming which daemon was killed.
 
-## AI providers and operational insights (#650)
-
-`/providers` and `/insights` are the Phase 4 **advisory** surfaces for AI-provider
-connections and evidence-backed operational findings. They never expose API keys,
-never mutate Gitea, and never authorize review, merge, or close.
-
-### Provider connections (`/providers`)
-
-Status is taken from the **worker registry** declaration (`webui/data/workers.registry.json`
-or `WEBUI_WORKER_REGISTRY`):
-
-| Field | Meaning |
-|-------|---------|
-| `connection_status` | `declared_available` or `declared_unavailable` from the registry `available` flag |
-| `models` | Declared model list only (not a live vendor enumeration) |
-| `worker_count` / `enabled_worker_count` | How many worker instances name this provider |
-| `secrets_exposed` | Always `false` — credentials are never loaded |
-
-Live executable health is **not** probed here (that belongs to the provider adapter
-framework). The page states this probe limit explicitly so a green badge is not
-misread as a process heartbeat.
-
-`GET /api/v1/providers` returns the same model (`schema_version: 1`). It answers
-`502` when the registry cannot be loaded so consumers cannot treat a fail-closed
-payload as “no providers configured”.
-
-### Operational insights (`/insights`)
-
-Insights are pure functions over durable console evidence:
-
-| Kind | Evidence source |
-|------|-----------------|
-| `blocked_queue_pressure` | Traffic control blocked bucket (issue/PR numbers + reasons) |
-| `controller_attention` | Traffic control `needs_controller` items |
-| `stale_runtime_risk` | System-health stale_runtime / mutation_safe |
-| `provider_without_workers` | Declared-available providers with zero workers |
-| `analytics_failure_pressure` | Analytics events with failure status (when loaded) |
-
-Rules:
-
-* Every insight carries at least one evidence ref (`kind` + `ref` + `detail`).
-  Evidence-less insights are refused, not emitted.
-* `advisory_only` is always true; `claims_action_completed` is always false.
-* Missing sources appear under `sources_unavailable` — never as a silent empty
-  “all clear”.
-* Titles and details pass through console redaction before display.
-
-`GET /api/v1/insights` exports the same model. The HTML page always renders
-interpretation limits so operators know these cards do not override workflow gates.
-
-## 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

docs/webui-requests.md

@@ -0,0 +1,160 @@
+# Web console requests: intent preview and workflow initiation (#643)
+
+**Phase 2. Preview is always live and always read-only. Initiation is wired but
+denied until an operator opts in.**
+
+Before this surface, starting role work meant pasting a prompt into a terminal
+and trusting the operator to have checked the allocator first. Nothing enforced
+that check, so two sessions could reach for the same issue and each believe it
+was theirs. This page replaces the paste with a *request*: a desired role, an
+issue or PR, and a stated intent, answered by an authorization decision and —
+on confirmation — an exclusive assignment from the allocator.
+
+| Concern | Module |
+|---------|--------|
+| Request model, preview, initiation | `webui/request_service.py` |
+| Form and preview rendering | `webui/request_views.py` |
+| Authorization | `webui/console_authz.py` (`initiate_workflow`) |
+| Audit | `webui/console_audit.py` |
+| Ownership substrate | `allocator_service.py` + `control_plane_db.py` |
+
+## Surfaces
+
+| Path | Method | Purpose |
+|------|--------|---------|
+| `/requests` | GET | Request form |
+| `/requests` | POST | Render an intent preview. **Never assigns.** |
+| `/api/v1/requests/preview` | POST | Intent preview as JSON |
+| `/api/v1/requests/apply` | POST | Initiate — confirmed, audited, allocator-owned |
+
+The HTML form has no initiate button on purpose. Initiating requires a
+confirmed POST to `/api/v1/requests/apply`, so a stray form submission cannot
+reserve work as a side effect.
+
+## The request
+
+```json
+{
+  "desired_role": "author",
+  "work_kind": "issue",
+  "work_number": 643,
+  "intent_summary": "implement request preview and initiation",
+  "remote": "prgs",
+  "org": "Scaled-Tech-Consulting",
+  "repo": "Gitea-Tools",
+  "expected_head_sha": null
+}
+```
+
+`desired_role` is one of `author`, `reviewer`, `merger`, `reconciler`,
+`controller`. `work_kind` is `issue` or `pr`. `remote`/`org`/`repo` default to
+the first project in the registry when omitted; when neither the request nor
+the registry resolves them, the request is rejected rather than pointed at some
+other repository. `intent_summary` is required — it is what the audit record
+states as the reason — and is truncated to 500 characters.
+
+Parsing rejects rather than corrects. An unknown role, an unknown work kind, a
+non-positive number, or a missing intent each return `400` with a `reason_code`
+and the offending `field`.
+
+## Preview
+
+Five checks, each with its own verdict, reason code, and detail:
+
+| Check | Passes when |
+|-------|-------------|
+| `authorization` | The console principal holds `operator` or above |
+| `capability` | The desired role maps to a declared profile and MCP namespace |
+| `lease_availability` | No active claim holds the work unit |
+| `next_safe_action` | The allocator would independently select this exact work unit |
+| `head_pin` | PR work resolves to a head SHA, and a supplied SHA still matches |
+
+A preview also returns the role's `allowed_actions` and `prohibited_actions`
+(from `allocator_service.ROLE_ACTIONS`), the `required_profile` and
+`required_namespace` the work must run under, and a `correlation_id` that ties
+the preview to its audit record and to any assignment that follows.
+
+Preview is read-only in the strict sense: it calls the allocator with
+`apply=false` and writes nothing but an audit line. An unauthorized principal
+never reaches the allocator or the control-plane DB at all, so a denial cannot
+be used to enumerate the queue.
+
+## Initiation
+
+`POST /api/v1/requests/apply` refuses in this order, and every refusal returns
+before any assignment is attempted:
+
+| Condition | Outcome | Status |
+|-----------|---------|--------|
+| Unparseable request | `invalid_request` | 400 |
+| Not authorized, or execution not wired | `denied` | 403 |
+| `confirm` not set | `denied` / `confirmation_required` | 409 |
+| Work unit already claimed | `blocked` / `duplicate_assignment` | 409 |
+| Allocator would select other work | `wait` / `not_next_safe_work` | 409 |
+| Allocator declines on apply | `blocked` or `wait` | 409 |
+| Evidence unavailable | `wait` / `evidence_unavailable` | 503 |
+| Assigned | `assigned_work` | 201 |
+
+A success returns the assignment plus a `handoff` block naming the profile, the
+namespace, and the actions that stay forbidden — enough for the operator to
+continue in the right MCP namespace without guessing.
+
+### Why apply runs the allocator twice
+
+The allocator is the only source of exclusive ownership (#600 / #613), and it
+selects work; it does not take orders. So `apply` runs a dry-run first and
+proceeds only when the allocator would independently pick the requested work
+unit. If it would not, the request reports `wait` and mutates nothing.
+
+A request is therefore a *confirmation* of the allocator's decision, never an
+override of it. The apply call carries the dry-run's
+`candidate_set_fingerprint` as a CAS pin (#776), so a queue that changed
+between the two calls fails closed rather than assigning against a stale view.
+The result is checked again on the way out: an assignment naming a different
+work unit is not read as success.
+
+### Fail-closed defaults
+
+- An unreadable control-plane DB denies. It is never treated as "nothing holds
+  this work unit".
+- An incomplete queue inventory denies (#758). Ranking a partial candidate set
+  can select the wrong work.
+- An allocator that raises denies.
+- PR work with no resolvable head SHA denies; a supplied SHA that no longer
+  matches denies with `head_moved`.
+
+## Enabling initiation
+
+Execution is wired off. Set `WEBUI_REQUESTS_EXECUTION=1` to enable it for the
+`initiate_workflow` action only — see
+[`webui-authz-audit.md`](webui-authz-audit.md) for why this is an
+action-scoped flag rather than a phase bump. With the variable unset, `apply`
+returns `403` with `reason_code: unauthorized` no matter who asks.
+
+Enabling execution does **not** enable approvals or merges. Those are phase 3
+console actions and remain forbidden in every path here; the console reserves
+work and hands off, and the MCP role profile enforces what that role may then
+do.
+
+## Audit
+
+Every preview and every apply emits a console audit record (schema in
+[`webui-authz-audit.md`](webui-authz-audit.md)):
+
+| Event | `result` |
+|-------|----------|
+| Preview | `previewed` |
+| Refusal at any stage | `denied` |
+| Assignment created | `succeeded` |
+
+`correlation.request_id` carries the request's `correlation_id`, and a
+successful record's `metadata` carries `assignment_id` and `lease_id`, so an
+assignment can be traced back to the intent that produced it. The operator's
+`intent_summary` travels in `metadata` and passes through the standard
+redaction pass before persistence like every other field.
+
+## Non-goals
+
+- No browser-initiated approve or merge, in this phase or any other.
+- No bypass of allocator exclusive ownership; no self-selection of work.
+- No auto-start from raw monitoring incidents (#612 stays downstream).

tests/test_allocator_service.py

@@ -7,6 +7,7 @@ import tempfile
 import threading
 import unittest
 from concurrent.futures import ThreadPoolExecutor, as_completed
+from datetime import datetime, timezone
 
 from allocator_service import (
     OUTCOME_ASSIGNED,
@@ -15,6 +16,7 @@ from allocator_service import (
     OUTCOME_PREVIEW,
     OUTCOME_WAIT,
     WorkCandidate,
+    _drop_expired_claims,
     allocate_next_work,
     candidate_from_dict,
     classify_skip,
@@ -362,5 +364,161 @@ class AllocatorServiceTest(unittest.TestCase):
         self.assertIn("unavailable", res["reasons"][0].lower())
 
 
+class SideEffectFreeAllocationTest(unittest.TestCase):
+    """``side_effect_free`` dry runs write nothing to the control plane (#643).
+
+    A plain ``apply=False`` still called ``upsert_session`` and
+    ``expire_stale_leases`` before the apply branch was consulted, so a caller
+    advertising a read-only preview mutated on every call — one unreferenced
+    session row per preview, plus a global lease sweep.
+    """
+
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def _alloc(self, **kwargs):
+        defaults = dict(
+            db=self.db,
+            session_id="s-preview",
+            role="author",
+            remote="prgs",
+            org="org",
+            repo="repo",
+            candidates=[
+                WorkCandidate(kind="issue", number=643, labels=("status:ready",))
+            ],
+            apply=False,
+            profile_name="prgs-author",
+            username="jcwalker3",
+        )
+        defaults.update(kwargs)
+        return allocate_next_work(**defaults)
+
+    def _session_ids(self) -> set[str]:
+        return {str(r.get("session_id")) for r in self.db.list_sessions()}
+
+    def test_side_effect_free_preview_writes_no_session_row(self):
+        before = self._session_ids()
+        result = self._alloc(side_effect_free=True)
+        self.assertEqual(result["outcome"], OUTCOME_PREVIEW)
+        self.assertEqual(self._session_ids(), before)
+        self.assertNotIn("s-preview", self._session_ids())
+
+    def test_plain_dry_run_still_registers_a_session(self):
+        # The default is unchanged for every existing caller.
+        self._alloc()
+        self.assertIn("s-preview", self._session_ids())
+
+    def test_repeated_previews_do_not_accumulate_rows(self):
+        for index in range(5):
+            self._alloc(side_effect_free=True, session_id=f"s-{index}")
+        self.assertEqual(self._session_ids(), set())
+
+    def test_side_effect_free_does_not_sweep_stale_leases(self):
+        self.db.upsert_session(session_id="owner", role="author", pid=1)
+        assigned = self.db.assign_and_lease(
+            session_id="owner",
+            role="author",
+            remote="prgs",
+            org="org",
+            repo="repo",
+            kind="issue",
+            number=999,
+            lease_ttl_seconds=-60,  # already expired
+        )
+        self.assertEqual(assigned.outcome, "assigned")
+
+        self._alloc(side_effect_free=True)
+
+        # The expired row is still 'active' in the DB: nothing swept it.
+        statuses = {
+            r["lease_id"]: r["status"]
+            for r in self.db.list_leases(
+                remote="prgs", org="org", repo="repo",
+                statuses=("active", "expired"),
+            )
+        }
+        self.assertEqual(statuses.get(assigned.lease_id), "active")
+
+    def test_expired_claims_are_filtered_in_memory_so_work_stays_selectable(self):
+        """The read-only mirror of the sweep: expired claims must not block."""
+        self.db.upsert_session(session_id="owner", role="author", pid=1)
+        self.db.assign_and_lease(
+            session_id="owner",
+            role="author",
+            remote="prgs",
+            org="org",
+            repo="repo",
+            kind="issue",
+            number=643,
+            lease_ttl_seconds=-60,  # expired: must not withhold #643
+        )
+        result = self._alloc(side_effect_free=True)
+        self.assertEqual(result["outcome"], OUTCOME_PREVIEW)
+        self.assertEqual(result["selected"]["number"], 643)
+
+    def test_a_live_claim_still_withholds_the_work(self):
+        self.db.upsert_session(session_id="owner", role="author", pid=1)
+        self.db.assign_and_lease(
+            session_id="owner",
+            role="author",
+            remote="prgs",
+            org="org",
+            repo="repo",
+            kind="issue",
+            number=643,
+            lease_ttl_seconds=3600,
+        )
+        result = self._alloc(side_effect_free=True)
+        self.assertNotEqual(result["outcome"], OUTCOME_ASSIGNED)
+        self.assertNotEqual((result.get("selected") or {}).get("number"), 643)
+
+    def test_side_effect_free_with_apply_fails_closed(self):
+        result = self._alloc(side_effect_free=True, apply=True)
+        self.assertFalse(result["success"])
+        self.assertEqual(result["outcome"], OUTCOME_NO_SAFE)
+        self.assertIsNone(result["assignment"])
+        self.assertIn("incompatible with apply", result["reasons"][0])
+        # And it reserved nothing.
+        self.assertEqual(
+            self.db.list_leases(remote="prgs", org="org", repo="repo"), []
+        )
+
+
+class DropExpiredClaimsTest(unittest.TestCase):
+    """The in-memory expiry filter behind side-effect-free previews (#643)."""
+
+    def test_unparseable_expiry_is_kept_rather_than_assumed_free(self):
+        claims = {
+            ("issue", 1): {"lease_id": "l1", "expires_at": "not-a-date"},
+            ("issue", 2): {"lease_id": "l2"},
+            ("issue", 3): {"lease_id": "l3", "expires_at": None},
+        }
+        self.assertEqual(_drop_expired_claims(claims), claims)
+
+    def test_expired_dropped_and_future_kept(self):
+        now = datetime(2026, 7, 25, 12, 0, tzinfo=timezone.utc)
+        claims = {
+            ("issue", 1): {"expires_at": "2026-07-25T11:59:59+00:00"},
+            ("issue", 2): {"expires_at": "2026-07-25T12:00:01+00:00"},
+            ("issue", 3): {"expires_at": "2026-07-25T12:00:00+00:00"},  # boundary
+        }
+        kept = _drop_expired_claims(claims, now=now)
+        self.assertEqual(set(kept), {("issue", 2)})
+
+    def test_naive_and_zulu_timestamps_are_treated_as_utc(self):
+        now = datetime(2026, 7, 25, 12, 0, tzinfo=timezone.utc)
+        claims = {
+            ("issue", 1): {"expires_at": "2026-07-25T11:00:00"},  # naive, past
+            ("issue", 2): {"expires_at": "2026-07-25T13:00:00Z"},  # zulu, future
+        }
+        kept = _drop_expired_claims(claims, now=now)
+        self.assertEqual(set(kept), {("issue", 2)})
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_webui_gitea_linkage.py

@@ -1,511 +0,0 @@
-"""Tests for the Gitea issue↔PR linkage console (#645, Phase 3).
-
-Covers the acceptance criteria of the issue:
-
-* AC1 — issue↔PR linkage is visible for the selected project/repo, in both
-  directions, with the evidence that produced each edge.
-* AC2 — the latest canonical handoff (CTH) is summarized for a focused thread.
-* AC3 — an external Gitea link appears only under the admin reveal opt-in.
-* AC4 — every case is driven by mocked Gitea payloads; no network.
-
-Plus the invariants this console must not violate: a partial or failed read is
-never rendered as "no link exists", an unfetched thread is never rendered as
-"no handoff", redaction happens before display, and the surface stays read-only.
-"""
-
-from __future__ import annotations
-
-import json
-import os
-import sys
-import unittest
-from pathlib import Path
-from unittest import mock
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-from tests.webui_testclient import TestClient
-
-from canonical_thread_handoff import format_cth_body
-from webui.app import create_app
-from webui.linkage_loader import (
-    EVIDENCE_BRANCH,
-    EVIDENCE_CLOSES,
-    EVIDENCE_REFERENCE,
-    HANDOFF_LOADED,
-    HANDOFF_NOT_LOADED,
-    HANDOFF_UNAVAILABLE,
-    LinkageSnapshot,
-    load_linkage_snapshot,
-    resolve_linkage,
-    resolve_pr_links,
-    snapshot_to_dict,
-    summarize_handoff,
-)
-from webui.linkage_views import render_linkage_page
-from webui.nav import nav_hrefs
-from webui.queue_loader import PaginationMeta
-
-
-def _pagination(*, complete: bool = True, count: int = 0) -> PaginationMeta:
-    return PaginationMeta(
-        page=1,
-        per_page=50,
-        returned_count=count,
-        has_more=not complete,
-        is_final_page=complete,
-        inventory_complete=complete,
-        pages_fetched=1,
-    )
-
-
-def _pr(
-    number: int,
-    *,
-    title: str = "",
-    body: str = "",
-    head: str = "",
-    state: str = "open",
-    labels: tuple[str, ...] = (),
-) -> dict:
-    return {
-        "number": number,
-        "title": title or f"pr {number}",
-        "body": body,
-        "state": state,
-        "head": {"ref": head},
-        "labels": [{"name": name} for name in labels],
-    }
-
-
-def _issue(
-    number: int,
-    *,
-    title: str = "",
-    state: str = "open",
-    labels: tuple[str, ...] = (),
-) -> dict:
-    return {
-        "number": number,
-        "title": title or f"issue {number}",
-        "state": state,
-        "labels": [{"name": name} for name in labels],
-    }
-
-
-def _fetcher(items: list[dict], *, complete: bool = True):
-    def _fetch(*_args, **_kwargs):
-        return items, _pagination(complete=complete, count=len(items))
-
-    return _fetch
-
-
-def _load(
-    issues: list[dict],
-    prs: list[dict],
-    *,
-    complete: bool = True,
-    **kwargs,
-) -> LinkageSnapshot:
-    return load_linkage_snapshot(
-        fetch_prs=_fetcher(prs, complete=complete),
-        fetch_issues=_fetcher(issues, complete=complete),
-        **kwargs,
-    )
-
-
-def _cth(comment_id: int, *, created_at: str, status: str, next_owner: str) -> dict:
-    return {
-        "id": comment_id,
-        "created_at": created_at,
-        "user": {"login": "jcwalker3"},
-        "body": format_cth_body(
-            cth_type="Author Handoff",
-            status=status,
-            next_owner=next_owner,
-            current_blocker="none",
-            decision="implemented",
-            proof="full suite green",
-            next_action="review PR",
-            ready_to_paste_prompt="Review PR #902 now.",
-        ),
-    }
-
-
-class TestLinkageEvidence(unittest.TestCase):
-    """AC1 — every edge records how it was found, and keeps all candidates."""
-
-    def test_closes_keyword_in_body_is_strongest_evidence(self):
-        links = resolve_pr_links(_pr(902, body="Closes #643"))
-        self.assertEqual([link.issue_number for link in links], [643])
-        self.assertEqual(links[0].evidence, (EVIDENCE_CLOSES,))
-        self.assertTrue(links[0].closes)
-
-    def test_closes_keyword_in_title_counts(self):
-        links = resolve_pr_links(_pr(902, title="feat(webui): preview (Closes #643)"))
-        self.assertEqual(links[0].evidence, (EVIDENCE_CLOSES,))
-
-    def test_canonical_branch_marker_links_without_a_keyword(self):
-        links = resolve_pr_links(_pr(902, head="feat/issue-643-request-preview"))
-        self.assertEqual([link.issue_number for link in links], [643])
-        self.assertEqual(links[0].evidence, (EVIDENCE_BRANCH,))
-        self.assertFalse(links[0].closes)
-
-    def test_non_canonical_branch_is_not_treated_as_a_marker(self):
-        self.assertEqual(resolve_pr_links(_pr(902, head="issue-643-preview")), ())
-
-    def test_bare_mention_is_recorded_as_the_weakest_evidence(self):
-        links = resolve_pr_links(_pr(902, body="context in #643"))
-        self.assertEqual(links[0].evidence, (EVIDENCE_REFERENCE,))
-        self.assertFalse(links[0].closes)
-
-    def test_several_evidence_kinds_merge_onto_one_edge(self):
-        links = resolve_pr_links(
-            _pr(902, body="Closes #643 — see #643", head="feat/issue-643-preview")
-        )
-        self.assertEqual(len(links), 1)
-        self.assertEqual(
-            links[0].evidence,
-            (EVIDENCE_CLOSES, EVIDENCE_BRANCH, EVIDENCE_REFERENCE),
-        )
-
-    def test_stronger_evidence_sorts_first(self):
-        links = resolve_pr_links(_pr(902, body="Closes #643, related #700"))
-        self.assertEqual([link.issue_number for link in links], [643, 700])
-
-    def test_self_reference_is_not_linkage(self):
-        links = resolve_pr_links(_pr(902, body="supersedes #902"))
-        self.assertEqual(links, ())
-
-    def test_every_candidate_is_kept_never_collapsed_to_a_guess(self):
-        links = resolve_pr_links(_pr(902, body="Closes #643\nCloses #644"))
-        self.assertEqual([link.issue_number for link in links], [643, 644])
-
-
-class TestLinkageIndex(unittest.TestCase):
-    def test_issue_direction_ignores_mention_only_edges(self):
-        index = resolve_linkage([_pr(902, body="context in #643")])
-        self.assertIsNone(index.issue_prs.get(643))
-        self.assertEqual(index.pr_links[902][0].evidence, (EVIDENCE_REFERENCE,))
-
-    def test_contested_issue_is_reported_when_two_prs_claim_it(self):
-        index = resolve_linkage(
-            [_pr(902, body="Closes #643"), _pr(903, head="feat/issue-643-again")]
-        )
-        self.assertEqual(index.contested_issues(), (643,))
-        self.assertEqual(index.issue_prs[643], (902, 903))
-
-    def test_single_claim_is_not_contested(self):
-        index = resolve_linkage([_pr(902, body="Closes #643")])
-        self.assertEqual(index.contested_issues(), ())
-
-    def test_ambiguous_when_two_issues_tie_at_the_strongest_evidence(self):
-        index = resolve_linkage([_pr(902, body="Closes #643\nCloses #644")])
-        self.assertTrue(index.ambiguous(902))
-
-    def test_weaker_candidate_alongside_a_stronger_one_is_not_ambiguous(self):
-        index = resolve_linkage([_pr(902, body="Closes #643, see #700")])
-        self.assertFalse(index.ambiguous(902))
-        self.assertEqual(index.primary_issue(902).issue_number, 643)
-
-    def test_malformed_pr_row_is_skipped_not_raised_on(self):
-        index = resolve_linkage([{"title": "no number"}, _pr(902, body="Closes #643")])
-        self.assertEqual(sorted(index.pr_links), [902])
-
-
-class TestLinkageSnapshot(unittest.TestCase):
-    """AC1 — linkage is visible per project/repo, in both directions."""
-
-    def test_both_directions_are_populated(self):
-        snapshot = _load([_issue(643)], [_pr(902, body="Closes #643")])
-        self.assertTrue(snapshot.ok)
-        self.assertEqual([node.number for node in snapshot.issues], [643])
-        self.assertEqual(snapshot.issues[0].linked_prs, (902,))
-        self.assertEqual(snapshot.prs[0].links[0].issue_number, 643)
-
-    def test_repo_scope_comes_from_the_registry_project(self):
-        snapshot = _load([], [])
-        self.assertIn("/", snapshot.repo_label)
-        self.assertTrue(snapshot.project_id)
-
-    def test_unknown_project_fails_closed_with_a_reason(self):
-        snapshot = _load([_issue(643)], [], project_id="no-such-project")
-        self.assertFalse(snapshot.ok)
-        self.assertIn("not found in registry", snapshot.fetch_error)
-        self.assertEqual(snapshot.issues, ())
-
-    def test_orphan_pr_is_identifiable(self):
-        snapshot = _load([], [_pr(902), _pr(903, body="Closes #643")])
-        self.assertEqual([node.number for node in snapshot.orphan_prs], [902])
-
-    def test_state_scope_defaults_to_open_and_is_reported(self):
-        self.assertEqual(_load([], []).state_scope, "open")
-        self.assertEqual(_load([], [], state="all").state_scope, "all")
-
-    def test_unsupported_state_falls_back_to_open(self):
-        self.assertEqual(_load([], [], state="../etc").state_scope, "open")
-
-    def test_state_is_passed_through_to_the_fetchers(self):
-        seen: list[str] = []
-
-        def _fetch(*_args, **kwargs):
-            seen.append(kwargs.get("state", ""))
-            return [], _pagination()
-
-        load_linkage_snapshot(state="all", fetch_prs=_fetch, fetch_issues=_fetch)
-        self.assertEqual(seen, ["all", "all"])
-
-
-class TestPartialInventoryIsNotAnAbsenceClaim(unittest.TestCase):
-    """An empty edge list from a partial read must never read as 'no link'."""
-
-    def test_incomplete_pagination_marks_links_non_authoritative(self):
-        snapshot = _load([_issue(643)], [], complete=False)
-        self.assertFalse(snapshot.inventory_complete)
-        self.assertFalse(snapshot.issues[0].links_authoritative)
-
-    def test_complete_pagination_marks_links_authoritative(self):
-        snapshot = _load([_issue(643)], [], complete=True)
-        self.assertTrue(snapshot.inventory_complete)
-        self.assertTrue(snapshot.issues[0].links_authoritative)
-
-    def test_partial_window_renders_a_qualified_empty_cell(self):
-        html = render_linkage_page(_load([_issue(643)], [], complete=False))
-        self.assertIn("none found (partial inventory)", html)
-
-    def test_complete_window_renders_a_plain_none(self):
-        html = render_linkage_page(_load([_issue(643)], [], complete=True))
-        self.assertNotIn("partial inventory", html)
-        self.assertIn(">none<", html)
-
-    def test_missing_credentials_fail_closed_without_a_table(self):
-        with mock.patch(
-            "webui.linkage_loader._offline_test_mode", return_value=False
-        ), mock.patch("webui.linkage_loader.get_auth_header", return_value=""):
-            snapshot = load_linkage_snapshot()
-        self.assertFalse(snapshot.ok)
-        self.assertIn("credentials unavailable", snapshot.fetch_error)
-        html = render_linkage_page(snapshot)
-        self.assertIn("Linkage unavailable", html)
-        self.assertNotIn("Issues → pull requests", html)
-
-    def test_fetch_failure_is_reported_not_raised(self):
-        def _boom(*_args, **_kwargs):
-            raise RuntimeError("gitea 502")
-
-        snapshot = load_linkage_snapshot(fetch_prs=_boom, fetch_issues=_boom)
-        self.assertFalse(snapshot.ok)
-        self.assertIn("Gitea fetch failed", snapshot.fetch_error)
-
-
-class TestHandoffSummary(unittest.TestCase):
-    """AC2 — the latest canonical handoff is summarized for a focused thread."""
-
-    def test_latest_cth_wins(self):
-        summary = summarize_handoff([
-            _cth(1, created_at="2026-07-24T10:00:00Z", status="in progress",
-                 next_owner="author"),
-            _cth(2, created_at="2026-07-25T10:00:00Z", status="PR-open",
-                 next_owner="reviewer"),
-        ])
-        self.assertEqual(summary.comment_id, 2)
-        self.assertEqual(summary.status, "PR-open")
-        self.assertEqual(summary.next_owner, "reviewer")
-        self.assertTrue(summary.cth_type_known)
-
-    def test_thread_without_a_cth_summarizes_to_none(self):
-        self.assertIsNone(summarize_handoff([{"id": 1, "body": "ordinary comment"}]))
-
-    def test_unknown_heading_is_reported_not_republished(self):
-        summary = summarize_handoff([
-            {
-                "id": 5,
-                "created_at": "2026-07-25T10:00:00Z",
-                "user": {"login": "someone"},
-                "body": "<!-- cth:v1 -->\n## CTH: Totally Made Up\n\nStatus: odd\n",
-            }
-        ])
-        self.assertFalse(summary.cth_type_known)
-        self.assertEqual(summary.cth_type, "unrecognized")
-        self.assertNotIn("Totally Made Up", json.dumps(summary.to_dict()))
-
-    def test_focused_pr_loads_its_handoff(self):
-        snapshot = _load(
-            [_issue(643)],
-            [_pr(902, body="Closes #643")],
-            pr=902,
-            comment_source=lambda kind, number: [
-                _cth(2, created_at="2026-07-25T10:00:00Z", status="PR-open",
-                     next_owner="reviewer")
-            ],
-        )
-        self.assertEqual(snapshot.handoff_status.state, HANDOFF_LOADED)
-        self.assertEqual(snapshot.focus, ("pr", 902))
-        self.assertEqual(snapshot.prs[0].handoff.status, "PR-open")
-
-    def test_unfocused_rows_report_not_loaded_never_none(self):
-        snapshot = _load(
-            [_issue(643)],
-            [_pr(902, body="Closes #643"), _pr(903)],
-            pr=902,
-            comment_source=lambda kind, number: [],
-        )
-        other = next(node for node in snapshot.prs if node.number == 903)
-        self.assertIsNone(other.handoff)
-        self.assertEqual(other.handoff_status.state, HANDOFF_NOT_LOADED)
-        self.assertIn("not loaded", render_linkage_page(snapshot))
-
-    def test_no_focus_means_no_thread_is_claimed_handoff_free(self):
-        snapshot = _load([_issue(643)], [])
-        self.assertEqual(snapshot.handoff_status.state, HANDOFF_NOT_LOADED)
-        self.assertIn("thread-scoped", snapshot.handoff_status.reason)
-
-    def test_comment_source_failure_degrades_only_the_handoff(self):
-        def _boom(_kind, _number):
-            raise RuntimeError("comments 500")
-
-        snapshot = _load(
-            [_issue(643)], [_pr(902, body="Closes #643")], pr=902, comment_source=_boom
-        )
-        self.assertTrue(snapshot.ok)
-        self.assertEqual(snapshot.handoff_status.state, HANDOFF_UNAVAILABLE)
-        self.assertEqual(snapshot.issues[0].linked_prs, (902,))
-        self.assertIn("unavailable", render_linkage_page(snapshot))
-
-    def test_loaded_thread_with_no_cth_says_so_explicitly(self):
-        snapshot = _load(
-            [_issue(643)],
-            [_pr(902, body="Closes #643")],
-            pr=902,
-            comment_source=lambda kind, number: [{"id": 1, "body": "hi"}],
-        )
-        self.assertIn(
-            "no Canonical Thread Handoff comment found", render_linkage_page(snapshot)
-        )
-
-
-class TestDeepLinks(unittest.TestCase):
-    """AC3 — an external Gitea link is emitted only when permitted."""
-
-    def test_deep_links_are_withheld_by_default(self):
-        with mock.patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": ""}):
-            snapshot = _load([_issue(643)], [])
-            html = render_linkage_page(snapshot)
-        self.assertFalse(snapshot.deep_links_enabled)
-        self.assertIsNone(snapshot.issues[0].deep_link)
-        self.assertIn("Gitea deep links are withheld", html)
-
-    def test_reveal_opt_in_emits_the_link(self):
-        with mock.patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}):
-            snapshot = _load([_issue(643)], [_pr(902, body="Closes #643")])
-            html = render_linkage_page(snapshot)
-        self.assertTrue(snapshot.deep_links_enabled)
-        self.assertIn("/issues/643", snapshot.issues[0].deep_link)
-        self.assertIn("/pulls/902", snapshot.prs[0].deep_link)
-        self.assertIn(f'href="{snapshot.issues[0].deep_link}"', html)
-
-
-class TestRedactionBoundary(unittest.TestCase):
-    def test_secret_shaped_title_is_redacted_before_display(self):
-        snapshot = _load(
-            [_issue(643, title="token=ghp_thisisnotarealsecretvalue0001")], []
-        )
-        payload = json.dumps(snapshot_to_dict(snapshot))
-        self.assertNotIn("ghp_thisisnotarealsecretvalue0001", payload)
-        self.assertNotIn(
-            "ghp_thisisnotarealsecretvalue0001", render_linkage_page(snapshot)
-        )
-
-    def test_handoff_fields_are_redacted(self):
-        comment = _cth(
-            2, created_at="2026-07-25T10:00:00Z", status="ok", next_owner="reviewer"
-        )
-        comment["body"] += "\nDecision: password=hunter2hunter2\n"
-        snapshot = _load(
-            [_issue(643)],
-            [_pr(902, body="Closes #643")],
-            pr=902,
-            comment_source=lambda kind, number: [comment],
-        )
-        self.assertNotIn("hunter2hunter2", json.dumps(snapshot_to_dict(snapshot)))
-        self.assertNotIn("hunter2hunter2", render_linkage_page(snapshot))
-
-    def test_html_escapes_markup_in_a_title(self):
-        snapshot = _load([_issue(643, title="<script>alert(1)</script>")], [])
-        html = render_linkage_page(snapshot)
-        self.assertNotIn("<script>alert(1)</script>", html)
-        self.assertIn("&lt;script&gt;", html)
-
-
-class TestLinkageRoutes(unittest.TestCase):
-    def setUp(self):
-        self.snapshot = _load(
-            [_issue(643, labels=("status:ready",))],
-            [_pr(902, body="Closes #643", labels=("status:pr-open",))],
-        )
-        self.client = TestClient(create_app())
-
-    def test_page_renders_both_tables(self):
-        with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
-            response = self.client.get("/gitea")
-        self.assertEqual(response.status_code, 200)
-        self.assertIn("Issues → pull requests", response.text)
-        self.assertIn("Pull requests → issues", response.text)
-        self.assertIn("#643", response.text)
-
-    def test_api_exports_the_same_model(self):
-        with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
-            response = self.client.get("/api/v1/gitea/linkage")
-        self.assertEqual(response.status_code, 200)
-        payload = response.json()
-        self.assertTrue(payload["ok"])
-        self.assertEqual(payload["issues"][0]["linked_prs"], [902])
-        self.assertEqual(payload["prs"][0]["links"][0]["issue_number"], 643)
-        self.assertEqual(payload["schema_version"], 1)
-
-    def test_api_declares_the_evidence_vocabulary(self):
-        with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
-            payload = self.client.get("/api/v1/gitea/linkage").json()
-        names = {entry["name"] for entry in payload["evidence_kinds"]}
-        self.assertEqual(names, {EVIDENCE_CLOSES, EVIDENCE_BRANCH, EVIDENCE_REFERENCE})
-
-    def test_api_fails_closed_with_a_non_200_when_the_read_failed(self):
-        failed = _load([], [], project_id="no-such-project")
-        with mock.patch("webui.app.load_linkage_snapshot", return_value=failed):
-            response = self.client.get("/api/v1/gitea/linkage")
-        self.assertEqual(response.status_code, 502)
-        self.assertFalse(response.json()["ok"])
-
-    def test_page_still_renders_when_the_read_failed(self):
-        failed = _load([], [], project_id="no-such-project")
-        with mock.patch("webui.app.load_linkage_snapshot", return_value=failed):
-            response = self.client.get("/gitea")
-        self.assertEqual(response.status_code, 200)
-        self.assertIn("Linkage unavailable", response.text)
-
-    def test_query_parameters_reach_the_loader(self):
-        with mock.patch(
-            "webui.app.load_linkage_snapshot", return_value=self.snapshot
-        ) as loader:
-            self.client.get("/gitea?project=gitea-tools&state=all&pr=902")
-        loader.assert_called_once()
-        args, kwargs = loader.call_args
-        self.assertEqual(args[0], "gitea-tools")
-        self.assertEqual(kwargs["state"], "all")
-        self.assertEqual(kwargs["pr"], 902)
-        self.assertIsNone(kwargs["issue"])
-
-    def test_surface_stays_read_only(self):
-        for path in ("/gitea", "/api/v1/gitea/linkage"):
-            with self.subTest(path=path):
-                self.assertEqual(self.client.post(path).status_code, 405)
-
-    def test_nav_exposes_the_linkage_page_as_live(self):
-        self.assertIn("/gitea", nav_hrefs())
-        home = self.client.get("/").text
-        self.assertIn('href="/gitea"', home)
-        self.assertIn(">Gitea<", home)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_webui_providers_insights.py

@@ -1,404 +0,0 @@
-"""Tests for AI-provider connections and evidence-backed insights (#650).
-
-Covers acceptance criteria:
-
-1. Provider connection status is redacted and accurate (declared registry only).
-2. At least three insight types with evidence citations.
-3. Insights never claim actions completed without proof.
-4. Evidence requirement is enforced (no evidence-less insights).
-5. Interpretation limits appear in docs-facing payloads.
-"""
-
-from __future__ import annotations
-
-import json
-import sys
-import unittest
-from dataclasses import dataclass
-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 webui.app import create_app
-from webui.insights_loader import (
-    CONFIDENCE_HIGH,
-    CONNECTION_DECLARED_AVAILABLE,
-    CONNECTION_DECLARED_UNAVAILABLE,
-    INSIGHT_BLOCKED_QUEUE,
-    INSIGHT_CONTROLLER_ATTENTION,
-    INSIGHT_PROVIDER_WITHOUT_WORKERS,
-    INSIGHT_STALE_RUNTIME,
-    ProviderConnection,
-    build_provider_connection,
-    generate_insights,
-    insight_blocked_queue,
-    insight_controller_attention,
-    insight_providers_without_workers,
-    insight_stale_runtime,
-    load_insights_snapshot,
-    load_provider_snapshot,
-    snapshot_insights_to_dict,
-    snapshot_providers_to_dict,
-)
-from webui.insights_views import render_insights_page, render_providers_page
-from webui.nav import STUB_PAGES, nav_hrefs
-from webui.worker_registry import ProviderRecord, WorkerRecord, ScheduleSpec, SchedulerSpec
-
-
-def _provider(
-    provider_id: str = "claude",
-    *,
-    available: bool = True,
-    models: tuple[str, ...] = ("claude-opus-4-8",),
-    notes: str = "",
-) -> ProviderRecord:
-    return ProviderRecord(
-        id=provider_id,
-        display_name=provider_id.title(),
-        vendor="TestVendor",
-        executable=provider_id,
-        available=available,
-        models=models,
-        notes=notes,
-    )
-
-
-def _worker(
-    worker_id: str = "claude-author",
-    *,
-    provider: str = "claude",
-    enabled: bool = True,
-) -> WorkerRecord:
-    return WorkerRecord(
-        id=worker_id,
-        display_name=worker_id,
-        provider=provider,
-        model="m1",
-        project="gitea-tools",
-        role="author",
-        namespace="gitea-author",
-        profile="prgs-author",
-        workflow="skills/llm-project-workflow/workflows/work-issue.md",
-        schedule=ScheduleSpec(kind="manual", seconds=None, expression=None),
-        timeout_seconds=3600,
-        enabled=enabled,
-        scheduler=SchedulerSpec(kind="manual", label=None),
-        notes="",
-    )
-
-
-@dataclass(frozen=True)
-class _TrafficItem:
-    kind: str
-    number: int
-    title: str = ""
-    traffic_state: str = "blocked"
-    expected_role: str = "author"
-    safe_for_roles: tuple[str, ...] = ()
-    badges: tuple[str, ...] = ()
-    block_reason: str | None = "dependency"
-
-
-@dataclass(frozen=True)
-class _Traffic:
-    blocked: tuple = ()
-    needs_controller: tuple = ()
-    inventory_complete: bool = True
-    fetch_error: str | None = None
-
-
-@dataclass(frozen=True)
-class _Stale:
-    daemon_head: str | None
-    checkout_head: str | None
-    remote_head: str | None
-    stale: bool
-    determinable: bool
-    mutation_safe: bool
-
-
-@dataclass(frozen=True)
-class _Health:
-    stale_runtime: _Stale | None
-
-
-class TestProviderConnections(unittest.TestCase):
-    def test_available_provider_status(self):
-        conn = build_provider_connection(_provider(available=True), (_worker(),))
-        self.assertEqual(conn.connection_status, CONNECTION_DECLARED_AVAILABLE)
-        self.assertTrue(conn.available_declared)
-        self.assertEqual(conn.worker_count, 1)
-        self.assertEqual(conn.enabled_worker_count, 1)
-        self.assertFalse(conn.to_dict()["secrets_exposed"])
-
-    def test_unavailable_provider_status(self):
-        conn = build_provider_connection(_provider(available=False), ())
-        self.assertEqual(conn.connection_status, CONNECTION_DECLARED_UNAVAILABLE)
-        self.assertEqual(conn.worker_count, 0)
-
-    def test_notes_are_redacted(self):
-        conn = build_provider_connection(
-            _provider(notes="token=ghp_thisisnotarealsecretvalue0001"),
-            (),
-        )
-        self.assertNotIn("ghp_thisisnotarealsecretvalue0001", conn.notes)
-        self.assertNotIn(
-            "ghp_thisisnotarealsecretvalue0001",
-            json.dumps(conn.to_dict()),
-        )
-
-    def test_load_provider_snapshot_from_injected_registry(self):
-        from webui.worker_registry import WorkerRegistry
-        from pathlib import Path
-
-        registry = WorkerRegistry(
-            version=1,
-            revision=3,
-            updated_at="2026-07-25T00:00:00Z",
-            providers=(_provider("claude"), _provider("grok", available=False)),
-            workers=(_worker(),),
-            source_path=Path("/tmp/workers.registry.json"),
-        )
-        snapshot = load_provider_snapshot(registry=registry)
-        self.assertTrue(snapshot.ok)
-        self.assertEqual(snapshot.registry_revision, 3)
-        ids = {p.provider_id for p in snapshot.providers}
-        self.assertEqual(ids, {"claude", "grok"})
-
-    def test_registry_failure_is_fail_closed(self):
-        def _boom():
-            raise RuntimeError("disk gone")
-
-        snapshot = load_provider_snapshot(registry_loader=_boom)
-        self.assertFalse(snapshot.ok)
-        self.assertIn("unavailable", snapshot.fetch_error or "")
-        self.assertEqual(snapshot.providers, ())
-
-
-class TestInsightGenerators(unittest.TestCase):
-    def test_blocked_queue_requires_evidence(self):
-        traffic = _Traffic(
-            blocked=(
-                _TrafficItem(kind="issue", number=647, block_reason="depends #646"),
-                _TrafficItem(kind="pr", number=902, block_reason="conflict"),
-            )
-        )
-        insight = insight_blocked_queue(traffic)
-        self.assertIsNotNone(insight)
-        self.assertEqual(insight.kind, INSIGHT_BLOCKED_QUEUE)
-        self.assertGreaterEqual(len(insight.evidence), 2)
-        self.assertTrue(insight.advisory_only)
-        self.assertFalse(insight.claims_action_completed)
-        refs = {e.ref for e in insight.evidence}
-        self.assertIn("#647", refs)
-        self.assertIn("#902", refs)
-
-    def test_empty_blocked_queue_yields_no_insight(self):
-        self.assertIsNone(insight_blocked_queue(_Traffic()))
-
-    def test_controller_attention_insight(self):
-        traffic = _Traffic(
-            needs_controller=(_TrafficItem(kind="issue", number=100, traffic_state="needs_controller"),)
-        )
-        insight = insight_controller_attention(traffic)
-        self.assertEqual(insight.kind, INSIGHT_CONTROLLER_ATTENTION)
-        self.assertEqual(insight.evidence[0].ref, "#100")
-
-    def test_stale_runtime_insight(self):
-        health = _Health(
-            stale_runtime=_Stale(
-                daemon_head="aaa",
-                checkout_head="bbb",
-                remote_head="ccc",
-                stale=True,
-                determinable=True,
-                mutation_safe=False,
-            )
-        )
-        insight = insight_stale_runtime(health)
-        self.assertEqual(insight.kind, INSIGHT_STALE_RUNTIME)
-        self.assertIn("stale", insight.evidence[0].detail)
-        self.assertFalse(insight.claims_action_completed)
-
-    def test_mutation_safe_runtime_yields_no_insight(self):
-        health = _Health(
-            stale_runtime=_Stale(
-                daemon_head="aaa",
-                checkout_head="aaa",
-                remote_head="aaa",
-                stale=False,
-                determinable=True,
-                mutation_safe=True,
-            )
-        )
-        self.assertIsNone(insight_stale_runtime(health))
-
-    def test_provider_without_workers(self):
-        providers = (
-            build_provider_connection(_provider("claude"), (_worker(),)),
-            build_provider_connection(_provider("grok"), ()),
-        )
-        insight = insight_providers_without_workers(providers)
-        self.assertEqual(insight.kind, INSIGHT_PROVIDER_WITHOUT_WORKERS)
-        self.assertEqual(insight.evidence[0].ref, "grok")
-
-    def test_generate_insights_composes_three_kinds(self):
-        from webui.worker_registry import WorkerRegistry
-        from pathlib import Path
-
-        registry = WorkerRegistry(
-            version=1,
-            revision=1,
-            updated_at="2026-07-25T00:00:00Z",
-            providers=(_provider("lonely"),),
-            workers=(),
-            source_path=Path("/tmp/w.json"),
-        )
-        provider_snapshot = load_provider_snapshot(registry=registry)
-        traffic = _Traffic(
-            blocked=(_TrafficItem(kind="issue", number=1),),
-            needs_controller=(_TrafficItem(kind="issue", number=2),),
-        )
-        health = _Health(
-            stale_runtime=_Stale("a", "b", "c", True, True, False)
-        )
-        insights, used, unavailable = generate_insights(
-            traffic=traffic,
-            health=health,
-            provider_snapshot=provider_snapshot,
-            analytics=None,
-        )
-        kinds = {i.kind for i in insights}
-        self.assertIn(INSIGHT_BLOCKED_QUEUE, kinds)
-        self.assertIn(INSIGHT_CONTROLLER_ATTENTION, kinds)
-        self.assertIn(INSIGHT_STALE_RUNTIME, kinds)
-        self.assertIn(INSIGHT_PROVIDER_WITHOUT_WORKERS, kinds)
-        self.assertGreaterEqual(len(kinds), 3)
-        self.assertIn("traffic", used)
-        self.assertIn("system_health", used)
-        self.assertIn("providers", used)
-        self.assertTrue(any(u["source"] == "analytics" for u in unavailable))
-        for insight in insights:
-            self.assertTrue(insight.advisory_only)
-            self.assertFalse(insight.claims_action_completed)
-            self.assertGreaterEqual(len(insight.evidence), 1)
-
-    def test_missing_source_is_reported_not_as_healthy_empty(self):
-        insights, used, unavailable = generate_insights(
-            traffic=None,
-            health=None,
-            provider_snapshot=None,
-            analytics=None,
-        )
-        self.assertEqual(insights, ())
-        self.assertEqual(used, ())
-        self.assertEqual(len(unavailable), 4)
-
-
-class TestViewsAndRoutes(unittest.TestCase):
-    def test_providers_page_renders_connections(self):
-        from webui.worker_registry import WorkerRegistry
-        from pathlib import Path
-
-        registry = WorkerRegistry(
-            version=1,
-            revision=1,
-            updated_at="2026-07-25T00:00:00Z",
-            providers=(_provider("claude"),),
-            workers=(_worker(),),
-            source_path=Path("/tmp/w.json"),
-        )
-        snapshot = load_provider_snapshot(registry=registry)
-        html = render_providers_page(snapshot)
-        self.assertIn("AI provider connections", html)
-        self.assertIn("claude", html)
-        self.assertIn("declared_available", html)
-        self.assertIn("Interpretation limits", html)
-        self.assertNotIn("api_key", html.lower())
-
-    def test_failed_provider_snapshot_renders_no_table(self):
-        snapshot = load_provider_snapshot(registry_loader=lambda: (_ for _ in ()).throw(RuntimeError("x")))
-        html = render_providers_page(snapshot)
-        self.assertIn("unavailable", html.lower())
-        self.assertNotIn("<tbody><tr><td><code>", html)
-
-    def test_insights_page_lists_evidence(self):
-        traffic = _Traffic(blocked=(_TrafficItem(kind="issue", number=42),))
-        snapshot = load_insights_snapshot(
-            traffic=traffic,
-            health=_Health(None),
-            provider_snapshot=load_provider_snapshot(
-                registry_loader=lambda: (_ for _ in ()).throw(RuntimeError("skip"))
-            ),
-            analytics=None,
-            load_live=False,
-        )
-        html = render_insights_page(snapshot)
-        self.assertIn("#42", html)
-        self.assertIn("advisory only", html.lower())
-        self.assertIn("Evidence", html)
-
-    def test_nav_exposes_live_insights_and_providers(self):
-        self.assertIn("/insights", nav_hrefs())
-        self.assertIn("/providers", nav_hrefs())
-        self.assertNotIn("/insights", STUB_PAGES)
-
-    def test_routes_are_read_only_and_export_json(self):
-        client = TestClient(create_app())
-        # Use live registry from package data — should be ok.
-        with mock.patch(
-            "webui.app.load_provider_snapshot",
-            return_value=load_provider_snapshot(
-                registry=__import__(
-                    "webui.worker_registry", fromlist=["load_registry"]
-                ).load_registry()
-            ),
-        ):
-            response = client.get("/providers")
-            self.assertEqual(response.status_code, 200)
-            self.assertIn("provider", response.text.lower())
-            api = client.get("/api/v1/providers")
-            self.assertEqual(api.status_code, 200)
-            payload = api.json()
-            self.assertTrue(payload["ok"])
-            self.assertIn("interpretation_limits", payload)
-            self.assertTrue(all(not p.get("secrets_exposed") for p in payload["providers"]))
-
-        with mock.patch(
-            "webui.app.load_insights_snapshot",
-            return_value=load_insights_snapshot(
-                traffic=_Traffic(blocked=(_TrafficItem(kind="issue", number=7),)),
-                health=_Health(None),
-                provider_snapshot=load_provider_snapshot(
-                    registry_loader=lambda: (_ for _ in ()).throw(RuntimeError("x"))
-                ),
-                analytics=None,
-                load_live=False,
-            ),
-        ):
-            page = client.get("/insights")
-            self.assertEqual(page.status_code, 200)
-            self.assertIn("#7", page.text)
-            api = client.get("/api/v1/insights")
-            self.assertEqual(api.status_code, 200)
-            body = api.json()
-            self.assertTrue(body["ok"])
-            self.assertTrue(all(i["advisory_only"] for i in body["insights"]))
-            self.assertTrue(all(not i["claims_action_completed"] for i in body["insights"]))
-            self.assertTrue(all(i["evidence"] for i in body["insights"]))
-
-        for path in ("/providers", "/api/v1/providers", "/insights", "/api/v1/insights"):
-            with self.subTest(path=path):
-                self.assertEqual(client.post(path).status_code, 405)
-
-    def test_home_nav_links_providers_and_insights(self):
-        home = TestClient(create_app()).get("/").text
-        self.assertIn('href="/providers"', home)
-        self.assertIn('href="/insights"', home)
-
-
-if __name__ == "__main__":
-    unittest.main()

webui/app.py

@@ -53,18 +53,6 @@ from webui.session_loader import (
     snapshot_to_dict as session_view_snapshot_to_dict,
 )
 from webui.session_views import render_sessions_page
-from webui.insights_loader import (
-    load_insights_snapshot,
-    load_provider_snapshot,
-    snapshot_insights_to_dict,
-    snapshot_providers_to_dict,
-)
-from webui.insights_views import render_insights_page, render_providers_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,
@@ -84,6 +72,8 @@ from webui.system_health import (
     snapshot_to_dict as system_health_to_dict,
 )
 from webui.system_health_views import render_system_health_page
+from webui import request_service
+from webui.request_views import render_requests_page
 
 _READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
 _AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
@@ -354,67 +344,6 @@ async def api_sessions(_request: Request) -> JSONResponse:
     return JSONResponse(session_view_snapshot_to_dict(load_session_view_snapshot()))
 
 
-async def providers(_request: Request) -> HTMLResponse:
-    """AI-provider connection status (#650) — declared registry only, no secrets."""
-    return HTMLResponse(render_providers_page(load_provider_snapshot()))
-
-
-async def api_v1_providers(_request: Request) -> JSONResponse:
-    """JSON export of declared AI-provider connections (#650)."""
-    snapshot = load_provider_snapshot()
-    return JSONResponse(
-        snapshot_providers_to_dict(snapshot),
-        status_code=200 if snapshot.ok else 502,
-    )
-
-
-async def insights(_request: Request) -> HTMLResponse:
-    """Evidence-backed operational insights (#650) — advisory only."""
-    return HTMLResponse(render_insights_page(load_insights_snapshot()))
-
-
-async def api_v1_insights(_request: Request) -> JSONResponse:
-    """JSON export of evidence-backed insights (#650)."""
-    snapshot = load_insights_snapshot()
-    return JSONResponse(
-        snapshot_insights_to_dict(snapshot),
-        status_code=200 if snapshot.ok else 502,
-    )
-
-
-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
@@ -812,6 +741,109 @@ async def api_v1_analytics_ingest(request: Request) -> JSONResponse:
     )
 
 
+def _default_request_scope() -> dict[str, str]:
+    """Resolve remote/org/repo from the project registry for request forms.
+
+    Returns an empty mapping when the registry cannot be read, which makes
+    ``parse_request`` reject a request that did not name its own scope rather
+    than letting it default to some other repository.
+    """
+    from webui.queue_loader import _host_from_url  # host normalisation helper
+
+    registry, error = _load_project_registry()
+    if error is not None or not registry.projects:
+        return {}
+    project = registry.projects[0]
+    host = _host_from_url(project.remote_host)
+    return {
+        "remote": _derive_remote(host),
+        "org": project.gitea_owner or "",
+        "repo": project.repo_name or "",
+    }
+
+
+async def _request_payload(request: Request) -> dict[str, object]:
+    """Read a request body as JSON or form-encoded. Never raises."""
+    content_type = (request.headers.get("content-type") or "").lower()
+    if "application/json" in content_type:
+        try:
+            body = await request.json()
+        except Exception:
+            return {}
+        return dict(body) if isinstance(body, dict) else {}
+    try:
+        form = await request.form()
+    except Exception:
+        return {}
+    return {key: form[key] for key in form}
+
+
+async def requests_page(request: Request) -> HTMLResponse:
+    """Operator request form and intent preview (#643).
+
+    POST here only ever *previews*. Initiation is a separate confirmed call to
+    ``/api/v1/requests/apply`` so that submitting this form cannot reserve
+    work as a side effect.
+    """
+    submitted: dict[str, object] = {}
+    preview = None
+    error = None
+    if request.method == "POST":
+        submitted = await _request_payload(request)
+        work_request, error = request_service.parse_request(
+            submitted, default_scope=_default_request_scope()
+        )
+        if work_request is not None:
+            preview = request_service.preview_request(
+                work_request,
+                principal=resolve_principal(headers=dict(request.headers)),
+            )
+    return HTMLResponse(
+        render_requests_page(
+            preview=preview, error=error, submitted=submitted
+        )
+    )
+
+
+async def api_v1_request_preview(request: Request) -> JSONResponse:
+    """Dry-run authorization and intent preview for a work request (#643)."""
+    payload = await _request_payload(request)
+    work_request, error = request_service.parse_request(
+        payload, default_scope=_default_request_scope()
+    )
+    if work_request is None:
+        return JSONResponse(error.to_dict(), status_code=400)
+    preview = request_service.preview_request(
+        work_request,
+        principal=resolve_principal(headers=dict(request.headers)),
+    )
+    return JSONResponse(
+        preview.to_dict(), status_code=200 if preview.authorized else 403
+    )
+
+
+async def api_v1_request_apply(request: Request) -> JSONResponse:
+    """Initiate a previewed work request through the allocator (#643).
+
+    Fail-closed at every step: unauthorized, unconfirmed, not-next-safe, and
+    already-claimed all return without attempting an assignment.
+    """
+    payload = await _request_payload(request)
+    work_request, error = request_service.parse_request(
+        payload, default_scope=_default_request_scope()
+    )
+    if work_request is None:
+        return JSONResponse(error.to_dict(), status_code=400)
+    confirm = _truthy_flag(str(payload.get("confirm") or ""))
+    result = request_service.apply_request(
+        work_request,
+        principal=resolve_principal(headers=dict(request.headers)),
+        confirm=confirm,
+    )
+    status = int(result.pop("status_code", 403))
+    return JSONResponse(result, status_code=status)
+
+
 async def method_not_allowed(request: Request, _exc: Exception) -> Response:
     path = request.url.path
     if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
@@ -858,12 +890,6 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
             Route("/api/sessions", api_sessions, methods=["GET"]),
             Route("/api/v1/sessions", api_sessions, methods=["GET"]),
             Route("/api/v1/timeline", api_v1_timeline, methods=["GET"]),
-            Route("/providers", providers, methods=["GET"]),
-            Route("/api/v1/providers", api_v1_providers, methods=["GET"]),
-            Route("/insights", insights, methods=["GET"]),
-            Route("/api/v1/insights", api_v1_insights, 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"]),
@@ -885,6 +911,17 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
                 api_action_attempt,
                 methods=["POST"],
             ),
+            Route("/requests", requests_page, methods=["GET", "POST"]),
+            Route(
+                "/api/v1/requests/preview",
+                api_v1_request_preview,
+                methods=["POST"],
+            ),
+            Route(
+                "/api/v1/requests/apply",
+                api_v1_request_apply,
+                methods=["POST"],
+            ),
             Route("/api/leases", api_leases, methods=["GET"]),
             Route("/api/v1/inventory", api_inventory, methods=["GET"]),
             Route(

webui/console_authz.py

@@ -115,6 +115,12 @@ class ConsoleAction:
     break_glass: bool
     phase: int
     summary: str
+    # Opt-in switch for an action whose execution path is genuinely wired
+    # ahead of its phase becoming globally active (#643). Naming a variable
+    # here enables nothing on its own: the variable must also be set in the
+    # environment. An action that leaves this ``None`` can only execute once
+    # ACTIVE_PHASE reaches its phase, exactly as before.
+    execution_env_flag: str | None = None
 
     @property
     def mcp_permission(self) -> str:
@@ -277,6 +283,27 @@ _ACTION_SPECS: tuple[ConsoleAction, ...] = (
         phase=2,
         summary="Restart one MCP namespace via the host supervisor.",
     ),
+    # #643: submit a work request — desired role, issue/PR, intent — and let
+    # the allocator reserve it. This is the one Phase 2 action whose execution
+    # path is actually implemented (``webui.request_service``), so it carries
+    # the opt-in flag; it stays denied until an operator sets that variable.
+    # Authority is operator-class because the outcome is a claim, not a Gitea
+    # verdict: initiating reviewer or merger *work* does not grant the right
+    # to approve or merge, which stays with the MCP role profile.
+    ConsoleAction(
+        action_id="initiate_workflow",
+        task_key="allocate_next_work",
+        action_class=CLASS_WRITE,
+        minimum_role=OPERATOR,
+        requires_confirmation=True,
+        dual_control=False,
+        break_glass=False,
+        phase=2,
+        summary=(
+            "Preview and initiate allocator-owned workflow work for a role."
+        ),
+        execution_env_flag="WEBUI_REQUESTS_EXECUTION",
+    ),
 )
 
 ACTIONS: dict[str, ConsoleAction] = {a.action_id: a for a in _ACTION_SPECS}
@@ -430,6 +457,33 @@ ALLOW_PREVIEW = "allowed_preview_only"
 # gated on this model landing; nothing here enables it.
 ACTIVE_PHASE = 1
 
+_TRUTHY = frozenset({"1", "true", "yes", "on"})
+
+
+def execution_wired(
+    action: ConsoleAction | None, env: dict[str, str] | None = None
+) -> bool:
+    """Whether *action* has a live execution path right now.
+
+    Two ways to be wired, and only two. The action's phase is active, or the
+    action declares an opt-in environment variable *and* that variable is set.
+    Everything else — including every action that never declares a flag — is
+    unwired, so the default across the registry stays deny.
+
+    Bumping ``ACTIVE_PHASE`` would enable execution for every action of that
+    phase at once. The per-action flag exists so a single implemented action
+    can go live without dragging its unimplemented phase-mates with it.
+    """
+    if action is None:
+        return False
+    if action.phase <= ACTIVE_PHASE:
+        return True
+    flag = (action.execution_env_flag or "").strip()
+    if not flag:
+        return False
+    source = env if env is not None else os.environ
+    return (source.get(flag) or "").strip().lower() in _TRUTHY
+
 
 @dataclass(frozen=True)
 class AuthorizationDecision:
@@ -469,16 +523,19 @@ def authorize(
     principal: Principal | None = None,
     *,
     for_execution: bool = False,
+    env: dict[str, str] | None = None,
 ) -> AuthorizationDecision:
     """Decide whether *principal* may invoke *action_id*. Deny by default.
 
     ``for_execution`` distinguishes a read-only preview from a real invocation.
-    Even an allowed decision reports ``execution_enabled=False`` while the
+    ``execution_enabled`` reports whether the action has a live execution path
-    console is in Phase 1, so no caller can read an allow as permission to
+    at all (:func:`execution_wired`) — for every action without an explicit
-    mutate.
+    opt-in flag that stays ``False`` while the console is in Phase 1, so no
+    caller can read an allow as permission to mutate.
     """
     who = principal if principal is not None else ANONYMOUS
     action = get_action(action_id)
+    wired = execution_wired(action, env)
 
     if action is None:
         return AuthorizationDecision(
@@ -497,7 +554,7 @@ def authorize(
         "requires_confirmation": action.requires_confirmation,
         "dual_control": action.dual_control,
         "break_glass": action.break_glass,
-        "execution_enabled": False,
+        "execution_enabled": wired,
     }
 
     if not who.authenticated:
@@ -530,13 +587,19 @@ def authorize(
             **base,
         )
 
-    if for_execution and action.phase > ACTIVE_PHASE:
+    if for_execution and not wired:
         return AuthorizationDecision(
             allowed=False,
             reason_code=DENY_PHASE_NOT_ACTIVE,
             detail=(
                 f"Action {action_id!r} belongs to phase {action.phase}; the "
-                f"console is in phase {ACTIVE_PHASE}. Execution is not wired."
+                f"console is in phase {ACTIVE_PHASE}"
+                + (
+                    f" and {action.execution_env_flag} is not set"
+                    if action.execution_env_flag
+                    else ""
+                )
+                + ". Execution is not wired."
             ),
             **base,
         )
@@ -545,8 +608,8 @@ def authorize(
         allowed=True,
         reason_code=ALLOW_PREVIEW,
         detail=(
-            "Principal holds the required role. Preview only — execution "
+            "Principal holds the required role. Execution proceeds only for an "
-            "remains disabled until the Phase 2 action framework ships."
+            "action with a wired execution path; everything else is preview."
         ),
         **base,
     )

webui/insights_loader.py

@@ -1,713 +0,0 @@
-"""AI-provider connections and evidence-backed operational insights (#650, Phase 4).
-
-Operators need two related, **advisory** surfaces:
-
-1. **Provider connection status** — which AI runtimes are *declared* in the
-   worker registry (#798), without ever exposing API keys or inventing a live
-   probe that this process cannot perform.
-2. **Evidence-backed insights** — short cards derived only from durable
-   console evidence (traffic, system health, analytics, the same registry).
-   Every insight carries explicit evidence refs (issue/PR/provider/event ids).
-   Insights never claim that a workflow action completed without proof, and
-   they never mutate anything.
-
-Design rules matching the rest of the console:
-
-- **Read-only.** No endpoint registered here mutates Gitea, the control plane,
-  or the registry.
-- **Advisory only.** Insights carry ``advisory_only=True`` and never emit an
-  "action completed" claim. The allocator, review, and merge paths remain the
-  only authorities for work selection and terminal state.
-- **Qualified absence.** When a source could not run, the insight list says so
-  rather than inventing an empty-and-healthy fleet or zero blocked items.
-- **Redaction.** Free-text titles, reasons, and notes pass through
-  ``webui.console_redaction`` before they leave this module.
-- **No secrets.** Provider records are taken from the credential-free worker
-  registry. Keys never appear in this surface.
-
-Non-goals (from the issue): free-form chatbot that overrides gates, secret
-provider keys in the UI, auto-merge or auto-close from insights.
-"""
-
-from __future__ import annotations
-
-import os
-from dataclasses import dataclass
-from typing import Any, Callable, Sequence
-
-from webui import console_redaction
-from webui.worker_registry import (
-    ProviderRecord,
-    WorkerRegistry,
-    WorkerRecord,
-    load_registry as load_worker_registry,
-    workers_for_provider,
-)
-
-INSIGHTS_SCHEMA_VERSION = 1
-
-# Provider connection vocabulary. Declared availability is not a live probe —
-# the worker registry owns the declaration, and adapters (#800) own live checks.
-CONNECTION_DECLARED_AVAILABLE = "declared_available"
-CONNECTION_DECLARED_UNAVAILABLE = "declared_unavailable"
-CONNECTION_REGISTRY_UNAVAILABLE = "registry_unavailable"
-
-# Insight kinds. Each generator is a pure function over one evidence source.
-INSIGHT_BLOCKED_QUEUE = "blocked_queue_pressure"
-INSIGHT_CONTROLLER_ATTENTION = "controller_attention"
-INSIGHT_STALE_RUNTIME = "stale_runtime_risk"
-INSIGHT_PROVIDER_WITHOUT_WORKERS = "provider_without_workers"
-INSIGHT_ANALYTICS_FAILURE_RATE = "analytics_failure_pressure"
-
-SEVERITY_INFO = "info"
-SEVERITY_WARN = "warn"
-SEVERITY_CRITICAL = "critical"
-SEVERITY_UNPROVEN = "unproven"
-
-CONFIDENCE_HIGH = "high"
-CONFIDENCE_MEDIUM = "medium"
-CONFIDENCE_LOW = "low"
-CONFIDENCE_UNPROVEN = "unproven"
-
-
-def _redact(value: Any) -> Any:
-    if value is None:
-        return None
-    return console_redaction.redact_text(str(value))
-
-
-def _offline_test_mode() -> bool:
-    return (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip().lower() in {
-        "1",
-        "true",
-        "yes",
-        "on",
-    }
-
-
-# --- Provider connection status ------------------------------------------------
-
-
-@dataclass(frozen=True)
-class ProviderConnection:
-    """One AI provider's declared connection status (no secrets, no live probe)."""
-
-    provider_id: str
-    display_name: str
-    vendor: str
-    executable: str
-    connection_status: str
-    available_declared: bool
-    models: tuple[str, ...]
-    worker_count: int
-    enabled_worker_count: int
-    notes: str
-    #: Explicit statement of what was *not* proven (live process health, etc.).
-    probe_limit: str
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "provider_id": self.provider_id,
-            "display_name": self.display_name,
-            "vendor": self.vendor,
-            "executable": self.executable,
-            "connection_status": self.connection_status,
-            "available_declared": self.available_declared,
-            "models": list(self.models),
-            "worker_count": self.worker_count,
-            "enabled_worker_count": self.enabled_worker_count,
-            "notes": self.notes,
-            "probe_limit": self.probe_limit,
-            # Always true for this surface: keys are never loaded.
-            "secrets_exposed": False,
-        }
-
-
-@dataclass(frozen=True)
-class ProviderSnapshot:
-    ok: bool
-    providers: tuple[ProviderConnection, ...] = ()
-    registry_revision: int | None = None
-    registry_path: str | None = None
-    fetch_error: str | None = None
-    schema_version: int = INSIGHTS_SCHEMA_VERSION
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "ok": self.ok,
-            "schema_version": self.schema_version,
-            "registry_revision": self.registry_revision,
-            "registry_path": self.registry_path,
-            "fetch_error": self.fetch_error,
-            "providers": [p.to_dict() for p in self.providers],
-            "interpretation_limits": [
-                "connection_status reflects the worker registry declaration only",
-                "no API keys or credential material are loaded or rendered",
-                "live executable health is not probed on this surface (#800 owns that)",
-            ],
-        }
-
-
-_PROBE_LIMIT = (
-    "Declared status only. This console does not probe the provider executable "
-    "or call vendor APIs; live health belongs to the provider adapter framework."
-)
-
-
-def connection_status_for(provider: ProviderRecord) -> str:
-    return (
-        CONNECTION_DECLARED_AVAILABLE
-        if provider.available
-        else CONNECTION_DECLARED_UNAVAILABLE
-    )
-
-
-def build_provider_connection(
-    provider: ProviderRecord,
-    workers: Sequence[WorkerRecord],
-) -> ProviderConnection:
-    enabled = sum(1 for worker in workers if worker.enabled)
-    return ProviderConnection(
-        provider_id=provider.id,
-        display_name=str(_redact(provider.display_name) or provider.id),
-        vendor=str(_redact(provider.vendor) or ""),
-        executable=str(_redact(provider.executable) or ""),
-        connection_status=connection_status_for(provider),
-        available_declared=bool(provider.available),
-        models=tuple(str(_redact(m) or m) for m in provider.models),
-        worker_count=len(workers),
-        enabled_worker_count=enabled,
-        notes=str(_redact(provider.notes) or ""),
-        probe_limit=_PROBE_LIMIT,
-    )
-
-
-def load_provider_snapshot(
-    *,
-    registry: WorkerRegistry | None = None,
-    registry_loader: Callable[[], WorkerRegistry] | None = None,
-) -> ProviderSnapshot:
-    """Load declared provider connections. Never raises for missing registry."""
-    if registry is None:
-        loader = registry_loader or load_worker_registry
-        try:
-            if _offline_test_mode() and registry_loader is None:
-                return ProviderSnapshot(
-                    ok=False,
-                    fetch_error=(
-                        "provider registry not loaded in offline test mode "
-                        "(inject a registry for unit tests)"
-                    ),
-                )
-            registry = loader()
-        except Exception as exc:  # fail soft — operator-visible reason
-            return ProviderSnapshot(
-                ok=False,
-                fetch_error=str(_redact(f"worker registry unavailable: {exc}")),
-            )
-
-    connections = tuple(
-        build_provider_connection(provider, workers_for_provider(registry, provider.id))
-        for provider in registry.providers
-    )
-    return ProviderSnapshot(
-        ok=True,
-        providers=connections,
-        registry_revision=registry.revision,
-        registry_path=str(registry.source_path),
-    )
-
-
-# --- Evidence-backed insights --------------------------------------------------
-
-
-@dataclass(frozen=True)
-class EvidenceRef:
-    """One durable reference an insight is allowed to cite."""
-
-    kind: str  # issue | pr | provider | health | analytics | traffic
-    ref: str
-    detail: str
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "kind": self.kind,
-            "ref": self.ref,
-            "detail": str(_redact(self.detail) or ""),
-        }
-
-
-@dataclass(frozen=True)
-class Insight:
-    """One advisory finding. Never a claim that an action completed."""
-
-    insight_id: str
-    kind: str
-    severity: str
-    confidence: str
-    title: str
-    summary: str
-    evidence: tuple[EvidenceRef, ...]
-    advisory_only: bool = True
-    claims_action_completed: bool = False
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "insight_id": self.insight_id,
-            "kind": self.kind,
-            "severity": self.severity,
-            "confidence": self.confidence,
-            "title": str(_redact(self.title) or ""),
-            "summary": str(_redact(self.summary) or ""),
-            "evidence": [item.to_dict() for item in self.evidence],
-            "advisory_only": self.advisory_only,
-            "claims_action_completed": self.claims_action_completed,
-        }
-
-
-@dataclass(frozen=True)
-class InsightsSnapshot:
-    ok: bool
-    insights: tuple[Insight, ...] = ()
-    sources_used: tuple[str, ...] = ()
-    sources_unavailable: tuple[dict[str, str], ...] = ()
-    fetch_error: str | None = None
-    schema_version: int = INSIGHTS_SCHEMA_VERSION
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "ok": self.ok,
-            "schema_version": self.schema_version,
-            "insights": [insight.to_dict() for insight in self.insights],
-            "sources_used": list(self.sources_used),
-            "sources_unavailable": list(self.sources_unavailable),
-            "fetch_error": self.fetch_error,
-            "interpretation_limits": [
-                "insights are advisory only and never authorize merge, review, or close",
-                "an insight without evidence refs is refused rather than emitted",
-                "a missing source is listed under sources_unavailable, not as an empty success",
-                "insights never claim a workflow action completed",
-            ],
-        }
-
-
-def _require_evidence(evidence: Sequence[EvidenceRef]) -> tuple[EvidenceRef, ...]:
-    """Fail closed: an insight with no evidence must not be emitted."""
-    items = tuple(evidence)
-    if not items:
-        raise ValueError("insight requires at least one evidence ref")
-    return items
-
-
-def insight_blocked_queue(traffic: Any) -> Insight | None:
-    """Traffic blocked bucket pressure with per-item evidence."""
-    blocked = tuple(getattr(traffic, "blocked", ()) or ())
-    if not blocked:
-        return None
-    evidence = []
-    for item in blocked[:20]:
-        kind = str(getattr(item, "kind", "issue") or "issue")
-        number = int(getattr(item, "number", 0) or 0)
-        if number <= 0:
-            continue
-        reason = getattr(item, "block_reason", None) or "blocked"
-        evidence.append(
-            EvidenceRef(
-                kind=kind,
-                ref=f"#{number}",
-                detail=f"traffic_state=blocked; reason={reason}",
-            )
-        )
-    if not evidence:
-        return None
-    count = len(blocked)
-    severity = SEVERITY_CRITICAL if count >= 10 else SEVERITY_WARN
-    return Insight(
-        insight_id=f"{INSIGHT_BLOCKED_QUEUE}:{count}",
-        kind=INSIGHT_BLOCKED_QUEUE,
-        severity=severity,
-        confidence=(
-            CONFIDENCE_HIGH
-            if getattr(traffic, "inventory_complete", False)
-            else CONFIDENCE_MEDIUM
-        ),
-        title=f"{count} blocked work item(s) in traffic control",
-        summary=(
-            f"Traffic control reports {count} blocked item(s). "
-            "This is an observation of the loaded window, not a claim that "
-            "any remediation ran."
-        ),
-        evidence=_require_evidence(evidence),
-    )
-
-
-def insight_controller_attention(traffic: Any) -> Insight | None:
-    needs = tuple(getattr(traffic, "needs_controller", ()) or ())
-    if not needs:
-        return None
-    evidence = []
-    for item in needs[:20]:
-        kind = str(getattr(item, "kind", "issue") or "issue")
-        number = int(getattr(item, "number", 0) or 0)
-        if number <= 0:
-            continue
-        evidence.append(
-            EvidenceRef(
-                kind=kind,
-                ref=f"#{number}",
-                detail="traffic_state=needs_controller",
-            )
-        )
-    if not evidence:
-        return None
-    count = len(needs)
-    return Insight(
-        insight_id=f"{INSIGHT_CONTROLLER_ATTENTION}:{count}",
-        kind=INSIGHT_CONTROLLER_ATTENTION,
-        severity=SEVERITY_WARN if count else SEVERITY_INFO,
-        confidence=(
-            CONFIDENCE_HIGH
-            if getattr(traffic, "inventory_complete", False)
-            else CONFIDENCE_MEDIUM
-        ),
-        title=f"{count} item(s) need controller attention",
-        summary=(
-            f"Traffic control marks {count} item(s) as needs_controller. "
-            "Advisory only — the controller allocator remains the authority "
-            "for routing."
-        ),
-        evidence=_require_evidence(evidence),
-    )
-
-
-def insight_stale_runtime(health: Any) -> Insight | None:
-    stale = getattr(health, "stale_runtime", None)
-    if stale is None:
-        return None
-    mutation_safe = bool(getattr(stale, "mutation_safe", False))
-    is_stale = bool(getattr(stale, "stale", False))
-    determinable = bool(getattr(stale, "determinable", False))
-    if mutation_safe and not is_stale:
-        return None
-    daemon = getattr(stale, "daemon_head", None) or "unknown"
-    checkout = getattr(stale, "checkout_head", None) or "unknown"
-    remote = getattr(stale, "remote_head", None) or "unknown"
-    if not determinable:
-        severity = SEVERITY_UNPROVEN
-        confidence = CONFIDENCE_UNPROVEN
-        title = "Runtime parity is not determinable"
-        summary = (
-            "System health could not prove mutation_safe. This is not proof "
-            "that the runtime is stale — only that parity was unproven."
-        )
-    else:
-        severity = SEVERITY_CRITICAL if is_stale else SEVERITY_WARN
-        confidence = CONFIDENCE_HIGH
-        title = "Stale or mutation-unsafe runtime"
-        summary = (
-            "System health reports a runtime that is not mutation_safe. "
-            "No restart or recovery is claimed by this insight."
-        )
-    return Insight(
-        insight_id=f"{INSIGHT_STALE_RUNTIME}:{daemon}:{checkout}",
-        kind=INSIGHT_STALE_RUNTIME,
-        severity=severity,
-        confidence=confidence,
-        title=title,
-        summary=summary,
-        evidence=_require_evidence(
-            (
-                EvidenceRef(
-                    kind="health",
-                    ref="stale_runtime",
-                    detail=(
-                        f"stale={is_stale}; mutation_safe={mutation_safe}; "
-                        f"determinable={determinable}; daemon={daemon}; "
-                        f"checkout={checkout}; remote={remote}"
-                    ),
-                ),
-            )
-        ),
-    )
-
-
-def insight_providers_without_workers(
-    providers: Sequence[ProviderConnection],
-) -> Insight | None:
-    lonely = [
-        provider
-        for provider in providers
-        if provider.available_declared and provider.worker_count == 0
-    ]
-    if not lonely:
-        return None
-    evidence = tuple(
-        EvidenceRef(
-            kind="provider",
-            ref=provider.provider_id,
-            detail=(
-                f"available_declared=true; worker_count=0; "
-                f"vendor={provider.vendor}"
-            ),
-        )
-        for provider in lonely
-    )
-    return Insight(
-        insight_id=f"{INSIGHT_PROVIDER_WITHOUT_WORKERS}:{len(lonely)}",
-        kind=INSIGHT_PROVIDER_WITHOUT_WORKERS,
-        severity=SEVERITY_INFO,
-        confidence=CONFIDENCE_HIGH,
-        title=f"{len(lonely)} declared-available provider(s) have no workers",
-        summary=(
-            "The worker registry declares these providers available but no "
-            "worker instance names them. This is a configuration observation, "
-            "not a claim that a provider process is running or idle."
-        ),
-        evidence=_require_evidence(evidence),
-    )
-
-
-def insight_analytics_failures(analytics: Any) -> Insight | None:
-    """Flag elevated non-ok stage status in analytics when events exist."""
-    if analytics is None or not getattr(analytics, "ok", False):
-        return None
-    events = tuple(getattr(analytics, "events", ()) or ())
-    if not events:
-        return None
-    failed = [
-        event
-        for event in events
-        if str(getattr(event, "status", "") or "").lower()
-        in {"error", "failed", "failure"}
-    ]
-    if not failed:
-        return None
-    # Cap evidence so a large window stays readable.
-    evidence = []
-    for event in failed[:20]:
-        usage_id = getattr(event, "usage_id", None)
-        issue = getattr(event, "issue_number", None)
-        pr = getattr(event, "pr_number", None)
-        if pr is not None:
-            ref_kind, ref = "pr", f"#{int(pr)}"
-        elif issue is not None:
-            ref_kind, ref = "issue", f"#{int(issue)}"
-        else:
-            ref_kind, ref = "analytics", f"usage:{usage_id}"
-        evidence.append(
-            EvidenceRef(
-                kind=ref_kind,
-                ref=ref,
-                detail=(
-                    f"status={getattr(event, 'status', '')}; "
-                    f"stage={getattr(event, 'stage', '')}; "
-                    f"model={getattr(event, 'model', '')}"
-                ),
-            )
-        )
-    if not evidence:
-        return None
-    rate = len(failed) / max(len(events), 1)
-    return Insight(
-        insight_id=f"{INSIGHT_ANALYTICS_FAILURE_RATE}:{len(failed)}:{len(events)}",
-        kind=INSIGHT_ANALYTICS_FAILURE_RATE,
-        severity=SEVERITY_WARN if rate >= 0.1 else SEVERITY_INFO,
-        confidence=CONFIDENCE_MEDIUM,
-        title=f"{len(failed)} analytics event(s) reported failure status",
-        summary=(
-            f"{len(failed)} of {len(events)} loaded analytics events carry a "
-            "failure status. Advisory only — this is not a gate decision."
-        ),
-        evidence=_require_evidence(evidence),
-    )
-
-
-def generate_insights(
-    *,
-    traffic: Any | None = None,
-    health: Any | None = None,
-    provider_snapshot: ProviderSnapshot | None = None,
-    analytics: Any | None = None,
-) -> tuple[tuple[Insight, ...], tuple[str, ...], tuple[dict[str, str], ...]]:
-    """Pure multi-source insight generation. Never mutates inputs."""
-    insights: list[Insight] = []
-    used: list[str] = []
-    unavailable: list[dict[str, str]] = []
-
-    if traffic is None:
-        unavailable.append(
-            {"source": "traffic", "reason": "traffic snapshot not supplied"}
-        )
-    elif getattr(traffic, "fetch_error", None):
-        unavailable.append(
-            {
-                "source": "traffic",
-                "reason": str(_redact(traffic.fetch_error) or "traffic fetch failed"),
-            }
-        )
-    else:
-        used.append("traffic")
-        for builder in (insight_blocked_queue, insight_controller_attention):
-            try:
-                item = builder(traffic)
-            except ValueError:
-                continue
-            if item is not None:
-                insights.append(item)
-
-    if health is None:
-        unavailable.append(
-            {"source": "system_health", "reason": "system health snapshot not supplied"}
-        )
-    else:
-        used.append("system_health")
-        try:
-            item = insight_stale_runtime(health)
-        except ValueError:
-            item = None
-        if item is not None:
-            insights.append(item)
-
-    if provider_snapshot is None:
-        unavailable.append(
-            {"source": "providers", "reason": "provider snapshot not supplied"}
-        )
-    elif not provider_snapshot.ok:
-        unavailable.append(
-            {
-                "source": "providers",
-                "reason": str(
-                    _redact(provider_snapshot.fetch_error)
-                    or "provider registry unavailable"
-                ),
-            }
-        )
-    else:
-        used.append("providers")
-        try:
-            item = insight_providers_without_workers(provider_snapshot.providers)
-        except ValueError:
-            item = None
-        if item is not None:
-            insights.append(item)
-
-    if analytics is None:
-        unavailable.append(
-            {"source": "analytics", "reason": "analytics snapshot not supplied"}
-        )
-    elif not getattr(analytics, "ok", False):
-        unavailable.append(
-            {
-                "source": "analytics",
-                "reason": str(
-                    _redact(getattr(analytics, "fetch_error", None))
-                    or "analytics snapshot not ok"
-                ),
-            }
-        )
-    else:
-        used.append("analytics")
-        try:
-            item = insight_analytics_failures(analytics)
-        except ValueError:
-            item = None
-        if item is not None:
-            insights.append(item)
-
-    # Stable ordering: severity then kind.
-    _sev_rank = {
-        SEVERITY_CRITICAL: 0,
-        SEVERITY_WARN: 1,
-        SEVERITY_INFO: 2,
-        SEVERITY_UNPROVEN: 3,
-    }
-    insights.sort(key=lambda i: (_sev_rank.get(i.severity, 9), i.kind, i.insight_id))
-    return tuple(insights), tuple(used), tuple(unavailable)
-
-
-def load_insights_snapshot(
-    *,
-    traffic: Any | None = None,
-    health: Any | None = None,
-    provider_snapshot: ProviderSnapshot | None = None,
-    analytics: Any | None = None,
-    load_live: bool = True,
-) -> InsightsSnapshot:
-    """Compose insights from injected or live console evidence sources."""
-    sources_unavailable: list[dict[str, str]] = []
-
-    if load_live and traffic is None and not _offline_test_mode():
-        try:
-            from webui.traffic_loader import load_traffic_snapshot
-
-            traffic = load_traffic_snapshot()
-        except Exception as exc:  # fail soft
-            sources_unavailable.append(
-                {
-                    "source": "traffic",
-                    "reason": str(_redact(f"traffic load failed: {exc}")),
-                }
-            )
-            traffic = None
-
-    if load_live and health is None and not _offline_test_mode():
-        try:
-            from webui.system_health import load_system_health
-
-            health = load_system_health()
-        except Exception as exc:
-            sources_unavailable.append(
-                {
-                    "source": "system_health",
-                    "reason": str(_redact(f"system health load failed: {exc}")),
-                }
-            )
-            health = None
-
-    if provider_snapshot is None:
-        provider_snapshot = load_provider_snapshot()
-
-    if load_live and analytics is None and not _offline_test_mode():
-        try:
-            from webui.analytics_loader import load_analytics
-
-            analytics = load_analytics()
-        except Exception as exc:
-            sources_unavailable.append(
-                {
-                    "source": "analytics",
-                    "reason": str(_redact(f"analytics load failed: {exc}")),
-                }
-            )
-            analytics = None
-
-    insights, used, unavailable = generate_insights(
-        traffic=traffic,
-        health=health,
-        provider_snapshot=provider_snapshot,
-        analytics=analytics,
-    )
-    merged_unavailable = tuple(sources_unavailable) + unavailable
-    # ok when at least one source contributed or we can honestly report absence.
-    ok = bool(used) or bool(merged_unavailable)
-    return InsightsSnapshot(
-        ok=ok,
-        insights=insights,
-        sources_used=used,
-        sources_unavailable=merged_unavailable,
-        fetch_error=None
-        if used
-        else (
-            "no evidence sources produced a usable snapshot"
-            if merged_unavailable
-            else "no insight sources ran"
-        ),
-    )
-
-
-def snapshot_providers_to_dict(snapshot: ProviderSnapshot) -> dict[str, Any]:
-    return snapshot.to_dict()
-
-
-def snapshot_insights_to_dict(snapshot: InsightsSnapshot) -> dict[str, Any]:
-    return snapshot.to_dict()

webui/insights_views.py

@@ -1,196 +0,0 @@
-"""HTML views for AI-provider connections and operational insights (#650)."""
-
-from __future__ import annotations
-
-from html import escape
-
-from webui.insights_loader import (
-    CONNECTION_DECLARED_AVAILABLE,
-    CONNECTION_DECLARED_UNAVAILABLE,
-    InsightsSnapshot,
-    ProviderSnapshot,
-    SEVERITY_CRITICAL,
-    SEVERITY_INFO,
-    SEVERITY_UNPROVEN,
-    SEVERITY_WARN,
-)
-from webui.layout import render_page
-
-_SEVERITY_CSS = {
-    SEVERITY_CRITICAL: "badge-blocked",
-    SEVERITY_WARN: "badge-health-degraded",
-    SEVERITY_INFO: "badge-health-ok",
-    SEVERITY_UNPROVEN: "badge-health-unproven",
-}
-
-_CONN_CSS = {
-    CONNECTION_DECLARED_AVAILABLE: "badge-health-ok",
-    CONNECTION_DECLARED_UNAVAILABLE: "badge-health-degraded",
-    "registry_unavailable": "badge-blocked",
-}
-
-
-def _badge(text: str, css: str) -> str:
-    return f'<span class="badge {css}">{escape(text)}</span>'
-
-
-def _limits_card(lines: list[str], *, title: str) -> str:
-    items = "".join(f"<li>{escape(line)}</li>" for line in lines)
-    return f"""<div class="prompt-card">
-  <h3>{escape(title)}</h3>
-  <ul class="reasons">{items}</ul>
-  <p class="muted">Advisory surface only — no review, merge, close, or provider
-  mutation is available here.</p>
-</div>"""
-
-
-def render_providers_page(snapshot: ProviderSnapshot) -> str:
-    """Render the AI-provider connections page."""
-    if not snapshot.ok:
-        body = f"""<h2>AI provider connections</h2>
-<p class="meta">Phase 4 read-only provider status (#650).</p>
-<div class="health-card health-stale">
-  <strong>Provider registry unavailable:</strong>
-  {escape(snapshot.fetch_error or "registry could not be loaded")}.
-  No connection table is rendered — an empty table would claim that no
-  providers are configured.
-</div>
-{_limits_card([
-    "connection_status reflects the worker registry declaration only",
-    "no API keys or credential material are loaded or rendered",
-    "live executable health is not probed on this surface",
-], title="Interpretation limits")}
-"""
-        return render_page(title="Providers", body_html=body)
-
-    rows = []
-    for provider in snapshot.providers:
-        models = (
-            ", ".join(f"<code>{escape(m)}</code>" for m in provider.models)
-            if provider.models
-            else '<span class="muted">none declared</span>'
-        )
-        rows.append(
-            "<tr>"
-            f"<td><code>{escape(provider.provider_id)}</code></td>"
-            f"<td>{escape(provider.display_name)}</td>"
-            f"<td>{escape(provider.vendor)}</td>"
-            f"<td><code>{escape(provider.executable)}</code></td>"
-            f"<td>{_badge(provider.connection_status, _CONN_CSS.get(provider.connection_status, 'badge-health-skipped'))}</td>"
-            f"<td>{provider.worker_count} "
-            f"({provider.enabled_worker_count} enabled)</td>"
-            f"<td>{models}</td>"
-            "</tr>"
-        )
-    table = (
-        "".join(rows)
-        if rows
-        else '<tr><td colspan="7" class="muted">No providers declared in the registry.</td></tr>'
-    )
-
-    body = f"""<h2>AI provider connections</h2>
-<p class="meta">Phase 4 read-only provider status (#650). Registry revision
-<code>{escape(str(snapshot.registry_revision))}</code>.
-Declared status only — secrets never load.</p>
-
-<div class="health-card">
-  <h3>Declared connections</h3>
-  <table class="registry">
-    <thead>
-      <tr>
-        <th>Provider</th><th>Name</th><th>Vendor</th><th>Executable</th>
-        <th>Connection</th><th>Workers</th><th>Models (declared)</th>
-      </tr>
-    </thead>
-    <tbody>{table}</tbody>
-  </table>
-  <p class="muted">{escape(snapshot.providers[0].probe_limit if snapshot.providers else "")}</p>
-</div>
-
-{_limits_card([
-    "connection_status reflects the worker registry declaration only",
-    "no API keys or credential material are loaded or rendered",
-    "live executable health is not probed on this surface (#800 owns that)",
-], title="Interpretation limits")}
-<p class="muted">Related: <a href="/insights">Operational insights</a> ·
-<a href="/analytics">Analytics</a></p>
-"""
-    return render_page(title="Providers", body_html=body)
-
-
-def _evidence_list(insight) -> str:
-    items = "".join(
-        f"<li><code>{escape(ref.kind)}:{escape(ref.ref)}</code> — "
-        f"{escape(ref.detail)}</li>"
-        for ref in insight.evidence
-    )
-    return f'<ul class="reasons">{items}</ul>'
-
-
-def render_insights_page(snapshot: InsightsSnapshot) -> str:
-    """Render the operational insights page."""
-    if not snapshot.ok and not snapshot.insights:
-        body = f"""<h2>Operational insights</h2>
-<p class="meta">Phase 4 evidence-backed insights (#650).</p>
-<div class="health-card health-stale">
-  <strong>Insights unavailable:</strong>
-  {escape(snapshot.fetch_error or "no sources ran")}.
-</div>
-{_limits_card([
-    "insights are advisory only and never authorize merge, review, or close",
-    "an insight without evidence refs is refused rather than emitted",
-], title="Interpretation limits")}
-"""
-        return render_page(title="Insights", body_html=body)
-
-    source_bits = []
-    if snapshot.sources_used:
-        source_bits.append(
-            "sources used: " + ", ".join(f"<code>{escape(s)}</code>" for s in snapshot.sources_used)
-        )
-    if snapshot.sources_unavailable:
-        missing = "; ".join(
-            f"{escape(item.get('source', '?'))}: {escape(item.get('reason', ''))}"
-            for item in snapshot.sources_unavailable
-        )
-        source_bits.append(f"sources unavailable: {missing}")
-
-    cards = []
-    for insight in snapshot.insights:
-        cards.append(
-            f"""<div class="prompt-card">
-  <h3>{_badge(insight.severity, _SEVERITY_CSS.get(insight.severity, "badge-health-skipped"))}
-  {escape(insight.title)}</h3>
-  <p class="meta"><code>{escape(insight.kind)}</code> · confidence
-  <code>{escape(insight.confidence)}</code> ·
-  {_badge("advisory only", "badge-health-skipped")} ·
-  {_badge("no action claimed", "badge-health-ok")}</p>
-  <p>{escape(insight.summary)}</p>
-  <h4>Evidence</h4>
-  {_evidence_list(insight)}
-</div>"""
-        )
-    if not cards:
-        cards.append(
-            '<div class="prompt-card"><p class="muted">No insights met the '
-            "evidence threshold in the loaded sources. That is not a claim "
-            "that the fleet is healthy — only that no qualifying pattern was "
-            "found.</p></div>"
-        )
-
-    body = f"""<h2>Operational insights</h2>
-<p class="meta">Phase 4 evidence-backed insights (#650). Derived only from
-durable console evidence; never invents policy or completes workflow actions.</p>
-<p class="muted">{" · ".join(source_bits) if source_bits else ""}</p>
-{"".join(cards)}
-{_limits_card([
-    "insights are advisory only and never authorize merge, review, or close",
-    "an insight without evidence refs is refused rather than emitted",
-    "a missing source is listed as unavailable, not as an empty success",
-    "insights never claim a workflow action completed",
-], title="Interpretation limits")}
-<p class="muted">Related: <a href="/providers">Provider connections</a> ·
-<a href="/traffic">Traffic</a> · <a href="/system-health">System health</a> ·
-<a href="/analytics">Analytics</a></p>
-"""
-    return render_page(title="Insights", body_html=body)

webui/linkage_loader.py

@@ -1,771 +0,0 @@
-"""Gitea issue↔PR linkage model for the console (#645, Phase 3).
-
-Operators lose context between an issue and the PR that closes it: which PR
-carries which issue, whether two PRs claim the same issue, and what the latest
-canonical handoff on that thread said. The evidence exists in Gitea, but only
-as free text scattered across PR titles, bodies, and branch names.
-
-This module resolves that linkage into one read-only model:
-
-* :func:`resolve_linkage` is a pure function from raw Gitea issue/PR payloads to
-  a :class:`LinkageIndex`. It records *how* each edge was found (a ``Closes #N``
-  keyword, the canonical ``feat/issue-N-…`` branch marker, or a bare ``#N``
-  body reference) and never collapses several candidates into one silent guess.
-* :func:`load_linkage_snapshot` scopes that index to a registry project and
-  optionally attaches the latest Canonical Thread Handoff (CTH) summary for one
-  focused issue or PR.
-
-Design rules, matching the rest of the console:
-
-- **Read-only.** Gitea is read through the shared authenticated helpers. No
-  endpoint here mutates anything, and no write action is registered.
-- **Qualified absence.** Linkage is a claim about a *loaded* window of Gitea.
-  When pagination did not complete, when credentials were unavailable, or when
-  only open items were fetched, the snapshot says so and every "no linked PR"
-  is marked non-authoritative. An empty edge list from a partial read is not
-  evidence that no link exists.
-- **Handoff is loaded, never assumed.** CTH comments are thread-scoped, so they
-  are fetched only for an explicitly focused issue or PR. Every other row
-  reports ``not_loaded`` rather than rendering as "no handoff".
-- **Redaction at the boundary.** Titles, labels, handoff fields, and error
-  reasons are free text from Gitea and cross :mod:`webui.console_redaction`
-  before they leave this module.
-- **Deep links are opt-in.** A link to the Gitea web UI is emitted only under
-  the ``GITEA_MCP_REVEAL_ENDPOINTS`` admin opt-in, exactly as the MCP tools
-  gate their own URL exposure.
-
-Non-goals (from the issue): no issue/PR editor, no browser review or merge, no
-reimplementation of Gitea search.
-"""
-
-from __future__ import annotations
-
-import os
-import re
-from dataclasses import dataclass
-from typing import Any, Callable, Iterable, Sequence
-
-from gitea_auth import api_fetch_page, get_auth_header, gitea_url, repo_api_url
-
-from webui import console_redaction
-from webui.project_registry import ProjectRecord, load_registry
-from webui.queue_loader import (
-    PaginationMeta,
-    _fetch_issues,
-    _fetch_prs,
-    _host_from_url,
-)
-
-#: Version of the serialized linkage contract. Bump on any breaking change.
-LINKAGE_SCHEMA_VERSION = 1
-
-# --- Linkage evidence -------------------------------------------------------
-# Ordered strongest to weakest. The strength ordering is what makes an
-# ambiguous PR detectable: two candidates at the same strength are a genuine
-# ambiguity, while a weaker candidate alongside a stronger one is not.
-EVIDENCE_CLOSES = "closes_keyword"
-EVIDENCE_BRANCH = "branch_marker"
-EVIDENCE_REFERENCE = "body_reference"
-
-EVIDENCE_ORDER: tuple[str, ...] = (
-    EVIDENCE_CLOSES,
-    EVIDENCE_BRANCH,
-    EVIDENCE_REFERENCE,
-)
-_EVIDENCE_RANK = {name: rank for rank, name in enumerate(EVIDENCE_ORDER)}
-
-EVIDENCE_DESCRIPTIONS: dict[str, str] = {
-    EVIDENCE_CLOSES: (
-        "the PR title or body declares 'closes/fixes/resolves #N' — Gitea itself "
-        "acts on this keyword, so it is the strongest available evidence"
-    ),
-    EVIDENCE_BRANCH: (
-        "the PR head branch carries the canonical issue marker "
-        "'(fix|feat|docs|chore)/issue-N-…' minted by the issue lock"
-    ),
-    EVIDENCE_REFERENCE: (
-        "the PR body mentions '#N' without a closing keyword; a mention is not "
-        "a claim that the PR closes that issue"
-    ),
-}
-
-_CLOSES_RE = re.compile(r"(?:closes|fixes|resolves)\s+#(\d+)", re.IGNORECASE)
-_REFERENCE_RE = re.compile(r"#(\d+)")
-_BRANCH_MARKER_RE = re.compile(
-    r"^(?:fix|feat|docs|chore)/issue-(\d+)(?:[-/]|$)", re.IGNORECASE
-)
-
-# Handoff-source states. ``not_loaded`` is deliberately distinct from "none
-# found": a row whose comments were never fetched proves nothing about whether
-# a handoff exists on that thread.
-HANDOFF_NOT_LOADED = "not_loaded"
-HANDOFF_LOADED = "loaded"
-HANDOFF_UNAVAILABLE = "unavailable"
-
-# Which item states were fetched. Linkage claims are scoped to this window.
-STATE_OPEN = "open"
-STATE_ALL = "all"
-_SUPPORTED_STATES = (STATE_OPEN, STATE_ALL)
-
-
-def _redact(value: Any) -> Any:
-    """Redact one free-text field, failing closed to the placeholder."""
-    if value is None:
-        return None
-    return console_redaction.redact_text(str(value))
-
-
-def deep_links_enabled(env: dict[str, str] | None = None) -> bool:
-    """Whether Gitea web-UI deep links may be emitted (admin/debug opt-in)."""
-    source = env if env is not None else os.environ
-    return (source.get("GITEA_MCP_REVEAL_ENDPOINTS") or "").strip().lower() in {
-        "1",
-        "true",
-        "yes",
-        "on",
-    }
-
-
-def _deep_link(host: str, org: str, repo: str, kind: str, number: int) -> str | None:
-    """Build a Gitea web link for one item, or None when reveal is not enabled."""
-    if not deep_links_enabled() or not (host and org and repo):
-        return None
-    segment = "pulls" if kind == "pr" else "issues"
-    try:
-        return gitea_url(host, f"/{org}/{repo}/{segment}/{int(number)}")
-    except Exception:
-        return None
-
-
-# --- Pure linkage resolution -------------------------------------------------
-
-
-@dataclass(frozen=True)
-class IssueLink:
-    """One resolved edge from a PR to an issue, with the evidence that found it."""
-
-    issue_number: int
-    evidence: tuple[str, ...]
-
-    @property
-    def strength(self) -> int:
-        """Rank of the strongest evidence backing this edge (lower is stronger)."""
-        return min(
-            (_EVIDENCE_RANK.get(name, len(EVIDENCE_ORDER)) for name in self.evidence),
-            default=len(EVIDENCE_ORDER),
-        )
-
-    @property
-    def closes(self) -> bool:
-        """True only when the PR *declares* it closes the issue."""
-        return EVIDENCE_CLOSES in self.evidence
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "issue_number": self.issue_number,
-            "evidence": list(self.evidence),
-            "closes": self.closes,
-        }
-
-
-def _sorted_links(links: Iterable[IssueLink]) -> tuple[IssueLink, ...]:
-    return tuple(sorted(links, key=lambda link: (link.strength, link.issue_number)))
-
-
-def resolve_pr_links(pr: dict[str, Any]) -> tuple[IssueLink, ...]:
-    """Resolve every issue a PR points at, strongest evidence first.
-
-    Every candidate is kept. Collapsing to a single "linked issue" is what makes
-    a mislinked or double-claimed PR invisible, so the caller decides what to do
-    with several candidates rather than being handed one guess.
-    """
-    found: dict[int, set[str]] = {}
-
-    def _add(number: Any, evidence: str) -> None:
-        try:
-            issue_number = int(number)
-        except (TypeError, ValueError):
-            return
-        if issue_number <= 0:
-            return
-        found.setdefault(issue_number, set()).add(evidence)
-
-    title = str(pr.get("title") or "")
-    body = str(pr.get("body") or "")
-    for text in (title, body):
-        for match in _CLOSES_RE.finditer(text):
-            _add(match.group(1), EVIDENCE_CLOSES)
-
-    head_ref = str((pr.get("head") or {}).get("ref") or "")
-    branch_match = _BRANCH_MARKER_RE.match(head_ref.strip())
-    if branch_match:
-        _add(branch_match.group(1), EVIDENCE_BRANCH)
-
-    # The ``#N`` inside "Closes #N" is the *same* textual occurrence as the
-    # closing keyword, not a second, independent mention. Blanking the closing
-    # phrases first keeps "mention" meaning what the legend says it means: a
-    # reference the PR made without claiming to close anything.
-    for match in _REFERENCE_RE.finditer(_CLOSES_RE.sub(" ", body)):
-        _add(match.group(1), EVIDENCE_REFERENCE)
-
-    # A PR's own number appearing in its body is self-reference, not linkage.
-    try:
-        found.pop(int(pr.get("number")), None)
-    except (TypeError, ValueError):
-        pass
-
-    return _sorted_links(
-        IssueLink(
-            issue_number=number,
-            evidence=tuple(name for name in EVIDENCE_ORDER if name in evidence),
-        )
-        for number, evidence in found.items()
-    )
-
-
-@dataclass(frozen=True)
-class LinkageIndex:
-    """Resolved linkage over one loaded window of issues and PRs."""
-
-    pr_links: dict[int, tuple[IssueLink, ...]]
-    issue_prs: dict[int, tuple[int, ...]]
-
-    def primary_issue(self, pr_number: int) -> IssueLink | None:
-        """The strongest edge for a PR, or None when it points at no issue."""
-        links = self.pr_links.get(int(pr_number)) or ()
-        return links[0] if links else None
-
-    def ambiguous(self, pr_number: int) -> bool:
-        """True when two or more issues tie at the PR's strongest evidence."""
-        links = self.pr_links.get(int(pr_number)) or ()
-        if len(links) < 2:
-            return False
-        best = links[0].strength
-        return sum(1 for link in links if link.strength == best) > 1
-
-    def contested_issues(self) -> tuple[int, ...]:
-        """Issues claimed by more than one PR in the loaded window."""
-        return tuple(
-            number for number, prs in sorted(self.issue_prs.items()) if len(prs) > 1
-        )
-
-
-def resolve_linkage(prs: Sequence[dict[str, Any]]) -> LinkageIndex:
-    """Build the bidirectional linkage index for a loaded window of PRs.
-
-    Only *closing* and *branch-marker* edges populate the issue→PR direction: a
-    bare ``#N`` mention is a reference, and treating it as "this PR is the work
-    for issue N" would invent contested issues out of ordinary cross-links. The
-    weaker edge stays visible on the PR→issue side, where it is labelled.
-    """
-    pr_links: dict[int, tuple[IssueLink, ...]] = {}
-    issue_prs: dict[int, list[int]] = {}
-    for pr in prs or []:
-        try:
-            pr_number = int(pr["number"])
-        except (KeyError, TypeError, ValueError):
-            continue
-        links = resolve_pr_links(pr)
-        pr_links[pr_number] = links
-        for link in links:
-            if link.evidence == (EVIDENCE_REFERENCE,):
-                continue
-            bucket = issue_prs.setdefault(link.issue_number, [])
-            if pr_number not in bucket:
-                bucket.append(pr_number)
-    return LinkageIndex(
-        pr_links=pr_links,
-        issue_prs={number: tuple(sorted(items)) for number, items in issue_prs.items()},
-    )
-
-
-# --- Canonical handoff summary ----------------------------------------------
-
-
-@dataclass(frozen=True)
-class HandoffSummary:
-    """The latest CTH comment on one thread, redacted for display."""
-
-    comment_id: int | None
-    created_at: str | None
-    author: str | None
-    cth_type: str
-    cth_type_known: bool
-    status: str | None
-    next_owner: str | None
-    current_blocker: str | None
-    decision: str | None
-    next_action: str | None
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "comment_id": self.comment_id,
-            "created_at": self.created_at,
-            "author": self.author,
-            "cth_type": self.cth_type,
-            "cth_type_known": self.cth_type_known,
-            "status": self.status,
-            "next_owner": self.next_owner,
-            "current_blocker": self.current_blocker,
-            "decision": self.decision,
-            "next_action": self.next_action,
-        }
-
-
-@dataclass(frozen=True)
-class HandoffStatus:
-    """Why a thread's handoff summary is present, absent, or unknown."""
-
-    state: str
-    reason: str | None = None
-    target: str | None = None
-
-    @property
-    def loaded(self) -> bool:
-        return self.state == HANDOFF_LOADED
-
-    def to_dict(self) -> dict[str, Any]:
-        return {"state": self.state, "reason": self.reason, "target": self.target}
-
-
-def summarize_handoff(comments: Sequence[dict[str, Any]]) -> HandoffSummary | None:
-    """Summarize the newest CTH comment in *comments*, or None when there is none.
-
-    Every field is redacted before it is returned: a handoff body is operator
-    free text that regularly quotes commands, and it is rendered verbatim on the
-    page this feeds.
-    """
-    from canonical_thread_handoff import find_latest_cth, is_known_cth_type
-
-    try:
-        latest = find_latest_cth(list(comments or []))
-    except Exception:
-        return None
-    if not latest:
-        return None
-    fields = latest.get("fields") or {}
-    cth_type = str(latest.get("cth_type") or "").strip()
-    known = is_known_cth_type(cth_type)
-    try:
-        comment_id: int | None = int(latest.get("comment_id"))
-    except (TypeError, ValueError):
-        comment_id = None
-    return HandoffSummary(
-        comment_id=comment_id,
-        created_at=_redact(latest.get("created_at")),
-        author=_redact(latest.get("author")),
-        # An unrecognised heading is reported as such rather than republished:
-        # the heading is free text, and CTH_TYPES is the only authority for what
-        # a handoff type may be.
-        cth_type=cth_type if known else "unrecognized",
-        cth_type_known=known,
-        status=_redact(fields.get("status")),
-        next_owner=_redact(fields.get("next owner")),
-        current_blocker=_redact(fields.get("current blocker")),
-        decision=_redact(fields.get("decision")),
-        next_action=_redact(fields.get("next action")),
-    )
-
-
-CommentSource = Callable[[str, int], list[dict[str, Any]]]
-
-
-def build_comment_source(host: str, org: str, repo: str) -> CommentSource | None:
-    """Build an authenticated ``(kind, number) -> comments`` fetcher, or None.
-
-    Returns None when the console is running in offline test mode or when no
-    credential is available for *host*, so the caller reports the handoff source
-    as unavailable instead of as an empty thread.
-    """
-    if _offline_test_mode() or not (host and org and repo):
-        return None
-    auth = get_auth_header(host)
-    if not auth:
-        return None
-
-    def _fetch(kind: str, number: int) -> list[dict[str, Any]]:
-        segment = "pulls" if kind == "pr" else "issues"
-        url = f"{repo_api_url(host, org, repo)}/{segment}/{int(number)}/comments"
-        comments: list[dict[str, Any]] = []
-        page = 1
-        while page <= 20:
-            raw, meta = api_fetch_page(url, auth, page=page, limit=50)
-            comments.extend(raw)
-            if bool(meta["is_final_page"]):
-                break
-            page += 1
-        return comments
-
-    return _fetch
-
-
-# --- Snapshot ----------------------------------------------------------------
-
-
-@dataclass(frozen=True)
-class LinkageNode:
-    """One issue or PR row with its resolved links and display metadata."""
-
-    kind: str
-    number: int
-    title: str
-    state: str
-    labels: tuple[str, ...] = ()
-    links: tuple[IssueLink, ...] = ()
-    linked_prs: tuple[int, ...] = ()
-    ambiguous: bool = False
-    contested: bool = False
-    deep_link: str | None = None
-    handoff: HandoffSummary | None = None
-    handoff_status: HandoffStatus = HandoffStatus(HANDOFF_NOT_LOADED)
-    links_authoritative: bool = True
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "kind": self.kind,
-            "number": self.number,
-            "title": self.title,
-            "state": self.state,
-            "labels": list(self.labels),
-            "links": [link.to_dict() for link in self.links],
-            "linked_prs": list(self.linked_prs),
-            "ambiguous": self.ambiguous,
-            "contested": self.contested,
-            "deep_link": self.deep_link,
-            "links_authoritative": self.links_authoritative,
-            "handoff": self.handoff.to_dict() if self.handoff else None,
-            "handoff_status": self.handoff_status.to_dict(),
-        }
-
-
-@dataclass(frozen=True)
-class LinkageSnapshot:
-    """One answered linkage query over a scoped window of a Gitea repo."""
-
-    ok: bool
-    project_id: str
-    repo_label: str
-    host: str
-    state_scope: str
-    issues: tuple[LinkageNode, ...] = ()
-    prs: tuple[LinkageNode, ...] = ()
-    contested_issues: tuple[int, ...] = ()
-    focus: tuple[str, int] | None = None
-    inventory_complete: bool = False
-    deep_links_enabled: bool = False
-    handoff_status: HandoffStatus = HandoffStatus(HANDOFF_NOT_LOADED)
-    fetch_error: str | None = None
-
-    @property
-    def orphan_prs(self) -> tuple[LinkageNode, ...]:
-        """PRs in the loaded window that point at no issue at all."""
-        return tuple(node for node in self.prs if not node.links)
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "ok": self.ok,
-            "schema_version": LINKAGE_SCHEMA_VERSION,
-            "project_id": self.project_id,
-            "repo": self.repo_label,
-            "state_scope": self.state_scope,
-            "inventory_complete": self.inventory_complete,
-            "deep_links_enabled": self.deep_links_enabled,
-            "focus": (
-                None
-                if self.focus is None
-                else {"kind": self.focus[0], "number": self.focus[1]}
-            ),
-            "handoff_source": self.handoff_status.to_dict(),
-            "fetch_error": self.fetch_error,
-            "contested_issues": list(self.contested_issues),
-            "issues": [node.to_dict() for node in self.issues],
-            "prs": [node.to_dict() for node in self.prs],
-            "evidence_kinds": [
-                {"name": name, "description": EVIDENCE_DESCRIPTIONS[name]}
-                for name in EVIDENCE_ORDER
-            ],
-        }
-
-
-def snapshot_to_dict(snapshot: LinkageSnapshot) -> dict[str, Any]:
-    """JSON-serializable export for ``/api/v1/gitea/linkage``."""
-    return snapshot.to_dict()
-
-
-def _offline_test_mode() -> bool:
-    return (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip().lower() in {
-        "1",
-        "true",
-        "yes",
-    }
-
-
-def _labels_of(item: dict[str, Any]) -> tuple[str, ...]:
-    return tuple(
-        str(_redact(label.get("name")))
-        for label in (item.get("labels") or [])
-        if label.get("name")
-    )
-
-
-def _failed_snapshot(
-    *,
-    project_id: str,
-    repo_label: str,
-    host: str,
-    state_scope: str,
-    reason: str,
-) -> LinkageSnapshot:
-    """A read that could not be answered. Never an empty-and-healthy snapshot."""
-    return LinkageSnapshot(
-        ok=False,
-        project_id=project_id,
-        repo_label=repo_label,
-        host=host,
-        state_scope=state_scope,
-        inventory_complete=False,
-        deep_links_enabled=deep_links_enabled(),
-        handoff_status=HandoffStatus(
-            HANDOFF_UNAVAILABLE, reason="linkage inventory could not be loaded"
-        ),
-        fetch_error=str(_redact(reason)),
-    )
-
-
-def _resolve_project(project_id: str | None) -> ProjectRecord | None:
-    registry = load_registry()
-    if project_id:
-        for entry in registry.projects:
-            if entry.id == project_id:
-                return entry
-        return None
-    return registry.projects[0] if registry.projects else None
-
-
-def _normalize_state(state: str | None) -> str:
-    text = (state or STATE_OPEN).strip().lower()
-    return text if text in _SUPPORTED_STATES else STATE_OPEN
-
-
-def load_linkage_snapshot(
-    project_id: str | None = None,
-    *,
-    state: str | None = None,
-    issue: int | None = None,
-    pr: int | None = None,
-    fetch_prs: Callable[..., tuple[list[dict], PaginationMeta]] | None = None,
-    fetch_issues: Callable[..., tuple[list[dict], PaginationMeta]] | None = None,
-    comment_source: CommentSource | None = None,
-) -> LinkageSnapshot:
-    """Load issue↔PR linkage for a registry project.
-
-    ``issue``/``pr`` focus one thread: the focused row is the only one whose
-    Canonical Thread Handoff comments are fetched, because handoff comments are
-    thread-scoped and loading them for a whole queue would be one request per
-    row. Every unfocused row reports its handoff as ``not_loaded``.
-    """
-    state_scope = _normalize_state(state)
-    try:
-        project = _resolve_project(project_id)
-    except Exception as exc:  # registry invalid — fail closed with the reason
-        return _failed_snapshot(
-            project_id=project_id or "",
-            repo_label="",
-            host="",
-            state_scope=state_scope,
-            reason=f"project registry unavailable: {exc}",
-        )
-
-    if project is None:
-        return _failed_snapshot(
-            project_id=project_id or "",
-            repo_label="",
-            host="",
-            state_scope=state_scope,
-            reason=(
-                f"project {project_id!r} not found in registry"
-                if project_id
-                else "no projects registered"
-            ),
-        )
-
-    host = _host_from_url(project.remote_host)
-    repo_label = f"{project.gitea_owner}/{project.repo_name}"
-    offline_test = _offline_test_mode()
-
-    def _empty_fetch(*_args, **_kwargs):
-        return [], PaginationMeta(
-            page=1,
-            per_page=50,
-            returned_count=0,
-            has_more=False,
-            is_final_page=True,
-            # An offline stub loaded nothing; claiming a complete inventory here
-            # would let the page assert that no issue has a linked PR.
-            inventory_complete=False,
-            pages_fetched=0,
-        )
-
-    pr_fetch = fetch_prs or (_empty_fetch if offline_test else _fetch_prs)
-    issue_fetch = fetch_issues or (_empty_fetch if offline_test else _fetch_issues)
-    using_live_fetch = not offline_test and (fetch_prs is None or fetch_issues is None)
-    auth = get_auth_header(host) if using_live_fetch else "test-auth"
-    if using_live_fetch and not auth:
-        return _failed_snapshot(
-            project_id=project.id,
-            repo_label=repo_label,
-            host=host,
-            state_scope=state_scope,
-            reason=(
-                f"Gitea credentials unavailable for {host}; linkage cannot be "
-                "loaded (fail closed — not rendering an empty linkage table)"
-            ),
-        )
-
-    try:
-        raw_prs, pr_pagination = pr_fetch(
-            host, project.gitea_owner, project.repo_name, auth, state=state_scope
-        )
-        raw_issues, issue_pagination = issue_fetch(
-            host, project.gitea_owner, project.repo_name, auth, state=state_scope
-        )
-    except Exception as exc:  # noqa: BLE001 — operator-visible fetch failure
-        return _failed_snapshot(
-            project_id=project.id,
-            repo_label=repo_label,
-            host=host,
-            state_scope=state_scope,
-            reason=f"Gitea fetch failed: {exc}",
-        )
-
-    inventory_complete = bool(
-        getattr(pr_pagination, "inventory_complete", False)
-        and getattr(issue_pagination, "inventory_complete", False)
-    )
-
-    index = resolve_linkage(raw_prs)
-    contested = index.contested_issues()
-
-    focus: tuple[str, int] | None = None
-    if pr is not None:
-        focus = ("pr", int(pr))
-    elif issue is not None:
-        focus = ("issue", int(issue))
-
-    unfocused_reason = (
-        "canonical handoff comments are thread-scoped; focus one issue or PR "
-        "to load its latest handoff"
-    )
-    handoff_status = HandoffStatus(HANDOFF_NOT_LOADED, reason=unfocused_reason)
-    focus_handoff: HandoffSummary | None = None
-    if focus is not None:
-        source = comment_source
-        if source is None and not offline_test:
-            source = build_comment_source(host, project.gitea_owner, project.repo_name)
-        target = f"{focus[0]}#{focus[1]}"
-        if source is None:
-            handoff_status = HandoffStatus(
-                HANDOFF_UNAVAILABLE,
-                reason="no authenticated comment source available for this read",
-                target=target,
-            )
-        else:
-            try:
-                focus_handoff = summarize_handoff(source(focus[0], focus[1]) or [])
-                handoff_status = HandoffStatus(HANDOFF_LOADED, target=target)
-            except Exception as exc:  # fail soft: degrade this source only
-                handoff_status = HandoffStatus(
-                    HANDOFF_UNAVAILABLE,
-                    reason=str(_redact(f"handoff fetch failed: {exc}")),
-                    target=target,
-                )
-
-    def _node_handoff(
-        kind: str, number: int
-    ) -> tuple[HandoffSummary | None, HandoffStatus]:
-        """Attach the handoff only to the focused row; qualify every other row."""
-        if focus == (kind, number):
-            return (focus_handoff, handoff_status)
-        return (
-            None,
-            HandoffStatus(
-                HANDOFF_NOT_LOADED,
-                reason=unfocused_reason if focus is None else "not the focused thread",
-            ),
-        )
-
-    def _number_of(raw: dict[str, Any]) -> int | None:
-        try:
-            return int(raw["number"])
-        except (KeyError, TypeError, ValueError):
-            return None
-
-    def _sort_key(raw: dict[str, Any]) -> int:
-        number = _number_of(raw)
-        return -1 if number is None else number
-
-    issue_nodes: list[LinkageNode] = []
-    for raw in sorted(raw_issues or [], key=_sort_key, reverse=True):
-        number = _number_of(raw)
-        if number is None:
-            continue
-        node_handoff, node_status = _node_handoff("issue", number)
-        linked_prs = index.issue_prs.get(number, ())
-        issue_nodes.append(
-            LinkageNode(
-                kind="issue",
-                number=number,
-                title=str(_redact(raw.get("title")) or ""),
-                state=str(raw.get("state") or ""),
-                labels=_labels_of(raw),
-                linked_prs=linked_prs,
-                contested=len(linked_prs) > 1,
-                deep_link=_deep_link(
-                    host, project.gitea_owner, project.repo_name, "issue", number
-                ),
-                handoff=node_handoff,
-                handoff_status=node_status,
-                links_authoritative=inventory_complete,
-            )
-        )
-
-    pr_nodes: list[LinkageNode] = []
-    for raw in sorted(raw_prs or [], key=_sort_key, reverse=True):
-        number = _number_of(raw)
-        if number is None:
-            continue
-        node_handoff, node_status = _node_handoff("pr", number)
-        links = index.pr_links.get(number, ())
-        pr_nodes.append(
-            LinkageNode(
-                kind="pr",
-                number=number,
-                title=str(_redact(raw.get("title")) or ""),
-                state=str(raw.get("state") or ""),
-                labels=_labels_of(raw),
-                links=links,
-                ambiguous=index.ambiguous(number),
-                contested=any(link.issue_number in contested for link in links),
-                deep_link=_deep_link(
-                    host, project.gitea_owner, project.repo_name, "pr", number
-                ),
-                handoff=node_handoff,
-                handoff_status=node_status,
-                links_authoritative=inventory_complete,
-            )
-        )
-
-    return LinkageSnapshot(
-        ok=True,
-        project_id=project.id,
-        repo_label=repo_label,
-        host=host,
-        state_scope=state_scope,
-        issues=tuple(issue_nodes),
-        prs=tuple(pr_nodes),
-        contested_issues=contested,
-        focus=focus,
-        inventory_complete=inventory_complete,
-        deep_links_enabled=deep_links_enabled(),
-        handoff_status=handoff_status,
-    )

webui/linkage_views.py

@@ -1,364 +0,0 @@
-"""HTML views for the Gitea issue↔PR linkage console (#645, Phase 3).
-
-Read-only renderer over :mod:`webui.linkage_loader`. The page's job is to make
-three things impossible to misread:
-
-* **why** an edge exists — every link carries its evidence badge, so a bare
-  ``#N`` mention never looks like a closing claim;
-* **what was not loaded** — a partial inventory, an unfocused thread, or an
-  unavailable handoff source renders as an explicit qualifier, never as an
-  affirmative "none";
-* **that nothing here mutates** — there is no review, merge, or edit control,
-  and the deep link out to Gitea appears only under the admin reveal opt-in.
-"""
-
-from __future__ import annotations
-
-from html import escape
-from typing import Sequence
-
-from webui.layout import render_page
-from webui.linkage_loader import (
-    EVIDENCE_BRANCH,
-    EVIDENCE_CLOSES,
-    EVIDENCE_DESCRIPTIONS,
-    EVIDENCE_ORDER,
-    EVIDENCE_REFERENCE,
-    HANDOFF_LOADED,
-    HANDOFF_NOT_LOADED,
-    HandoffSummary,
-    LinkageNode,
-    LinkageSnapshot,
-)
-
-_EVIDENCE_CSS = {
-    EVIDENCE_CLOSES: "badge-health-ok",
-    EVIDENCE_BRANCH: "badge-health-skipped",
-    EVIDENCE_REFERENCE: "badge-health-unproven",
-}
-
-_EVIDENCE_LABEL = {
-    EVIDENCE_CLOSES: "closes",
-    EVIDENCE_BRANCH: "branch",
-    EVIDENCE_REFERENCE: "mention",
-}
-
-
-def _badge(text: str, css: str) -> str:
-    return f'<span class="badge {css}">{escape(text)}</span>'
-
-
-def _labels(names: Sequence[str]) -> str:
-    if not names:
-        return '<span class="muted">—</span>'
-    return " ".join(_badge(name, "badge-health-skipped") for name in names)
-
-
-def _ref(node: LinkageNode) -> str:
-    """Render an item reference, hyperlinked only when deep links are revealed."""
-    label = f"#{node.number}"
-    if node.deep_link:
-        return f'<a href="{escape(node.deep_link)}"><code>{escape(label)}</code></a>'
-    return f"<code>{escape(label)}</code>"
-
-
-def _scope_card(snapshot: LinkageSnapshot) -> str:
-    focus = (
-        "none"
-        if snapshot.focus is None
-        else f"{snapshot.focus[0]}#{snapshot.focus[1]}"
-    )
-    completeness = (
-        _badge("complete", "badge-health-ok")
-        if snapshot.inventory_complete
-        else _badge("partial", "badge-health-degraded")
-    )
-    links_note = (
-        "Every linkage edge below is a claim about this loaded window only."
-        if snapshot.inventory_complete
-        else (
-            "Pagination did not complete for this window, so an empty link list "
-            "means <em>none found in what was loaded</em> — not that no link exists."
-        )
-    )
-    deep_links = (
-        _badge("enabled", "badge-health-ok")
-        if snapshot.deep_links_enabled
-        else _badge("hidden", "badge-health-skipped")
-    )
-    return f"""<div class="health-card">
-  <h3>Scope</h3>
-  <table class="detail">
-    <tr><th>Project</th><td><code>{escape(snapshot.project_id or "—")}</code></td></tr>
-    <tr><th>Repository</th><td><code>{escape(snapshot.repo_label or "—")}</code></td></tr>
-    <tr><th>Item state</th><td><code>{escape(snapshot.state_scope)}</code></td></tr>
-    <tr><th>Focused thread</th><td><code>{escape(focus)}</code></td></tr>
-    <tr><th>Inventory</th><td>{completeness}</td></tr>
-    <tr><th>Gitea deep links</th><td>{deep_links}</td></tr>
-  </table>
-  <p class="muted">{links_note}</p>
-</div>"""
-
-
-def _error_card(snapshot: LinkageSnapshot) -> str:
-    if snapshot.ok and not snapshot.fetch_error:
-        return ""
-    return (
-        '<div class="health-card health-stale"><strong>Linkage unavailable:</strong> '
-        f"{escape(snapshot.fetch_error or 'the linkage read did not complete')}. "
-        "No linkage table is rendered: an empty table would read as "
-        "<em>no issue is linked to any PR</em>, which this read cannot claim."
-        "</div>"
-    )
-
-
-def _contested_card(snapshot: LinkageSnapshot) -> str:
-    if not snapshot.contested_issues:
-        return ""
-    refs = ", ".join(f"<code>#{number}</code>" for number in snapshot.contested_issues)
-    return (
-        '<div class="health-card health-stale">'
-        f"<strong>Contested issues:</strong> {refs}. More than one PR in this "
-        "window claims each of these — duplicate work or a superseded PR. "
-        "Resolution stays in Gitea and the workflow; this console only reports it."
-        "</div>"
-    )
-
-
-def _evidence_badges(evidence: Sequence[str]) -> str:
-    return " ".join(
-        _badge(
-            _EVIDENCE_LABEL.get(name, name),
-            _EVIDENCE_CSS.get(name, "badge-health-skipped"),
-        )
-        for name in EVIDENCE_ORDER
-        if name in evidence
-    )
-
-
-def _handoff_inline(handoff: HandoffSummary) -> str:
-    type_css = "badge-health-ok" if handoff.cth_type_known else "badge-health-degraded"
-    return (
-        f'{_badge(handoff.cth_type or "—", type_css)}'
-        f'<div class="muted" style="font-size:0.82rem;">'
-        f'{escape(handoff.status or "—")} → {escape(handoff.next_owner or "—")}</div>'
-    )
-
-
-def _handoff_cell(node: LinkageNode) -> str:
-    """Render the handoff column, distinguishing 'none found' from 'not loaded'."""
-    status = node.handoff_status
-    if status.state == HANDOFF_LOADED:
-        if node.handoff is None:
-            return '<span class="muted">no canonical handoff on this thread</span>'
-        return _handoff_inline(node.handoff)
-    if status.state == HANDOFF_NOT_LOADED:
-        return (
-            f'{_badge("not loaded", "badge-health-skipped")}'
-            f'<div class="muted" style="font-size:0.82rem;">'
-            f'{escape(status.reason or "")}</div>'
-        )
-    return (
-        f'{_badge("unavailable", "badge-health-degraded")}'
-        f'<div class="muted" style="font-size:0.82rem;">'
-        f'{escape(status.reason or "")}</div>'
-    )
-
-
-def _issue_rows(snapshot: LinkageSnapshot) -> str:
-    rows = []
-    for node in snapshot.issues:
-        if node.linked_prs:
-            linked = ", ".join(f"<code>#{number}</code>" for number in node.linked_prs)
-            if node.contested:
-                linked += " " + _badge("contested", "badge-blocked")
-        elif node.links_authoritative:
-            linked = '<span class="muted">none</span>'
-        else:
-            # The distinction an operator needs: nothing found in a window that
-            # was not fully loaded is not the same as nothing existing.
-            linked = '<span class="muted">none found (partial inventory)</span>'
-        rows.append(
-            "<tr>"
-            f"<td>{_ref(node)}</td>"
-            f"<td>{escape(node.title)}</td>"
-            f"<td>{escape(node.state or '—')}</td>"
-            f"<td>{_labels(node.labels)}</td>"
-            f"<td>{linked}</td>"
-            f"<td>{_handoff_cell(node)}</td>"
-            "</tr>"
-        )
-    if not rows:
-        return '<tr><td colspan="6" class="muted">No issues in the loaded window.</td></tr>'
-    return "".join(rows)
-
-
-def _pr_rows(snapshot: LinkageSnapshot) -> str:
-    rows = []
-    for node in snapshot.prs:
-        if node.links:
-            linked = "".join(
-                f"<div><code>#{link.issue_number}</code> "
-                f"{_evidence_badges(link.evidence)}</div>"
-                for link in node.links
-            )
-            if node.ambiguous:
-                linked += _badge("ambiguous", "badge-blocked")
-            if node.contested:
-                linked += " " + _badge("contested", "badge-blocked")
-        elif node.links_authoritative:
-            linked = '<span class="muted">no issue reference</span>'
-        else:
-            linked = '<span class="muted">none found (partial inventory)</span>'
-        rows.append(
-            "<tr>"
-            f"<td>{_ref(node)}</td>"
-            f"<td>{escape(node.title)}</td>"
-            f"<td>{escape(node.state or '—')}</td>"
-            f"<td>{_labels(node.labels)}</td>"
-            f"<td>{linked}</td>"
-            f"<td>{_handoff_cell(node)}</td>"
-            "</tr>"
-        )
-    if not rows:
-        return (
-            '<tr><td colspan="6" class="muted">No pull requests in the loaded '
-            "window.</td></tr>"
-        )
-    return "".join(rows)
-
-
-def _focus_card(snapshot: LinkageSnapshot) -> str:
-    """Render the focused thread's latest canonical handoff, when one was loaded."""
-    if snapshot.focus is None:
-        return f"""<div class="prompt-card">
-  <h3>Canonical handoff</h3>
-  <p class="muted">{escape(snapshot.handoff_status.reason or "")}
-  Add <code>?issue=N</code> or <code>?pr=N</code> to load the latest
-  Canonical Thread Handoff for one thread.</p>
-</div>"""
-
-    kind, number = snapshot.focus
-    target = f"{kind} #{number}"
-    if not snapshot.handoff_status.loaded:
-        return f"""<div class="prompt-card">
-  <h3>Canonical handoff — {escape(target)}</h3>
-  <p class="muted">{_badge("unavailable", "badge-health-degraded")}
-  {escape(snapshot.handoff_status.reason or "handoff source did not run")}.
-  This is not evidence that the thread carries no handoff.</p>
-</div>"""
-
-    handoff = next(
-        (
-            node.handoff
-            for node in (snapshot.issues + snapshot.prs)
-            if node.kind == kind and node.number == number and node.handoff
-        ),
-        None,
-    )
-    if handoff is None:
-        return f"""<div class="prompt-card">
-  <h3>Canonical handoff — {escape(target)}</h3>
-  <p class="muted">Comments loaded; no Canonical Thread Handoff comment found on
-  this thread.</p>
-</div>"""
-
-    type_css = "badge-health-ok" if handoff.cth_type_known else "badge-health-degraded"
-    unknown_note = (
-        ""
-        if handoff.cth_type_known
-        else (
-            '<p class="muted">The comment\'s heading is not a declared CTH type, '
-            "so it is reported as unrecognized rather than republished.</p>"
-        )
-    )
-    return f"""<div class="prompt-card">
-  <h3>Canonical handoff — {escape(target)}</h3>
-  <p class="meta">{_badge(handoff.cth_type or "—", type_css)}
-  by <code>{escape(handoff.author or "unknown")}</code>
-  at <code>{escape(handoff.created_at or "unknown")}</code></p>
-  {unknown_note}
-  <table class="detail">
-    <tr><th>Status</th><td>{escape(handoff.status or "—")}</td></tr>
-    <tr><th>Next owner</th><td>{escape(handoff.next_owner or "—")}</td></tr>
-    <tr><th>Current blocker</th><td>{escape(handoff.current_blocker or "—")}</td></tr>
-    <tr><th>Decision</th><td>{escape(handoff.decision or "—")}</td></tr>
-    <tr><th>Next action</th><td>{escape(handoff.next_action or "—")}</td></tr>
-  </table>
-  <p class="muted">Full event history:
-  <a href="/api/v1/timeline?{escape(kind)}={number}"><code>/api/v1/timeline</code></a></p>
-</div>"""
-
-
-def _legend_card(snapshot: LinkageSnapshot) -> str:
-    items = "".join(
-        f"<li>{_badge(_EVIDENCE_LABEL[name], _EVIDENCE_CSS[name])} — "
-        f"{escape(EVIDENCE_DESCRIPTIONS[name])}</li>"
-        for name in EVIDENCE_ORDER
-    )
-    reveal_note = (
-        "Gitea deep links are shown because the "
-        "<code>GITEA_MCP_REVEAL_ENDPOINTS</code> admin opt-in is set."
-        if snapshot.deep_links_enabled
-        else (
-            "Gitea deep links are withheld. Set "
-            "<code>GITEA_MCP_REVEAL_ENDPOINTS=1</code> server-side to reveal "
-            "them; item numbers stay usable without them."
-        )
-    )
-    return f"""<div class="prompt-card">
-  <h3>How an edge was found</h3>
-  <ul class="reasons">{items}</ul>
-  <p class="muted">{reveal_note}</p>
-  <p class="muted">Read-only surface: no issue or PR editing, no review, and no merge.
-  JSON export: <a href="/api/v1/gitea/linkage"><code>/api/v1/gitea/linkage</code></a></p>
-</div>"""
-
-
-def render_linkage_page(snapshot: LinkageSnapshot) -> str:
-    """Render the full HTML page for the Gitea linkage console."""
-    if not snapshot.ok:
-        return render_page(
-            title="Gitea linkage",
-            body_html=f"""<h2>Gitea issue and PR linkage</h2>
-<p class="meta">Phase 3 read-only linkage console (#645).</p>
-{_error_card(snapshot)}
-{_scope_card(snapshot)}""",
-        )
-
-    body = f"""<h2>Gitea issue and PR linkage</h2>
-<p class="meta">Phase 3 read-only linkage console (#645). Gitea remains the source of
-truth; this page reads it and never writes to it.</p>
-{_scope_card(snapshot)}
-{_contested_card(snapshot)}
-
-<div class="prompt-card">
-  <h3>Issues → pull requests</h3>
-  <table class="registry">
-    <thead>
-      <tr>
-        <th>Issue</th><th>Title</th><th>State</th><th>Labels</th>
-        <th>Linked PRs</th><th>Latest handoff</th>
-      </tr>
-    </thead>
-    <tbody>{_issue_rows(snapshot)}</tbody>
-  </table>
-</div>
-
-<div class="prompt-card">
-  <h3>Pull requests → issues</h3>
-  <table class="registry">
-    <thead>
-      <tr>
-        <th>PR</th><th>Title</th><th>State</th><th>Labels</th>
-        <th>Linked issues</th><th>Latest handoff</th>
-      </tr>
-    </thead>
-    <tbody>{_pr_rows(snapshot)}</tbody>
-  </table>
-</div>
-
-{_focus_card(snapshot)}
-{_legend_card(snapshot)}
-"""
-    return render_page(title="Gitea linkage", body_html=body)

webui/nav.py

@@ -4,11 +4,11 @@ Single source of truth for the console navigation so ``webui/layout.py`` and
 the ``webui/app.py`` route table stay aligned with epic #631. Read-only: every
 destination is a GET view or a Phase 1 placeholder. No mutation links.
 
-Nav groups follow the #631 information architecture: Health, Traffic,
+Nav groups follow the #631 Phase 1 information architecture: Health, Traffic,
 Runtime/Sessions, Projects, Inventory, Timeline, Policy (placeholder), and
-Gitea linkage (#645) plus Phase 4 Insights/Providers (#650). Later-phase
+Insights (placeholder). Later-phase surfaces are declared as ``stub`` items and
-surfaces are declared as ``stub`` items and backed by ``STUB_PAGES`` so their
+backed by ``STUB_PAGES`` so their nav links resolve to a graceful placeholder
-nav links resolve to a graceful placeholder instead of a 404.
+instead of a 404.
 """
 
 from __future__ import annotations
@@ -45,6 +45,7 @@ NAV_GROUPS: tuple[NavGroup, ...] = (
         NavItem("/queue", "Queue"),
         NavItem("/leases", "Leases"),
         NavItem("/actions", "Actions"),
+        NavItem("/requests", "Requests"),
     )),
     NavGroup("Runtime/Sessions", (
         NavItem("/runtime", "Runtime health"),
@@ -60,16 +61,12 @@ 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"),
     )),
     NavGroup("Insights", (
-        NavItem("/insights", "Insights"),
+        NavItem("/insights", "Insights", "stub"),
-        NavItem("/providers", "Providers"),
         NavItem("/analytics", "Analytics"),
         NavItem("/audit", "Audit"),
     )),
@@ -93,6 +90,11 @@ STUB_PAGES: dict[str, tuple[str, str]] = {
         "Policy",
         "Capability and role policy surface. Placeholder until a later phase.",
     ),
+    "/insights": (
+        "Insights",
+        "Aggregate operational insights and trends. Placeholder until a later "
+        "phase.",
+    ),
 }
 
 

webui/queue_loader.py

@@ -211,19 +211,6 @@ def _pagination_from_pages(
     )
 
 
-_SUPPORTED_FETCH_STATES = ("open", "closed", "all")
-
-
-def _safe_state(state: str | None) -> str:
-    """Constrain a caller-supplied item state before it reaches a query string.
-
-    The value is interpolated into the Gitea URL, so an unrecognised state falls
-    back to ``open`` rather than being passed through.
-    """
-    text = (state or "open").strip().lower()
-    return text if text in _SUPPORTED_FETCH_STATES else "open"
-
-
 def _fetch_prs(
     host: str,
     org: str,
@@ -231,15 +218,8 @@ def _fetch_prs(
     auth: str,
     *,
     per_page: int = 50,
-    state: str = "open",
 ) -> tuple[list[dict], PaginationMeta]:
-    """Fetch PRs in *state* (``open``, ``closed``, or ``all``).
+    url = f"{repo_api_url(host, org, repo)}/pulls?state=open"
-
-    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
@@ -268,13 +248,8 @@ def _fetch_issues(
     auth: str,
     *,
     per_page: int = 50,
-    state: str = "open",
 ) -> tuple[list[dict], PaginationMeta]:
-    """Fetch issues in *state* (``open``, ``closed``, or ``all``); see :func:`_fetch_prs`."""
+    url = f"{repo_api_url(host, org, repo)}/issues?state=open&type=issues"
-    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

webui/request_views.py

@@ -0,0 +1,164 @@
+"""HTML views for the operator request surface (#643).
+
+The form is deliberately a *preview* form. It has no initiate button, because
+initiating requires a confirmed POST to ``/api/v1/requests/apply`` and a stray
+form submission must not be able to produce one by accident.
+
+Nothing rendered here is trusted input: every interpolated value is escaped,
+and the page renders only values the service already produced rather than
+echoing a raw request body back.
+"""
+
+from __future__ import annotations
+
+import html
+import json
+from typing import Any
+
+from webui.layout import render_page
+from webui.request_service import (
+    REQUESTABLE_ROLES,
+    WORK_KINDS,
+    RequestError,
+    RequestPreview,
+)
+
+REQUESTS_PATH = "/requests"
+PREVIEW_API_PATH = "/api/v1/requests/preview"
+APPLY_API_PATH = "/api/v1/requests/apply"
+
+
+def _escape(text: Any) -> str:
+    return html.escape(str(text if text is not None else ""), quote=True)
+
+
+REQUEST_PAGE_STYLES = """
+<style>
+  .request-form { display: grid; gap: 0.75rem; max-width: 44rem; }
+  .request-form label { display: grid; gap: 0.25rem; font-size: 0.9rem; }
+  .request-check { margin: 0.35rem 0; }
+  .request-check .verdict-ok { color: var(--accent); }
+  .request-check .verdict-fail { color: #d14; }
+  .request-prohibited code { margin-right: 0.4rem; }
+</style>
+"""
+
+
+def _options(values: tuple[str, ...], selected: Any) -> str:
+    return "".join(
+        f"<option value='{_escape(value)}'"
+        + (" selected" if selected == value else "")
+        + f">{_escape(value)}</option>"
+        for value in values
+    )
+
+
+def _form(values: dict[str, Any] | None = None) -> str:
+    current = dict(values or {})
+    number = current.get("work_number")
+    return (
+        f"<form class='request-form' method='post' action='{REQUESTS_PATH}'>"
+        "<label>Desired role<select name='desired_role'>"
+        f"{_options(REQUESTABLE_ROLES, current.get('desired_role'))}"
+        "</select></label>"
+        "<label>Work kind<select name='work_kind'>"
+        f"{_options(WORK_KINDS, current.get('work_kind'))}"
+        "</select></label>"
+        "<label>Issue or PR number"
+        "<input type='number' name='work_number' min='1' required "
+        f"value='{_escape(number) if number else ''}'></label>"
+        "<label>Intent summary"
+        "<input type='text' name='intent_summary' maxlength='500' required "
+        f"value='{_escape(current.get('intent_summary'))}'></label>"
+        "<label>Expected head SHA <span class='muted'>(PR work only)</span>"
+        "<input type='text' name='expected_head_sha' "
+        f"value='{_escape(current.get('expected_head_sha'))}'></label>"
+        "<button type='submit' class='copy-btn'>Preview request</button>"
+        "<p class='muted meta'>Preview is read-only and creates no assignment. "
+        f"Initiating requires a confirmed POST to <code>{APPLY_API_PATH}</code>."
+        "</p>"
+        "</form>"
+    )
+
+
+def _checks_block(preview: RequestPreview) -> str:
+    rows = []
+    for check in preview.checks:
+        verdict = "PASS" if check.ok else "FAIL"
+        css = "verdict-ok" if check.ok else "verdict-fail"
+        rows.append(
+            "<li class='request-check'>"
+            f"<span class='{css}'><strong>{verdict}</strong></span> "
+            f"<code>{_escape(check.name)}</code> — {_escape(check.detail)} "
+            f"<span class='muted meta'>({_escape(check.reason_code)})</span>"
+            "</li>"
+        )
+    return "<ul>" + "".join(rows) + "</ul>"
+
+
+def _preview_block(preview: RequestPreview) -> str:
+    verdict = "AUTHORIZED" if preview.authorized else "DENIED"
+    prohibited = "".join(
+        f"<code>{_escape(action)}</code>" for action in preview.prohibited_actions
+    )
+    request = preview.request
+    evidence = json.dumps(preview.allocator_evidence, indent=2, default=str)
+    return (
+        "<h3>Intent preview</h3>"
+        f"<p><strong>{verdict}</strong> — {_escape(preview.detail)}</p>"
+        "<p class='meta'>"
+        f"Role <code>{_escape(request.desired_role)}</code> · "
+        f"{_escape(request.work_kind)} <code>{_escape(request.display_ref)}</code>"
+        f" · profile <code>{_escape(preview.required_profile)}</code> · "
+        f"namespace <code>{_escape(preview.required_namespace)}</code> · "
+        f"permission <code>{_escape(preview.required_permission)}</code>"
+        "</p>"
+        f"<p>Intent: {_escape(request.intent_summary)}</p>"
+        f"{_checks_block(preview)}"
+        f"<p><strong>Next safe action:</strong> "
+        f"{_escape(preview.next_safe_action)}</p>"
+        "<p class='request-prohibited'><strong>Prohibited for this role:</strong> "
+        + (prohibited or "<span class='muted'>none declared</span>")
+        + "</p>"
+        "<p class='muted meta'>Correlation id "
+        f"<code>{_escape(preview.correlation_id)}</code></p>"
+        "<details><summary>Allocator evidence</summary>"
+        f"<pre class='prompt-text'>{_escape(evidence)}</pre>"
+        "</details>"
+    )
+
+
+def _error_block(error: RequestError) -> str:
+    field = (
+        f"<p class='meta'>Field: <code>{_escape(error.field_name)}</code></p>"
+        if error.field_name
+        else ""
+    )
+    return (
+        "<h3>Request rejected</h3>"
+        f"<p><strong>{_escape(error.reason_code)}</strong> — "
+        f"{_escape(error.detail)}</p>{field}"
+    )
+
+
+def render_requests_page(
+    *,
+    preview: RequestPreview | None = None,
+    error: RequestError | None = None,
+    submitted: dict[str, Any] | None = None,
+) -> str:
+    """Render the request form, plus a preview or rejection when one exists."""
+    body = (
+        "<h2>Requests</h2>"
+        "<p>Submit a work request — desired role, issue or PR, and intent — "
+        "and see whether it would be authorized before anything is reserved. "
+        "Initiation goes through the allocator (#600/#613); this console never "
+        "self-selects work, never approves, and never merges.</p>"
+        + _form(submitted)
+        + (_error_block(error) if error is not None else "")
+        + (_preview_block(preview) if preview is not None else "")
+        + f"<p class='meta'><a href='{PREVIEW_API_PATH}'>Preview API</a> · "
+        "<a href='/api/console/security-model'>RBAC model</a></p>"
+        + REQUEST_PAGE_STYLES
+    )
+    return render_page(title="Requests", body_html=body)

webui/traffic_loader.py

@@ -201,6 +201,16 @@ def _candidates_from_queue_snapshot(q_snap: QueueSnapshot) -> list[WorkCandidate
     return candidates
 
 
+def candidates_from_queue_snapshot(q_snap: QueueSnapshot) -> list[WorkCandidate]:
+    """Public alias for :func:`_candidates_from_queue_snapshot` (#643).
+
+    The request-initiation service ranks the same candidate set this view
+    renders, so both must agree on how a queue row becomes a candidate. One
+    construction, two callers — not two that can drift apart.
+    """
+    return _candidates_from_queue_snapshot(q_snap)
+
+
 def _claim_lease_records(inventory: dict[str, Any] | None) -> list[dict[str, Any]]:
     """Normalize ``build_claim_inventory`` entries into lease records.