"""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 inspect 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 _new_session_id() -> str: """Mint the control-plane session id for one request flow (#643). Minted once per flow and reused by the dry-run and the apply, so a lease the apply creates is owned by an id the caller still holds and can release. When each call minted its own id, an apply wrote two session rows and bound the lease to the second — leaving it with no owner able to reclaim it. """ return f"webui-request-{uuid.uuid4().hex[:12]}" 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, session_id: str | None = None, ) -> RequestPreview: """Build the read-only intent preview for *request*. Never mutates. "Never mutates" is now literal on the default path: the allocator runs ``side_effect_free``, so no control-plane session row is written and no lease sweep runs. *session_id* is threaded from an enclosing apply so both halves of that flow share one control-plane identity (#643). """ 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, session_id=session_id ) 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. One exception is unavoidable and is compensated rather than prevented: the allocator can commit an assignment and only then reveal, on egress, that it selected a different work unit than the one requested. That assignment is released before refusing, and ``mutation_performed`` reports whether any durable control-plane state survives this call — ``False`` once the stray assignment is gone, ``True`` with an explicit ``orphaned_assignment`` and reclaim action when the release did not succeed (#643). """ who = principal or console_authz.ANONYMOUS corr = correlation_id or _correlation_id() # One identity for the whole flow, so a lease the apply creates is owned by # an id this function still holds and can release. session_id = _new_session_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, mutation_performed: bool = False, ) -> 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, # True only when durable control-plane state survives this refusal — # an assignment the allocator created that could not be released. "mutation_performed": bool(mutation_performed), "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, session_id=session_id, ) 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, session_id=session_id, ) if not allocation: # "No result" is not proof of "no write": allocate_next_work only catches # three exception types, so anything raised after assign_and_lease # committed lands here with a lease already durable. Release whatever # this flow's session owns before refusing. sweep = _sweep_session_leases(request, session_id=session_id) extra: dict[str, Any] = {} detail = "the allocator returned no result; nothing was assigned." if sweep.get("released"): extra["compensation"] = sweep detail = ( "the allocator returned no result after creating an assignment; " "the assignment was released and nothing remains assigned." ) elif sweep.get("error"): extra["compensation"] = sweep return _refuse( OUTCOME_WAIT, REASON_EVIDENCE_UNAVAILABLE, detail, status=503, extra=extra or None, ) assignment = allocation.get("assignment") or None outcome = _clean(allocation.get("outcome")) committed = bool(outcome == allocator_service.OUTCOME_ASSIGNED and assignment) assigned = bool( committed 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, } extra = {"allocator_evidence": _allocator_evidence(allocation)} detail = ( f"the allocator returned {outcome or 'no outcome'} rather than " "an assignment for this work unit; nothing was assigned." ) if committed: # The allocator assigned a *different* unit than the one requested. # That assignment is real and durable; releasing it is the whole # difference between a refusal and a silent leak. compensation = _release_stray_assignment( request, allocation, session_id=session_id ) extra["compensation"] = compensation if compensation.get("released"): detail = ( "the allocator assigned a different work unit than the one " "requested; that assignment was released and nothing " "remains assigned." ) else: extra["orphaned_assignment"] = compensation return _refuse( OUTCOME_BLOCKED if blocked else OUTCOME_WAIT, REASON_ALLOCATOR_OUTCOME, ( "the allocator assigned a different work unit than the " "one requested and it could not be released; it must be " "reclaimed explicitly." ), status=409, extra=extra, mutation_performed=True, ) return _refuse( OUTCOME_BLOCKED if blocked else OUTCOME_WAIT, REASON_ALLOCATOR_OUTCOME, detail, status=409, extra=extra, ) _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, session_id: str | None = None, ) -> dict[str, Any] | None: fn = allocator or default_allocator kwargs: dict[str, Any] = { "request": request, "apply": apply, "expected_candidate_set_fingerprint": expected_candidate_set_fingerprint, } if session_id: # Injected allocators in tests predate this argument; only pass it to # callables that accept it so a fake signature is never broken. if _accepts_session_id(fn): kwargs["session_id"] = session_id try: result = fn(**kwargs) except Exception: # noqa: BLE001 — an allocator failure denies, never proceeds return None return result if isinstance(result, dict) else None def _accepts_session_id(fn: AllocatorFn) -> bool: try: params = inspect.signature(fn).parameters except (TypeError, ValueError): # builtins / C callables return False if "session_id" in params: return True return any(p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values()) def _release_stray_assignment( request: WorkRequest, allocation: Mapping[str, Any], *, session_id: str | None, ) -> dict[str, Any]: """Release an assignment the allocator committed for the wrong work unit. The apply path can commit an assignment and then discover, on egress, that the allocator selected a *different* unit than the one requested — the CAS fingerprint hashes only ``{kind, number}`` plus exclusions, so a lease taken on the requested unit inside the window leaves the fingerprint identical and the pin passes while the selection moves on. Without compensation that lease is orphaned: owned by a session nothing heartbeats, and invisible because the response says nothing was mutated. Returns a record describing what was attempted so the caller can report it, including a reclaim action when the release itself did not succeed. """ assignment = allocation.get("assignment") or {} lease_id = _clean(assignment.get("lease_id")) or None assignment_id = _clean(assignment.get("assignment_id")) or None owner = _clean(allocation.get("session_id")) or session_id or None selected = allocation.get("selected") or {} record: dict[str, Any] = { "attempted": False, "released": False, "lease_id": lease_id, "assignment_id": assignment_id, "owner_session_id": owner, "selected": { "kind": _clean(selected.get("kind")).lower() or None, "number": selected.get("number"), }, } if not lease_id: # Nothing durable to release (or the allocator reported no lease id); # still surface the assignment id so a leak is never silent. record["detail"] = ( "the allocator reported an assignment with no lease id; nothing " "could be released automatically." ) if assignment_id: record["reclaim_action"] = _reclaim_action(record) return record if not owner: record["detail"] = ( "the owning session id is unknown; the assignment cannot be " "released automatically." ) record["reclaim_action"] = _reclaim_action(record) return record record["attempted"] = True try: import control_plane_db control_plane_db.ControlPlaneDB().release_lease(lease_id, session_id=owner) except Exception as exc: # noqa: BLE001 — a failed release must be reported record["error"] = str(exc) record["detail"] = ( "the allocator created an assignment for a different work unit and " "releasing it failed; it must be reclaimed explicitly." ) record["reclaim_action"] = _reclaim_action(record) return record record["released"] = True record["detail"] = ( "the allocator created an assignment for a different work unit; it was " "released, so no assignment persists from this request." ) return record def _reclaim_action(record: Mapping[str, Any]) -> dict[str, Any]: """The explicit operator action for an assignment that outlived its request.""" lease_id = record.get("lease_id") return { "tool": "gitea_release_workflow_lease", "lease_id": lease_id, "session_id": record.get("owner_session_id"), "assignment_id": record.get("assignment_id"), "instructions": ( "A control-plane assignment was created for a work unit other than " "the requested one and could not be released automatically. Call " f"gitea_release_workflow_lease(lease_id={lease_id!r}, " f"session_id={record.get('owner_session_id')!r}) to reclaim it, or " "wait for the lease TTL to expire." ), } def _sweep_session_leases( request: WorkRequest, *, session_id: str | None ) -> dict[str, Any]: """Release any lease this flow's session owns after an indeterminate apply. ``_run_allocator`` returns ``None`` for *any* exception, and ``allocate_next_work`` only catches ``InvalidWorkKindError``, ``LeaseRequiredError`` and ``ControlPlaneError`` — so an unexpected error raised *after* ``assign_and_lease`` committed surfaces as "no result" with a lease already written. Because the session id is now stable across the flow, that lease is findable: anything this session owns after a failed apply is by definition unclaimed by anyone, so it is released. """ record: dict[str, Any] = {"attempted": False, "released": [], "session_id": session_id} if not session_id: return record record["attempted"] = True try: import control_plane_db db = control_plane_db.ControlPlaneDB() leases = db.list_leases( remote=request.remote, org=request.org, repo=request.repo, statuses=("active",), ) for row in leases: if _clean(row.get("session_id")) != session_id: continue lease_id = _clean(row.get("lease_id")) or None if not lease_id: continue db.release_lease(lease_id, session_id=session_id) record["released"].append( { "lease_id": lease_id, "work_kind": row.get("work_kind"), "work_number": row.get("work_number"), } ) except Exception as exc: # noqa: BLE001 — best-effort; never mask the refusal record["error"] = str(exc) return record 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, session_id: 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. *session_id* is supplied by the caller so the dry-run and the apply share one control-plane identity; the lease an apply creates is then owned by an id the request flow still holds. A dry run additionally goes through the allocator's ``side_effect_free`` path, so a preview writes no session row and sweeps no leases (#643). """ 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=session_id or _new_session_id(), 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, side_effect_free=not apply, ) 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, )