Author SHA1 Message Date
sysadminandClaude Opus 4.8 433f66add8 feat(webui): request preview, authorization, and workflow initiation (Closes #643)
Operators had to paste a role prompt into a terminal to start work, and
nothing enforced that the allocator had been consulted first, so two sessions
could reach for the same issue and each believe it was theirs. This adds a
request surface: 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.

Preview (POST /api/v1/requests/preview, and the /requests form) runs five
checks and reports authorize/deny with a reason for each: console
authorization, capability resolution for the desired role, lease availability,
whether the allocator would independently select this work unit, and head
pinning for PR work. It is read-only - it calls the allocator with apply=false
and writes only an audit line. An unauthorized principal never reaches the
allocator or the control-plane DB, so a denial cannot enumerate the queue.

Initiation (POST /api/v1/requests/apply) never assigns the requested item
directly. It runs a dry-run first and proceeds only when the allocator would
independently pick that exact work unit, carrying the dry-run's
candidate_set_fingerprint as a CAS pin; otherwise it returns wait or blocked
and mutates nothing. An active claim on the work unit rejects a duplicate
assign before one is attempted. A returned assignment carries a handoff block
naming the required profile, namespace, and the actions that stay forbidden.

Authorization reuses the #633 model rather than adding a second one. The new
initiate_workflow action is operator-class because its outcome is a claim, not
a Gitea verdict: requesting reviewer or merger work reserves that work but
grants no right to approve or merge. Execution is gated by a new per-action
execution_env_flag (WEBUI_REQUESTS_EXECUTION), deliberately in place of raising
ACTIVE_PHASE - a phase bump would enable execution for every phase-2 action at
once, including ones whose execution path is not implemented. Actions that
declare no flag are unchanged and still report execution_enabled false.

Every preview and apply emits a console audit record correlated to the
resulting assignment by correlation.request_id.

Fail-closed throughout: an unreadable control-plane DB, an incomplete queue
inventory (#758), an allocator that raises, an unpinned PR head, a moved PR
head, and an unconfirmed apply all deny without mutating.

Files:
- webui/request_service.py (new) - request model, preview, initiation
- webui/request_views.py (new) - form and preview rendering, escaped
- tests/test_webui_request_initiation.py (new) - 52 tests
- webui/console_authz.py - initiate_workflow action, execution_wired()
- webui/app.py - /requests, /api/v1/requests/preview, /api/v1/requests/apply
- webui/nav.py - Requests nav entry
- webui/traffic_loader.py - public candidates_from_queue_snapshot alias
- docs/webui-requests.md (new), docs/webui-authz-audit.md

Validation: full suite on this branch 5242 passed, 6 skipped, 899 subtests, 23
failed. Clean master baseline at 2f4dec83 in an equivalent branches/ worktree:
5190 passed, 6 skipped, 867 subtests, the same 23 tests failed. The branch adds
52 passing tests and introduces no new full-suite failure signature.

Closes #643

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-25 01:47:38 -04:00
sysadmin 2f4dec8323 Merge pull request 'feat(restart): pre-restart drain proof and hard gate (Closes #661)' (#882) from feat/issue-661-drain-proof-hard-gate into master 2026-07-24 23:40:02 -05:00
sysadmin 3a9d634c17 fix(drain-proof): bind acknowledgement coverage to session identity (#661)
Review 582 (REQUEST_CHANGES at 95178349) found a residual fail-open of the
same class the PR set out to close. Acknowledgement coverage was decided by
comparing a count against a count:

    covers_live_sessions = live_count_known and acked_count >= sessions_live_other

Nothing bound an acknowledgement to the identity of a session that actually
owed one, so acknowledgements supplied for the requesting session and for a
session that does not exist satisfied the obligations of two live sessions
that never answered - minting a clean, correctly signed proof and an allow
verdict from the restart gate.

Coverage is now derived from authoritative impact-report evidence:

- New `_required_ack_sessions()` derives the required session ids from the
  report itself, via `ack_state` keys and/or `affected_sessions` filtered on
  `live and not is_requester`. The requester is excluded only on explicit
  `is_requester` evidence, never inferred.
- When both views are present they must name the same set, and the result is
  reconciled against `counts.sessions_live_other`. Missing, malformed,
  duplicated, contradictory, or unreconcilable identity evidence fails closed
  and outranks every permitting path, including the timeout policy.
- Coverage requires every required id to carry an explicit acknowledgement
  token keyed by that id. Acknowledgements for the requester, for unknown
  ids, or for fabricated ids never increase coverage.
- Caller-supplied acknowledgement cardinality is no longer proof of anything.

Failure propagates unchanged through `acks_or_timeout` -> `proof.clean` ->
`failed_checks` -> `gate_apply_restart` verdict `deny` / `allow=False`.

The earlier missing-acknowledgement remediation is preserved in full: absent,
None, non-mapping, empty, partial, stale, and unparseable acks still fail
closed, `ack_timeout_policy_applied` stays strict `value is True`, and the
legitimate zero-live-sessions and explicit-timeout paths still pass.

Reviewer's reproduction, before and after this commit:

    sessions_live_other = 2
    report ack_state    = {'other-0': 'pending', 'other-1': 'pending'}
    supplied acks       = {'req': 'ack', 'totally-bogus-session': 'ack'}

    before: acks_or_timeout = True  | proof.clean = True  | gate allow
    after:  acks_or_timeout = False | proof.clean = False | gate deny

Tests: 22 new cases in `AcknowledgementIdentityBindingTests` covering the
reviewer's exact exploit, wrong-ids-with-sufficient-count, partial identity
match, requester-only acks, fabricated ids, unproven per-session states,
missing/malformed/contradictory identity evidence, count mismatch, and the
preserved success paths.

Verification:
- `pytest tests/test_drain_proof.py` -> 61 passed, 56 subtests
  (baseline at 95178349: 39 passed, 26 subtests)
- Restart surface (6 modules) -> 154 passed, 68 subtests, exit 0
  (baseline at 95178349: 132 passed, 38 subtests)
- Full `pytest tests/` -> 5177 passed vs baseline 5155 passed; the 23
  failures are identical in both runs and pre-exist at 95178349.

Scope: drain_proof.py, tests/test_drain_proof.py.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01VRUZAf3Fr5n3kqhhiayN6C
(cherry picked from commit 4193b63f415b066ee292386c2c89bc3d2651a0cc)
2026-07-25 00:04:17 -04:00
jcwalker3 2068bae341 Merge branch 'master' into feat/issue-661-drain-proof-hard-gate 2026-07-24 22:34:27 -05:00
sysadmin 7af40fb5ff Merge pull request 'fix(allocator): exclude vision/roadmap/umbrella coordination containers (Closes #854)' (#883) from fix/issue-854-semantic-container-exclusion into master 2026-07-24 22:27:58 -05:00
sysadmin 9517834913 Merge commit '578c44b685a7ff5b01006c5e398bfac9863e0d8d' into feat/issue-661-drain-proof-hard-gate 2026-07-24 22:37:49 -04:00
sysadminandClaude Opus 5 824c42f7e3 fix(drain): fail closed on missing or unproven acknowledgement evidence (#661)
The acks_or_timeout check treated an absent `acks` key as proof that no
session needed to acknowledge: `drain_state.get("acks") or {}` collapsed
absent, None, and empty into the same value, and the resulting empty mapping
satisfied `no_sessions_to_ack`. The impact report's counts.sessions_live_other
was never consulted, so absence of evidence was read as evidence of absence.

Reproduced at head 1cbbde0089: with
sessions_live_other = 3 and the acknowledgement key absent, acks_or_timeout
passed with detail "no other live sessions required to acknowledge", the proof
minted clean, and gate_apply_restart returned verdict allow — a restart
authorized against three live sessions with zero acknowledgement evidence, and
the resulting artifact carried a valid signature.

Whether acknowledgement is required is now derived from the impact report,
never from the shape of the drain state:

- _live_session_count() reads counts.sessions_live_other and returns None for a
  missing, malformed, negative, or bool value, so an unreadable report fails
  closed instead of reading as "nobody was live".
- Absent, None, non-mapping, empty, partially-covering, and unparseable or
  stale acknowledgement data all fail closed while live sessions require
  acknowledgement.
- _is_acknowledged() no longer coerces with str(); only an explicit
  "ack"/"acked"/"acknowledged" string counts, so None, timestamps, and
  "pending"/"stale" markers are never read as an acknowledgement.
- Present-but-unacknowledged entries fail closed even when the report claims
  zero live sessions: that contradiction is not safe to resolve in favour of
  the restart.
- ack_timeout_policy_applied stays strict (`value is True`), so an absent, null,
  or non-boolean value cannot open the gate on its own.

The genuine no-other-live-sessions case still passes, now justified by the
report proving sessions_live_other == 0 rather than by the absence of data.

Adds AcknowledgementFailClosedTests: 14 cases / 26 subtests covering missing,
null, empty, malformed, stale, partial-coverage, and unproven-count inputs,
the valid-acknowledgement and zero-live-session paths, timeout-policy
strictness, and that a failed check blocks proof.clean, verification, and the
restart gate.

Restart-surface suite: 132 passed, 38 subtests (branch baseline 118 passed,
12 subtests; +14 new tests, no regressions).

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01VEaP3TohHLFWkp3Z2mmuZw
2026-07-24 22:36:25 -04:00
jcwalker3 578c44b685 Merge branch 'master' into feat/issue-661-drain-proof-hard-gate 2026-07-24 21:28:04 -05:00
jcwalker3 3b68d15593 Merge branch 'master' into fix/issue-854-semantic-container-exclusion 2026-07-24 21:27:55 -05:00
sysadmin a4c73766f4 Merge pull request 'feat(webui): Workflow traffic-control view (Phase 1) (Closes #640)' (#885) from issue-640 into master 2026-07-24 21:10:51 -05:00
jcwalker3 9f686253eb Merge branch 'master' into feat/issue-661-drain-proof-hard-gate 2026-07-24 21:06:49 -05:00
jcwalker3 b2e28428a4 Merge branch 'master' into fix/issue-854-semantic-container-exclusion 2026-07-24 21:06:41 -05:00
sysadmin 1cbbde0089 feat(restart): pre-restart drain proof and hard gate (#661)
Add `drain_proof.py`: a machine-verifiable DrainProof artifact plus a
fail-closed verifier and the hard gate the sanctioned restart-apply path
must consult, so a restart can never proceed on a stale or false "ready"
claim (#655 umbrella, child of #658 coordinator / #659 drain / #660
checkpoints).

- DrainProof: HMAC-SHA256 keyed proof-id over a canonical body using a
  per-process secret -> non-forgeable within the process; a proof minted in
  a prior daemon process will not verify after restart. Short TTL (120s).
- build_drain_proof(): mints the proof from the #658 impact report + the
  drain-mode outcomes. Checklist: no in-flight mutations, assignments
  stopped, checkpoints complete, handoffs ok, leases handled, acks-or-
  timeout. Every check fails closed on missing/ambiguous evidence; the
  no-in-flight-mutations and leases-handled checks are derived from the
  authoritative impact report, not self-reported.
- verify_drain_proof(): fail-closed — rejects missing, malformed, expired,
  signature-mismatched (forged/tampered/prior-process), unclean, or
  stale-fingerprint proofs; recomputes cleanliness from the checks rather
  than trusting the flag.
- gate_apply_restart(): allow only on a valid clean proof; deny -> durable
  incident descriptor; break-glass is the only bypass and is never silent.
- Checkpoint completeness is a supplied input, not a hard dependency on the
  (still-unmerged #660) checkpoint schema.

Wire the gate into gitea_request_mcp_restart: dry_run=False now enforces the
hard gate (drain_proof_json required; break-glass via request_break_glass +
GITEA_BREAKGLASS_RESTART_AUTHORIZATION env). The tool still performs no
actual restart — execution remains a further child.

Tests: tests/test_drain_proof.py — 25 cases covering AC#1-4 (apply without
proof denied, successful drain verifiable, open unsafe mutation fails,
pass/fail/expired), forgery/tamper/wrong-secret/stale-fingerprint rejection,
break-glass bypass, and secret hygiene. 25/25 pass (coordinator suite
unaffected: 40/40 together).

Links #652 #653 #655 #658 #659 #660.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01E7Fv9Bp2XWgvaWa4M1kdR7
(cherry picked from commit e7bcc952bb3e820fda95acbecefeaebfa5f8fcff)
2026-07-24 17:17:13 -04:00
sysadmin 0a78da39e5 fix(allocator): exclude vision/roadmap/umbrella coordination containers (#854)
#844 only caught epic-shaped child-only records. Live allocation still
selected product vision (#652), phased roadmap (#653), and umbrella (#655)
as implement targets. Extend pre-rank semantic classification with body
markers and container labels for those coordination records, keep title-
only and incidental mentions eligible, and add a live-equivalent canary.

Closes #854
2026-07-24 17:17:05 -04:00
14 changed files with 4886 additions and 42 deletions
+81 -17
View File
@@ -133,10 +133,15 @@ ROLE_ACTIONS: dict[str, tuple[tuple[str, ...], tuple[str, ...]]] = {
# Body phrases that prove an issue is an implementation container, not a
# unit of direct author work (#844). Matched case-insensitively against the
# issue body. Title alone is never sufficient (ordinary issues may mention
# "epic" incidentally).
# unit of direct author work (#844 / #854). Matched case-insensitively against
# the issue body. Title alone is never sufficient (ordinary issues may mention
# "epic", "roadmap", "vision", or "umbrella" incidentally).
#
# #854 extends the #844 marker set so product-vision (#652), phased-roadmap
# (#653), and umbrella (#655) coordination records — which do not use the word
# "epic" — are classified with the same semantic exclusion as epic containers.
_CHILD_ONLY_BODY_MARKERS: tuple[str, ...] = (
# Epic / child-only (#844, live #631)
"implementation is delivered via child issues only",
"implementation is delivered through child issues only",
"implementation is delivered via child issues",
@@ -150,9 +155,26 @@ _CHILD_ONLY_BODY_MARKERS: tuple[str, ...] = (
"coordination container",
"child-only container",
"implementation is delegated to child",
# Vision / roadmap / umbrella coordination (#854, live #652/#653/#655).
# Prefer authoritative non-implementation / child-only scope language over
# bare words like "roadmap" so ordinary implementable issues that mention
# a parent vision or roadmap stay eligible.
"do not implement features on this issue",
"implementing features on this roadmap issue",
"implementation is via linked children only",
"no product feature claimed complete on this issue alone",
"phased delivery roadmap and epic sequencing",
"this issue is the enduring source of truth",
"enduring source of truth for the",
"canonical product vision — enduring source of truth",
"canonical product vision - enduring source of truth",
"state: vision-active",
"state: roadmap-active",
)
# Explicit epic / umbrella labels (structured evidence preferred over title).
# Explicit epic / umbrella / vision / roadmap labels (structured evidence
# preferred over title). Tracker alone is *not* included — ordinary issues
# may carry a tracker label without being non-implementable containers.
_EPIC_LABELS: frozenset[str] = frozenset(
{
"type:epic",
@@ -161,6 +183,16 @@ _EPIC_LABELS: frozenset[str] = frozenset(
"scope:epic",
"type:umbrella",
"umbrella",
"kind:umbrella",
"scope:umbrella",
"type:vision",
"vision",
"kind:vision",
"scope:vision",
"type:roadmap",
"roadmap",
"kind:roadmap",
"scope:roadmap",
}
)
@@ -222,16 +254,45 @@ class WorkCandidate:
}
def _title_container_prefix(title_l: str) -> str | None:
"""Return a coordination-title prefix token if *title_l* uses one (#854).
Title prefixes alone never exclude; they only corroborate body/label
evidence. Ordinary issues may say "roadmap" or "vision" mid-title.
"""
for prefix, token in (
("epic:", "title_epic_prefix"),
("epic ", "title_epic_prefix"),
("umbrella:", "title_umbrella_prefix"),
("umbrella ", "title_umbrella_prefix"),
("roadmap:", "title_roadmap_prefix"),
("roadmap ", "title_roadmap_prefix"),
("product vision:", "title_vision_prefix"),
("product vision ", "title_vision_prefix"),
("vision:", "title_vision_prefix"),
("vision ", "title_vision_prefix"),
):
if title_l.startswith(prefix):
return token
return None
def classify_epic_or_child_only_container(
c: WorkCandidate,
) -> tuple[bool, str | None]:
"""Return whether *c* is an epic / child-only implementation container (#844).
"""Return whether *c* is a non-implementable coordination container (#844/#854).
Exclusion uses structured evidence first (labels, body scope language).
A bare title containing the word "epic" is **not** enough — ordinary
implementable issues may mention epics incidentally. A title that is
explicitly prefixed ``Epic:`` only counts when the body also proves
child-only / no-direct-implementation scope (or an epic label is present).
A bare title containing the words "epic", "roadmap", "vision", or
"umbrella" is **not** enough — ordinary implementable issues may mention
those terms incidentally. Explicit title prefixes (``Epic:``, ``Roadmap:``,
``Product vision:``, ``Umbrella:``) only count when the body also proves
child-only / no-direct-implementation scope (or a container label is
present).
Covers epic, product-vision, phased-roadmap, umbrella, and child-only
records so the allocator never assigns coordination containers as direct
author work.
PRs are never classified as containers here (they already have a head).
"""
@@ -245,25 +306,28 @@ def classify_epic_or_child_only_container(
title_l = title.lower()
body_hits = [m for m in _CHILD_ONLY_BODY_MARKERS if m in body_l]
title_epic_prefix = title_l.startswith("epic:") or title_l.startswith("epic ")
title_prefix = _title_container_prefix(title_l)
if epic_label:
detail = f"label={epic_label[0]}"
if body_hits:
detail = f"{detail}; body_marker={body_hits[0]!r}"
if title_prefix:
detail = f"{title_prefix}; {detail}"
return True, detail
if body_hits:
# Body proves child-only / umbrella scope. Title "Epic:" is corroborating
# but not required — containers without the word still exclude.
# Body proves child-only / vision / roadmap / umbrella scope. Title
# prefixes are corroborating but not required — containers without the
# title word still exclude.
detail = f"body_marker={body_hits[0]!r}"
if title_epic_prefix:
detail = f"title_epic_prefix; {detail}"
if title_prefix:
detail = f"{title_prefix}; {detail}"
return True, detail
# Title-only "Epic:" without body scope evidence is insufficient (#844 AC:
# eligibility does not rely solely on the word "Epic" in a title).
# Similarly, incidental "epic" mid-title without markers stays eligible.
# Title-only coordination prefix without body scope evidence is
# insufficient (#844/#854 AC: eligibility does not rely solely on a title
# word). Incidental mid-title mentions without markers stay eligible.
return False, None
+41 -9
View File
@@ -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.
+160
View File
@@ -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).
+1015
View File
File diff suppressed because it is too large Load Diff
+58 -8
View File
@@ -2068,6 +2068,7 @@ import lease_lifecycle # noqa: E402
import lease_policy # noqa: E402
import workflow_dashboard # noqa: E402 # #605 live queue/lease dashboard
import restart_coordinator # noqa: E402 # #658 MCP restart coordinator/impact
import drain_proof # noqa: E402 # #661 pre-restart drain proof and hard gate
import incident_bridge # noqa: E402
import sentry_observability # noqa: E402 (#606 optional Sentry observability)
import sentry_incident_bridge # noqa: E402 (#607 Sentry→Gitea incident bridge)
@@ -22342,6 +22343,8 @@ def gitea_request_mcp_restart(
request_override: bool = False,
session_id: str | None = None,
limit: int = 200,
drain_proof_json: str | None = None,
request_break_glass: bool = False,
) -> dict:
"""Evaluate a proposed MCP restart and return an impact preview (#658).
@@ -22351,10 +22354,15 @@ def gitea_request_mcp_restart(
verdict, so the console (#642/#652) and operators can see what a restart
would disrupt *before* any concurrent LLM work is destroyed.
This tool is **dry-run and never restarts anything.** The mutative apply
path is a separate child gated by a drain proof (non-goal here); calling
with ``dry_run=False`` still performs no restart and reports that apply is
not yet available.
This tool **never restarts a process.** In dry-run (the default) it returns
only the impact preview. With ``dry_run=False`` it enforces the #661 hard
gate: the apply request must present a valid, unexpired, clean drain proof
(``drain_proof_json``) or it is denied and a durable incident descriptor is
returned under ``incident``. Break-glass is the only bypass and is honoured
only when ``request_break_glass`` is set *and* the environment carries
``GITEA_BREAKGLASS_RESTART_AUTHORIZATION``. Even an authorized gate performs
no restart here; actual execution is a further child. The gate outcome is
reported under ``apply_gate`` / ``apply_authorized``.
Operator override authority is read from the process environment
(``GITEA_OPERATOR_RESTART_OVERRIDE_AUTHORIZATION``), never self-asserted by
@@ -22473,12 +22481,54 @@ def gitea_request_mcp_restart(
payload["requesting_session_id"] = sid
payload["operator_override_requested"] = bool(request_override)
payload["operator_override_authorized"] = operator_authorized
# Actual restart execution remains a further child; this tool never restarts
# a process. What #661 adds is the *hard gate*: an apply request (dry_run
# False) must present a valid, unexpired, clean drain proof, or it is denied
# and a durable incident is raised. Break-glass is the only bypass and its
# authorization is read from the environment, never self-asserted.
payload["apply_supported"] = False
if not dry_run:
payload["reasons"] = list(payload.get("reasons") or []) + [
"apply requested but not supported: sanctioned restart apply is "
"gated by a drain proof (separate child); no restart performed (#658)"
]
proof_obj: dict | None = None
proof_parse_error: str | None = None
if drain_proof_json:
try:
parsed = json.loads(drain_proof_json)
proof_obj = parsed if isinstance(parsed, dict) else None
if proof_obj is None:
proof_parse_error = "drain_proof_json is not a JSON object"
except (ValueError, TypeError) as exc:
proof_parse_error = f"invalid drain_proof_json: {_redact(str(exc))}"
break_glass_authorized = bool(
(
os.environ.get("GITEA_BREAKGLASS_RESTART_AUTHORIZATION") or ""
).strip()
)
break_glass = bool(request_break_glass and break_glass_authorized)
expected_fp = drain_proof.impact_fingerprint(report.as_dict())
gate = drain_proof.gate_apply_restart(
proof=proof_obj,
break_glass=break_glass,
expected_impact_fingerprint=expected_fp,
requesting_session_id=sid,
)
gate_payload = gate.as_dict()
if proof_parse_error and not break_glass:
gate_payload["reasons"] = [proof_parse_error] + list(
gate_payload.get("reasons") or []
)
payload["apply_gate"] = gate_payload
payload["apply_authorized"] = gate.allow
payload["break_glass_requested"] = bool(request_break_glass)
payload["break_glass_authorized"] = break_glass_authorized
# Even an authorized gate performs no restart here: execution is a later
# child. The gate proves the apply path *would* be permitted.
payload["reasons"] = list(payload.get("reasons") or []) + list(
gate_payload.get("reasons") or []
)
if not gate.allow and gate.incident is not None:
payload["incident"] = gate.incident
return payload
+907
View File
@@ -0,0 +1,907 @@
"""Tests for the pre-restart drain proof and hard gate (#661).
Covers the acceptance criteria:
1. Restart apply without a proof fails closed.
2. A successful drain produces a verifiable proof.
3. An open unsafe mutation makes the proof fail (multi-session fixture).
4. Pass / fail / expired verification paths.
Plus the security posture: forged/tampered proofs are rejected, break-glass is
the only bypass and is never silent, a stale blast-radius fingerprint rejects a
proof, and no per-process secret ever leaks into a serialized artifact.
"""
from __future__ import annotations
import os
import unittest
from datetime import datetime, timedelta, timezone
import drain_proof as dp
import restart_coordinator as rc
NOW = datetime(2026, 7, 24, 6, 0, 0, tzinfo=timezone.utc)
SECRET = b"unit-test-drain-proof-secret-0123456789abcdef"
def _live_pid() -> int:
return os.getpid()
def _clean_drain_state() -> dict:
"""Every drain action succeeded, no sessions outstanding."""
return {
"assignments_stopped": True,
"checkpoints_complete": True,
"handoffs_verified": True,
"leases_handled": True,
"acks": {}, # no other live sessions to acknowledge
"ack_timeout_policy_applied": False,
}
def _safe_report() -> dict:
"""Impact report with no other live work: a restart here is safe."""
report = rc.evaluate_restart_impact(
{"sessions": [], "leases": [], "inventory_complete": True},
now=NOW,
requesting_session_id="prgs-controller-1-req",
)
return report.as_dict()
def _unsafe_mutation_report() -> dict:
"""Multi-session report: a second session holds a live author mutation."""
sessions = [
{
"session_id": "prgs-controller-1-req",
"role": "controller",
"profile": "prgs-controller",
"pid": _live_pid(),
"status": "active",
"last_heartbeat_at": NOW.isoformat(),
},
{
"session_id": "prgs-author-99",
"role": "author",
"profile": "prgs-author",
"pid": _live_pid(),
"status": "active",
"last_heartbeat_at": NOW.isoformat(),
},
]
leases = [
{
"lease_id": "lease-mut",
"session_id": "prgs-author-99",
"role": "author",
"phase": "implementing",
"work_kind": "issue",
"work_number": 661,
"worktree_path": "branches/issue-661",
"freshness": {"freshness": "active"},
}
]
report = rc.evaluate_restart_impact(
{"sessions": sessions, "leases": leases, "inventory_complete": True},
now=NOW,
requesting_session_id="prgs-controller-1-req",
)
return report.as_dict()
class BuildDrainProofTests(unittest.TestCase):
def test_clean_drain_produces_verifiable_clean_proof(self):
"""AC#2: a successful drain produces a verifiable proof."""
proof = dp.build_drain_proof(
impact_report=_safe_report(),
drain_state=_clean_drain_state(),
requesting_session_id="prgs-controller-1-req",
now=NOW,
secret=SECRET,
)
self.assertTrue(proof.clean)
self.assertEqual(proof.failed_checks, [])
self.assertEqual(
{c.name for c in proof.checks}, set(dp.REQUIRED_CHECKS)
)
result = dp.verify_drain_proof(
proof.as_dict(), now=NOW, secret=SECRET
)
self.assertTrue(result.valid, result.reasons)
self.assertFalse(result.expired)
self.assertFalse(result.tampered)
def test_open_mutation_makes_proof_unclean(self):
"""AC#3: an unsafe mutation still in flight fails the proof."""
proof = dp.build_drain_proof(
impact_report=_unsafe_mutation_report(),
drain_state=_clean_drain_state(),
now=NOW,
secret=SECRET,
)
self.assertFalse(proof.clean)
self.assertIn(dp.CHECK_NO_INFLIGHT_MUTATIONS, proof.failed_checks)
# Leases-handled also fails: the report still shows a disruptive lease.
self.assertIn(dp.CHECK_LEASES_HANDLED, proof.failed_checks)
result = dp.verify_drain_proof(proof.as_dict(), now=NOW, secret=SECRET)
self.assertFalse(result.valid)
def test_incomplete_inventory_fails_no_mutations_check(self):
proof = dp.build_drain_proof(
impact_report={"inventory_complete": False},
drain_state=_clean_drain_state(),
now=NOW,
secret=SECRET,
)
self.assertFalse(proof.clean)
self.assertIn(dp.CHECK_NO_INFLIGHT_MUTATIONS, proof.failed_checks)
def test_missing_checkpoint_flag_fails_closed(self):
state = _clean_drain_state()
del state["checkpoints_complete"]
proof = dp.build_drain_proof(
impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET
)
self.assertFalse(proof.clean)
self.assertIn(dp.CHECK_CHECKPOINTS_COMPLETE, proof.failed_checks)
def test_non_true_flags_fail_closed(self):
"""A truthy-but-not-True value (e.g. the string 'yes') must not pass."""
state = _clean_drain_state()
state["assignments_stopped"] = "yes"
proof = dp.build_drain_proof(
impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET
)
self.assertIn(dp.CHECK_ASSIGNMENTS_STOPPED, proof.failed_checks)
def test_ack_timeout_policy_satisfies_ack_check(self):
state = _clean_drain_state()
state["acks"] = {"prgs-author-99": "pending"}
state["ack_timeout_policy_applied"] = True
proof = dp.build_drain_proof(
impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET
)
names = {c.name: c.passed for c in proof.checks}
self.assertTrue(names[dp.CHECK_ACKS_OR_TIMEOUT])
def test_outstanding_acks_without_timeout_fail(self):
state = _clean_drain_state()
state["acks"] = {"prgs-author-99": "pending"}
state["ack_timeout_policy_applied"] = False
proof = dp.build_drain_proof(
impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET
)
self.assertIn(dp.CHECK_ACKS_OR_TIMEOUT, proof.failed_checks)
def test_all_acked_satisfies_ack_check(self):
state = _clean_drain_state()
state["acks"] = {"prgs-author-99": "acked", "prgs-author-2": "acknowledged"}
proof = dp.build_drain_proof(
impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET
)
names = {c.name: c.passed for c in proof.checks}
self.assertTrue(names[dp.CHECK_ACKS_OR_TIMEOUT])
class VerifyDrainProofTests(unittest.TestCase):
def _clean_proof_dict(self) -> dict:
return dp.build_drain_proof(
impact_report=_safe_report(),
drain_state=_clean_drain_state(),
now=NOW,
secret=SECRET,
).as_dict()
def test_missing_proof_is_invalid(self):
result = dp.verify_drain_proof(None, now=NOW, secret=SECRET)
self.assertFalse(result.valid)
self.assertIsNone(result.proof_id)
def test_expired_proof_is_invalid(self):
"""AC#4: an expired proof fails verification."""
proof = self._clean_proof_dict()
later = NOW + timedelta(seconds=dp.DEFAULT_PROOF_TTL_SECONDS + 1)
result = dp.verify_drain_proof(proof, now=later, secret=SECRET)
self.assertFalse(result.valid)
self.assertTrue(result.expired)
def test_proof_valid_just_before_expiry(self):
proof = self._clean_proof_dict()
almost = NOW + timedelta(seconds=dp.DEFAULT_PROOF_TTL_SECONDS - 1)
result = dp.verify_drain_proof(proof, now=almost, secret=SECRET)
self.assertTrue(result.valid, result.reasons)
def test_wrong_secret_rejected(self):
"""A proof minted in a prior process (different secret) will not verify."""
proof = self._clean_proof_dict()
result = dp.verify_drain_proof(proof, now=NOW, secret=b"other-secret")
self.assertFalse(result.valid)
self.assertTrue(result.tampered)
def test_flipping_clean_flag_is_detected(self):
"""Forging clean=True on an unclean proof breaks the signature."""
unclean = dp.build_drain_proof(
impact_report=_unsafe_mutation_report(),
drain_state=_clean_drain_state(),
now=NOW,
secret=SECRET,
).as_dict()
self.assertFalse(unclean["clean"])
unclean["clean"] = True # forge
result = dp.verify_drain_proof(unclean, now=NOW, secret=SECRET)
self.assertFalse(result.valid)
self.assertTrue(result.tampered)
def test_tampering_a_check_is_detected(self):
unclean = dp.build_drain_proof(
impact_report=_unsafe_mutation_report(),
drain_state=_clean_drain_state(),
now=NOW,
secret=SECRET,
).as_dict()
for c in unclean["checks"]:
if c["name"] == dp.CHECK_NO_INFLIGHT_MUTATIONS:
c["passed"] = True # forge the failing check to pass
result = dp.verify_drain_proof(unclean, now=NOW, secret=SECRET)
self.assertFalse(result.valid)
self.assertTrue(result.tampered)
def test_missing_required_check_rejected(self):
proof = self._clean_proof_dict()
proof["checks"] = [
c for c in proof["checks"] if c["name"] != dp.CHECK_HANDOFFS_OK
]
result = dp.verify_drain_proof(proof, now=NOW, secret=SECRET)
self.assertFalse(result.valid)
def test_stale_fingerprint_rejected(self):
proof = self._clean_proof_dict()
result = dp.verify_drain_proof(
proof,
now=NOW,
secret=SECRET,
expected_impact_fingerprint="deadbeef",
)
self.assertFalse(result.valid)
def test_matching_fingerprint_accepted(self):
report = _safe_report()
proof = dp.build_drain_proof(
impact_report=report,
drain_state=_clean_drain_state(),
now=NOW,
secret=SECRET,
).as_dict()
fp = dp.impact_fingerprint(report)
result = dp.verify_drain_proof(
proof, now=NOW, secret=SECRET, expected_impact_fingerprint=fp
)
self.assertTrue(result.valid, result.reasons)
class GateApplyRestartTests(unittest.TestCase):
def _clean_proof_dict(self) -> dict:
return dp.build_drain_proof(
impact_report=_safe_report(),
drain_state=_clean_drain_state(),
now=NOW,
secret=SECRET,
).as_dict()
def test_apply_without_proof_denied(self):
"""AC#1: restart apply without a proof fails closed + raises incident."""
decision = dp.gate_apply_restart(proof=None, now=NOW, secret=SECRET)
self.assertFalse(decision.allow)
self.assertEqual(decision.verdict, dp.GATE_DENY)
self.assertIsNotNone(decision.incident)
self.assertEqual(
decision.incident["kind"], "restart_drain_gate_denied"
)
def test_apply_with_valid_proof_allowed(self):
decision = dp.gate_apply_restart(
proof=self._clean_proof_dict(), now=NOW, secret=SECRET
)
self.assertTrue(decision.allow)
self.assertEqual(decision.verdict, dp.GATE_ALLOW)
self.assertIsNone(decision.incident)
def test_apply_with_expired_proof_denied_with_incident(self):
later = NOW + timedelta(seconds=dp.DEFAULT_PROOF_TTL_SECONDS + 5)
decision = dp.gate_apply_restart(
proof=self._clean_proof_dict(), now=later, secret=SECRET
)
self.assertFalse(decision.allow)
self.assertIsNotNone(decision.incident)
def test_apply_with_unclean_proof_denied(self):
"""AC#3 at the gate: an unsafe-mutation proof is denied."""
unclean = dp.build_drain_proof(
impact_report=_unsafe_mutation_report(),
drain_state=_clean_drain_state(),
now=NOW,
secret=SECRET,
).as_dict()
decision = dp.gate_apply_restart(proof=unclean, now=NOW, secret=SECRET)
self.assertFalse(decision.allow)
self.assertIsNotNone(decision.incident)
def test_break_glass_allows_without_proof_but_records_bypass(self):
decision = dp.gate_apply_restart(
proof=None, now=NOW, secret=SECRET, break_glass=True
)
self.assertTrue(decision.allow)
self.assertEqual(decision.verdict, dp.GATE_BREAK_GLASS)
self.assertTrue(decision.break_glass)
self.assertIsNone(decision.incident)
self.assertTrue(decision.audit_record["break_glass"])
def test_denied_gate_carries_stale_fingerprint_reason(self):
decision = dp.gate_apply_restart(
proof=self._clean_proof_dict(),
now=NOW,
secret=SECRET,
expected_impact_fingerprint="not-the-fingerprint",
)
self.assertFalse(decision.allow)
class SecretHygieneTests(unittest.TestCase):
def test_secret_never_serialized(self):
proof = dp.build_drain_proof(
impact_report=_safe_report(),
drain_state=_clean_drain_state(),
now=NOW,
secret=SECRET,
)
blob = dp._canonical(proof.as_dict())
self.assertNotIn(SECRET.decode(), blob)
# The signature is a hex digest, not the raw secret.
self.assertNotIn(SECRET.hex(), blob)
def test_incident_descriptor_has_no_secret(self):
decision = dp.gate_apply_restart(proof=None, now=NOW, secret=SECRET)
blob = dp._canonical(decision.incident)
self.assertNotIn(SECRET.decode(), blob)
def _drained_report_with_live_sessions(count: int) -> dict:
"""Report with ``count`` other live sessions but nothing in flight.
Every other checklist item passes against this report, so a failure
isolates the acknowledgement check rather than tripping on mutations.
"""
sessions = [
{
"session_id": "prgs-controller-1-req",
"role": "controller",
"profile": "prgs-controller",
"pid": _live_pid(),
"status": "active",
"last_heartbeat_at": NOW.isoformat(),
}
]
for index in range(count):
sessions.append(
{
"session_id": f"prgs-author-{index}",
"role": "author",
"profile": "prgs-author",
"pid": _live_pid(),
"status": "active",
"last_heartbeat_at": NOW.isoformat(),
}
)
report = rc.evaluate_restart_impact(
{"sessions": sessions, "leases": [], "inventory_complete": True},
now=NOW,
requesting_session_id="prgs-controller-1-req",
)
return report.as_dict()
class AcknowledgementFailClosedTests(unittest.TestCase):
"""Acknowledgement evidence must fail closed unless explicitly verified.
Regression cover for the reviewed fail-open on PR #882: an absent ``acks``
key collapsed to ``{}`` and was read as "no other live sessions required to
acknowledge", so a proof minted clean and the restart gate allowed while the
impact report still showed other live sessions.
"""
def _state(self, **overrides) -> dict:
state = _clean_drain_state()
state.pop("acks", None)
state["ack_timeout_policy_applied"] = False
state.update(overrides)
return state
def _acks_check(self, proof) -> dp.DrainCheck:
return next(c for c in proof.checks if c.name == dp.CHECK_ACKS_OR_TIMEOUT)
def _build(self, report: dict, state: dict):
return dp.build_drain_proof(
impact_report=report, drain_state=state, now=NOW, secret=SECRET
)
def assertAcksFailClosed(self, report: dict, state: dict) -> None:
proof = self._build(report, state)
self.assertFalse(self._acks_check(proof).passed)
self.assertIn(dp.CHECK_ACKS_OR_TIMEOUT, proof.failed_checks)
self.assertFalse(proof.clean)
# --- missing / null / empty / malformed ------------------------------
def test_missing_acks_key_with_live_sessions_fails_closed(self):
"""The exact reviewed defect: absent key, three other live sessions."""
report = _drained_report_with_live_sessions(3)
self.assertEqual(report["counts"]["sessions_live_other"], 3)
state = self._state()
self.assertNotIn("acks", state)
proof = self._build(report, state)
check = self._acks_check(proof)
self.assertFalse(check.passed)
self.assertNotIn("no other live sessions", check.detail)
self.assertIn("fail closed", check.detail)
self.assertFalse(proof.clean)
self.assertEqual(proof.failed_checks, [dp.CHECK_ACKS_OR_TIMEOUT])
def test_none_acks_with_live_sessions_fails_closed(self):
self.assertAcksFailClosed(
_drained_report_with_live_sessions(2), self._state(acks=None)
)
def test_empty_acks_with_live_sessions_fails_closed(self):
self.assertAcksFailClosed(
_drained_report_with_live_sessions(1), self._state(acks={})
)
def test_malformed_acks_fail_closed(self):
for malformed in ([], "ack", 7, ("ack",), True):
with self.subTest(malformed=malformed):
self.assertAcksFailClosed(
_drained_report_with_live_sessions(1),
self._state(acks=malformed),
)
# --- stale / unproven values -----------------------------------------
def test_stale_or_unproven_ack_values_fail_closed(self):
for value in ("pending", "stale", "unknown", "", None, True, 1, NOW):
with self.subTest(value=value):
self.assertAcksFailClosed(
_drained_report_with_live_sessions(1),
self._state(acks={"prgs-author-0": value}),
)
def test_partial_coverage_fails_closed(self):
"""Fewer acknowledgements than the report's live-session count."""
self.assertAcksFailClosed(
_drained_report_with_live_sessions(3),
self._state(acks={"prgs-author-0": "ack"}),
)
def test_one_unacked_entry_among_many_fails_closed(self):
self.assertAcksFailClosed(
_drained_report_with_live_sessions(2),
self._state(acks={"prgs-author-0": "ack", "prgs-author-1": "pending"}),
)
def test_unproven_live_session_count_fails_closed(self):
"""A missing or malformed count cannot prove nobody had to acknowledge."""
malformed_counts = (
None,
{},
{"sessions_live_other": None},
{"sessions_live_other": "3"},
{"sessions_live_other": -1},
{"sessions_live_other": True},
)
for counts in malformed_counts:
with self.subTest(counts=counts):
report = _drained_report_with_live_sessions(0)
if counts is None:
report.pop("counts", None)
else:
report["counts"] = counts
self.assertAcksFailClosed(report, self._state())
# --- valid evidence still passes -------------------------------------
def test_complete_valid_acks_pass(self):
report = _drained_report_with_live_sessions(2)
state = self._state(
acks={"prgs-author-0": "ack", "prgs-author-1": "acknowledged"}
)
proof = self._build(report, state)
self.assertTrue(self._acks_check(proof).passed)
self.assertTrue(proof.clean)
self.assertEqual(proof.failed_checks, [])
def test_no_other_live_sessions_still_passes(self):
"""Intended behavior retained: zero live sessions needs no acks."""
report = _drained_report_with_live_sessions(0)
self.assertEqual(report["counts"]["sessions_live_other"], 0)
proof = self._build(report, self._state())
check = self._acks_check(proof)
self.assertTrue(check.passed)
self.assertIn("sessions_live_other=0", check.detail)
self.assertTrue(proof.clean)
# --- timeout policy cannot become a second fail-open ------------------
def test_unproven_timeout_policy_cannot_open_the_gate(self):
for value in (None, "true", "yes", 1, "True", [], {}):
with self.subTest(value=value):
self.assertAcksFailClosed(
_drained_report_with_live_sessions(2),
self._state(ack_timeout_policy_applied=value),
)
def test_explicit_timeout_policy_permits(self):
proof = self._build(
_drained_report_with_live_sessions(2),
self._state(ack_timeout_policy_applied=True),
)
check = self._acks_check(proof)
self.assertTrue(check.passed)
self.assertIn("timeout policy", check.detail)
self.assertTrue(proof.clean)
# --- the gate itself must deny ---------------------------------------
def test_failed_ack_check_denies_the_restart_gate(self):
report = _drained_report_with_live_sessions(3)
proof = self._build(report, self._state())
self.assertFalse(proof.clean)
decision = dp.gate_apply_restart(
proof=proof.as_dict(),
now=NOW,
secret=SECRET,
expected_impact_fingerprint=dp.impact_fingerprint(report),
)
self.assertFalse(decision.allow)
self.assertEqual(decision.verdict, dp.GATE_DENY)
self.assertIsNotNone(decision.incident)
def test_unclean_ack_proof_fails_verification(self):
report = _drained_report_with_live_sessions(3)
proof = self._build(report, self._state())
result = dp.verify_drain_proof(
proof.as_dict(),
now=NOW,
secret=SECRET,
expected_impact_fingerprint=dp.impact_fingerprint(report),
)
self.assertFalse(result.valid)
self.assertFalse(result.clean)
def _identity_report(*, requester: str, others: tuple[str, ...]) -> dict:
"""Report with explicitly named requester and other live sessions.
Unlike :func:`_drained_report_with_live_sessions`, the session ids are
chosen by the caller so a test can supply acknowledgements for the *wrong*
identities while keeping the count correct.
"""
sessions = [
{
"session_id": requester,
"role": "controller",
"profile": "prgs-controller",
"pid": _live_pid(),
"status": "active",
"last_heartbeat_at": NOW.isoformat(),
}
]
for session_id in others:
sessions.append(
{
"session_id": session_id,
"role": "author",
"profile": "prgs-author",
"pid": _live_pid(),
"status": "active",
"last_heartbeat_at": NOW.isoformat(),
}
)
report = rc.evaluate_restart_impact(
{"sessions": sessions, "leases": [], "inventory_complete": True},
now=NOW,
requesting_session_id=requester,
)
return report.as_dict()
class AcknowledgementIdentityBindingTests(unittest.TestCase):
"""Acknowledgement coverage must be bound to session identity, not counted.
Regression cover for the second reviewed fail-open on PR #882 (review 582,
blocker B1): coverage compared ``acked_count`` against
``counts.sessions_live_other``, so acknowledgements supplied for the
requesting session and for ids that do not exist satisfied the obligations
of the live sessions that never answered. The required identities are
carried by the report itself — ``ack_state`` keys and ``affected_sessions``
filtered on ``live and not is_requester`` — and only an acknowledgement
keyed by one of those ids may count for it.
"""
def _state(self, **overrides) -> dict:
state = _clean_drain_state()
state.pop("acks", None)
state["ack_timeout_policy_applied"] = False
state.update(overrides)
return state
def _acks_check(self, proof) -> dp.DrainCheck:
return next(c for c in proof.checks if c.name == dp.CHECK_ACKS_OR_TIMEOUT)
def _build(self, report: dict, state: dict):
return dp.build_drain_proof(
impact_report=report, drain_state=state, now=NOW, secret=SECRET
)
def assertAcksFailClosed(self, report: dict, state: dict) -> dp.DrainCheck:
"""Failure must propagate through the check, the proof, and the gate."""
proof = self._build(report, state)
check = self._acks_check(proof)
self.assertFalse(check.passed)
self.assertFalse(proof.clean)
self.assertIn(dp.CHECK_ACKS_OR_TIMEOUT, proof.failed_checks)
decision = dp.gate_apply_restart(
proof=proof.as_dict(),
now=NOW,
secret=SECRET,
expected_impact_fingerprint=dp.impact_fingerprint(report),
)
self.assertEqual(decision.verdict, dp.GATE_DENY)
self.assertFalse(decision.allow)
return check
# --- the reviewer's exact reproduction --------------------------------
def test_requester_plus_unknown_id_cannot_satisfy_two_live_sessions(self):
"""Review 582 B1 verbatim: requester + a nonexistent session.
``sessions_live_other=2`` with ``ack_state`` naming ``other-0`` and
``other-1``; the drain state supplies an acknowledgement from the
requesting session itself and from a session that does not exist. The
count matches, the identities do not.
"""
report = _identity_report(requester="req", others=("other-0", "other-1"))
self.assertEqual(report["counts"]["sessions_live_other"], 2)
self.assertEqual(
report["ack_state"], {"other-0": "pending", "other-1": "pending"}
)
state = self._state(acks={"req": "ack", "totally-bogus-session": "ack"})
check = self.assertAcksFailClosed(report, state)
self.assertIn("other-0", check.detail)
self.assertIn("other-1", check.detail)
self.assertIn("fail closed", check.detail)
# --- wrong / unknown / requester identities ---------------------------
def test_sufficient_count_of_wrong_ids_fails_closed(self):
"""Right cardinality, wrong identities: two acks, neither required."""
report = _identity_report(requester="req", others=("other-0", "other-1"))
state = self._state(acks={"ghost-a": "ack", "ghost-b": "ack"})
check = self.assertAcksFailClosed(report, state)
self.assertIn("do not count", check.detail)
def test_more_acks_than_required_still_fails_on_wrong_ids(self):
"""Coverage cannot be bought with volume: five acks, none required."""
report = _identity_report(requester="req", others=("other-0", "other-1"))
state = self._state(acks={f"ghost-{i}": "acknowledged" for i in range(5)})
self.assertAcksFailClosed(report, state)
def test_partial_identity_match_fails_closed(self):
"""One required id acknowledged, the rest padded with unknown ids."""
report = _identity_report(
requester="req", others=("other-0", "other-1", "other-2")
)
state = self._state(
acks={"other-0": "ack", "ghost-1": "ack", "ghost-2": "ack"}
)
check = self.assertAcksFailClosed(report, state)
self.assertIn("other-1", check.detail)
self.assertIn("other-2", check.detail)
def test_requester_ack_never_satisfies_another_sessions_obligation(self):
"""The requester is excluded from the required set and stays excluded."""
report = _identity_report(requester="req", others=("other-0",))
requester_rows = [s for s in report["affected_sessions"] if s["is_requester"]]
self.assertEqual([s["session_id"] for s in requester_rows], ["req"])
self.assertNotIn("req", report["ack_state"])
check = self.assertAcksFailClosed(report, self._state(acks={"req": "ack"}))
self.assertIn("other-0", check.detail)
def test_fabricated_ids_do_not_count_toward_coverage(self):
report = _identity_report(requester="req", others=("other-0",))
for bogus in ("", " ", "other-0 extra", "OTHER-0", "other-01", "0"):
with self.subTest(bogus=bogus):
self.assertAcksFailClosed(report, self._state(acks={bogus: "ack"}))
# --- per-session state must be explicitly valid ------------------------
def test_unproven_per_session_states_fail_closed(self):
"""A required id present but not explicitly acknowledged fails closed."""
report = _identity_report(requester="req", others=("other-0", "other-1"))
for value in ("pending", "stale", "unknown", "", None, True, 1, NOW):
with self.subTest(value=value):
self.assertAcksFailClosed(
report,
self._state(acks={"other-0": "ack", "other-1": value}),
)
def test_report_ack_state_placeholder_is_never_read_as_an_ack(self):
"""``ack_state`` values are the report's own placeholders, not evidence."""
report = _identity_report(requester="req", others=("other-0",))
report["ack_state"] = {"other-0": "ack"}
self.assertAcksFailClosed(report, self._state())
# --- missing / malformed / contradictory identity evidence -------------
def test_missing_identity_evidence_fails_closed(self):
report = _identity_report(requester="req", others=("other-0",))
report.pop("ack_state", None)
report.pop("affected_sessions", None)
check = self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"}))
self.assertIn("no session-identity evidence", check.detail)
def test_malformed_ack_state_fails_closed(self):
for malformed in ([], "other-0", 7, None, ("other-0",)):
with self.subTest(malformed=malformed):
report = _identity_report(requester="req", others=("other-0",))
report["ack_state"] = malformed
self.assertAcksFailClosed(
report, self._state(acks={"other-0": "ack"})
)
def test_non_string_ack_state_key_fails_closed(self):
report = _identity_report(requester="req", others=("other-0",))
report["ack_state"] = {7: "pending"}
self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"}))
def test_malformed_affected_sessions_fails_closed(self):
for malformed in ("sessions", 7, {"session_id": "other-0"}, [None], [7]):
with self.subTest(malformed=malformed):
report = _identity_report(requester="req", others=("other-0",))
report.pop("ack_state", None)
report["affected_sessions"] = malformed
self.assertAcksFailClosed(
report, self._state(acks={"other-0": "ack"})
)
def test_affected_sessions_without_explicit_booleans_fails_closed(self):
"""``live``/``is_requester`` must be real booleans, never inferred."""
report = _identity_report(requester="req", others=("other-0",))
report.pop("ack_state", None)
for row in report["affected_sessions"]:
if row["session_id"] == "other-0":
row["is_requester"] = "false"
self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"}))
def test_affected_sessions_missing_live_flag_fails_closed(self):
report = _identity_report(requester="req", others=("other-0",))
report.pop("ack_state", None)
for row in report["affected_sessions"]:
row.pop("live", None)
self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"}))
def test_contradictory_ack_state_and_affected_sessions_fails_closed(self):
"""Both views present and disagreeing is unresolvable, not a tie-break."""
report = _identity_report(requester="req", others=("other-0", "other-1"))
report["ack_state"] = {"other-0": "pending", "other-9": "pending"}
check = self.assertAcksFailClosed(
report, self._state(acks={"other-0": "ack", "other-9": "ack"})
)
self.assertIn("contradicts itself", check.detail)
def test_identity_count_mismatch_fails_closed(self):
"""Identity evidence that cannot be reconciled with the count denies."""
report = _identity_report(requester="req", others=("other-0", "other-1"))
report["counts"] = dict(report["counts"], sessions_live_other=1)
check = self.assertAcksFailClosed(
report, self._state(acks={"other-0": "ack", "other-1": "ack"})
)
self.assertIn("cannot be reconciled", check.detail)
def test_broken_identity_evidence_outranks_timeout_policy(self):
"""The sanctioned timeout path cannot paper over an unreadable report."""
report = _identity_report(requester="req", others=("other-0",))
report["ack_state"] = "not-a-mapping"
self.assertAcksFailClosed(report, self._state(ack_timeout_policy_applied=True))
# --- legitimate success is preserved -----------------------------------
def test_every_required_session_acknowledged_passes(self):
report = _identity_report(
requester="req", others=("other-0", "other-1", "other-2")
)
state = self._state(
acks={
"other-0": "ack",
"other-1": "acked",
"other-2": "acknowledged",
}
)
proof = self._build(report, state)
check = self._acks_check(proof)
self.assertTrue(check.passed)
self.assertTrue(proof.clean)
self.assertEqual(proof.failed_checks, [])
self.assertIn("acknowledged by identity", check.detail)
decision = dp.gate_apply_restart(
proof=proof.as_dict(),
now=NOW,
secret=SECRET,
expected_impact_fingerprint=dp.impact_fingerprint(report),
)
self.assertEqual(decision.verdict, dp.GATE_ALLOW)
self.assertTrue(decision.allow)
def test_required_session_ack_tolerates_surrounding_whitespace(self):
report = _identity_report(requester="req", others=("other-0",))
proof = self._build(report, self._state(acks={" other-0 ": " ACK "}))
self.assertTrue(self._acks_check(proof).passed)
self.assertTrue(proof.clean)
def test_no_other_live_sessions_still_passes_with_identity_evidence(self):
report = _identity_report(requester="req", others=())
self.assertEqual(report["counts"]["sessions_live_other"], 0)
self.assertEqual(report["ack_state"], {})
proof = self._build(report, self._state())
check = self._acks_check(proof)
self.assertTrue(check.passed)
self.assertIn("sessions_live_other=0", check.detail)
self.assertTrue(proof.clean)
def test_explicit_timeout_policy_retains_intended_behavior(self):
"""Valid, correctly typed timeout evidence still permits the check."""
report = _identity_report(requester="req", others=("other-0", "other-1"))
proof = self._build(report, self._state(ack_timeout_policy_applied=True))
check = self._acks_check(proof)
self.assertTrue(check.passed)
self.assertIn("timeout policy", check.detail)
self.assertTrue(proof.clean)
def test_timeout_policy_still_strictly_typed_under_identity_binding(self):
report = _identity_report(requester="req", others=("other-0",))
for value in (None, "true", "True", 1, [], {}):
with self.subTest(value=value):
self.assertAcksFailClosed(
report, self._state(ack_timeout_policy_applied=value)
)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,378 @@
"""Allocator semantic container exclusion for vision/roadmap/umbrella (#854).
#844 excluded epic / child-only containers (live #631) but product-vision
(#652), phased-roadmap (#653), and umbrella (#655) coordination records still
ranked as implementable work. This module is the live-equivalent canary:
* #631 / #652 / #653 / #655-shaped records are all excluded in one inventory.
* Independently executable children remain eligible and can be selected.
* Ordinary issues that merely mention vision / roadmap / umbrella stay eligible.
* Excluded containers never receive assignments or workflow leases.
* Structured skip reason ``epic_or_child_only_container`` is reported.
* Candidate-set fingerprint remains stable after exclusions.
"""
from __future__ import annotations
import os
import tempfile
import unittest
from allocator_service import (
OUTCOME_ASSIGNED,
OUTCOME_PREVIEW,
SKIP_EPIC_OR_CHILD_ONLY_CONTAINER,
WorkCandidate,
allocate_next_work,
candidate_set_fingerprint,
classify_epic_or_child_only_container,
)
from control_plane_db import ControlPlaneDB
REMOTE = "prgs"
ORG = "Scaled-Tech-Consulting"
REPO = "Gitea-Tools"
# Minimal bodies mirroring live coordination records (not full issue text).
_EPIC_631_BODY = """
## Scope (umbrella)
This epic owns the **product roadmap and linkage** for the Web Console.
Implementation is delivered via child issues only.
## Explicit non-goals
* Do not implement product features in this epic issue itself.
* No product feature implementation is claimed complete solely on this epic.
"""
_VISION_652_BODY = """
## Canonical product vision — enduring source of truth
**This issue is the enduring source of truth for the MCP Control Plane Web Console product vision.**
## Implementation linkage
* **Do not implement features on this issue.**
* Sequencing: roadmap issue + #631 children.
## Canonical issue state
```text
STATE: vision-active
WHO_IS_NEXT: controller (triage/ordering) / author (implementation of linked children only)
```
"""
_ROADMAP_653_BODY = """
## Purpose
This issue is the **phased delivery roadmap and epic sequencing** for the MCP Control Plane Web Console.
## Non-goals
* Implementing features on this roadmap issue.
* Deleting vision items by omitting them from phases without #652 change log.
## Canonical issue state
```text
STATE: roadmap-active
WHO_IS_NEXT: author
```
"""
_UMBRELLA_655_BODY = """
## Scope (umbrella)
This issue owns the **canonical restart-governance program**. Implementation is via linked children only.
## Acceptance criteria (umbrella)
6. No product feature claimed complete on this issue alone.
"""
_CHILD_BODY = """
## Problem
Operators need a workflow-event timeline model for Phase 1.
## Acceptance criteria
- [ ] Timeline model API exists
"""
def _issue(
number: int,
*,
title: str = "",
body: str = "",
labels: tuple[str, ...] = ("status:ready", "type:feature"),
priority: int = 20,
) -> WorkCandidate:
return WorkCandidate(
kind="issue",
number=number,
state="open",
labels=labels,
title=title or f"issue {number}",
body=body,
priority=priority,
)
def _live_shaped_containers() -> list[WorkCandidate]:
return [
_issue(
631,
title="Epic: MCP Control Plane Web Console",
body=_EPIC_631_BODY,
),
_issue(
652,
title="Product vision: MCP Control Plane Web Console (canonical)",
body=_VISION_652_BODY,
),
_issue(
653,
title="Roadmap: MCP Control Plane Web Console (phased delivery)",
body=_ROADMAP_653_BODY,
),
_issue(
655,
title="Umbrella: Governed MCP restart coordination and zero-disruption recovery",
body=_UMBRELLA_655_BODY,
),
]
class ClassifySemanticContainersTest(unittest.TestCase):
def test_652_vision_is_container(self) -> None:
c = _issue(
652,
title="Product vision: MCP Control Plane Web Console (canonical)",
body=_VISION_652_BODY,
)
is_c, detail = classify_epic_or_child_only_container(c)
self.assertTrue(is_c)
self.assertIsNotNone(detail)
self.assertIn("body_marker", detail or "")
def test_653_roadmap_is_container(self) -> None:
c = _issue(
653,
title="Roadmap: MCP Control Plane Web Console (phased delivery)",
body=_ROADMAP_653_BODY,
)
is_c, detail = classify_epic_or_child_only_container(c)
self.assertTrue(is_c)
self.assertIn("body_marker", detail or "")
def test_655_umbrella_is_container(self) -> None:
c = _issue(
655,
title="Umbrella: Governed MCP restart coordination and zero-disruption recovery",
body=_UMBRELLA_655_BODY,
)
is_c, detail = classify_epic_or_child_only_container(c)
self.assertTrue(is_c)
self.assertIn("body_marker", detail or "")
def test_631_still_container_after_854(self) -> None:
c = _issue(
631,
title="Epic: MCP Control Plane Web Console",
body=_EPIC_631_BODY,
)
is_c, detail = classify_epic_or_child_only_container(c)
self.assertTrue(is_c)
self.assertIn("body_marker", detail or "")
def test_incidental_vision_roadmap_umbrella_words_not_container(self) -> None:
cases = (
(
"Document vision handoff conventions",
"Update docs so implementable issues that mention a vision "
"remain independently executable.",
),
(
"Clarify roadmap sequencing notes",
"Write a short note about how the roadmap issue relates to children.",
),
(
"Umbrella recovery checklist for authors",
"Authors should still implement the concrete recovery fix here.",
),
(
"Product vision wording in the help text",
"Fix a typo in the operator-facing help string that says product vision.",
),
)
for title, body in cases:
with self.subTest(title=title):
c = _issue(900, title=title, body=body)
is_c, detail = classify_epic_or_child_only_container(c)
self.assertFalse(is_c)
self.assertIsNone(detail)
def test_title_prefix_alone_not_container(self) -> None:
for title in (
"Epic: something mentioned only in title",
"Roadmap: title only without body scope",
"Product vision: title only without body scope",
"Umbrella: title only without body scope",
):
with self.subTest(title=title):
c = _issue(
901,
title=title,
body="Implement a concrete fix for the allocator skip list.",
)
is_c, detail = classify_epic_or_child_only_container(c)
self.assertFalse(is_c)
self.assertIsNone(detail)
def test_roadmap_label_alone_is_container(self) -> None:
c = _issue(
902,
title="Console delivery sequencing",
body="Track phased delivery only.",
labels=("status:ready", "type:roadmap"),
)
is_c, detail = classify_epic_or_child_only_container(c)
self.assertTrue(is_c)
self.assertIn("type:roadmap", detail or "")
def test_child_referencing_parent_policy_stays_eligible(self) -> None:
"""Children may quote parent policy without becoming containers."""
c = _issue(
637,
title="Web Console: Workflow-event timeline model (Phase 1)",
body=(
_CHILD_BODY
+ "\n\nParent #652 says do not implement on the vision issue; "
"this child is the implementable unit."
),
)
is_c, _ = classify_epic_or_child_only_container(c)
self.assertFalse(is_c)
class AllocateSemanticContainerExclusionTest(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
def _alloc(self, candidates, **kwargs):
defaults = dict(
session_id="sess-854",
role="author",
remote=REMOTE,
org=ORG,
repo=REPO,
profile_name="prgs-author",
username="jcwalker3",
claims={},
apply=False,
)
defaults.update(kwargs)
return allocate_next_work(self.db, candidates=candidates, **defaults)
def test_live_equivalent_canary_excludes_all_containers_selects_child(self) -> None:
containers = _live_shaped_containers()
child = _issue(
637,
title="Web Console: Workflow-event timeline model (Phase 1)",
body=_CHILD_BODY,
)
inventory = containers + [child]
res = self._alloc(inventory, apply=False)
self.assertTrue(res["success"], res)
self.assertEqual(res["outcome"], OUTCOME_PREVIEW)
self.assertEqual(res["selected"]["number"], 637)
skipped = {s["number"]: s for s in res["skipped"]}
for number in (631, 652, 653, 655):
self.assertIn(number, skipped, res["skipped"])
self.assertEqual(
skipped[number]["reason_code"],
SKIP_EPIC_OR_CHILD_ONLY_CONTAINER,
)
self.assertIn(
SKIP_EPIC_OR_CHILD_ONLY_CONTAINER, skipped[number]["reason"]
)
def test_containers_cannot_receive_assignment_or_lease(self) -> None:
containers = _live_shaped_containers()
res = self._alloc(containers, apply=True)
self.assertTrue(res["success"], res)
self.assertNotEqual(res["outcome"], OUTCOME_ASSIGNED)
self.assertIsNone(res.get("assignment"))
self.assertIsNone(res.get("selected"))
skipped = {s["number"]: s for s in res["skipped"]}
for number in (631, 652, 653, 655):
self.assertEqual(
skipped[number]["reason_code"],
SKIP_EPIC_OR_CHILD_ONLY_CONTAINER,
)
leases = []
if hasattr(self.db, "list_active_leases"):
leases = self.db.list_active_leases(
remote=REMOTE, org=ORG, repo=REPO
)
if not leases and hasattr(self.db, "list_leases"):
leases = self.db.list_leases(remote=REMOTE, org=ORG, repo=REPO)
for lease in leases or []:
work_number = (
lease.get("work_number") if isinstance(lease, dict) else None
)
self.assertNotIn(work_number, {631, 652, 653, 655})
def test_apply_selects_child_not_container(self) -> None:
containers = _live_shaped_containers()
child = _issue(
637,
title="Web Console: Workflow-event timeline model (Phase 1)",
body=_CHILD_BODY,
)
res = self._alloc(containers + [child], apply=True)
self.assertTrue(res["success"], res)
self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
self.assertEqual(res["selected"]["number"], 637)
self.assertEqual(res["assignment"]["work_number"], 637)
def test_fingerprint_stable_with_containers_present(self) -> None:
containers = _live_shaped_containers()
child = _issue(
637,
title="Web Console: Workflow-event timeline model (Phase 1)",
body=_CHILD_BODY,
)
inventory = containers + [child]
fp_before = candidate_set_fingerprint(inventory)
res = self._alloc(inventory, apply=False)
self.assertTrue(res["success"], res)
self.assertEqual(res["selected"]["number"], 637)
# Allocator reports the same CAS fingerprint for the full candidate set.
reported = res.get("candidate_set_fingerprint")
self.assertEqual(reported, fp_before)
# Re-fingerprint of the same inventory is byte-stable.
self.assertEqual(candidate_set_fingerprint(inventory), fp_before)
def test_incidental_mentions_remain_eligible(self) -> None:
ordinary = _issue(
700,
title="Document roadmap handoff conventions",
body="Write runbook text about vision vs roadmap vs child issues.",
)
res = self._alloc([ordinary], apply=False)
self.assertTrue(res["success"], res)
self.assertEqual(res["selected"]["number"], 700)
self.assertEqual(res["skipped"], [])
if __name__ == "__main__":
unittest.main()
+917
View File
@@ -0,0 +1,917 @@
"""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 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",
}
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."""
def _drifting(*, request, apply, expected_candidate_set_fingerprint=None):
return {
"outcome": (
allocator_service.OUTCOME_ASSIGNED
if apply
else allocator_service.OUTCOME_PREVIEW
),
"selected": _selection(number=999 if apply else 643),
"assignment": {"assignment_id": "asn-wrong"} if apply else None,
"candidate_set_fingerprint": "fp-test",
}
result = request_service.apply_request(
_request(number=643),
principal=_principal(console_authz.OPERATOR),
confirm=True,
allocator=_drifting,
claims_source=_no_claims,
)
self.assertFalse(result["ok"])
self.assertIsNone(result["assignment"])
self.assertFalse(result["mutation_performed"])
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 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="<script>alert(1)</script>"),
principal=_principal(console_authz.OPERATOR),
allocator=_fake_allocator(),
claims_source=_no_claims,
audit=False,
)
html = render_requests_page(preview=preview)
self.assertNotIn("<script>alert(1)</script>", html)
self.assertIn("&lt;script&gt;", 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()
+116
View File
@@ -67,6 +67,8 @@ from webui.system_health import (
snapshot_to_dict as system_health_to_dict,
)
from webui.system_health_views import render_system_health_page
from webui import request_service
from webui.request_views import render_requests_page
_READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
_AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
@@ -722,6 +724,109 @@ async def api_v1_analytics_ingest(request: Request) -> JSONResponse:
)
def _default_request_scope() -> dict[str, str]:
"""Resolve remote/org/repo from the project registry for request forms.
Returns an empty mapping when the registry cannot be read, which makes
``parse_request`` reject a request that did not name its own scope rather
than letting it default to some other repository.
"""
from webui.queue_loader import _host_from_url # host normalisation helper
registry, error = _load_project_registry()
if error is not None or not registry.projects:
return {}
project = registry.projects[0]
host = _host_from_url(project.remote_host)
return {
"remote": _derive_remote(host),
"org": project.gitea_owner or "",
"repo": project.repo_name or "",
}
async def _request_payload(request: Request) -> dict[str, object]:
"""Read a request body as JSON or form-encoded. Never raises."""
content_type = (request.headers.get("content-type") or "").lower()
if "application/json" in content_type:
try:
body = await request.json()
except Exception:
return {}
return dict(body) if isinstance(body, dict) else {}
try:
form = await request.form()
except Exception:
return {}
return {key: form[key] for key in form}
async def requests_page(request: Request) -> HTMLResponse:
"""Operator request form and intent preview (#643).
POST here only ever *previews*. Initiation is a separate confirmed call to
``/api/v1/requests/apply`` so that submitting this form cannot reserve
work as a side effect.
"""
submitted: dict[str, object] = {}
preview = None
error = None
if request.method == "POST":
submitted = await _request_payload(request)
work_request, error = request_service.parse_request(
submitted, default_scope=_default_request_scope()
)
if work_request is not None:
preview = request_service.preview_request(
work_request,
principal=resolve_principal(headers=dict(request.headers)),
)
return HTMLResponse(
render_requests_page(
preview=preview, error=error, submitted=submitted
)
)
async def api_v1_request_preview(request: Request) -> JSONResponse:
"""Dry-run authorization and intent preview for a work request (#643)."""
payload = await _request_payload(request)
work_request, error = request_service.parse_request(
payload, default_scope=_default_request_scope()
)
if work_request is None:
return JSONResponse(error.to_dict(), status_code=400)
preview = request_service.preview_request(
work_request,
principal=resolve_principal(headers=dict(request.headers)),
)
return JSONResponse(
preview.to_dict(), status_code=200 if preview.authorized else 403
)
async def api_v1_request_apply(request: Request) -> JSONResponse:
"""Initiate a previewed work request through the allocator (#643).
Fail-closed at every step: unauthorized, unconfirmed, not-next-safe, and
already-claimed all return without attempting an assignment.
"""
payload = await _request_payload(request)
work_request, error = request_service.parse_request(
payload, default_scope=_default_request_scope()
)
if work_request is None:
return JSONResponse(error.to_dict(), status_code=400)
confirm = _truthy_flag(str(payload.get("confirm") or ""))
result = request_service.apply_request(
work_request,
principal=resolve_principal(headers=dict(request.headers)),
confirm=confirm,
)
status = int(result.pop("status_code", 403))
return JSONResponse(result, status_code=status)
async def method_not_allowed(request: Request, _exc: Exception) -> Response:
path = request.url.path
if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
@@ -786,6 +891,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(
+71 -8
View File
@@ -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,
)
+1
View File
@@ -45,6 +45,7 @@ NAV_GROUPS: tuple[NavGroup, ...] = (
NavItem("/queue", "Queue"),
NavItem("/leases", "Leases"),
NavItem("/actions", "Actions"),
NavItem("/requests", "Requests"),
)),
NavGroup("Runtime/Sessions", (
NavItem("/runtime", "Runtime health"),
+967
View File
@@ -0,0 +1,967 @@
"""Operator work-request preview and initiation (#643, Phase 2).
An operator's alternative to pasting a role prompt into a terminal. A
*request* names three things — the role to run as, the issue or PR to run
against, and what the operator intends — and this module answers two questions
about it:
* **Preview** (:func:`preview_request`) — would that request be authorized,
is the work unit actually free, is it the next safe thing that role should
touch, and which actions stay prohibited? Read-only, always. It creates no
assignment and never mutates.
* **Initiate** (:func:`apply_request`) — turn an authorized request into an
*exclusive assignment*, and only ever through the allocator.
Three invariants hold and are the reason this module exists rather than a
direct call to :func:`allocator_service.allocate_next_work` from a route:
1. **The allocator remains the only source of exclusive ownership** (#600 /
#613). ``apply`` never assigns the requested item directly. It runs a
dry-run first and proceeds only when the allocator would independently pick
that exact item; otherwise it reports ``wait`` and mutates nothing. A
request is therefore a *confirmation* of the allocator's decision, never an
override of it.
2. **Duplicate assignment is rejected before it is attempted.** An active
claim on the work unit — held by any session, this one included — blocks.
3. **Fail closed at every unknown.** An unparseable request, an unavailable
control-plane DB, an incomplete queue inventory, or an unresolved
authorization all deny. There is no branch that proceeds on missing
evidence.
Authorization comes from :mod:`webui.console_authz` (``initiate_workflow``)
and every outcome is audited through :mod:`webui.console_audit`, correlated to
the resulting assignment by ``correlation_id``.
"""
from __future__ import annotations
import uuid
from dataclasses import dataclass, field
from typing import Any, Callable, Mapping, Sequence
import allocator_service
from task_capability_map import required_permission, required_role
from webui import console_audit, console_authz
# The console action this module is gated by. Registered in console_authz.
ACTION_ID = "initiate_workflow"
KIND_ISSUE = "issue"
KIND_PR = "pr"
WORK_KINDS: tuple[str, ...] = (KIND_ISSUE, KIND_PR)
REQUESTABLE_ROLES: tuple[str, ...] = (
allocator_service.ROLE_AUTHOR,
allocator_service.ROLE_REVIEWER,
allocator_service.ROLE_MERGER,
allocator_service.ROLE_RECONCILER,
allocator_service.ROLE_CONTROLLER,
)
# Intent is operator prose echoed back into an audit record. Bounded so a
# pasted transcript cannot bloat the append-only log.
MAX_INTENT_CHARS = 500
# --- Outcomes ---------------------------------------------------------------
OUTCOME_ASSIGNED = allocator_service.OUTCOME_ASSIGNED
OUTCOME_WAIT = allocator_service.OUTCOME_WAIT
OUTCOME_BLOCKED = "blocked"
OUTCOME_DENIED = "denied"
OUTCOME_INVALID = "invalid_request"
OUTCOME_PREVIEW = allocator_service.OUTCOME_PREVIEW
# --- Reason codes -----------------------------------------------------------
REASON_AUTHORIZED = "request_authorized"
REASON_PREVIEW_OK = "preview_authorized"
REASON_UNAUTHORIZED = "unauthorized"
REASON_NOT_NEXT_SAFE = "not_next_safe_work"
REASON_DUPLICATE_ASSIGNMENT = "duplicate_assignment"
REASON_CONFIRMATION_REQUIRED = "confirmation_required"
REASON_EVIDENCE_UNAVAILABLE = "evidence_unavailable"
REASON_ALLOCATOR_OUTCOME = "allocator_declined"
# --- Check names ------------------------------------------------------------
CHECK_AUTHORIZATION = "authorization"
CHECK_CAPABILITY = "capability"
CHECK_LEASE_AVAILABILITY = "lease_availability"
CHECK_NEXT_SAFE_ACTION = "next_safe_action"
CHECK_HEAD_PIN = "head_pin"
# --- Request model ----------------------------------------------------------
@dataclass(frozen=True)
class WorkRequest:
"""One operator request: a role, a work unit, and a stated intent."""
desired_role: str
work_kind: str
work_number: int
intent_summary: str
remote: str
org: str
repo: str
expected_head_sha: str | None = None
@property
def work_key(self) -> tuple[str, int]:
return (self.work_kind, self.work_number)
@property
def display_ref(self) -> str:
return f"#{self.work_number}"
def to_dict(self) -> dict[str, Any]:
return {
"desired_role": self.desired_role,
"work_kind": self.work_kind,
"work_number": self.work_number,
"intent_summary": self.intent_summary,
"remote": self.remote,
"org": self.org,
"repo": self.repo,
"expected_head_sha": self.expected_head_sha,
}
@dataclass(frozen=True)
class RequestError:
"""A rejected request, with the field that caused the rejection."""
reason_code: str
detail: str
field_name: str | None = None
def to_dict(self) -> dict[str, Any]:
return {
"ok": False,
"outcome": OUTCOME_INVALID,
"reason_code": self.reason_code,
"detail": self.detail,
"field": self.field_name,
}
def _clean(value: Any) -> str:
return str(value or "").strip()
def parse_request(
payload: Mapping[str, Any] | None,
*,
default_scope: Mapping[str, str] | None = None,
) -> tuple[WorkRequest | None, RequestError | None]:
"""Validate an operator payload into a :class:`WorkRequest`.
Returns ``(request, None)`` or ``(None, error)``. Never raises and never
guesses: an unknown role, an unknown work kind, or a non-positive number is
an error rather than a silently corrected value.
"""
body = dict(payload or {})
scope = dict(default_scope or {})
role = _clean(body.get("desired_role") or body.get("role")).lower()
if role not in REQUESTABLE_ROLES:
return None, RequestError(
reason_code="unknown_role",
detail=(
f"desired_role must be one of {', '.join(REQUESTABLE_ROLES)}; "
f"got {role or '(empty)'!r}."
),
field_name="desired_role",
)
kind = _clean(body.get("work_kind") or body.get("kind")).lower()
if kind not in WORK_KINDS:
return None, RequestError(
reason_code="unknown_work_kind",
detail=(
f"work_kind must be 'issue' or 'pr'; got {kind or '(empty)'!r}."
),
field_name="work_kind",
)
raw_number = body.get("work_number")
if raw_number is None:
raw_number = (
body.get("pr_number") if kind == KIND_PR else body.get("issue_number")
)
if raw_number is None:
raw_number = body.get("number")
try:
number = int(str(raw_number).strip())
except (TypeError, ValueError):
return None, RequestError(
reason_code="invalid_work_number",
detail=f"work_number must be an integer; got {raw_number!r}.",
field_name="work_number",
)
if number <= 0:
return None, RequestError(
reason_code="invalid_work_number",
detail="work_number must be a positive issue or PR number.",
field_name="work_number",
)
intent = _clean(body.get("intent_summary") or body.get("intent"))
if not intent:
return None, RequestError(
reason_code="missing_intent",
detail="intent_summary is required so the audit record states why.",
field_name="intent_summary",
)
intent = intent[:MAX_INTENT_CHARS]
remote = _clean(body.get("remote")) or _clean(scope.get("remote"))
org = _clean(body.get("org")) or _clean(scope.get("org"))
repo = _clean(body.get("repo")) or _clean(scope.get("repo"))
if not (remote and org and repo):
return None, RequestError(
reason_code="scope_unresolved",
detail=(
"remote, org, and repo could not be resolved from the request "
"or the project registry."
),
field_name="repo",
)
head = _clean(body.get("expected_head_sha")) or None
return (
WorkRequest(
desired_role=role,
work_kind=kind,
work_number=number,
intent_summary=intent,
remote=remote,
org=org,
repo=repo,
expected_head_sha=head,
),
None,
)
# --- Preview ----------------------------------------------------------------
@dataclass(frozen=True)
class RequestCheck:
"""One named precondition and its verdict."""
name: str
ok: bool
reason_code: str
detail: str
evidence: dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> dict[str, Any]:
return {
"name": self.name,
"ok": self.ok,
"reason_code": self.reason_code,
"detail": self.detail,
"evidence": dict(self.evidence),
}
@dataclass(frozen=True)
class RequestPreview:
"""The full intent preview for one request. Read-only in every field."""
request: WorkRequest
authorized: bool
reason_code: str
detail: str
authorization: dict[str, Any]
checks: tuple[RequestCheck, ...]
prohibited_actions: tuple[str, ...]
allowed_actions: tuple[str, ...]
next_safe_action: str
required_profile: str
required_namespace: str
required_permission: str
correlation_id: str
allocator_evidence: dict[str, Any] = field(default_factory=dict)
@property
def failed_checks(self) -> tuple[RequestCheck, ...]:
return tuple(c for c in self.checks if not c.ok)
def to_dict(self) -> dict[str, Any]:
return {
"ok": self.authorized,
"outcome": OUTCOME_PREVIEW,
"dry_run": True,
"mutation_performed": False,
"authorized": self.authorized,
"reason_code": self.reason_code,
"detail": self.detail,
"request": self.request.to_dict(),
"authorization": dict(self.authorization),
"checks": [c.to_dict() for c in self.checks],
"failed_checks": [c.name for c in self.failed_checks],
"prohibited_actions": list(self.prohibited_actions),
"allowed_actions": list(self.allowed_actions),
"next_safe_action": self.next_safe_action,
"required_profile": self.required_profile,
"required_namespace": self.required_namespace,
"required_permission": self.required_permission,
"correlation_id": self.correlation_id,
"allocator_evidence": dict(self.allocator_evidence),
}
AllocatorFn = Callable[..., dict[str, Any] | None]
ClaimsFn = Callable[["WorkRequest"], Mapping[tuple[str, int], dict[str, Any]]]
def _correlation_id() -> str:
return f"req-{uuid.uuid4().hex}"
def _selection_matches(
selection: Mapping[str, Any] | None, request: WorkRequest
) -> bool:
if not selection:
return False
kind = _clean(selection.get("kind")).lower()
try:
number_int = int(selection.get("number"))
except (TypeError, ValueError):
return False
return (kind, number_int) == request.work_key
def _authorization_check(
decision: console_authz.AuthorizationDecision,
) -> RequestCheck:
return RequestCheck(
name=CHECK_AUTHORIZATION,
ok=bool(decision.allowed),
reason_code=decision.reason_code,
detail=decision.detail,
evidence={
"subject": decision.principal.subject,
"role": decision.principal.role,
"required_role": decision.required_role,
"identity_source": decision.principal.identity_source,
},
)
def _capability_check(request: WorkRequest) -> RequestCheck:
"""Whether the requested role maps to a declared MCP capability.
The console never invents an authority: the permission and role come from
``task_capability_map`` via the same ``allocate_next_work`` task the MCP
allocator gates on.
"""
# The remote-prefixed hint keeps a dadeschools request from being told to
# run under a prgs profile; ``required_profile_for_role`` preserves the
# prefix when one is present and falls back to its own default otherwise.
profile_hint = f"{request.remote}-{request.desired_role}"
try:
profile = allocator_service.required_profile_for_role(
request.desired_role, profile_name=profile_hint
)
namespace = allocator_service.required_namespace_for_role(
request.desired_role, profile_name=profile_hint
)
except Exception as exc: # noqa: BLE001 — an unresolved role is a denial
return RequestCheck(
name=CHECK_CAPABILITY,
ok=False,
reason_code="capability_unresolved",
detail=(
f"no profile/namespace maps to role {request.desired_role!r}: "
f"{exc}"
),
)
resolved = bool(profile and namespace)
return RequestCheck(
name=CHECK_CAPABILITY,
ok=resolved,
reason_code="capability_resolved" if resolved else "capability_unresolved",
detail=(
f"role {request.desired_role!r} runs under profile {profile!r} in "
f"MCP namespace {namespace!r}."
),
evidence={
"required_profile": profile,
"required_namespace": namespace,
"required_permission": required_permission("allocate_next_work"),
"capability_role": required_role("allocate_next_work"),
},
)
def _lease_check(
request: WorkRequest,
claims: Mapping[tuple[str, int], dict[str, Any]] | None,
) -> RequestCheck:
"""Whether the work unit is free of an active claim.
``claims is None`` means the control-plane DB could not be read. That is a
failure, not an absence of claims: an unreadable substrate must never read
as "nothing holds this".
"""
if claims is None:
return RequestCheck(
name=CHECK_LEASE_AVAILABILITY,
ok=False,
reason_code=REASON_EVIDENCE_UNAVAILABLE,
detail=(
"active-claim inventory is unavailable; refusing to treat an "
"unreadable control-plane DB as an unclaimed work unit."
),
)
claim = claims.get(request.work_key)
if claim:
return RequestCheck(
name=CHECK_LEASE_AVAILABILITY,
ok=False,
reason_code=REASON_DUPLICATE_ASSIGNMENT,
detail=(
f"{request.work_kind} {request.display_ref} already carries an "
f"active {claim.get('role') or 'unknown'} lease."
),
evidence={
"lease_id": claim.get("lease_id"),
"session_id": claim.get("session_id"),
"role": claim.get("role"),
"expires_at": claim.get("expires_at"),
},
)
return RequestCheck(
name=CHECK_LEASE_AVAILABILITY,
ok=True,
reason_code="lease_available",
detail=f"no active lease holds {request.work_kind} {request.display_ref}.",
)
def _next_safe_action_check(
request: WorkRequest, allocation: Mapping[str, Any] | None
) -> RequestCheck:
"""Whether the allocator would independently select this exact work unit."""
if not allocation:
return RequestCheck(
name=CHECK_NEXT_SAFE_ACTION,
ok=False,
reason_code=REASON_EVIDENCE_UNAVAILABLE,
detail="allocator dry-run produced no result; refusing to proceed.",
)
selection = allocation.get("selected") or {}
outcome = _clean(allocation.get("outcome"))
if not _selection_matches(selection, request):
chosen = (
f"{_clean(selection.get('kind')) or 'unknown'} #{selection.get('number')}"
if selection
else "nothing"
)
return RequestCheck(
name=CHECK_NEXT_SAFE_ACTION,
ok=False,
reason_code=REASON_NOT_NEXT_SAFE,
detail=(
f"the allocator would select {chosen} for role "
f"{request.desired_role!r}, not {request.work_kind} "
f"{request.display_ref}. Requests confirm the allocator's "
"decision; they never override it."
),
evidence={
"allocator_outcome": outcome,
"allocator_selection": dict(selection),
"reasons": list(allocation.get("reasons") or ()),
},
)
return RequestCheck(
name=CHECK_NEXT_SAFE_ACTION,
ok=True,
reason_code="next_safe_work",
detail=(
f"the allocator selects {request.work_kind} {request.display_ref} "
f"for role {request.desired_role!r}."
),
evidence={
"allocator_outcome": outcome,
"selected_action": _clean(selection.get("selected_action")),
"expected_role_next": _clean(selection.get("expected_role_next")),
},
)
def _head_pin_check(
request: WorkRequest, allocation: Mapping[str, Any] | None
) -> RequestCheck:
"""PR work must be pinned to a head SHA; issue work has nothing to pin."""
if request.work_kind != KIND_PR:
return RequestCheck(
name=CHECK_HEAD_PIN,
ok=True,
reason_code="head_pin_not_applicable",
detail="issue work carries no head SHA to pin.",
)
selection = (allocation or {}).get("selected") or {}
allocator_head = _clean(selection.get("head_sha")) or None
if not allocator_head:
return RequestCheck(
name=CHECK_HEAD_PIN,
ok=False,
reason_code=REASON_EVIDENCE_UNAVAILABLE,
detail=(
"the allocator reported no head SHA for this PR; PR work "
"cannot be initiated unpinned."
),
)
if request.expected_head_sha and request.expected_head_sha != allocator_head:
return RequestCheck(
name=CHECK_HEAD_PIN,
ok=False,
reason_code="head_moved",
detail=(
"the requested head SHA does not match the PR's current head; "
"re-preview against the live head before initiating."
),
evidence={
"requested_head_sha": request.expected_head_sha,
"current_head_sha": allocator_head,
},
)
return RequestCheck(
name=CHECK_HEAD_PIN,
ok=True,
reason_code="head_pinned",
detail=f"PR {request.display_ref} is pinned at {allocator_head}.",
evidence={"head_sha": allocator_head},
)
def _next_safe_action_text(
request: WorkRequest, checks: Sequence[RequestCheck], authorized: bool
) -> str:
if authorized:
return (
f"Confirm and initiate {request.desired_role} work on "
f"{request.work_kind} {request.display_ref} via the allocator."
)
for check in checks:
if not check.ok:
return f"Resolve {check.name}: {check.detail}"
return "No safe action; the request is not authorized."
def preview_request(
request: WorkRequest,
*,
principal: console_authz.Principal | None = None,
allocator: AllocatorFn | None = None,
claims_source: ClaimsFn | None = None,
correlation_id: str | None = None,
audit: bool = True,
) -> RequestPreview:
"""Build the read-only intent preview for *request*. Never mutates."""
who = principal or console_authz.ANONYMOUS
corr = correlation_id or _correlation_id()
decision = console_authz.authorize(ACTION_ID, who, for_execution=False)
allocation: dict[str, Any] | None = None
claims: Mapping[tuple[str, int], dict[str, Any]] | None = None
checks: list[RequestCheck] = [_authorization_check(decision)]
if decision.allowed:
# An unauthorized principal never reaches the allocator or the
# control-plane DB: a denial must not double as a queue oracle.
allocation = _run_allocator(request, allocator, apply=False)
claims = _load_claims(request, claims_source)
checks.append(_capability_check(request))
checks.append(_lease_check(request, claims))
checks.append(_next_safe_action_check(request, allocation))
checks.append(_head_pin_check(request, allocation))
authorized = all(c.ok for c in checks)
allowed_actions, prohibited_actions = allocator_service.role_actions(
request.desired_role
)
capability = next((c for c in checks if c.name == CHECK_CAPABILITY), None)
evidence = capability.evidence if capability else {}
if authorized:
reason_code = REASON_PREVIEW_OK
detail = (
"Request is authorized. Preview only — nothing has been assigned."
)
else:
first_failure = next(c for c in checks if not c.ok)
reason_code, detail = first_failure.reason_code, first_failure.detail
preview = RequestPreview(
request=request,
authorized=authorized,
reason_code=reason_code,
detail=detail,
authorization=decision.to_dict(),
checks=tuple(checks),
prohibited_actions=tuple(prohibited_actions),
allowed_actions=tuple(allowed_actions),
next_safe_action=_next_safe_action_text(request, checks, authorized),
required_profile=str(evidence.get("required_profile") or ""),
required_namespace=str(evidence.get("required_namespace") or ""),
required_permission=str(evidence.get("required_permission") or ""),
correlation_id=corr,
allocator_evidence=_allocator_evidence(allocation),
)
if audit:
_audit(
request,
result=console_audit.RESULT_PREVIEWED,
decision=decision,
principal=who,
reason_code=reason_code,
detail=detail,
correlation_id=corr,
metadata={
"intent_summary": request.intent_summary,
"desired_role": request.desired_role,
"authorized": authorized,
"failed_checks": [c.name for c in preview.failed_checks],
"phase": "preview",
},
)
return preview
# --- Initiation -------------------------------------------------------------
def apply_request(
request: WorkRequest,
*,
principal: console_authz.Principal | None = None,
confirm: bool = False,
allocator: AllocatorFn | None = None,
claims_source: ClaimsFn | None = None,
correlation_id: str | None = None,
) -> dict[str, Any]:
"""Initiate *request* as an exclusive assignment, or refuse.
The only path to an assignment is the allocator agreeing, on a dry-run,
that this work unit is what the requested role should take next. Every
refusal returns before any mutation is attempted.
"""
who = principal or console_authz.ANONYMOUS
corr = correlation_id or _correlation_id()
execution_decision = console_authz.authorize(ACTION_ID, who, for_execution=True)
authorization = execution_decision.to_dict()
def _refuse(
outcome: str,
reason_code: str,
detail: str,
*,
status: int,
extra: dict[str, Any] | None = None,
) -> dict[str, Any]:
_audit(
request,
result=console_audit.RESULT_DENIED,
decision=execution_decision,
principal=who,
reason_code=reason_code,
detail=detail,
correlation_id=corr,
metadata={
"intent_summary": request.intent_summary,
"desired_role": request.desired_role,
"phase": "apply",
"outcome": outcome,
},
)
payload: dict[str, Any] = {
"ok": False,
"outcome": outcome,
"reason_code": reason_code,
"detail": detail,
"request": request.to_dict(),
"authorization": authorization,
"assignment": None,
"correlation_id": corr,
"mutation_performed": False,
"status_code": status,
}
payload.update(extra or {})
return payload
if not (execution_decision.allowed and execution_decision.execution_enabled):
return _refuse(
OUTCOME_DENIED,
REASON_UNAUTHORIZED,
execution_decision.detail,
status=403,
)
# Confirmation is a property of the action in the RBAC model, so it is read
# from there rather than assumed here.
action = console_authz.get_action(ACTION_ID)
if action is not None and action.requires_confirmation and not confirm:
return _refuse(
OUTCOME_DENIED,
REASON_CONFIRMATION_REQUIRED,
(
"This action requires explicit confirmation. Re-submit with "
"confirm=true after reviewing the preview."
),
status=409,
)
preview = preview_request(
request,
principal=who,
allocator=allocator,
claims_source=claims_source,
correlation_id=corr,
audit=False,
)
if not preview.authorized:
outcome = (
OUTCOME_BLOCKED
if preview.reason_code == REASON_DUPLICATE_ASSIGNMENT
else OUTCOME_WAIT
)
return _refuse(
outcome,
preview.reason_code,
preview.detail,
status=409,
extra={"preview": preview.to_dict()},
)
fingerprint = (
_clean(preview.allocator_evidence.get("candidate_set_fingerprint")) or None
)
allocation = _run_allocator(
request,
allocator,
apply=True,
expected_candidate_set_fingerprint=fingerprint,
)
if not allocation:
return _refuse(
OUTCOME_WAIT,
REASON_EVIDENCE_UNAVAILABLE,
"the allocator returned no result; nothing was assigned.",
status=503,
)
assignment = allocation.get("assignment") or None
outcome = _clean(allocation.get("outcome"))
assigned = bool(
outcome == allocator_service.OUTCOME_ASSIGNED
and assignment
and _selection_matches(allocation.get("selected"), request)
)
if not assigned:
blocked = outcome in {
allocator_service.OUTCOME_BLOCKED_LEASE,
allocator_service.OUTCOME_BLOCKED_TERMINAL,
allocator_service.OUTCOME_BLOCKED_EXCLUDED_OWN_LEASE,
}
return _refuse(
OUTCOME_BLOCKED if blocked else OUTCOME_WAIT,
REASON_ALLOCATOR_OUTCOME,
(
f"the allocator returned {outcome or 'no outcome'} rather than "
"an assignment for this work unit; nothing was assigned."
),
status=409,
extra={"allocator_evidence": _allocator_evidence(allocation)},
)
_audit(
request,
result=console_audit.RESULT_SUCCEEDED,
decision=execution_decision,
principal=who,
reason_code=REASON_AUTHORIZED,
detail=(
f"assigned {request.work_kind} {request.display_ref} to role "
f"{request.desired_role}."
),
correlation_id=corr,
metadata={
"intent_summary": request.intent_summary,
"desired_role": request.desired_role,
"phase": "apply",
"outcome": OUTCOME_ASSIGNED,
"assignment_id": assignment.get("assignment_id"),
"lease_id": assignment.get("lease_id"),
},
)
return {
"ok": True,
"outcome": OUTCOME_ASSIGNED,
"reason_code": REASON_AUTHORIZED,
"detail": (
"Exclusive assignment created via the allocator. Continue in the "
f"{preview.required_namespace or 'assigned'} MCP namespace."
),
"request": request.to_dict(),
"authorization": authorization,
"assignment": dict(assignment),
"handoff": {
"assignment_id": assignment.get("assignment_id"),
"lease_id": assignment.get("lease_id"),
"session_id": assignment.get("session_id"),
"required_profile": preview.required_profile,
"required_namespace": preview.required_namespace,
"allowed_actions": list(preview.allowed_actions),
"forbidden_actions": list(preview.prohibited_actions),
"expected_head_sha": assignment.get("expected_head_sha"),
},
"correlation_id": corr,
"mutation_performed": True,
"status_code": 201,
"allocator_evidence": _allocator_evidence(allocation),
}
# --- Adapters ---------------------------------------------------------------
def _allocator_evidence(allocation: Mapping[str, Any] | None) -> dict[str, Any]:
"""Reduce an allocator result to the non-secret fields worth surfacing."""
if not allocation:
return {}
return {
"outcome": allocation.get("outcome"),
"selected": allocation.get("selected"),
"reasons": list(allocation.get("reasons") or ()),
"candidate_set_fingerprint": allocation.get("candidate_set_fingerprint"),
"candidate_count": allocation.get("candidate_count"),
"inventory_complete": allocation.get("inventory_complete"),
"selection_policy": allocation.get("selection_policy"),
"substrate": allocation.get("substrate"),
}
def _run_allocator(
request: WorkRequest,
allocator: AllocatorFn | None,
*,
apply: bool,
expected_candidate_set_fingerprint: str | None = None,
) -> dict[str, Any] | None:
fn = allocator or default_allocator
try:
result = fn(
request=request,
apply=apply,
expected_candidate_set_fingerprint=expected_candidate_set_fingerprint,
)
except Exception: # noqa: BLE001 — an allocator failure denies, never proceeds
return None
return result if isinstance(result, dict) else None
def _load_claims(
request: WorkRequest, claims_source: ClaimsFn | None
) -> Mapping[tuple[str, int], dict[str, Any]] | None:
fn = claims_source or default_claims_source
try:
claims = fn(request)
except Exception: # noqa: BLE001 — an unreadable substrate is a denial
return None
return claims if isinstance(claims, Mapping) else None
def default_claims_source(
request: WorkRequest,
) -> Mapping[tuple[str, int], dict[str, Any]]:
"""Live active-claim inventory from the #613 control-plane DB."""
import control_plane_db
db = control_plane_db.ControlPlaneDB()
return db.list_active_claims(
remote=request.remote, org=request.org, repo=request.repo
)
def default_allocator(
*,
request: WorkRequest,
apply: bool,
expected_candidate_set_fingerprint: str | None = None,
) -> dict[str, Any] | None:
"""Run the real allocator over the live queue for *request*'s scope.
An incomplete candidate inventory returns ``None`` rather than a ranking
over a partial set (#758): selecting from a short list can pick the wrong
work unit, so the request denies instead.
"""
import control_plane_db
from webui.queue_loader import load_queue_snapshot
from webui.traffic_loader import candidates_from_queue_snapshot
snapshot = load_queue_snapshot()
if snapshot.fetch_error:
return None
for pagination in (snapshot.pr_pagination, snapshot.issue_pagination):
if pagination is not None and not pagination.inventory_complete:
return None
candidates = candidates_from_queue_snapshot(snapshot)
db = control_plane_db.ControlPlaneDB()
result = allocator_service.allocate_next_work(
db,
session_id=f"webui-request-{uuid.uuid4().hex[:12]}",
role=request.desired_role,
remote=request.remote,
org=request.org,
repo=request.repo,
candidates=candidates,
apply=bool(apply),
allocation_mode="role_scoped",
expected_candidate_set_fingerprint=expected_candidate_set_fingerprint,
)
if isinstance(result, dict):
result.setdefault("candidate_count", len(candidates))
result.setdefault("inventory_complete", True)
result.setdefault("selection_policy", allocator_service.SELECTION_POLICY)
return result
# --- Audit ------------------------------------------------------------------
def _audit(
request: WorkRequest,
*,
result: str,
decision: console_authz.AuthorizationDecision,
principal: console_authz.Principal,
reason_code: str,
detail: str,
correlation_id: str,
metadata: dict[str, Any],
) -> dict[str, Any]:
return console_audit.record_event(
action_id=ACTION_ID,
result=result,
decision=decision,
principal=principal,
target={
"kind": request.work_kind,
"ref": request.display_ref,
"remote": request.remote,
"org": request.org,
"repo": request.repo,
},
reason_code=reason_code,
request_id=correlation_id,
detail=detail,
metadata=metadata,
)
+164
View File
@@ -0,0 +1,164 @@
"""HTML views for the operator request surface (#643).
The form is deliberately a *preview* form. It has no initiate button, because
initiating requires a confirmed POST to ``/api/v1/requests/apply`` and a stray
form submission must not be able to produce one by accident.
Nothing rendered here is trusted input: every interpolated value is escaped,
and the page renders only values the service already produced rather than
echoing a raw request body back.
"""
from __future__ import annotations
import html
import json
from typing import Any
from webui.layout import render_page
from webui.request_service import (
REQUESTABLE_ROLES,
WORK_KINDS,
RequestError,
RequestPreview,
)
REQUESTS_PATH = "/requests"
PREVIEW_API_PATH = "/api/v1/requests/preview"
APPLY_API_PATH = "/api/v1/requests/apply"
def _escape(text: Any) -> str:
return html.escape(str(text if text is not None else ""), quote=True)
REQUEST_PAGE_STYLES = """
<style>
.request-form { display: grid; gap: 0.75rem; max-width: 44rem; }
.request-form label { display: grid; gap: 0.25rem; font-size: 0.9rem; }
.request-check { margin: 0.35rem 0; }
.request-check .verdict-ok { color: var(--accent); }
.request-check .verdict-fail { color: #d14; }
.request-prohibited code { margin-right: 0.4rem; }
</style>
"""
def _options(values: tuple[str, ...], selected: Any) -> str:
return "".join(
f"<option value='{_escape(value)}'"
+ (" selected" if selected == value else "")
+ f">{_escape(value)}</option>"
for value in values
)
def _form(values: dict[str, Any] | None = None) -> str:
current = dict(values or {})
number = current.get("work_number")
return (
f"<form class='request-form' method='post' action='{REQUESTS_PATH}'>"
"<label>Desired role<select name='desired_role'>"
f"{_options(REQUESTABLE_ROLES, current.get('desired_role'))}"
"</select></label>"
"<label>Work kind<select name='work_kind'>"
f"{_options(WORK_KINDS, current.get('work_kind'))}"
"</select></label>"
"<label>Issue or PR number"
"<input type='number' name='work_number' min='1' required "
f"value='{_escape(number) if number else ''}'></label>"
"<label>Intent summary"
"<input type='text' name='intent_summary' maxlength='500' required "
f"value='{_escape(current.get('intent_summary'))}'></label>"
"<label>Expected head SHA <span class='muted'>(PR work only)</span>"
"<input type='text' name='expected_head_sha' "
f"value='{_escape(current.get('expected_head_sha'))}'></label>"
"<button type='submit' class='copy-btn'>Preview request</button>"
"<p class='muted meta'>Preview is read-only and creates no assignment. "
f"Initiating requires a confirmed POST to <code>{APPLY_API_PATH}</code>."
"</p>"
"</form>"
)
def _checks_block(preview: RequestPreview) -> str:
rows = []
for check in preview.checks:
verdict = "PASS" if check.ok else "FAIL"
css = "verdict-ok" if check.ok else "verdict-fail"
rows.append(
"<li class='request-check'>"
f"<span class='{css}'><strong>{verdict}</strong></span> "
f"<code>{_escape(check.name)}</code> — {_escape(check.detail)} "
f"<span class='muted meta'>({_escape(check.reason_code)})</span>"
"</li>"
)
return "<ul>" + "".join(rows) + "</ul>"
def _preview_block(preview: RequestPreview) -> str:
verdict = "AUTHORIZED" if preview.authorized else "DENIED"
prohibited = "".join(
f"<code>{_escape(action)}</code>" for action in preview.prohibited_actions
)
request = preview.request
evidence = json.dumps(preview.allocator_evidence, indent=2, default=str)
return (
"<h3>Intent preview</h3>"
f"<p><strong>{verdict}</strong> — {_escape(preview.detail)}</p>"
"<p class='meta'>"
f"Role <code>{_escape(request.desired_role)}</code> · "
f"{_escape(request.work_kind)} <code>{_escape(request.display_ref)}</code>"
f" · profile <code>{_escape(preview.required_profile)}</code> · "
f"namespace <code>{_escape(preview.required_namespace)}</code> · "
f"permission <code>{_escape(preview.required_permission)}</code>"
"</p>"
f"<p>Intent: {_escape(request.intent_summary)}</p>"
f"{_checks_block(preview)}"
f"<p><strong>Next safe action:</strong> "
f"{_escape(preview.next_safe_action)}</p>"
"<p class='request-prohibited'><strong>Prohibited for this role:</strong> "
+ (prohibited or "<span class='muted'>none declared</span>")
+ "</p>"
"<p class='muted meta'>Correlation id "
f"<code>{_escape(preview.correlation_id)}</code></p>"
"<details><summary>Allocator evidence</summary>"
f"<pre class='prompt-text'>{_escape(evidence)}</pre>"
"</details>"
)
def _error_block(error: RequestError) -> str:
field = (
f"<p class='meta'>Field: <code>{_escape(error.field_name)}</code></p>"
if error.field_name
else ""
)
return (
"<h3>Request rejected</h3>"
f"<p><strong>{_escape(error.reason_code)}</strong> — "
f"{_escape(error.detail)}</p>{field}"
)
def render_requests_page(
*,
preview: RequestPreview | None = None,
error: RequestError | None = None,
submitted: dict[str, Any] | None = None,
) -> str:
"""Render the request form, plus a preview or rejection when one exists."""
body = (
"<h2>Requests</h2>"
"<p>Submit a work request — desired role, issue or PR, and intent — "
"and see whether it would be authorized before anything is reserved. "
"Initiation goes through the allocator (#600/#613); this console never "
"self-selects work, never approves, and never merges.</p>"
+ _form(submitted)
+ (_error_block(error) if error is not None else "")
+ (_preview_block(preview) if preview is not None else "")
+ f"<p class='meta'><a href='{PREVIEW_API_PATH}'>Preview API</a> · "
"<a href='/api/console/security-model'>RBAC model</a></p>"
+ REQUEST_PAGE_STYLES
)
return render_page(title="Requests", body_html=body)
+10
View File
@@ -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.