Gitea-Tools/webui/request_service.py

"""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,
    )