diff --git a/allocator_service.py b/allocator_service.py
index 9f32fd9..a085d25 100644
--- a/allocator_service.py
+++ b/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]}"
- try:
- db.upsert_session(
- session_id=session_id,
- role=role_norm,
- profile=profile_name,
- pid=os.getpid(),
- controller_instance_id=controller_instance_id,
- )
- except Exception as exc: # noqa: BLE001 — surface structured
+ # A side-effect-free run may never reserve: reserving is a write, and the
+ # flag is the caller's assertion that this call writes nothing (#643).
+ if side_effect_free and apply:
return {
"success": False,
"outcome": OUTCOME_NO_SAFE,
+ "apply": True,
"reasons": [
- f"failed to register session in control-plane DB: {exc} "
- "(fail closed, #613)"
+ "side_effect_free is incompatible with apply=True; an "
+ "assignment is a write (fail closed, #643)"
],
"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",
- }
+ session_id = (session_id or "").strip() or f"alloc-{uuid.uuid4().hex[:12]}"
+ if not side_effect_free:
+ try:
+ db.upsert_session(
+ session_id=session_id,
+ role=role_norm,
+ profile=profile_name,
+ pid=os.getpid(),
+ controller_instance_id=controller_instance_id,
+ )
+ except Exception as exc: # noqa: BLE001 — surface structured
+ return {
+ "success": False,
+ "outcome": OUTCOME_NO_SAFE,
+ "reasons": [
+ f"failed to register session in control-plane DB: {exc} "
+ "(fail closed, #613)"
+ ],
+ "skipped": [],
+ "assignment": None,
+ "substrate": "control_plane_db",
+ }
+
+ # Expire stale leases globally before selection.
+ try:
+ db.expire_stale_leases()
+ except Exception as exc: # noqa: BLE001
+ return {
+ "success": False,
+ "outcome": OUTCOME_NO_SAFE,
+ "reasons": [f"lease expiry failed: {exc} (fail closed)"],
+ "skipped": [],
+ "assignment": None,
+ "substrate": "control_plane_db",
+ }
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)
diff --git a/docs/webui-authz-audit.md b/docs/webui-authz-audit.md
index 2dcffac..ff9cc38 100644
--- a/docs/webui-authz-audit.md
+++ b/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
-`execution_enabled: false` while the console is in Phase 1, so no caller can
-read an allow as permission to mutate.
+There is no implicit allow branch.
+
+`execution_enabled` on the decision reports whether the action has a live
+execution path at all, and is computed by `execution_wired(action)`. There are
+exactly two ways to be wired:
+
+1. the action's `phase` is at or below `ACTIVE_PHASE`; or
+2. the action declares an `execution_env_flag` **and** that variable is set.
+
+Every action that declares no flag therefore reports `execution_enabled: false`
+while the console is in Phase 1, so no caller can read an allow as permission
+to mutate. The per-action flag exists because raising `ACTIVE_PHASE` would
+enable execution for every action of that phase at once, including ones whose
+execution path is not implemented. One implemented action goes live on its own
+flag instead of dragging its unimplemented phase-mates with it.
+
+`initiate_workflow` is the only action that currently declares a flag
+(`WEBUI_REQUESTS_EXECUTION`), and it stays denied until an operator sets it.
## 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
-confirmation and dual-control flow the matrix already declares, emit a
-`succeeded` or `failed` record alongside the `gitea_audit` mutation record, and
-keep `viewer` unable to reach any of it. Turning on execution without the
-confirmation flow contradicts a declared requirement and is a review failure,
-not a shortcut.
+A Phase 2 action must: use `execution_wired` rather than a private enable flag,
+implement the confirmation and dual-control flow the matrix already declares,
+emit a `succeeded` or `failed` record alongside the `gitea_audit` mutation
+record, and keep `viewer` unable to reach any of it. Turning on execution
+without the confirmation flow contradicts a declared requirement and is a
+review failure, not a shortcut.
+
+Raising `ACTIVE_PHASE` remains the way to open a whole phase at once, and is
+deliberately *not* what #643 did: an action-scoped opt-in cannot enable an
+action whose execution path nobody wrote.
## 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.
diff --git a/docs/webui-local-dev.md b/docs/webui-local-dev.md
index 3e01401..5e987ae 100644
--- a/docs/webui-local-dev.md
+++ b/docs/webui-local-dev.md
@@ -80,6 +80,8 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
| `/sessions` | Runtime and session view (#641) — health + inventory sessions/namespaces/worktrees |
| `/api/sessions` | JSON export for the runtime/session view |
| `/api/v1/sessions` | Versioned alias of `/api/sessions` |
+| `/gitea` | Gitea issue↔PR linkage console (#645) — both directions, with the evidence for each edge |
+| `/api/v1/gitea/linkage` | JSON linkage export; `502` when the read could not be answered |
| `/inventory` | Phase 1 shell stub — unified inventory (backed by #636) |
| `/timeline` | Phase 1 shell stub — workflow event timeline |
| `/policy` | Phase 1 shell stub — capability/role policy placeholder |
@@ -327,6 +329,68 @@ Honesty rules specific to this view:
The write-time redactor is a narrow denylist and is not relied on. The field
itself is kept — it is the `#630` evidence naming which daemon was killed.
+## Gitea issue/PR linkage (#645)
+
+`/gitea` is the Phase 3 read-only linkage console: which PR carries which issue,
+which issues are claimed by more than one PR, and what the latest Canonical
+Thread Handoff on a thread said. Gitea remains the source of truth — this
+surface reads it and never writes to it. There is no issue/PR editor, no review,
+and no merge control.
+
+Query parameters (all optional):
+
+| Parameter | Meaning |
+|-----------|---------|
+| `project` | Registry project id to scope the read (default: first registry entry) |
+| `state` | `open` (default) or `all`; `all` widens the window to merged/closed items, where a landed edge lives |
+| `issue=N` / `pr=N` | Focus one thread and load *its* latest canonical handoff |
+
+`GET /api/v1/gitea/linkage` returns the same model as JSON
+(`schema_version: 1`). It answers `502` when the read could not be answered, so
+an automated consumer cannot mistake a fail-closed payload for "no links exist".
+The HTML page always answers `200` and renders the reason instead — an operator
+view must show why a read failed rather than withhold the page.
+
+### How an edge is found
+
+Each edge carries the evidence that produced it, strongest first:
+
+| Evidence | Meaning |
+|----------|---------|
+| `closes_keyword` | The PR title or body declares `closes/fixes/resolves #N`. Gitea itself acts on this keyword. |
+| `branch_marker` | The PR head branch carries the canonical `(fix\|feat\|docs\|chore)/issue-N-…` marker minted by the issue lock. |
+| `body_reference` | The PR body mentions `#N` with no closing keyword. A mention is not a claim to close. |
+
+Only closing and branch-marker edges populate the **issue → PR** direction: a
+bare mention is a cross-link, and counting it as ownership would invent
+contested issues out of ordinary references. The mention stays visible on the
+**PR → issue** side, labelled as such. A PR whose two strongest edges tie is
+flagged `ambiguous`; an issue claimed by two PRs is flagged `contested`.
+
+### Honesty rules specific to this view
+
+* **A partial read never reads as an absence.** Linkage is a claim about the
+ loaded window only. When pagination did not complete, every empty edge cell
+ renders `none found (partial inventory)` rather than `none`, and the JSON
+ carries `inventory_complete: false` plus per-row `links_authoritative: false`.
+* **A failed read renders no table at all.** Missing credentials, an unknown
+ project, or a fetch error produce `ok: false` with a reason. An empty linkage
+ table would assert that no issue is linked to any PR, which such a read is not
+ in a position to claim.
+* **Handoffs are loaded, never assumed.** CTH comments are thread-scoped, so
+ only the focused issue or PR has its comments fetched. Every other row reports
+ `not_loaded` with the reason; a thread whose comments *were* loaded and carried
+ no CTH says exactly that. A comment-source failure degrades the handoff alone —
+ the linkage tables still render.
+* **Unrecognised handoff headings are reported, not republished.** A `## CTH:`
+ heading outside `CTH_TYPES` renders as `unrecognized`.
+* **Redaction precedes display.** Titles, labels, handoff fields, and error
+ reasons pass through `webui.console_redaction` before serialization, and the
+ page HTML-escapes everything it renders.
+* **Deep links are opt-in.** A link out to the Gitea web UI appears only when
+ `GITEA_MCP_REVEAL_ENDPOINTS=1` is set server-side, matching how the MCP tools
+ gate URL exposure. Item numbers stay usable without it.
+
## System-health dashboard (#639)
`/system-health` renders the same snapshot the `/api/v1/system/health` API
diff --git a/docs/webui-requests.md b/docs/webui-requests.md
new file mode 100644
index 0000000..b8fbdb2
--- /dev/null
+++ b/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).
diff --git a/tests/test_allocator_service.py b/tests/test_allocator_service.py
index 3e252a1..5c37d0d 100644
--- a/tests/test_allocator_service.py
+++ b/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()
diff --git a/tests/test_webui_gitea_linkage.py b/tests/test_webui_gitea_linkage.py
new file mode 100644
index 0000000..7997e19
--- /dev/null
+++ b/tests/test_webui_gitea_linkage.py
@@ -0,0 +1,511 @@
+"""Tests for the Gitea issue↔PR linkage console (#645, Phase 3).
+
+Covers the acceptance criteria of the issue:
+
+* AC1 — issue↔PR linkage is visible for the selected project/repo, in both
+ directions, with the evidence that produced each edge.
+* AC2 — the latest canonical handoff (CTH) is summarized for a focused thread.
+* AC3 — an external Gitea link appears only under the admin reveal opt-in.
+* AC4 — every case is driven by mocked Gitea payloads; no network.
+
+Plus the invariants this console must not violate: a partial or failed read is
+never rendered as "no link exists", an unfetched thread is never rendered as
+"no handoff", redaction happens before display, and the surface stays read-only.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from tests.webui_testclient import TestClient
+
+from canonical_thread_handoff import format_cth_body
+from webui.app import create_app
+from webui.linkage_loader import (
+ EVIDENCE_BRANCH,
+ EVIDENCE_CLOSES,
+ EVIDENCE_REFERENCE,
+ HANDOFF_LOADED,
+ HANDOFF_NOT_LOADED,
+ HANDOFF_UNAVAILABLE,
+ LinkageSnapshot,
+ load_linkage_snapshot,
+ resolve_linkage,
+ resolve_pr_links,
+ snapshot_to_dict,
+ summarize_handoff,
+)
+from webui.linkage_views import render_linkage_page
+from webui.nav import nav_hrefs
+from webui.queue_loader import PaginationMeta
+
+
+def _pagination(*, complete: bool = True, count: int = 0) -> PaginationMeta:
+ return PaginationMeta(
+ page=1,
+ per_page=50,
+ returned_count=count,
+ has_more=not complete,
+ is_final_page=complete,
+ inventory_complete=complete,
+ pages_fetched=1,
+ )
+
+
+def _pr(
+ number: int,
+ *,
+ title: str = "",
+ body: str = "",
+ head: str = "",
+ state: str = "open",
+ labels: tuple[str, ...] = (),
+) -> dict:
+ return {
+ "number": number,
+ "title": title or f"pr {number}",
+ "body": body,
+ "state": state,
+ "head": {"ref": head},
+ "labels": [{"name": name} for name in labels],
+ }
+
+
+def _issue(
+ number: int,
+ *,
+ title: str = "",
+ state: str = "open",
+ labels: tuple[str, ...] = (),
+) -> dict:
+ return {
+ "number": number,
+ "title": title or f"issue {number}",
+ "state": state,
+ "labels": [{"name": name} for name in labels],
+ }
+
+
+def _fetcher(items: list[dict], *, complete: bool = True):
+ def _fetch(*_args, **_kwargs):
+ return items, _pagination(complete=complete, count=len(items))
+
+ return _fetch
+
+
+def _load(
+ issues: list[dict],
+ prs: list[dict],
+ *,
+ complete: bool = True,
+ **kwargs,
+) -> LinkageSnapshot:
+ return load_linkage_snapshot(
+ fetch_prs=_fetcher(prs, complete=complete),
+ fetch_issues=_fetcher(issues, complete=complete),
+ **kwargs,
+ )
+
+
+def _cth(comment_id: int, *, created_at: str, status: str, next_owner: str) -> dict:
+ return {
+ "id": comment_id,
+ "created_at": created_at,
+ "user": {"login": "jcwalker3"},
+ "body": format_cth_body(
+ cth_type="Author Handoff",
+ status=status,
+ next_owner=next_owner,
+ current_blocker="none",
+ decision="implemented",
+ proof="full suite green",
+ next_action="review PR",
+ ready_to_paste_prompt="Review PR #902 now.",
+ ),
+ }
+
+
+class TestLinkageEvidence(unittest.TestCase):
+ """AC1 — every edge records how it was found, and keeps all candidates."""
+
+ def test_closes_keyword_in_body_is_strongest_evidence(self):
+ links = resolve_pr_links(_pr(902, body="Closes #643"))
+ self.assertEqual([link.issue_number for link in links], [643])
+ self.assertEqual(links[0].evidence, (EVIDENCE_CLOSES,))
+ self.assertTrue(links[0].closes)
+
+ def test_closes_keyword_in_title_counts(self):
+ links = resolve_pr_links(_pr(902, title="feat(webui): preview (Closes #643)"))
+ self.assertEqual(links[0].evidence, (EVIDENCE_CLOSES,))
+
+ def test_canonical_branch_marker_links_without_a_keyword(self):
+ links = resolve_pr_links(_pr(902, head="feat/issue-643-request-preview"))
+ self.assertEqual([link.issue_number for link in links], [643])
+ self.assertEqual(links[0].evidence, (EVIDENCE_BRANCH,))
+ self.assertFalse(links[0].closes)
+
+ def test_non_canonical_branch_is_not_treated_as_a_marker(self):
+ self.assertEqual(resolve_pr_links(_pr(902, head="issue-643-preview")), ())
+
+ def test_bare_mention_is_recorded_as_the_weakest_evidence(self):
+ links = resolve_pr_links(_pr(902, body="context in #643"))
+ self.assertEqual(links[0].evidence, (EVIDENCE_REFERENCE,))
+ self.assertFalse(links[0].closes)
+
+ def test_several_evidence_kinds_merge_onto_one_edge(self):
+ links = resolve_pr_links(
+ _pr(902, body="Closes #643 — see #643", head="feat/issue-643-preview")
+ )
+ self.assertEqual(len(links), 1)
+ self.assertEqual(
+ links[0].evidence,
+ (EVIDENCE_CLOSES, EVIDENCE_BRANCH, EVIDENCE_REFERENCE),
+ )
+
+ def test_stronger_evidence_sorts_first(self):
+ links = resolve_pr_links(_pr(902, body="Closes #643, related #700"))
+ self.assertEqual([link.issue_number for link in links], [643, 700])
+
+ def test_self_reference_is_not_linkage(self):
+ links = resolve_pr_links(_pr(902, body="supersedes #902"))
+ self.assertEqual(links, ())
+
+ def test_every_candidate_is_kept_never_collapsed_to_a_guess(self):
+ links = resolve_pr_links(_pr(902, body="Closes #643\nCloses #644"))
+ self.assertEqual([link.issue_number for link in links], [643, 644])
+
+
+class TestLinkageIndex(unittest.TestCase):
+ def test_issue_direction_ignores_mention_only_edges(self):
+ index = resolve_linkage([_pr(902, body="context in #643")])
+ self.assertIsNone(index.issue_prs.get(643))
+ self.assertEqual(index.pr_links[902][0].evidence, (EVIDENCE_REFERENCE,))
+
+ def test_contested_issue_is_reported_when_two_prs_claim_it(self):
+ index = resolve_linkage(
+ [_pr(902, body="Closes #643"), _pr(903, head="feat/issue-643-again")]
+ )
+ self.assertEqual(index.contested_issues(), (643,))
+ self.assertEqual(index.issue_prs[643], (902, 903))
+
+ def test_single_claim_is_not_contested(self):
+ index = resolve_linkage([_pr(902, body="Closes #643")])
+ self.assertEqual(index.contested_issues(), ())
+
+ def test_ambiguous_when_two_issues_tie_at_the_strongest_evidence(self):
+ index = resolve_linkage([_pr(902, body="Closes #643\nCloses #644")])
+ self.assertTrue(index.ambiguous(902))
+
+ def test_weaker_candidate_alongside_a_stronger_one_is_not_ambiguous(self):
+ index = resolve_linkage([_pr(902, body="Closes #643, see #700")])
+ self.assertFalse(index.ambiguous(902))
+ self.assertEqual(index.primary_issue(902).issue_number, 643)
+
+ def test_malformed_pr_row_is_skipped_not_raised_on(self):
+ index = resolve_linkage([{"title": "no number"}, _pr(902, body="Closes #643")])
+ self.assertEqual(sorted(index.pr_links), [902])
+
+
+class TestLinkageSnapshot(unittest.TestCase):
+ """AC1 — linkage is visible per project/repo, in both directions."""
+
+ def test_both_directions_are_populated(self):
+ snapshot = _load([_issue(643)], [_pr(902, body="Closes #643")])
+ self.assertTrue(snapshot.ok)
+ self.assertEqual([node.number for node in snapshot.issues], [643])
+ self.assertEqual(snapshot.issues[0].linked_prs, (902,))
+ self.assertEqual(snapshot.prs[0].links[0].issue_number, 643)
+
+ def test_repo_scope_comes_from_the_registry_project(self):
+ snapshot = _load([], [])
+ self.assertIn("/", snapshot.repo_label)
+ self.assertTrue(snapshot.project_id)
+
+ def test_unknown_project_fails_closed_with_a_reason(self):
+ snapshot = _load([_issue(643)], [], project_id="no-such-project")
+ self.assertFalse(snapshot.ok)
+ self.assertIn("not found in registry", snapshot.fetch_error)
+ self.assertEqual(snapshot.issues, ())
+
+ def test_orphan_pr_is_identifiable(self):
+ snapshot = _load([], [_pr(902), _pr(903, body="Closes #643")])
+ self.assertEqual([node.number for node in snapshot.orphan_prs], [902])
+
+ def test_state_scope_defaults_to_open_and_is_reported(self):
+ self.assertEqual(_load([], []).state_scope, "open")
+ self.assertEqual(_load([], [], state="all").state_scope, "all")
+
+ def test_unsupported_state_falls_back_to_open(self):
+ self.assertEqual(_load([], [], state="../etc").state_scope, "open")
+
+ def test_state_is_passed_through_to_the_fetchers(self):
+ seen: list[str] = []
+
+ def _fetch(*_args, **kwargs):
+ seen.append(kwargs.get("state", ""))
+ return [], _pagination()
+
+ load_linkage_snapshot(state="all", fetch_prs=_fetch, fetch_issues=_fetch)
+ self.assertEqual(seen, ["all", "all"])
+
+
+class TestPartialInventoryIsNotAnAbsenceClaim(unittest.TestCase):
+ """An empty edge list from a partial read must never read as 'no link'."""
+
+ def test_incomplete_pagination_marks_links_non_authoritative(self):
+ snapshot = _load([_issue(643)], [], complete=False)
+ self.assertFalse(snapshot.inventory_complete)
+ self.assertFalse(snapshot.issues[0].links_authoritative)
+
+ def test_complete_pagination_marks_links_authoritative(self):
+ snapshot = _load([_issue(643)], [], complete=True)
+ self.assertTrue(snapshot.inventory_complete)
+ self.assertTrue(snapshot.issues[0].links_authoritative)
+
+ def test_partial_window_renders_a_qualified_empty_cell(self):
+ html = render_linkage_page(_load([_issue(643)], [], complete=False))
+ self.assertIn("none found (partial inventory)", html)
+
+ def test_complete_window_renders_a_plain_none(self):
+ html = render_linkage_page(_load([_issue(643)], [], complete=True))
+ self.assertNotIn("partial inventory", html)
+ self.assertIn(">none<", html)
+
+ def test_missing_credentials_fail_closed_without_a_table(self):
+ with mock.patch(
+ "webui.linkage_loader._offline_test_mode", return_value=False
+ ), mock.patch("webui.linkage_loader.get_auth_header", return_value=""):
+ snapshot = load_linkage_snapshot()
+ self.assertFalse(snapshot.ok)
+ self.assertIn("credentials unavailable", snapshot.fetch_error)
+ html = render_linkage_page(snapshot)
+ self.assertIn("Linkage unavailable", html)
+ self.assertNotIn("Issues → pull requests", html)
+
+ def test_fetch_failure_is_reported_not_raised(self):
+ def _boom(*_args, **_kwargs):
+ raise RuntimeError("gitea 502")
+
+ snapshot = load_linkage_snapshot(fetch_prs=_boom, fetch_issues=_boom)
+ self.assertFalse(snapshot.ok)
+ self.assertIn("Gitea fetch failed", snapshot.fetch_error)
+
+
+class TestHandoffSummary(unittest.TestCase):
+ """AC2 — the latest canonical handoff is summarized for a focused thread."""
+
+ def test_latest_cth_wins(self):
+ summary = summarize_handoff([
+ _cth(1, created_at="2026-07-24T10:00:00Z", status="in progress",
+ next_owner="author"),
+ _cth(2, created_at="2026-07-25T10:00:00Z", status="PR-open",
+ next_owner="reviewer"),
+ ])
+ self.assertEqual(summary.comment_id, 2)
+ self.assertEqual(summary.status, "PR-open")
+ self.assertEqual(summary.next_owner, "reviewer")
+ self.assertTrue(summary.cth_type_known)
+
+ def test_thread_without_a_cth_summarizes_to_none(self):
+ self.assertIsNone(summarize_handoff([{"id": 1, "body": "ordinary comment"}]))
+
+ def test_unknown_heading_is_reported_not_republished(self):
+ summary = summarize_handoff([
+ {
+ "id": 5,
+ "created_at": "2026-07-25T10:00:00Z",
+ "user": {"login": "someone"},
+ "body": "\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="")], [])
+ html = render_linkage_page(snapshot)
+ self.assertNotIn("", html)
+ self.assertIn("<script>", html)
+
+
+class TestLinkageRoutes(unittest.TestCase):
+ def setUp(self):
+ self.snapshot = _load(
+ [_issue(643, labels=("status:ready",))],
+ [_pr(902, body="Closes #643", labels=("status:pr-open",))],
+ )
+ self.client = TestClient(create_app())
+
+ def test_page_renders_both_tables(self):
+ with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
+ response = self.client.get("/gitea")
+ self.assertEqual(response.status_code, 200)
+ self.assertIn("Issues → pull requests", response.text)
+ self.assertIn("Pull requests → issues", response.text)
+ self.assertIn("#643", response.text)
+
+ def test_api_exports_the_same_model(self):
+ with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
+ response = self.client.get("/api/v1/gitea/linkage")
+ self.assertEqual(response.status_code, 200)
+ payload = response.json()
+ self.assertTrue(payload["ok"])
+ self.assertEqual(payload["issues"][0]["linked_prs"], [902])
+ self.assertEqual(payload["prs"][0]["links"][0]["issue_number"], 643)
+ self.assertEqual(payload["schema_version"], 1)
+
+ def test_api_declares_the_evidence_vocabulary(self):
+ with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
+ payload = self.client.get("/api/v1/gitea/linkage").json()
+ names = {entry["name"] for entry in payload["evidence_kinds"]}
+ self.assertEqual(names, {EVIDENCE_CLOSES, EVIDENCE_BRANCH, EVIDENCE_REFERENCE})
+
+ def test_api_fails_closed_with_a_non_200_when_the_read_failed(self):
+ failed = _load([], [], project_id="no-such-project")
+ with mock.patch("webui.app.load_linkage_snapshot", return_value=failed):
+ response = self.client.get("/api/v1/gitea/linkage")
+ self.assertEqual(response.status_code, 502)
+ self.assertFalse(response.json()["ok"])
+
+ def test_page_still_renders_when_the_read_failed(self):
+ failed = _load([], [], project_id="no-such-project")
+ with mock.patch("webui.app.load_linkage_snapshot", return_value=failed):
+ response = self.client.get("/gitea")
+ self.assertEqual(response.status_code, 200)
+ self.assertIn("Linkage unavailable", response.text)
+
+ def test_query_parameters_reach_the_loader(self):
+ with mock.patch(
+ "webui.app.load_linkage_snapshot", return_value=self.snapshot
+ ) as loader:
+ self.client.get("/gitea?project=gitea-tools&state=all&pr=902")
+ loader.assert_called_once()
+ args, kwargs = loader.call_args
+ self.assertEqual(args[0], "gitea-tools")
+ self.assertEqual(kwargs["state"], "all")
+ self.assertEqual(kwargs["pr"], 902)
+ self.assertIsNone(kwargs["issue"])
+
+ def test_surface_stays_read_only(self):
+ for path in ("/gitea", "/api/v1/gitea/linkage"):
+ with self.subTest(path=path):
+ self.assertEqual(self.client.post(path).status_code, 405)
+
+ def test_nav_exposes_the_linkage_page_as_live(self):
+ self.assertIn("/gitea", nav_hrefs())
+ home = self.client.get("/").text
+ self.assertIn('href="/gitea"', home)
+ self.assertIn(">Gitea<", home)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_webui_request_initiation.py b/tests/test_webui_request_initiation.py
new file mode 100644
index 0000000..e05978c
--- /dev/null
+++ b/tests/test_webui_request_initiation.py
@@ -0,0 +1,1228 @@
+"""Request preview, authorization, and workflow initiation tests (#643).
+
+Covers each acceptance criterion:
+
+* AC1 — preview shows authorize/deny with reasons.
+* AC2 — apply creates an exclusive assignment or returns wait/blocked.
+* AC3 — duplicate assign rejected.
+* AC4 — preview / apply / deny / collision are all exercised.
+* AC5 — the UI never renders a secret, and messaging stays brief.
+
+Required tests named in the issue: allocator integration with fakes, and
+gated-action tests. The allocator is injected as a fake throughout so no test
+touches Gitea or reserves real work; one class asserts the *real* default
+allocator refuses an incomplete inventory rather than ranking a partial set.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import json
+import os
+import pathlib
+import sys
+import tempfile
+import unittest
+from typing import Any
+from unittest import mock
+
+from tests.webui_testclient import TestClient
+
+sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))
+
+import allocator_service # noqa: E402
+from webui import console_audit, console_authz, request_service # noqa: E402
+from webui.app import create_app # noqa: E402
+from webui.console_redaction import scan_for_secrets # noqa: E402
+from webui.request_views import render_requests_page # noqa: E402
+
+EXEC_FLAG = "WEBUI_REQUESTS_EXECUTION"
+
+SCOPE = {
+ "remote": "prgs",
+ "org": "Scaled-Tech-Consulting",
+ "repo": "Gitea-Tools",
+}
+
+
+def _principal(role: str) -> console_authz.Principal:
+ return console_authz.Principal(
+ subject=f"{role}@example.com",
+ role=role,
+ identity_source=console_authz.IDENTITY_ACCESS_PROXY,
+ authenticated=True,
+ )
+
+
+def _request(
+ *,
+ role: str = "author",
+ kind: str = "issue",
+ number: int = 643,
+ intent: str = "implement request preview and initiation",
+ head: str | None = None,
+) -> request_service.WorkRequest:
+ parsed, error = request_service.parse_request(
+ {
+ "desired_role": role,
+ "work_kind": kind,
+ "work_number": number,
+ "intent_summary": intent,
+ "expected_head_sha": head,
+ **SCOPE,
+ }
+ )
+ assert error is None, error
+ assert parsed is not None
+ return parsed
+
+
+def _selection(
+ *, kind: str = "issue", number: int = 643, head_sha: str | None = None
+) -> dict[str, Any]:
+ return {
+ "kind": kind,
+ "number": number,
+ "title": "Web Console: Requests, intent preview, authorization",
+ "head_sha": head_sha,
+ "selected_action": "implement",
+ "expected_role_next": "author",
+ }
+
+
+@contextlib.contextmanager
+def _patched_control_plane(
+ *,
+ release_sink: list[tuple[str, str]] | None = None,
+ release_error: Exception | None = None,
+ leases=None,
+):
+ """Stand in for ``control_plane_db`` so compensation paths are observable.
+
+ The service imports the module inside the function, so patching
+ ``sys.modules`` is what intercepts it. No real DB is opened.
+ """
+ module = mock.MagicMock()
+ db = mock.MagicMock()
+
+ def _release(lease_id, *, session_id):
+ if release_error is not None:
+ raise release_error
+ if release_sink is not None:
+ release_sink.append((lease_id, session_id))
+
+ db.release_lease.side_effect = _release
+ db.list_leases.side_effect = leases or (lambda **_kwargs: [])
+ module.ControlPlaneDB.return_value = db
+ with mock.patch.dict(sys.modules, {"control_plane_db": module}):
+ yield db
+
+
+def _drifting_allocator(*, lease_id: str | None = "lease-wrong"):
+ """An allocator that previews the requested unit but assigns another.
+
+ This is the #643 B1 race in miniature: the CAS fingerprint hashes only
+ ``{kind, number}``, so a lease taken on the requested unit inside the window
+ leaves the fingerprint identical while the selection moves on.
+ """
+ assignment: dict[str, Any] = {"assignment_id": "asn-wrong"}
+ if lease_id:
+ assignment["lease_id"] = lease_id
+
+ def _drifting(
+ *, request, apply, expected_candidate_set_fingerprint=None, session_id=None
+ ):
+ return {
+ "outcome": (
+ allocator_service.OUTCOME_ASSIGNED
+ if apply
+ else allocator_service.OUTCOME_PREVIEW
+ ),
+ "selected": _selection(number=999 if apply else 643),
+ "assignment": dict(assignment) if apply else None,
+ "candidate_set_fingerprint": "fp-test",
+ "session_id": session_id,
+ }
+
+ return _drifting
+
+
+def _fake_allocator(
+ *,
+ selection: dict[str, Any] | None = None,
+ preview_outcome: str = allocator_service.OUTCOME_PREVIEW,
+ apply_outcome: str = allocator_service.OUTCOME_ASSIGNED,
+ assignment: dict[str, Any] | None = None,
+ calls: list[dict[str, Any]] | None = None,
+):
+ """Build an allocator double that records how it was called."""
+ chosen = selection if selection is not None else _selection()
+ made = (
+ assignment
+ if assignment is not None
+ else {
+ "assignment_id": "asn-test-0001",
+ "lease_id": "lease-test-0001",
+ "session_id": "webui-request-test",
+ "expected_head_sha": chosen.get("head_sha"),
+ }
+ )
+
+ def _allocator(*, request, apply, expected_candidate_set_fingerprint=None):
+ if calls is not None:
+ calls.append(
+ {
+ "apply": apply,
+ "role": request.desired_role,
+ "fingerprint": expected_candidate_set_fingerprint,
+ }
+ )
+ return {
+ "outcome": apply_outcome if apply else preview_outcome,
+ "selected": dict(chosen),
+ "reasons": ["fake allocator"],
+ "candidate_set_fingerprint": "fp-test",
+ "candidate_count": 3,
+ "inventory_complete": True,
+ "selection_policy": allocator_service.SELECTION_POLICY,
+ "substrate": "control_plane_db",
+ "assignment": dict(made) if apply else None,
+ }
+
+ return _allocator
+
+
+def _no_claims(_request):
+ return {}
+
+
+def _claimed(role: str = "author"):
+ def _source(request):
+ return {
+ request.work_key: {
+ "lease_id": "lease-foreign-9999",
+ "session_id": "prgs-author-999-foreign",
+ "role": role,
+ "expires_at": "2026-07-25T09:10:37Z",
+ }
+ }
+
+ return _source
+
+
+class TestRequestParsing(unittest.TestCase):
+ """The request model rejects rather than guesses."""
+
+ def test_valid_request_round_trips(self):
+ req = _request()
+ self.assertEqual(req.work_key, ("issue", 643))
+ self.assertEqual(req.display_ref, "#643")
+ self.assertEqual(req.to_dict()["desired_role"], "author")
+
+ def test_unknown_role_rejected(self):
+ parsed, error = request_service.parse_request(
+ {
+ "desired_role": "admin",
+ "work_kind": "issue",
+ "work_number": 1,
+ "intent_summary": "x",
+ **SCOPE,
+ }
+ )
+ self.assertIsNone(parsed)
+ self.assertEqual(error.reason_code, "unknown_role")
+ self.assertEqual(error.field_name, "desired_role")
+
+ def test_unknown_work_kind_rejected(self):
+ parsed, error = request_service.parse_request(
+ {
+ "desired_role": "author",
+ "work_kind": "branch",
+ "work_number": 1,
+ "intent_summary": "x",
+ **SCOPE,
+ }
+ )
+ self.assertIsNone(parsed)
+ self.assertEqual(error.reason_code, "unknown_work_kind")
+
+ def test_non_positive_number_rejected(self):
+ for value in (0, -3):
+ with self.subTest(value=value):
+ parsed, error = request_service.parse_request(
+ {
+ "desired_role": "author",
+ "work_kind": "issue",
+ "work_number": value,
+ "intent_summary": "x",
+ **SCOPE,
+ }
+ )
+ self.assertIsNone(parsed)
+ self.assertEqual(error.reason_code, "invalid_work_number")
+
+ def test_missing_intent_rejected(self):
+ parsed, error = request_service.parse_request(
+ {
+ "desired_role": "author",
+ "work_kind": "issue",
+ "work_number": 1,
+ **SCOPE,
+ }
+ )
+ self.assertIsNone(parsed)
+ self.assertEqual(error.reason_code, "missing_intent")
+
+ def test_intent_is_bounded(self):
+ req = _request(intent="x" * 5000)
+ self.assertEqual(len(req.intent_summary), request_service.MAX_INTENT_CHARS)
+
+ def test_unresolved_scope_rejected(self):
+ parsed, error = request_service.parse_request(
+ {
+ "desired_role": "author",
+ "work_kind": "issue",
+ "work_number": 1,
+ "intent_summary": "x",
+ }
+ )
+ self.assertIsNone(parsed)
+ self.assertEqual(error.reason_code, "scope_unresolved")
+
+ def test_default_scope_fills_missing_fields(self):
+ parsed, error = request_service.parse_request(
+ {
+ "desired_role": "author",
+ "work_kind": "issue",
+ "work_number": 7,
+ "intent_summary": "x",
+ },
+ default_scope=SCOPE,
+ )
+ self.assertIsNone(error)
+ self.assertEqual(parsed.repo, "Gitea-Tools")
+
+
+class TestPreviewAuthorizeDeny(unittest.TestCase):
+ """AC1 — preview shows authorize/deny with reasons."""
+
+ def test_authorized_preview_names_every_check(self):
+ preview = request_service.preview_request(
+ _request(),
+ principal=_principal(console_authz.OPERATOR),
+ allocator=_fake_allocator(),
+ claims_source=_no_claims,
+ audit=False,
+ )
+ self.assertTrue(preview.authorized)
+ self.assertEqual(
+ {c.name for c in preview.checks},
+ {
+ request_service.CHECK_AUTHORIZATION,
+ request_service.CHECK_CAPABILITY,
+ request_service.CHECK_LEASE_AVAILABILITY,
+ request_service.CHECK_NEXT_SAFE_ACTION,
+ request_service.CHECK_HEAD_PIN,
+ },
+ )
+ self.assertEqual(preview.required_profile, "prgs-author")
+ self.assertEqual(preview.required_namespace, "gitea-author")
+
+ def test_every_check_carries_a_reason(self):
+ preview = request_service.preview_request(
+ _request(),
+ principal=_principal(console_authz.OPERATOR),
+ allocator=_fake_allocator(),
+ claims_source=_no_claims,
+ audit=False,
+ )
+ for check in preview.checks:
+ with self.subTest(check=check.name):
+ self.assertTrue(check.reason_code.strip())
+ self.assertTrue(check.detail.strip())
+
+ def test_anonymous_preview_denied_with_reason(self):
+ preview = request_service.preview_request(
+ _request(),
+ allocator=_fake_allocator(),
+ claims_source=_no_claims,
+ audit=False,
+ )
+ self.assertFalse(preview.authorized)
+ self.assertEqual(preview.reason_code, console_authz.DENY_UNAUTHENTICATED)
+
+ def test_viewer_preview_denied_for_insufficient_role(self):
+ preview = request_service.preview_request(
+ _request(),
+ principal=_principal(console_authz.VIEWER),
+ allocator=_fake_allocator(),
+ claims_source=_no_claims,
+ audit=False,
+ )
+ self.assertFalse(preview.authorized)
+ self.assertEqual(preview.reason_code, console_authz.DENY_INSUFFICIENT_ROLE)
+
+ def test_denied_preview_never_reaches_the_allocator(self):
+ """A denial must not double as a queue oracle."""
+ calls: list[dict[str, Any]] = []
+ request_service.preview_request(
+ _request(),
+ principal=_principal(console_authz.VIEWER),
+ allocator=_fake_allocator(calls=calls),
+ claims_source=_no_claims,
+ audit=False,
+ )
+ self.assertEqual(calls, [])
+
+ def test_preview_lists_prohibited_actions(self):
+ preview = request_service.preview_request(
+ _request(role="author"),
+ principal=_principal(console_authz.OPERATOR),
+ allocator=_fake_allocator(),
+ claims_source=_no_claims,
+ audit=False,
+ )
+ self.assertIn("merge", preview.prohibited_actions)
+ self.assertIn("approve", preview.prohibited_actions)
+
+ def test_preview_reports_next_safe_action(self):
+ preview = request_service.preview_request(
+ _request(),
+ principal=_principal(console_authz.OPERATOR),
+ allocator=_fake_allocator(),
+ claims_source=_no_claims,
+ audit=False,
+ )
+ self.assertIn("issue #643", preview.next_safe_action)
+
+ def test_preview_never_mutates(self):
+ calls: list[dict[str, Any]] = []
+ request_service.preview_request(
+ _request(),
+ principal=_principal(console_authz.OPERATOR),
+ allocator=_fake_allocator(calls=calls),
+ claims_source=_no_claims,
+ audit=False,
+ )
+ self.assertEqual([c["apply"] for c in calls], [False])
+
+
+class TestPreviewFailClosed(unittest.TestCase):
+ """Missing evidence denies; it never reads as an absence of obstacles."""
+
+ def test_unreadable_claim_inventory_denies(self):
+ def _boom(_request):
+ raise RuntimeError("db unavailable")
+
+ preview = request_service.preview_request(
+ _request(),
+ principal=_principal(console_authz.OPERATOR),
+ allocator=_fake_allocator(),
+ claims_source=_boom,
+ audit=False,
+ )
+ self.assertFalse(preview.authorized)
+ self.assertEqual(
+ preview.reason_code, request_service.REASON_EVIDENCE_UNAVAILABLE
+ )
+
+ def test_allocator_failure_denies(self):
+ def _boom(**_kwargs):
+ raise RuntimeError("allocator exploded")
+
+ preview = request_service.preview_request(
+ _request(),
+ principal=_principal(console_authz.OPERATOR),
+ allocator=_boom,
+ claims_source=_no_claims,
+ audit=False,
+ )
+ self.assertFalse(preview.authorized)
+ self.assertEqual(
+ preview.reason_code, request_service.REASON_EVIDENCE_UNAVAILABLE
+ )
+
+ def test_allocator_selecting_other_work_denies(self):
+ preview = request_service.preview_request(
+ _request(number=643),
+ principal=_principal(console_authz.OPERATOR),
+ allocator=_fake_allocator(selection=_selection(number=999)),
+ claims_source=_no_claims,
+ audit=False,
+ )
+ self.assertFalse(preview.authorized)
+ self.assertEqual(preview.reason_code, request_service.REASON_NOT_NEXT_SAFE)
+ self.assertIn("#999", preview.detail)
+
+ def test_pr_without_head_sha_denies(self):
+ preview = request_service.preview_request(
+ _request(role="reviewer", kind="pr", number=898),
+ principal=_principal(console_authz.OPERATOR),
+ allocator=_fake_allocator(
+ selection=_selection(kind="pr", number=898, head_sha=None)
+ ),
+ claims_source=_no_claims,
+ audit=False,
+ )
+ self.assertFalse(preview.authorized)
+ self.assertEqual(
+ preview.reason_code, request_service.REASON_EVIDENCE_UNAVAILABLE
+ )
+
+ def test_pr_head_moved_denies(self):
+ preview = request_service.preview_request(
+ _request(role="reviewer", kind="pr", number=898, head="a" * 40),
+ principal=_principal(console_authz.OPERATOR),
+ allocator=_fake_allocator(
+ selection=_selection(kind="pr", number=898, head_sha="b" * 40)
+ ),
+ claims_source=_no_claims,
+ audit=False,
+ )
+ self.assertFalse(preview.authorized)
+ self.assertEqual(preview.reason_code, "head_moved")
+
+ def test_pr_head_matching_passes(self):
+ preview = request_service.preview_request(
+ _request(role="reviewer", kind="pr", number=898, head="b" * 40),
+ principal=_principal(console_authz.OPERATOR),
+ allocator=_fake_allocator(
+ selection=_selection(kind="pr", number=898, head_sha="b" * 40)
+ ),
+ claims_source=_no_claims,
+ audit=False,
+ )
+ self.assertTrue(preview.authorized)
+
+
+class TestApplyExecutionGate(unittest.TestCase):
+ """Execution stays wired off unless an operator opts in explicitly."""
+
+ def test_action_is_registered_and_unwired_by_default(self):
+ action = console_authz.get_action(request_service.ACTION_ID)
+ self.assertIsNotNone(action)
+ self.assertEqual(action.phase, 2)
+ self.assertEqual(action.minimum_role, console_authz.OPERATOR)
+ self.assertTrue(action.requires_confirmation)
+ self.assertFalse(console_authz.execution_wired(action, env={}))
+
+ def test_flag_named_but_unset_does_not_wire(self):
+ action = console_authz.get_action(request_service.ACTION_ID)
+ self.assertFalse(console_authz.execution_wired(action, env={EXEC_FLAG: "no"}))
+ self.assertTrue(console_authz.execution_wired(action, env={EXEC_FLAG: "1"}))
+
+ def test_opting_in_wires_only_this_action(self):
+ env = {EXEC_FLAG: "1"}
+ for action_id, action in console_authz.ACTIONS.items():
+ with self.subTest(action=action_id):
+ self.assertEqual(
+ console_authz.execution_wired(action, env=env),
+ action_id == request_service.ACTION_ID,
+ )
+
+ def test_apply_denied_while_unwired(self):
+ with mock.patch.dict(os.environ, {EXEC_FLAG: ""}):
+ result = request_service.apply_request(
+ _request(),
+ principal=_principal(console_authz.OPERATOR),
+ confirm=True,
+ allocator=_fake_allocator(),
+ claims_source=_no_claims,
+ )
+ self.assertFalse(result["ok"])
+ self.assertEqual(result["outcome"], request_service.OUTCOME_DENIED)
+ self.assertEqual(result["reason_code"], request_service.REASON_UNAUTHORIZED)
+ self.assertFalse(result["mutation_performed"])
+
+
+class TestApplyOutcomes(unittest.TestCase):
+ """AC2/AC3/AC4 — assignment, wait, blocked, and duplicate rejection."""
+
+ def setUp(self):
+ patcher = mock.patch.dict(os.environ, {EXEC_FLAG: "1"})
+ patcher.start()
+ self.addCleanup(patcher.stop)
+
+ def test_apply_creates_exclusive_assignment(self):
+ calls: list[dict[str, Any]] = []
+ result = request_service.apply_request(
+ _request(),
+ principal=_principal(console_authz.OPERATOR),
+ confirm=True,
+ allocator=_fake_allocator(calls=calls),
+ claims_source=_no_claims,
+ )
+ self.assertTrue(result["ok"])
+ self.assertEqual(result["outcome"], allocator_service.OUTCOME_ASSIGNED)
+ self.assertEqual(result["assignment"]["assignment_id"], "asn-test-0001")
+ self.assertTrue(result["mutation_performed"])
+ self.assertEqual(result["status_code"], 201)
+ # Dry-run first, then apply — never apply alone.
+ self.assertEqual([c["apply"] for c in calls], [False, True])
+ # The apply call carries the fingerprint the dry-run produced.
+ self.assertEqual(calls[1]["fingerprint"], "fp-test")
+
+ def test_assignment_returns_a_role_handoff(self):
+ result = request_service.apply_request(
+ _request(),
+ principal=_principal(console_authz.OPERATOR),
+ confirm=True,
+ allocator=_fake_allocator(),
+ claims_source=_no_claims,
+ )
+ handoff = result["handoff"]
+ self.assertEqual(handoff["required_profile"], "prgs-author")
+ self.assertEqual(handoff["required_namespace"], "gitea-author")
+ self.assertEqual(handoff["assignment_id"], "asn-test-0001")
+ self.assertIn("merge", handoff["forbidden_actions"])
+
+ def test_unconfirmed_apply_refuses_before_the_allocator(self):
+ calls: list[dict[str, Any]] = []
+ result = request_service.apply_request(
+ _request(),
+ principal=_principal(console_authz.OPERATOR),
+ confirm=False,
+ allocator=_fake_allocator(calls=calls),
+ claims_source=_no_claims,
+ )
+ self.assertFalse(result["ok"])
+ self.assertEqual(
+ result["reason_code"], request_service.REASON_CONFIRMATION_REQUIRED
+ )
+ self.assertEqual(calls, [])
+
+ def test_duplicate_assignment_rejected(self):
+ """AC3 — an active lease on the work unit blocks a second assign."""
+ calls: list[dict[str, Any]] = []
+ result = request_service.apply_request(
+ _request(),
+ principal=_principal(console_authz.OPERATOR),
+ confirm=True,
+ allocator=_fake_allocator(calls=calls),
+ claims_source=_claimed(),
+ )
+ self.assertFalse(result["ok"])
+ self.assertEqual(result["outcome"], request_service.OUTCOME_BLOCKED)
+ self.assertEqual(
+ result["reason_code"], request_service.REASON_DUPLICATE_ASSIGNMENT
+ )
+ self.assertFalse(result["mutation_performed"])
+ # The dry-run ran; the apply never did.
+ self.assertEqual([c["apply"] for c in calls], [False])
+
+ def test_not_next_safe_work_returns_wait_without_applying(self):
+ calls: list[dict[str, Any]] = []
+ result = request_service.apply_request(
+ _request(number=643),
+ principal=_principal(console_authz.OPERATOR),
+ confirm=True,
+ allocator=_fake_allocator(
+ selection=_selection(number=999), calls=calls
+ ),
+ claims_source=_no_claims,
+ )
+ self.assertFalse(result["ok"])
+ self.assertEqual(result["outcome"], request_service.OUTCOME_WAIT)
+ self.assertEqual(result["reason_code"], request_service.REASON_NOT_NEXT_SAFE)
+ self.assertEqual([c["apply"] for c in calls], [False])
+
+ def test_allocator_declining_on_apply_returns_blocked(self):
+ result = request_service.apply_request(
+ _request(),
+ principal=_principal(console_authz.OPERATOR),
+ confirm=True,
+ allocator=_fake_allocator(
+ apply_outcome=allocator_service.OUTCOME_BLOCKED_LEASE,
+ assignment={},
+ ),
+ claims_source=_no_claims,
+ )
+ self.assertFalse(result["ok"])
+ self.assertEqual(result["outcome"], request_service.OUTCOME_BLOCKED)
+ self.assertEqual(
+ result["reason_code"], request_service.REASON_ALLOCATOR_OUTCOME
+ )
+ self.assertFalse(result["mutation_performed"])
+
+ def test_allocator_drift_on_apply_is_not_read_as_an_assignment(self):
+ """The apply call must return *this* work unit, not a substitute.
+
+ Drift is not merely refused: the allocator has already committed the
+ substitute assignment by the time egress rejects it, so the refusal must
+ also release it. Asserting only ``mutation_performed is False`` would
+ pass just as well against a leak.
+ """
+ released: list[tuple[str, str]] = []
+
+ with _patched_control_plane(release_sink=released):
+ result = request_service.apply_request(
+ _request(number=643),
+ principal=_principal(console_authz.OPERATOR),
+ confirm=True,
+ allocator=_drifting_allocator(),
+ claims_source=_no_claims,
+ )
+
+ self.assertFalse(result["ok"])
+ self.assertIsNone(result["assignment"])
+ # The substitute assignment was released, so nothing durable survives.
+ self.assertEqual(len(released), 1)
+ self.assertEqual(released[0][0], "lease-wrong")
+ compensation = result["compensation"]
+ self.assertTrue(compensation["released"])
+ self.assertEqual(compensation["lease_id"], "lease-wrong")
+ self.assertEqual(compensation["assignment_id"], "asn-wrong")
+ self.assertEqual(compensation["selected"]["number"], 999)
+ self.assertFalse(result["mutation_performed"])
+ self.assertNotIn("orphaned_assignment", result)
+
+ def test_drift_whose_release_fails_reports_the_orphan_and_a_reclaim(self):
+ """A release that fails must surface the leak, never swallow it."""
+ with _patched_control_plane(release_error=RuntimeError("db is read-only")):
+ result = request_service.apply_request(
+ _request(number=643),
+ principal=_principal(console_authz.OPERATOR),
+ confirm=True,
+ allocator=_drifting_allocator(),
+ claims_source=_no_claims,
+ )
+
+ self.assertFalse(result["ok"])
+ self.assertIsNone(result["assignment"])
+ # A lease really is out there; saying "nothing changed" would be a lie.
+ self.assertTrue(result["mutation_performed"])
+ orphan = result["orphaned_assignment"]
+ self.assertFalse(orphan["released"])
+ self.assertTrue(orphan["attempted"])
+ self.assertIn("db is read-only", orphan["error"])
+ reclaim = orphan["reclaim_action"]
+ self.assertEqual(reclaim["tool"], "gitea_release_workflow_lease")
+ self.assertEqual(reclaim["lease_id"], "lease-wrong")
+
+ def test_drift_without_a_lease_id_still_surfaces_a_reclaim(self):
+ """An assignment with no lease id cannot be released — say so."""
+ result = request_service.apply_request(
+ _request(number=643),
+ principal=_principal(console_authz.OPERATOR),
+ confirm=True,
+ allocator=_drifting_allocator(lease_id=None),
+ claims_source=_no_claims,
+ )
+ self.assertFalse(result["ok"])
+ self.assertTrue(result["mutation_performed"])
+ orphan = result["orphaned_assignment"]
+ self.assertFalse(orphan["attempted"])
+ self.assertEqual(orphan["assignment_id"], "asn-wrong")
+ self.assertIn("reclaim_action", orphan)
+
+ def test_exception_after_commit_releases_the_session_lease(self):
+ """A post-commit exception surfaces as no result — with a live lease.
+
+ ``allocate_next_work`` catches only three exception types, so anything
+ else raised after ``assign_and_lease`` committed reaches the caller as
+ ``None`` while the lease is durable. The stable per-flow session id is
+ what makes that lease findable.
+ """
+ released: list[tuple[str, str]] = []
+ seen: list[str | None] = []
+
+ def _explodes_after_commit(
+ *, request, apply, expected_candidate_set_fingerprint=None, session_id=None
+ ):
+ seen.append(session_id)
+ if not apply:
+ return {
+ "outcome": allocator_service.OUTCOME_PREVIEW,
+ "selected": _selection(number=643),
+ "assignment": None,
+ "candidate_set_fingerprint": "fp-test",
+ }
+ raise KeyError("selection['required_profile']")
+
+ def _leases(**_kwargs):
+ return [
+ {
+ "lease_id": "lease-committed",
+ "session_id": seen[-1],
+ "work_kind": "issue",
+ "work_number": 643,
+ }
+ ]
+
+ with _patched_control_plane(release_sink=released, leases=_leases):
+ result = request_service.apply_request(
+ _request(number=643),
+ principal=_principal(console_authz.OPERATOR),
+ confirm=True,
+ allocator=_explodes_after_commit,
+ claims_source=_no_claims,
+ )
+
+ self.assertFalse(result["ok"])
+ self.assertEqual(
+ result["reason_code"], request_service.REASON_EVIDENCE_UNAVAILABLE
+ )
+ self.assertEqual(len(released), 1)
+ self.assertEqual(released[0][0], "lease-committed")
+ self.assertEqual(
+ result["compensation"]["released"][0]["lease_id"], "lease-committed"
+ )
+ self.assertFalse(result["mutation_performed"])
+
+ def test_one_session_id_spans_the_dry_run_and_the_apply(self):
+ """Both halves of an apply share one control-plane identity."""
+ seen: list[str | None] = []
+
+ def _recording(
+ *, request, apply, expected_candidate_set_fingerprint=None, session_id=None
+ ):
+ seen.append(session_id)
+ return {
+ "outcome": (
+ allocator_service.OUTCOME_ASSIGNED
+ if apply
+ else allocator_service.OUTCOME_PREVIEW
+ ),
+ "selected": _selection(number=643),
+ "assignment": (
+ {
+ "assignment_id": "asn-ok",
+ "lease_id": "lease-ok",
+ "expected_head_sha": None,
+ }
+ if apply
+ else None
+ ),
+ "candidate_set_fingerprint": "fp-test",
+ "session_id": session_id,
+ }
+
+ result = request_service.apply_request(
+ _request(number=643),
+ principal=_principal(console_authz.OPERATOR),
+ confirm=True,
+ allocator=_recording,
+ claims_source=_no_claims,
+ )
+ self.assertTrue(result["ok"])
+ self.assertEqual(len(seen), 2)
+ self.assertTrue(all(s for s in seen))
+ self.assertEqual(seen[0], seen[1], "dry-run and apply must share one id")
+ self.assertTrue(seen[0].startswith("webui-request-"))
+
+ def test_viewer_cannot_apply(self):
+ result = request_service.apply_request(
+ _request(),
+ principal=_principal(console_authz.VIEWER),
+ confirm=True,
+ allocator=_fake_allocator(),
+ claims_source=_no_claims,
+ )
+ self.assertFalse(result["ok"])
+ self.assertEqual(result["reason_code"], request_service.REASON_UNAUTHORIZED)
+
+ def test_anonymous_cannot_apply(self):
+ result = request_service.apply_request(
+ _request(),
+ confirm=True,
+ allocator=_fake_allocator(),
+ claims_source=_no_claims,
+ )
+ self.assertFalse(result["ok"])
+ self.assertFalse(result["mutation_performed"])
+
+
+class TestDefaultAllocatorPastTheEarlyReturns(unittest.TestCase):
+ """``default_allocator`` beyond its two fail-closed guards (#643).
+
+ Both prior tests returned before ``ControlPlaneDB`` was ever constructed, so
+ the session id, the ``side_effect_free`` routing and the CAS round-trip had
+ no coverage at all — which is how a preview that writes session rows shipped.
+ """
+
+ def setUp(self):
+ from webui.queue_loader import QueueSnapshot
+
+ self.snapshot = QueueSnapshot(
+ project_id="p",
+ repo_label="r",
+ prs=(),
+ issues=(),
+ pr_pagination=None,
+ issue_pagination=None,
+ )
+
+ @contextlib.contextmanager
+ def _harness(self):
+ """Run the real ``default_allocator`` against a recorded allocator call."""
+ calls: dict[str, Any] = {}
+
+ def _allocate(db, **kwargs):
+ calls.update(kwargs)
+ return {"outcome": allocator_service.OUTCOME_PREVIEW}
+
+ module = mock.MagicMock()
+ with mock.patch.dict(sys.modules, {"control_plane_db": module}), \
+ mock.patch(
+ "webui.queue_loader.load_queue_snapshot",
+ return_value=self.snapshot,
+ ), \
+ mock.patch(
+ "webui.traffic_loader.candidates_from_queue_snapshot",
+ return_value=[],
+ ), \
+ mock.patch.object(
+ allocator_service, "allocate_next_work", side_effect=_allocate
+ ):
+ yield calls
+
+ def test_preview_runs_side_effect_free_and_never_applies(self):
+ with self._harness() as calls:
+ result = request_service.default_allocator(
+ request=_request(), apply=False
+ )
+ self.assertIsNotNone(result)
+ self.assertTrue(calls["side_effect_free"])
+ self.assertFalse(calls["apply"])
+
+ def test_apply_is_not_side_effect_free(self):
+ with self._harness() as calls:
+ request_service.default_allocator(request=_request(), apply=True)
+ self.assertFalse(calls["side_effect_free"])
+ self.assertTrue(calls["apply"])
+
+ def test_caller_session_id_is_passed_through_verbatim(self):
+ with self._harness() as calls:
+ request_service.default_allocator(
+ request=_request(), apply=True, session_id="webui-request-fixed"
+ )
+ self.assertEqual(calls["session_id"], "webui-request-fixed")
+
+ def test_absent_session_id_is_minted_with_the_expected_shape(self):
+ with self._harness() as calls:
+ request_service.default_allocator(request=_request(), apply=False)
+ self.assertTrue(str(calls["session_id"]).startswith("webui-request-"))
+
+ def test_scope_and_fingerprint_reach_the_allocator(self):
+ with self._harness() as calls:
+ request_service.default_allocator(
+ request=_request(number=664),
+ apply=False,
+ expected_candidate_set_fingerprint="fp-pinned",
+ )
+ self.assertEqual(calls["expected_candidate_set_fingerprint"], "fp-pinned")
+ self.assertEqual(calls["remote"], SCOPE["remote"])
+ self.assertEqual(calls["org"], SCOPE["org"])
+ self.assertEqual(calls["repo"], SCOPE["repo"])
+ self.assertEqual(calls["allocation_mode"], "role_scoped")
+
+
+class TestDefaultClaimsSource(unittest.TestCase):
+ """``default_claims_source`` had no test at all (#643)."""
+
+ def test_claims_are_read_for_the_request_scope(self):
+ db = mock.MagicMock()
+ db.list_active_claims.return_value = {("issue", 643): {"lease_id": "l1"}}
+ module = mock.MagicMock()
+ module.ControlPlaneDB.return_value = db
+ with mock.patch.dict(sys.modules, {"control_plane_db": module}):
+ claims = request_service.default_claims_source(_request())
+ self.assertEqual(claims, {("issue", 643): {"lease_id": "l1"}})
+ db.list_active_claims.assert_called_once_with(
+ remote=SCOPE["remote"], org=SCOPE["org"], repo=SCOPE["repo"]
+ )
+
+ def test_an_unreadable_substrate_denies_rather_than_returning_empty(self):
+ module = mock.MagicMock()
+ module.ControlPlaneDB.side_effect = RuntimeError("no db")
+ with mock.patch.dict(sys.modules, {"control_plane_db": module}):
+ # _load_claims converts the failure into None, which fails the
+ # lease check closed; an empty mapping would read as "nothing
+ # claimed" and wrongly authorize.
+ claims = request_service._load_claims(
+ _request(), request_service.default_claims_source
+ )
+ self.assertIsNone(claims)
+
+
+class TestAllocatorIntegrationFakes(unittest.TestCase):
+ """The real default allocator refuses a partial inventory (#758)."""
+
+ def test_incomplete_inventory_returns_none(self):
+ from webui.queue_loader import PaginationMeta, QueueSnapshot
+
+ snapshot = QueueSnapshot(
+ project_id="p",
+ repo_label="r",
+ prs=(),
+ issues=(),
+ pr_pagination=PaginationMeta(
+ page=1,
+ per_page=50,
+ returned_count=50,
+ has_more=True,
+ is_final_page=False,
+ inventory_complete=False,
+ pages_fetched=1,
+ ),
+ issue_pagination=None,
+ )
+ with mock.patch(
+ "webui.queue_loader.load_queue_snapshot", return_value=snapshot
+ ):
+ result = request_service.default_allocator(
+ request=_request(), apply=False
+ )
+ self.assertIsNone(result)
+
+ def test_fetch_error_returns_none(self):
+ from webui.queue_loader import QueueSnapshot
+
+ snapshot = QueueSnapshot(
+ project_id="p",
+ repo_label="r",
+ prs=(),
+ issues=(),
+ pr_pagination=None,
+ issue_pagination=None,
+ fetch_error="no credentials",
+ )
+ with mock.patch(
+ "webui.queue_loader.load_queue_snapshot", return_value=snapshot
+ ):
+ result = request_service.default_allocator(
+ request=_request(), apply=True
+ )
+ self.assertIsNone(result)
+
+
+class TestAuditRecords(unittest.TestCase):
+ """Every preview and apply is auditable, correlated, and redacted."""
+
+ def setUp(self):
+ handle = tempfile.NamedTemporaryFile(
+ mode="w", suffix=".jsonl", delete=False
+ )
+ handle.close()
+ self.sink = handle.name
+ self.addCleanup(
+ lambda: os.path.exists(self.sink) and os.remove(self.sink)
+ )
+ patcher = mock.patch.dict(
+ os.environ, {console_audit.AUDIT_LOG_ENV: self.sink}
+ )
+ patcher.start()
+ self.addCleanup(patcher.stop)
+
+ def _records(self) -> list[dict[str, Any]]:
+ with open(self.sink, encoding="utf-8") as handle:
+ return [json.loads(line) for line in handle if line.strip()]
+
+ def test_preview_is_audited_with_a_correlation_id(self):
+ preview = request_service.preview_request(
+ _request(),
+ principal=_principal(console_authz.OPERATOR),
+ allocator=_fake_allocator(),
+ claims_source=_no_claims,
+ )
+ records = self._records()
+ self.assertEqual(len(records), 1)
+ record = records[0]
+ self.assertEqual(record["action"], request_service.ACTION_ID)
+ self.assertEqual(record["result"], console_audit.RESULT_PREVIEWED)
+ self.assertEqual(
+ record["correlation"]["request_id"], preview.correlation_id
+ )
+ self.assertEqual(record["target"]["ref"], "#643")
+
+ def test_denied_apply_is_audited(self):
+ with mock.patch.dict(os.environ, {EXEC_FLAG: ""}):
+ request_service.apply_request(
+ _request(),
+ principal=_principal(console_authz.VIEWER),
+ confirm=True,
+ allocator=_fake_allocator(),
+ claims_source=_no_claims,
+ )
+ self.assertEqual(self._records()[-1]["result"], console_audit.RESULT_DENIED)
+
+ def test_assignment_is_audited_and_correlated(self):
+ with mock.patch.dict(os.environ, {EXEC_FLAG: "1"}):
+ result = request_service.apply_request(
+ _request(),
+ principal=_principal(console_authz.OPERATOR),
+ confirm=True,
+ allocator=_fake_allocator(),
+ claims_source=_no_claims,
+ )
+ record = self._records()[-1]
+ self.assertEqual(record["result"], console_audit.RESULT_SUCCEEDED)
+ self.assertEqual(
+ record["correlation"]["request_id"], result["correlation_id"]
+ )
+ self.assertEqual(
+ record["metadata"]["assignment_id"],
+ result["assignment"]["assignment_id"],
+ )
+
+ def test_intent_bearing_a_secret_is_not_persisted_raw(self):
+ request_service.preview_request(
+ _request(intent="use token=ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"),
+ principal=_principal(console_authz.OPERATOR),
+ allocator=_fake_allocator(),
+ claims_source=_no_claims,
+ )
+ for record in self._records():
+ with self.subTest(event=record.get("event_id")):
+ self.assertFalse(scan_for_secrets(record))
+
+
+class TestRequestRoutes(unittest.TestCase):
+ """The HTTP surface: form page, preview API, apply API."""
+
+ def setUp(self):
+ self.client = TestClient(create_app())
+
+ def test_requests_page_renders_form(self):
+ response = self.client.get("/requests")
+ self.assertEqual(response.status_code, 200)
+ body = response.text
+ self.assertIn("Requests", body)
+ self.assertIn("desired_role", body)
+ self.assertIn("intent_summary", body)
+
+ def test_requests_page_is_linked_from_nav(self):
+ from webui.nav import nav_hrefs
+
+ self.assertIn("/requests", nav_hrefs())
+
+ def test_preview_api_rejects_an_invalid_request(self):
+ response = self.client.post(
+ "/api/v1/requests/preview",
+ json={
+ "desired_role": "wizard",
+ "work_kind": "issue",
+ "work_number": 1,
+ "intent_summary": "x",
+ **SCOPE,
+ },
+ )
+ self.assertEqual(response.status_code, 400)
+ self.assertEqual(response.json()["reason_code"], "unknown_role")
+
+ def test_preview_api_denies_anonymous(self):
+ response = self.client.post(
+ "/api/v1/requests/preview",
+ json={
+ "desired_role": "author",
+ "work_kind": "issue",
+ "work_number": 643,
+ "intent_summary": "x",
+ **SCOPE,
+ },
+ )
+ self.assertEqual(response.status_code, 403)
+ payload = response.json()
+ self.assertFalse(payload["authorized"])
+ self.assertFalse(payload["mutation_performed"])
+
+ def test_apply_api_denies_anonymous(self):
+ response = self.client.post(
+ "/api/v1/requests/apply",
+ json={
+ "desired_role": "author",
+ "work_kind": "issue",
+ "work_number": 643,
+ "intent_summary": "x",
+ "confirm": True,
+ **SCOPE,
+ },
+ )
+ self.assertEqual(response.status_code, 403)
+ payload = response.json()
+ self.assertFalse(payload["ok"])
+ self.assertFalse(payload["mutation_performed"])
+ self.assertIsNone(payload["assignment"])
+
+ def test_apply_api_rejects_an_invalid_request(self):
+ response = self.client.post(
+ "/api/v1/requests/apply",
+ json={
+ "desired_role": "author",
+ "work_kind": "issue",
+ "work_number": -1,
+ "intent_summary": "x",
+ **SCOPE,
+ },
+ )
+ self.assertEqual(response.status_code, 400)
+
+ def test_request_apis_are_post_only(self):
+ """GET is not a way in. The app's 405 handler renders a read-only
+ method against a write route as 404, so that is what is asserted."""
+ for path in ("/api/v1/requests/preview", "/api/v1/requests/apply"):
+ with self.subTest(path=path):
+ self.assertEqual(self.client.get(path).status_code, 404)
+
+ def test_form_post_previews_and_never_assigns(self):
+ response = self.client.post(
+ "/requests",
+ data={
+ "desired_role": "author",
+ "work_kind": "issue",
+ "work_number": "643",
+ "intent_summary": "implement the request surface",
+ "remote": "prgs",
+ "org": "Scaled-Tech-Consulting",
+ "repo": "Gitea-Tools",
+ },
+ )
+ self.assertEqual(response.status_code, 200)
+ self.assertIn("Intent preview", response.text)
+
+
+class TestRenderingSafety(unittest.TestCase):
+ """AC5 — the page escapes hostile input and shows no secret."""
+
+ def test_intent_is_escaped(self):
+ preview = request_service.preview_request(
+ _request(intent=""),
+ principal=_principal(console_authz.OPERATOR),
+ allocator=_fake_allocator(),
+ claims_source=_no_claims,
+ audit=False,
+ )
+ html = render_requests_page(preview=preview)
+ self.assertNotIn("", html)
+ self.assertIn("<script>", html)
+
+ def test_page_renders_a_denial_without_a_preview(self):
+ _, error = request_service.parse_request(
+ {
+ "desired_role": "wizard",
+ "work_kind": "issue",
+ "work_number": 1,
+ "intent_summary": "x",
+ **SCOPE,
+ }
+ )
+ html = render_requests_page(error=error)
+ self.assertIn("Request rejected", html)
+ self.assertIn("unknown_role", html)
+
+ def test_page_shows_no_credential_material(self):
+ preview = request_service.preview_request(
+ _request(),
+ principal=_principal(console_authz.OPERATOR),
+ allocator=_fake_allocator(),
+ claims_source=_no_claims,
+ audit=False,
+ )
+ html = render_requests_page(preview=preview)
+ for needle in ("token=", "Bearer ", "password"):
+ with self.subTest(needle=needle):
+ self.assertNotIn(needle, html)
+
+
+if __name__ == "__main__": # pragma: no cover
+ unittest.main()
diff --git a/webui/app.py b/webui/app.py
index 3840874..3115eaf 100644
--- a/webui/app.py
+++ b/webui/app.py
@@ -53,6 +53,11 @@ from webui.session_loader import (
snapshot_to_dict as session_view_snapshot_to_dict,
)
from webui.session_views import render_sessions_page
+from webui.linkage_loader import (
+ load_linkage_snapshot,
+ snapshot_to_dict as linkage_snapshot_to_dict,
+)
+from webui.linkage_views import render_linkage_page
from webui.inventory import (
SECTION_NAMES as _INVENTORY_SECTIONS,
load_inventory_snapshot,
@@ -77,6 +82,8 @@ from webui.notifications import (
snapshot_to_dict as notifications_snapshot_to_dict,
)
from webui.notification_views import render_notifications_page
+from webui import request_service
+from webui.request_views import render_requests_page
_READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
_AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
@@ -347,6 +354,41 @@ async def api_sessions(_request: Request) -> JSONResponse:
return JSONResponse(session_view_snapshot_to_dict(load_session_view_snapshot()))
+def _linkage_snapshot(request: Request):
+ """Load one linkage snapshot from the request's scope and focus parameters."""
+ return load_linkage_snapshot(
+ request.query_params.get("project") or None,
+ state=request.query_params.get("state"),
+ issue=_query_int(request, "issue"),
+ pr=_query_int(request, "pr"),
+ )
+
+
+async def gitea_linkage(request: Request) -> HTMLResponse:
+ """Gitea issue↔PR linkage console (#645) — read-only.
+
+ Always 200, including on a failed read: this is an operator view, and it
+ must render *why* linkage could not be loaded rather than withhold the page.
+ The snapshot itself carries ``ok=False`` and the page refuses to draw a
+ linkage table it cannot stand behind.
+ """
+ return HTMLResponse(render_linkage_page(_linkage_snapshot(request)))
+
+
+async def api_v1_gitea_linkage(request: Request) -> JSONResponse:
+ """JSON export of the issue↔PR linkage model (#645).
+
+ Unlike the HTML view, the API answers with 502 when the snapshot could not
+ be loaded, so an automated consumer cannot read a fail-closed payload as a
+ successful "no links exist" result.
+ """
+ snapshot = _linkage_snapshot(request)
+ return JSONResponse(
+ linkage_snapshot_to_dict(snapshot),
+ status_code=200 if snapshot.ok else 502,
+ )
+
+
async def _parse_audit_form(request: Request) -> tuple[str, str | None]:
if request.method == "GET":
return "", None
@@ -760,6 +802,108 @@ async def api_notifications(request: Request) -> JSONResponse:
data = notifications_snapshot_to_dict(snap)
return JSONResponse(data)
+def _default_request_scope() -> dict[str, str]:
+ """Resolve remote/org/repo from the project registry for request forms.
+
+ Returns an empty mapping when the registry cannot be read, which makes
+ ``parse_request`` reject a request that did not name its own scope rather
+ than letting it default to some other repository.
+ """
+ from webui.queue_loader import _host_from_url # host normalisation helper
+
+ registry, error = _load_project_registry()
+ if error is not None or not registry.projects:
+ return {}
+ project = registry.projects[0]
+ host = _host_from_url(project.remote_host)
+ return {
+ "remote": _derive_remote(host),
+ "org": project.gitea_owner or "",
+ "repo": project.repo_name or "",
+ }
+
+
+async def _request_payload(request: Request) -> dict[str, object]:
+ """Read a request body as JSON or form-encoded. Never raises."""
+ content_type = (request.headers.get("content-type") or "").lower()
+ if "application/json" in content_type:
+ try:
+ body = await request.json()
+ except Exception:
+ return {}
+ return dict(body) if isinstance(body, dict) else {}
+ try:
+ form = await request.form()
+ except Exception:
+ return {}
+ return {key: form[key] for key in form}
+
+
+async def requests_page(request: Request) -> HTMLResponse:
+ """Operator request form and intent preview (#643).
+
+ POST here only ever *previews*. Initiation is a separate confirmed call to
+ ``/api/v1/requests/apply`` so that submitting this form cannot reserve
+ work as a side effect.
+ """
+ submitted: dict[str, object] = {}
+ preview = None
+ error = None
+ if request.method == "POST":
+ submitted = await _request_payload(request)
+ work_request, error = request_service.parse_request(
+ submitted, default_scope=_default_request_scope()
+ )
+ if work_request is not None:
+ preview = request_service.preview_request(
+ work_request,
+ principal=resolve_principal(headers=dict(request.headers)),
+ )
+ return HTMLResponse(
+ render_requests_page(
+ preview=preview, error=error, submitted=submitted
+ )
+ )
+
+
+async def api_v1_request_preview(request: Request) -> JSONResponse:
+ """Dry-run authorization and intent preview for a work request (#643)."""
+ payload = await _request_payload(request)
+ work_request, error = request_service.parse_request(
+ payload, default_scope=_default_request_scope()
+ )
+ if work_request is None:
+ return JSONResponse(error.to_dict(), status_code=400)
+ preview = request_service.preview_request(
+ work_request,
+ principal=resolve_principal(headers=dict(request.headers)),
+ )
+ return JSONResponse(
+ preview.to_dict(), status_code=200 if preview.authorized else 403
+ )
+
+
+async def api_v1_request_apply(request: Request) -> JSONResponse:
+ """Initiate a previewed work request through the allocator (#643).
+
+ Fail-closed at every step: unauthorized, unconfirmed, not-next-safe, and
+ already-claimed all return without attempting an assignment.
+ """
+ payload = await _request_payload(request)
+ work_request, error = request_service.parse_request(
+ payload, default_scope=_default_request_scope()
+ )
+ if work_request is None:
+ return JSONResponse(error.to_dict(), status_code=400)
+ confirm = _truthy_flag(str(payload.get("confirm") or ""))
+ result = request_service.apply_request(
+ work_request,
+ principal=resolve_principal(headers=dict(request.headers)),
+ confirm=confirm,
+ )
+ status = int(result.pop("status_code", 403))
+ return JSONResponse(result, status_code=status)
+
async def method_not_allowed(request: Request, _exc: Exception) -> Response:
path = request.url.path
@@ -810,6 +954,8 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
Route("/api/sessions", api_sessions, methods=["GET"]),
Route("/api/v1/sessions", api_sessions, methods=["GET"]),
Route("/api/v1/timeline", api_v1_timeline, methods=["GET"]),
+ Route("/gitea", gitea_linkage, methods=["GET"]),
+ Route("/api/v1/gitea/linkage", api_v1_gitea_linkage, methods=["GET"]),
Route("/analytics", analytics, methods=["GET"]),
Route("/api/analytics", api_v1_analytics, methods=["GET"]),
Route("/api/v1/analytics", api_v1_analytics, methods=["GET"]),
@@ -831,6 +977,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(
diff --git a/webui/console_authz.py b/webui/console_authz.py
index 0c51037..282d831 100644
--- a/webui/console_authz.py
+++ b/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
- console is in Phase 1, so no caller can read an allow as permission to
- mutate.
+ ``execution_enabled`` reports whether the action has a live execution path
+ at all (:func:`execution_wired`) — for every action without an explicit
+ opt-in flag that stays ``False`` while the console is in Phase 1, so no
+ caller can read an allow as permission to mutate.
"""
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 "
- "remains disabled until the Phase 2 action framework ships."
+ "Principal holds the required role. Execution proceeds only for an "
+ "action with a wired execution path; everything else is preview."
),
**base,
)
diff --git a/webui/linkage_loader.py b/webui/linkage_loader.py
new file mode 100644
index 0000000..49e4f7e
--- /dev/null
+++ b/webui/linkage_loader.py
@@ -0,0 +1,771 @@
+"""Gitea issue↔PR linkage model for the console (#645, Phase 3).
+
+Operators lose context between an issue and the PR that closes it: which PR
+carries which issue, whether two PRs claim the same issue, and what the latest
+canonical handoff on that thread said. The evidence exists in Gitea, but only
+as free text scattered across PR titles, bodies, and branch names.
+
+This module resolves that linkage into one read-only model:
+
+* :func:`resolve_linkage` is a pure function from raw Gitea issue/PR payloads to
+ a :class:`LinkageIndex`. It records *how* each edge was found (a ``Closes #N``
+ keyword, the canonical ``feat/issue-N-…`` branch marker, or a bare ``#N``
+ body reference) and never collapses several candidates into one silent guess.
+* :func:`load_linkage_snapshot` scopes that index to a registry project and
+ optionally attaches the latest Canonical Thread Handoff (CTH) summary for one
+ focused issue or PR.
+
+Design rules, matching the rest of the console:
+
+- **Read-only.** Gitea is read through the shared authenticated helpers. No
+ endpoint here mutates anything, and no write action is registered.
+- **Qualified absence.** Linkage is a claim about a *loaded* window of Gitea.
+ When pagination did not complete, when credentials were unavailable, or when
+ only open items were fetched, the snapshot says so and every "no linked PR"
+ is marked non-authoritative. An empty edge list from a partial read is not
+ evidence that no link exists.
+- **Handoff is loaded, never assumed.** CTH comments are thread-scoped, so they
+ are fetched only for an explicitly focused issue or PR. Every other row
+ reports ``not_loaded`` rather than rendering as "no handoff".
+- **Redaction at the boundary.** Titles, labels, handoff fields, and error
+ reasons are free text from Gitea and cross :mod:`webui.console_redaction`
+ before they leave this module.
+- **Deep links are opt-in.** A link to the Gitea web UI is emitted only under
+ the ``GITEA_MCP_REVEAL_ENDPOINTS`` admin opt-in, exactly as the MCP tools
+ gate their own URL exposure.
+
+Non-goals (from the issue): no issue/PR editor, no browser review or merge, no
+reimplementation of Gitea search.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from dataclasses import dataclass
+from typing import Any, Callable, Iterable, Sequence
+
+from gitea_auth import api_fetch_page, get_auth_header, gitea_url, repo_api_url
+
+from webui import console_redaction
+from webui.project_registry import ProjectRecord, load_registry
+from webui.queue_loader import (
+ PaginationMeta,
+ _fetch_issues,
+ _fetch_prs,
+ _host_from_url,
+)
+
+#: Version of the serialized linkage contract. Bump on any breaking change.
+LINKAGE_SCHEMA_VERSION = 1
+
+# --- Linkage evidence -------------------------------------------------------
+# Ordered strongest to weakest. The strength ordering is what makes an
+# ambiguous PR detectable: two candidates at the same strength are a genuine
+# ambiguity, while a weaker candidate alongside a stronger one is not.
+EVIDENCE_CLOSES = "closes_keyword"
+EVIDENCE_BRANCH = "branch_marker"
+EVIDENCE_REFERENCE = "body_reference"
+
+EVIDENCE_ORDER: tuple[str, ...] = (
+ EVIDENCE_CLOSES,
+ EVIDENCE_BRANCH,
+ EVIDENCE_REFERENCE,
+)
+_EVIDENCE_RANK = {name: rank for rank, name in enumerate(EVIDENCE_ORDER)}
+
+EVIDENCE_DESCRIPTIONS: dict[str, str] = {
+ EVIDENCE_CLOSES: (
+ "the PR title or body declares 'closes/fixes/resolves #N' — Gitea itself "
+ "acts on this keyword, so it is the strongest available evidence"
+ ),
+ EVIDENCE_BRANCH: (
+ "the PR head branch carries the canonical issue marker "
+ "'(fix|feat|docs|chore)/issue-N-…' minted by the issue lock"
+ ),
+ EVIDENCE_REFERENCE: (
+ "the PR body mentions '#N' without a closing keyword; a mention is not "
+ "a claim that the PR closes that issue"
+ ),
+}
+
+_CLOSES_RE = re.compile(r"(?:closes|fixes|resolves)\s+#(\d+)", re.IGNORECASE)
+_REFERENCE_RE = re.compile(r"#(\d+)")
+_BRANCH_MARKER_RE = re.compile(
+ r"^(?:fix|feat|docs|chore)/issue-(\d+)(?:[-/]|$)", re.IGNORECASE
+)
+
+# Handoff-source states. ``not_loaded`` is deliberately distinct from "none
+# found": a row whose comments were never fetched proves nothing about whether
+# a handoff exists on that thread.
+HANDOFF_NOT_LOADED = "not_loaded"
+HANDOFF_LOADED = "loaded"
+HANDOFF_UNAVAILABLE = "unavailable"
+
+# Which item states were fetched. Linkage claims are scoped to this window.
+STATE_OPEN = "open"
+STATE_ALL = "all"
+_SUPPORTED_STATES = (STATE_OPEN, STATE_ALL)
+
+
+def _redact(value: Any) -> Any:
+ """Redact one free-text field, failing closed to the placeholder."""
+ if value is None:
+ return None
+ return console_redaction.redact_text(str(value))
+
+
+def deep_links_enabled(env: dict[str, str] | None = None) -> bool:
+ """Whether Gitea web-UI deep links may be emitted (admin/debug opt-in)."""
+ source = env if env is not None else os.environ
+ return (source.get("GITEA_MCP_REVEAL_ENDPOINTS") or "").strip().lower() in {
+ "1",
+ "true",
+ "yes",
+ "on",
+ }
+
+
+def _deep_link(host: str, org: str, repo: str, kind: str, number: int) -> str | None:
+ """Build a Gitea web link for one item, or None when reveal is not enabled."""
+ if not deep_links_enabled() or not (host and org and repo):
+ return None
+ segment = "pulls" if kind == "pr" else "issues"
+ try:
+ return gitea_url(host, f"/{org}/{repo}/{segment}/{int(number)}")
+ except Exception:
+ return None
+
+
+# --- Pure linkage resolution -------------------------------------------------
+
+
+@dataclass(frozen=True)
+class IssueLink:
+ """One resolved edge from a PR to an issue, with the evidence that found it."""
+
+ issue_number: int
+ evidence: tuple[str, ...]
+
+ @property
+ def strength(self) -> int:
+ """Rank of the strongest evidence backing this edge (lower is stronger)."""
+ return min(
+ (_EVIDENCE_RANK.get(name, len(EVIDENCE_ORDER)) for name in self.evidence),
+ default=len(EVIDENCE_ORDER),
+ )
+
+ @property
+ def closes(self) -> bool:
+ """True only when the PR *declares* it closes the issue."""
+ return EVIDENCE_CLOSES in self.evidence
+
+ def to_dict(self) -> dict[str, Any]:
+ return {
+ "issue_number": self.issue_number,
+ "evidence": list(self.evidence),
+ "closes": self.closes,
+ }
+
+
+def _sorted_links(links: Iterable[IssueLink]) -> tuple[IssueLink, ...]:
+ return tuple(sorted(links, key=lambda link: (link.strength, link.issue_number)))
+
+
+def resolve_pr_links(pr: dict[str, Any]) -> tuple[IssueLink, ...]:
+ """Resolve every issue a PR points at, strongest evidence first.
+
+ Every candidate is kept. Collapsing to a single "linked issue" is what makes
+ a mislinked or double-claimed PR invisible, so the caller decides what to do
+ with several candidates rather than being handed one guess.
+ """
+ found: dict[int, set[str]] = {}
+
+ def _add(number: Any, evidence: str) -> None:
+ try:
+ issue_number = int(number)
+ except (TypeError, ValueError):
+ return
+ if issue_number <= 0:
+ return
+ found.setdefault(issue_number, set()).add(evidence)
+
+ title = str(pr.get("title") or "")
+ body = str(pr.get("body") or "")
+ for text in (title, body):
+ for match in _CLOSES_RE.finditer(text):
+ _add(match.group(1), EVIDENCE_CLOSES)
+
+ head_ref = str((pr.get("head") or {}).get("ref") or "")
+ branch_match = _BRANCH_MARKER_RE.match(head_ref.strip())
+ if branch_match:
+ _add(branch_match.group(1), EVIDENCE_BRANCH)
+
+ # The ``#N`` inside "Closes #N" is the *same* textual occurrence as the
+ # closing keyword, not a second, independent mention. Blanking the closing
+ # phrases first keeps "mention" meaning what the legend says it means: a
+ # reference the PR made without claiming to close anything.
+ for match in _REFERENCE_RE.finditer(_CLOSES_RE.sub(" ", body)):
+ _add(match.group(1), EVIDENCE_REFERENCE)
+
+ # A PR's own number appearing in its body is self-reference, not linkage.
+ try:
+ found.pop(int(pr.get("number")), None)
+ except (TypeError, ValueError):
+ pass
+
+ return _sorted_links(
+ IssueLink(
+ issue_number=number,
+ evidence=tuple(name for name in EVIDENCE_ORDER if name in evidence),
+ )
+ for number, evidence in found.items()
+ )
+
+
+@dataclass(frozen=True)
+class LinkageIndex:
+ """Resolved linkage over one loaded window of issues and PRs."""
+
+ pr_links: dict[int, tuple[IssueLink, ...]]
+ issue_prs: dict[int, tuple[int, ...]]
+
+ def primary_issue(self, pr_number: int) -> IssueLink | None:
+ """The strongest edge for a PR, or None when it points at no issue."""
+ links = self.pr_links.get(int(pr_number)) or ()
+ return links[0] if links else None
+
+ def ambiguous(self, pr_number: int) -> bool:
+ """True when two or more issues tie at the PR's strongest evidence."""
+ links = self.pr_links.get(int(pr_number)) or ()
+ if len(links) < 2:
+ return False
+ best = links[0].strength
+ return sum(1 for link in links if link.strength == best) > 1
+
+ def contested_issues(self) -> tuple[int, ...]:
+ """Issues claimed by more than one PR in the loaded window."""
+ return tuple(
+ number for number, prs in sorted(self.issue_prs.items()) if len(prs) > 1
+ )
+
+
+def resolve_linkage(prs: Sequence[dict[str, Any]]) -> LinkageIndex:
+ """Build the bidirectional linkage index for a loaded window of PRs.
+
+ Only *closing* and *branch-marker* edges populate the issue→PR direction: a
+ bare ``#N`` mention is a reference, and treating it as "this PR is the work
+ for issue N" would invent contested issues out of ordinary cross-links. The
+ weaker edge stays visible on the PR→issue side, where it is labelled.
+ """
+ pr_links: dict[int, tuple[IssueLink, ...]] = {}
+ issue_prs: dict[int, list[int]] = {}
+ for pr in prs or []:
+ try:
+ pr_number = int(pr["number"])
+ except (KeyError, TypeError, ValueError):
+ continue
+ links = resolve_pr_links(pr)
+ pr_links[pr_number] = links
+ for link in links:
+ if link.evidence == (EVIDENCE_REFERENCE,):
+ continue
+ bucket = issue_prs.setdefault(link.issue_number, [])
+ if pr_number not in bucket:
+ bucket.append(pr_number)
+ return LinkageIndex(
+ pr_links=pr_links,
+ issue_prs={number: tuple(sorted(items)) for number, items in issue_prs.items()},
+ )
+
+
+# --- Canonical handoff summary ----------------------------------------------
+
+
+@dataclass(frozen=True)
+class HandoffSummary:
+ """The latest CTH comment on one thread, redacted for display."""
+
+ comment_id: int | None
+ created_at: str | None
+ author: str | None
+ cth_type: str
+ cth_type_known: bool
+ status: str | None
+ next_owner: str | None
+ current_blocker: str | None
+ decision: str | None
+ next_action: str | None
+
+ def to_dict(self) -> dict[str, Any]:
+ return {
+ "comment_id": self.comment_id,
+ "created_at": self.created_at,
+ "author": self.author,
+ "cth_type": self.cth_type,
+ "cth_type_known": self.cth_type_known,
+ "status": self.status,
+ "next_owner": self.next_owner,
+ "current_blocker": self.current_blocker,
+ "decision": self.decision,
+ "next_action": self.next_action,
+ }
+
+
+@dataclass(frozen=True)
+class HandoffStatus:
+ """Why a thread's handoff summary is present, absent, or unknown."""
+
+ state: str
+ reason: str | None = None
+ target: str | None = None
+
+ @property
+ def loaded(self) -> bool:
+ return self.state == HANDOFF_LOADED
+
+ def to_dict(self) -> dict[str, Any]:
+ return {"state": self.state, "reason": self.reason, "target": self.target}
+
+
+def summarize_handoff(comments: Sequence[dict[str, Any]]) -> HandoffSummary | None:
+ """Summarize the newest CTH comment in *comments*, or None when there is none.
+
+ Every field is redacted before it is returned: a handoff body is operator
+ free text that regularly quotes commands, and it is rendered verbatim on the
+ page this feeds.
+ """
+ from canonical_thread_handoff import find_latest_cth, is_known_cth_type
+
+ try:
+ latest = find_latest_cth(list(comments or []))
+ except Exception:
+ return None
+ if not latest:
+ return None
+ fields = latest.get("fields") or {}
+ cth_type = str(latest.get("cth_type") or "").strip()
+ known = is_known_cth_type(cth_type)
+ try:
+ comment_id: int | None = int(latest.get("comment_id"))
+ except (TypeError, ValueError):
+ comment_id = None
+ return HandoffSummary(
+ comment_id=comment_id,
+ created_at=_redact(latest.get("created_at")),
+ author=_redact(latest.get("author")),
+ # An unrecognised heading is reported as such rather than republished:
+ # the heading is free text, and CTH_TYPES is the only authority for what
+ # a handoff type may be.
+ cth_type=cth_type if known else "unrecognized",
+ cth_type_known=known,
+ status=_redact(fields.get("status")),
+ next_owner=_redact(fields.get("next owner")),
+ current_blocker=_redact(fields.get("current blocker")),
+ decision=_redact(fields.get("decision")),
+ next_action=_redact(fields.get("next action")),
+ )
+
+
+CommentSource = Callable[[str, int], list[dict[str, Any]]]
+
+
+def build_comment_source(host: str, org: str, repo: str) -> CommentSource | None:
+ """Build an authenticated ``(kind, number) -> comments`` fetcher, or None.
+
+ Returns None when the console is running in offline test mode or when no
+ credential is available for *host*, so the caller reports the handoff source
+ as unavailable instead of as an empty thread.
+ """
+ if _offline_test_mode() or not (host and org and repo):
+ return None
+ auth = get_auth_header(host)
+ if not auth:
+ return None
+
+ def _fetch(kind: str, number: int) -> list[dict[str, Any]]:
+ segment = "pulls" if kind == "pr" else "issues"
+ url = f"{repo_api_url(host, org, repo)}/{segment}/{int(number)}/comments"
+ comments: list[dict[str, Any]] = []
+ page = 1
+ while page <= 20:
+ raw, meta = api_fetch_page(url, auth, page=page, limit=50)
+ comments.extend(raw)
+ if bool(meta["is_final_page"]):
+ break
+ page += 1
+ return comments
+
+ return _fetch
+
+
+# --- Snapshot ----------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class LinkageNode:
+ """One issue or PR row with its resolved links and display metadata."""
+
+ kind: str
+ number: int
+ title: str
+ state: str
+ labels: tuple[str, ...] = ()
+ links: tuple[IssueLink, ...] = ()
+ linked_prs: tuple[int, ...] = ()
+ ambiguous: bool = False
+ contested: bool = False
+ deep_link: str | None = None
+ handoff: HandoffSummary | None = None
+ handoff_status: HandoffStatus = HandoffStatus(HANDOFF_NOT_LOADED)
+ links_authoritative: bool = True
+
+ def to_dict(self) -> dict[str, Any]:
+ return {
+ "kind": self.kind,
+ "number": self.number,
+ "title": self.title,
+ "state": self.state,
+ "labels": list(self.labels),
+ "links": [link.to_dict() for link in self.links],
+ "linked_prs": list(self.linked_prs),
+ "ambiguous": self.ambiguous,
+ "contested": self.contested,
+ "deep_link": self.deep_link,
+ "links_authoritative": self.links_authoritative,
+ "handoff": self.handoff.to_dict() if self.handoff else None,
+ "handoff_status": self.handoff_status.to_dict(),
+ }
+
+
+@dataclass(frozen=True)
+class LinkageSnapshot:
+ """One answered linkage query over a scoped window of a Gitea repo."""
+
+ ok: bool
+ project_id: str
+ repo_label: str
+ host: str
+ state_scope: str
+ issues: tuple[LinkageNode, ...] = ()
+ prs: tuple[LinkageNode, ...] = ()
+ contested_issues: tuple[int, ...] = ()
+ focus: tuple[str, int] | None = None
+ inventory_complete: bool = False
+ deep_links_enabled: bool = False
+ handoff_status: HandoffStatus = HandoffStatus(HANDOFF_NOT_LOADED)
+ fetch_error: str | None = None
+
+ @property
+ def orphan_prs(self) -> tuple[LinkageNode, ...]:
+ """PRs in the loaded window that point at no issue at all."""
+ return tuple(node for node in self.prs if not node.links)
+
+ def to_dict(self) -> dict[str, Any]:
+ return {
+ "ok": self.ok,
+ "schema_version": LINKAGE_SCHEMA_VERSION,
+ "project_id": self.project_id,
+ "repo": self.repo_label,
+ "state_scope": self.state_scope,
+ "inventory_complete": self.inventory_complete,
+ "deep_links_enabled": self.deep_links_enabled,
+ "focus": (
+ None
+ if self.focus is None
+ else {"kind": self.focus[0], "number": self.focus[1]}
+ ),
+ "handoff_source": self.handoff_status.to_dict(),
+ "fetch_error": self.fetch_error,
+ "contested_issues": list(self.contested_issues),
+ "issues": [node.to_dict() for node in self.issues],
+ "prs": [node.to_dict() for node in self.prs],
+ "evidence_kinds": [
+ {"name": name, "description": EVIDENCE_DESCRIPTIONS[name]}
+ for name in EVIDENCE_ORDER
+ ],
+ }
+
+
+def snapshot_to_dict(snapshot: LinkageSnapshot) -> dict[str, Any]:
+ """JSON-serializable export for ``/api/v1/gitea/linkage``."""
+ return snapshot.to_dict()
+
+
+def _offline_test_mode() -> bool:
+ return (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip().lower() in {
+ "1",
+ "true",
+ "yes",
+ }
+
+
+def _labels_of(item: dict[str, Any]) -> tuple[str, ...]:
+ return tuple(
+ str(_redact(label.get("name")))
+ for label in (item.get("labels") or [])
+ if label.get("name")
+ )
+
+
+def _failed_snapshot(
+ *,
+ project_id: str,
+ repo_label: str,
+ host: str,
+ state_scope: str,
+ reason: str,
+) -> LinkageSnapshot:
+ """A read that could not be answered. Never an empty-and-healthy snapshot."""
+ return LinkageSnapshot(
+ ok=False,
+ project_id=project_id,
+ repo_label=repo_label,
+ host=host,
+ state_scope=state_scope,
+ inventory_complete=False,
+ deep_links_enabled=deep_links_enabled(),
+ handoff_status=HandoffStatus(
+ HANDOFF_UNAVAILABLE, reason="linkage inventory could not be loaded"
+ ),
+ fetch_error=str(_redact(reason)),
+ )
+
+
+def _resolve_project(project_id: str | None) -> ProjectRecord | None:
+ registry = load_registry()
+ if project_id:
+ for entry in registry.projects:
+ if entry.id == project_id:
+ return entry
+ return None
+ return registry.projects[0] if registry.projects else None
+
+
+def _normalize_state(state: str | None) -> str:
+ text = (state or STATE_OPEN).strip().lower()
+ return text if text in _SUPPORTED_STATES else STATE_OPEN
+
+
+def load_linkage_snapshot(
+ project_id: str | None = None,
+ *,
+ state: str | None = None,
+ issue: int | None = None,
+ pr: int | None = None,
+ fetch_prs: Callable[..., tuple[list[dict], PaginationMeta]] | None = None,
+ fetch_issues: Callable[..., tuple[list[dict], PaginationMeta]] | None = None,
+ comment_source: CommentSource | None = None,
+) -> LinkageSnapshot:
+ """Load issue↔PR linkage for a registry project.
+
+ ``issue``/``pr`` focus one thread: the focused row is the only one whose
+ Canonical Thread Handoff comments are fetched, because handoff comments are
+ thread-scoped and loading them for a whole queue would be one request per
+ row. Every unfocused row reports its handoff as ``not_loaded``.
+ """
+ state_scope = _normalize_state(state)
+ try:
+ project = _resolve_project(project_id)
+ except Exception as exc: # registry invalid — fail closed with the reason
+ return _failed_snapshot(
+ project_id=project_id or "",
+ repo_label="",
+ host="",
+ state_scope=state_scope,
+ reason=f"project registry unavailable: {exc}",
+ )
+
+ if project is None:
+ return _failed_snapshot(
+ project_id=project_id or "",
+ repo_label="",
+ host="",
+ state_scope=state_scope,
+ reason=(
+ f"project {project_id!r} not found in registry"
+ if project_id
+ else "no projects registered"
+ ),
+ )
+
+ host = _host_from_url(project.remote_host)
+ repo_label = f"{project.gitea_owner}/{project.repo_name}"
+ offline_test = _offline_test_mode()
+
+ def _empty_fetch(*_args, **_kwargs):
+ return [], PaginationMeta(
+ page=1,
+ per_page=50,
+ returned_count=0,
+ has_more=False,
+ is_final_page=True,
+ # An offline stub loaded nothing; claiming a complete inventory here
+ # would let the page assert that no issue has a linked PR.
+ inventory_complete=False,
+ pages_fetched=0,
+ )
+
+ pr_fetch = fetch_prs or (_empty_fetch if offline_test else _fetch_prs)
+ issue_fetch = fetch_issues or (_empty_fetch if offline_test else _fetch_issues)
+ using_live_fetch = not offline_test and (fetch_prs is None or fetch_issues is None)
+ auth = get_auth_header(host) if using_live_fetch else "test-auth"
+ if using_live_fetch and not auth:
+ return _failed_snapshot(
+ project_id=project.id,
+ repo_label=repo_label,
+ host=host,
+ state_scope=state_scope,
+ reason=(
+ f"Gitea credentials unavailable for {host}; linkage cannot be "
+ "loaded (fail closed — not rendering an empty linkage table)"
+ ),
+ )
+
+ try:
+ raw_prs, pr_pagination = pr_fetch(
+ host, project.gitea_owner, project.repo_name, auth, state=state_scope
+ )
+ raw_issues, issue_pagination = issue_fetch(
+ host, project.gitea_owner, project.repo_name, auth, state=state_scope
+ )
+ except Exception as exc: # noqa: BLE001 — operator-visible fetch failure
+ return _failed_snapshot(
+ project_id=project.id,
+ repo_label=repo_label,
+ host=host,
+ state_scope=state_scope,
+ reason=f"Gitea fetch failed: {exc}",
+ )
+
+ inventory_complete = bool(
+ getattr(pr_pagination, "inventory_complete", False)
+ and getattr(issue_pagination, "inventory_complete", False)
+ )
+
+ index = resolve_linkage(raw_prs)
+ contested = index.contested_issues()
+
+ focus: tuple[str, int] | None = None
+ if pr is not None:
+ focus = ("pr", int(pr))
+ elif issue is not None:
+ focus = ("issue", int(issue))
+
+ unfocused_reason = (
+ "canonical handoff comments are thread-scoped; focus one issue or PR "
+ "to load its latest handoff"
+ )
+ handoff_status = HandoffStatus(HANDOFF_NOT_LOADED, reason=unfocused_reason)
+ focus_handoff: HandoffSummary | None = None
+ if focus is not None:
+ source = comment_source
+ if source is None and not offline_test:
+ source = build_comment_source(host, project.gitea_owner, project.repo_name)
+ target = f"{focus[0]}#{focus[1]}"
+ if source is None:
+ handoff_status = HandoffStatus(
+ HANDOFF_UNAVAILABLE,
+ reason="no authenticated comment source available for this read",
+ target=target,
+ )
+ else:
+ try:
+ focus_handoff = summarize_handoff(source(focus[0], focus[1]) or [])
+ handoff_status = HandoffStatus(HANDOFF_LOADED, target=target)
+ except Exception as exc: # fail soft: degrade this source only
+ handoff_status = HandoffStatus(
+ HANDOFF_UNAVAILABLE,
+ reason=str(_redact(f"handoff fetch failed: {exc}")),
+ target=target,
+ )
+
+ def _node_handoff(
+ kind: str, number: int
+ ) -> tuple[HandoffSummary | None, HandoffStatus]:
+ """Attach the handoff only to the focused row; qualify every other row."""
+ if focus == (kind, number):
+ return (focus_handoff, handoff_status)
+ return (
+ None,
+ HandoffStatus(
+ HANDOFF_NOT_LOADED,
+ reason=unfocused_reason if focus is None else "not the focused thread",
+ ),
+ )
+
+ def _number_of(raw: dict[str, Any]) -> int | None:
+ try:
+ return int(raw["number"])
+ except (KeyError, TypeError, ValueError):
+ return None
+
+ def _sort_key(raw: dict[str, Any]) -> int:
+ number = _number_of(raw)
+ return -1 if number is None else number
+
+ issue_nodes: list[LinkageNode] = []
+ for raw in sorted(raw_issues or [], key=_sort_key, reverse=True):
+ number = _number_of(raw)
+ if number is None:
+ continue
+ node_handoff, node_status = _node_handoff("issue", number)
+ linked_prs = index.issue_prs.get(number, ())
+ issue_nodes.append(
+ LinkageNode(
+ kind="issue",
+ number=number,
+ title=str(_redact(raw.get("title")) or ""),
+ state=str(raw.get("state") or ""),
+ labels=_labels_of(raw),
+ linked_prs=linked_prs,
+ contested=len(linked_prs) > 1,
+ deep_link=_deep_link(
+ host, project.gitea_owner, project.repo_name, "issue", number
+ ),
+ handoff=node_handoff,
+ handoff_status=node_status,
+ links_authoritative=inventory_complete,
+ )
+ )
+
+ pr_nodes: list[LinkageNode] = []
+ for raw in sorted(raw_prs or [], key=_sort_key, reverse=True):
+ number = _number_of(raw)
+ if number is None:
+ continue
+ node_handoff, node_status = _node_handoff("pr", number)
+ links = index.pr_links.get(number, ())
+ pr_nodes.append(
+ LinkageNode(
+ kind="pr",
+ number=number,
+ title=str(_redact(raw.get("title")) or ""),
+ state=str(raw.get("state") or ""),
+ labels=_labels_of(raw),
+ links=links,
+ ambiguous=index.ambiguous(number),
+ contested=any(link.issue_number in contested for link in links),
+ deep_link=_deep_link(
+ host, project.gitea_owner, project.repo_name, "pr", number
+ ),
+ handoff=node_handoff,
+ handoff_status=node_status,
+ links_authoritative=inventory_complete,
+ )
+ )
+
+ return LinkageSnapshot(
+ ok=True,
+ project_id=project.id,
+ repo_label=repo_label,
+ host=host,
+ state_scope=state_scope,
+ issues=tuple(issue_nodes),
+ prs=tuple(pr_nodes),
+ contested_issues=contested,
+ focus=focus,
+ inventory_complete=inventory_complete,
+ deep_links_enabled=deep_links_enabled(),
+ handoff_status=handoff_status,
+ )
diff --git a/webui/linkage_views.py b/webui/linkage_views.py
new file mode 100644
index 0000000..fcc06fa
--- /dev/null
+++ b/webui/linkage_views.py
@@ -0,0 +1,364 @@
+"""HTML views for the Gitea issue↔PR linkage console (#645, Phase 3).
+
+Read-only renderer over :mod:`webui.linkage_loader`. The page's job is to make
+three things impossible to misread:
+
+* **why** an edge exists — every link carries its evidence badge, so a bare
+ ``#N`` mention never looks like a closing claim;
+* **what was not loaded** — a partial inventory, an unfocused thread, or an
+ unavailable handoff source renders as an explicit qualifier, never as an
+ affirmative "none";
+* **that nothing here mutates** — there is no review, merge, or edit control,
+ and the deep link out to Gitea appears only under the admin reveal opt-in.
+"""
+
+from __future__ import annotations
+
+from html import escape
+from typing import Sequence
+
+from webui.layout import render_page
+from webui.linkage_loader import (
+ EVIDENCE_BRANCH,
+ EVIDENCE_CLOSES,
+ EVIDENCE_DESCRIPTIONS,
+ EVIDENCE_ORDER,
+ EVIDENCE_REFERENCE,
+ HANDOFF_LOADED,
+ HANDOFF_NOT_LOADED,
+ HandoffSummary,
+ LinkageNode,
+ LinkageSnapshot,
+)
+
+_EVIDENCE_CSS = {
+ EVIDENCE_CLOSES: "badge-health-ok",
+ EVIDENCE_BRANCH: "badge-health-skipped",
+ EVIDENCE_REFERENCE: "badge-health-unproven",
+}
+
+_EVIDENCE_LABEL = {
+ EVIDENCE_CLOSES: "closes",
+ EVIDENCE_BRANCH: "branch",
+ EVIDENCE_REFERENCE: "mention",
+}
+
+
+def _badge(text: str, css: str) -> str:
+ return f'{escape(text)}'
+
+
+def _labels(names: Sequence[str]) -> str:
+ if not names:
+ return '—'
+ 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'{escape(label)}'
+ return f"{escape(label)}"
+
+
+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 none found in what was loaded — 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"""
| Project | {escape(snapshot.project_id or "—")} |
|---|---|
| Repository | {escape(snapshot.repo_label or "—")} |
| Item state | {escape(snapshot.state_scope)} |
| Focused thread | {escape(focus)} |
| Inventory | {completeness} |
| Gitea deep links | {deep_links} |
{links_note}
+#{number}" for number in snapshot.contested_issues)
+ return (
+ '#{number}" for number in node.linked_prs)
+ if node.contested:
+ linked += " " + _badge("contested", "badge-blocked")
+ elif node.links_authoritative:
+ linked = 'none'
+ else:
+ # The distinction an operator needs: nothing found in a window that
+ # was not fully loaded is not the same as nothing existing.
+ linked = 'none found (partial inventory)'
+ rows.append(
+ "#{link.issue_number} "
+ f"{_evidence_badges(link.evidence)}{escape(snapshot.handoff_status.reason or "")}
+ Add ?issue=N or ?pr=N to load the latest
+ Canonical Thread Handoff for one thread.
{_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.
+Comments loaded; no Canonical Thread Handoff comment found on + this thread.
+The comment\'s heading is not a declared CTH type, ' + "so it is reported as unrecognized rather than republished.
" + ) + ) + return f"""| Status | {escape(handoff.status or "—")} |
|---|---|
| Next owner | {escape(handoff.next_owner or "—")} |
| Current blocker | {escape(handoff.current_blocker or "—")} |
| Decision | {escape(handoff.decision or "—")} |
| Next action | {escape(handoff.next_action or "—")} |
Full event history:
+ /api/v1/timeline
GITEA_MCP_REVEAL_ENDPOINTS admin opt-in is set."
+ if snapshot.deep_links_enabled
+ else (
+ "Gitea deep links are withheld. Set "
+ "GITEA_MCP_REVEAL_ENDPOINTS=1 server-side to reveal "
+ "them; item numbers stay usable without them."
+ )
+ )
+ return f"""{reveal_note}
+Read-only surface: no issue or PR editing, no review, and no merge.
+ JSON export: /api/v1/gitea/linkage
| Issue | Title | State | Labels | +Linked PRs | Latest handoff | +
|---|
| PR | Title | State | Labels | +Linked issues | Latest handoff | +
|---|
{_escape(check.name)} — {_escape(check.detail)} "
+ f""
+ "{_escape(action)}" for action in preview.prohibited_actions
+ )
+ request = preview.request
+ evidence = json.dumps(preview.allocator_evidence, indent=2, default=str)
+ return (
+ "{verdict} — {_escape(preview.detail)}
" + "" + f"Intent: {_escape(request.intent_summary)}
" + f"{_checks_block(preview)}" + f"Next safe action: " + f"{_escape(preview.next_safe_action)}
" + "Prohibited for this role: " + + (prohibited or "none declared") + + "
" + "" + "{_escape(evidence)}"
+ "{_escape(error.reason_code)} — " + f"{_escape(error.detail)}
{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 = ( + "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.
" + + _form(submitted) + + (_error_block(error) if error is not None else "") + + (_preview_block(preview) if preview is not None else "") + + f"" + + REQUEST_PAGE_STYLES + ) + return render_page(title="Requests", body_html=body) diff --git a/webui/traffic_loader.py b/webui/traffic_loader.py index 912d10e..63993ff 100644 --- a/webui/traffic_loader.py +++ b/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.