dify/api/services/dify_builder/native_testing/authorization.py

"""Scoped approval and atomic append-only dispatch spends.

Every real transport attempt (including OAuth, redirects and retries) needs its
own persisted native cursor. A permit authorizes exactly one dispatch; uncertain
or failed attempts never refund quota and cannot be reissued under another key.
"""

from collections.abc import Callable
from datetime import datetime, timedelta
from typing import Literal, overload
from uuid import uuid4

from pydantic import ValidationError
from sqlalchemy import select
from sqlalchemy.orm import Session

from models.dify_builder import DifyBuilderExecutionRequest
from models.dify_builder_native_tests import (
    NativeEffectAttempt,
    NativeEffectDispatchClaim,
    NativeEffectGrant,
    NativeEffectOperation,
    NativeEffectOutcome,
    NativeInvocationState,
    NativePendingAction,
    NativeTestScope,
)
from runtime_contracts.builder.execution import BuilderExecutionPolicyError, canonical_digest
from runtime_contracts.builder.native.contracts import (
    ApprovalWait,
    DispatchPermit,
    EffectDecision,
    EffectIntent,
    EffectReceipt,
    NativeTestContextRef,
    Outcome,
    PolicyStop,
    RecordedResult,
    operation_digest,
)
from services.dify_builder.native_testing.pending_intents import PendingDecisionGuard
from services.dify_builder.native_testing.private_reads import validate_invocation_workflow, validate_private_read
from services.dify_builder.native_testing.public_contracts import NativeTestPublicReceipt
from services.dify_builder.native_testing.repository import (
    BudgetReservation,
    NativeBinding,
    OwnedNativeTest,
    OwnedNativeTestLoader,
    admitted_workflow_revision,
    native_workflow_revision,
    utc_now,
)
from services.dify_builder.native_testing.transitions import (
    dispatch_readiness,
    pending_reason,
    require_ancestor_authority,
    validate_pending_fence,
    validate_pending_history,
)


def validate_binding(intent: EffectIntent, binding: NativeBinding) -> None:
    if canonical_digest(intent.model_dump(mode="json")) != binding.intent_digest or (
        intent.kind,
        intent.resource_ref,
        intent.implementation_revision,
        intent.account_ref,
        intent.capability_digest,
    ) != (
        binding.kind,
        binding.resource_ref,
        binding.implementation_revision,
        binding.account_ref,
        binding.capability_digest,
    ):
        raise BuilderExecutionPolicyError("native_binding_mismatch")
    if binding.destructive:
        raise BuilderExecutionPolicyError("destructive_effect_forbidden")


def task_grant_allowed(binding: NativeBinding, *, task_approved: bool, confirmed_model: bool = False) -> bool:
    if confirmed_model and binding.kind == "model":
        return task_approved and not binding.destructive
    return (
        task_approved
        and binding.kind in {"code", "file"}
        and binding.isolated
        and not binding.egress
        and not binding.new_data
        and not binding.destructive
    )


def _scope_digest(intent: EffectIntent, binding: NativeBinding) -> str:
    # Legacy batches exclude logical occurrence and the new default field so
    # their immutable grant digests stay byte-for-byte compatible. Operation
    # approval includes the actual key and explicit server-selected mode.
    batch = binding.approval_scope == "batch"
    binding_exclusions = {"intent_digest", "execution_id", "invocation_digest"}
    if batch:
        binding_exclusions.add("approval_scope")
    return canonical_digest(
        {
            "intent": intent.model_dump(mode="json", exclude={"key"} if batch else set()),
            "binding": binding.model_dump(mode="json", exclude=binding_exclusions),
        }
    )


def _owner(owned: OwnedNativeTest) -> dict[str, str]:
    return {"tenant_id": owned.tenant_id, "request_id": owned.ref.request_id, "invocation_id": owned.ref.invocation_id}


def _owned_query(model, owned: OwnedNativeTest):
    return select(model).where(
        model.tenant_id == owned.tenant_id,
        model.request_id == owned.ref.request_id,
        model.invocation_id == owned.ref.invocation_id,
    )


def _completed_config_read_readiness(session: Session, owned: OwnedNativeTest) -> None:
    """SQL-only authority for a dependency of an already spent, live shell.

    A later valid approval fence does not revoke the original command. It still
    fences every new dispatch. Other waits and nonactive owners grant no read.
    Scope cancellation/expiry and the exact original permit are checked by the
    enclosing completed-claim validator, before and after every external read.
    """
    states = {
        state.invocation_id: state
        for state in session.scalars(
            select(NativeInvocationState).where(
                NativeInvocationState.request_id == owned.ref.request_id,
                NativeInvocationState.tenant_id == owned.tenant_id,
            )
        )
    }
    roots = [state for state in states.values() if state.parent_invocation_id is None]
    if len(roots) != 1 or (roots[0].app_id, roots[0].workflow_id) != (owned.app_id, owned.workflow_id):
        raise BuilderExecutionPolicyError("native_dispatch_closed")
    selected, seen = owned.ref.invocation_id, set()
    while selected is not None:
        state = states.get(selected)
        if (
            selected in seen
            or state is None
            or state.state not in {"running", "waiting"}
            or type(state.pending_ids) is not list
            or state.wait_reason not in {None, "approval"}
            or (state.state == "waiting" and state.wait_reason is None)
        ):
            raise BuilderExecutionPolicyError("native_dispatch_closed")
        seen.add(selected)
        ref = owned.ref.model_copy(update={"invocation_id": selected})
        if state.wait_reason == "approval":
            rows = session.scalars(
                select(NativePendingAction).where(
                    NativePendingAction.tenant_id == owned.tenant_id,
                    NativePendingAction.request_id == owned.ref.request_id,
                    NativePendingAction.invocation_id == selected,
                    NativePendingAction.id.in_(state.pending_ids),
                )
            ).all()
            pending = validate_pending_fence(
                rows, state=state, ref=ref, tenant_id=owned.tenant_id, pending_ids=state.pending_ids
            )
            if pending_reason(pending) != "approval":
                raise BuilderExecutionPolicyError("native_pending_mismatch")
        elif state.pending_ids:
            validate_pending_history(session, ref=ref, tenant_id=owned.tenant_id, state=state, states=states)
        selected = state.parent_invocation_id


class EffectAuthorizationService:
    def __init__(self, loader: OwnedNativeTestLoader, *, clock: Callable[[], datetime] = utc_now, state_store=None):
        self.loader = loader
        self.clock = clock
        self.state_store = state_store

    def _resolver(self, ref, key):
        from services.dify_builder.native_testing.agent_operations import AgentOperationBindingResolver

        resolver = self.loader.resolver
        if self.state_store is not None and isinstance(resolver, AgentOperationBindingResolver):
            return resolver.for_operation(store=self.state_store, ref=ref, key=key)
        return resolver

    def _binding(
        self, session: Session, owned: OwnedNativeTest, intent: EffectIntent, *, resolver=None
    ) -> NativeBinding:
        if intent.key.invocation_id != owned.ref.invocation_id:
            raise BuilderExecutionPolicyError("native_binding_mismatch")
        raw = owned.bindings.get(operation_digest(intent.key))
        if raw is None:
            raise BuilderExecutionPolicyError("native_cursor_mismatch")
        try:
            frozen = NativeBinding.model_validate(raw)
            current = (resolver or self.loader.resolver).resolve_binding(session=session, owned=owned, key=intent.key)
            current = NativeBinding.model_validate(current.model_dump(mode="json"))
        except (ValidationError, AttributeError, TypeError):
            raise BuilderExecutionPolicyError("native_binding_mismatch") from None
        if current != frozen:
            raise BuilderExecutionPolicyError("native_binding_drift")
        validate_binding(intent, current)
        return current

    def before_call(self, *, ref: NativeTestContextRef, intent: EffectIntent) -> EffectDecision:
        try:
            # Revalidate DTOs after in-process model_copy/model_construct boundaries.
            ref = NativeTestContextRef.model_validate(ref.model_dump(mode="json"))
            intent = EffectIntent.model_validate(intent.model_dump(mode="json"))
            self.loader.load(ref=ref)
            resolver = self._resolver(ref, intent.key)
            with self.loader.locked(ref=ref) as (session, owned, scope):
                result = self._claim(session, owned, scope, intent, resolver=resolver)
            # The transaction committed before the permit crosses the I/O boundary.
            return result
        except BuilderExecutionPolicyError as error:
            return PolicyStop(reason_code=error.reason_code)
        except (ValidationError, AttributeError, TypeError):
            return PolicyStop(reason_code="invalid_native_effect")

    def _claim(
        self, session: Session, owned: OwnedNativeTest, scope: NativeTestScope, intent: EffectIntent, *, resolver=None
    ) -> EffectDecision:
        now = self.clock()
        operation = session.scalar(
            _owned_query(NativeEffectOperation, owned).where(
                NativeEffectOperation.operation_digest == operation_digest(intent.key)
            )
        )
        digest = canonical_digest(intent.model_dump(mode="json"))
        if operation is not None and operation.intent_digest != digest:
            return PolicyStop(reason_code="native_intent_conflict")
        binding = self._binding(session, owned, intent, resolver=resolver)
        scoped = _scope_digest(intent, binding)
        if operation is not None and operation.state != "pending":
            return self._existing(session, owned, operation, now)
        if now >= scope.expires_at:
            return PolicyStop(reason_code="native_scope_expired")
        grant = session.scalar(
            select(NativeEffectGrant).where(
                NativeEffectGrant.tenant_id == owned.tenant_id,
                NativeEffectGrant.request_id == owned.ref.request_id,
                NativeEffectGrant.scope_digest == scoped,
            )
        )
        if grant is not None and grant.decision == "denied":
            return PolicyStop(reason_code="effect_denied")
        fence = dispatch_readiness(session, ref=owned.ref, tenant_id=owned.tenant_id, now=now, coalesce_self=True)
        if fence is not None:
            from services.dify_builder.native_testing.agent_operations import confirmed_deferred_tool_approval

            if not confirmed_deferred_tool_approval(
                session=session, owned=owned, intent=intent, binding=binding, fence=fence
            ):
                return ApprovalWait(pending_operation_id=fence.id, wait_reason=fence.category)
        if operation is None:
            operation = NativeEffectOperation(
                **_owner(owned),
                operation_digest=operation_digest(intent.key),
                intent_digest=digest,
                scope_digest=scoped,
                state="pending",
                result_ref=None,
            )
            session.add(operation)
            session.flush()
        if fence is not None:
            return self._pending(session, owned, operation)
        # Only the server-owned confirmed Agent producer can bind this exact
        # selected model to an already confirmed outer task. Tool calls still
        # require their actual-argument operation grant.
        from services.dify_builder.native_testing.agent_operations import confirmed_task_model

        if grant is None and task_grant_allowed(
            binding,
            task_approved=owned.task_approved,
            confirmed_model=confirmed_task_model(session=session, owned=owned, intent=intent),
        ):
            grant = NativeEffectGrant(
                tenant_id=owned.tenant_id,
                request_id=owned.ref.request_id,
                actor_id=owned.actor_id,
                scope_digest=scoped,
                decision="approved",
                expires_at=scope.expires_at,
                max_calls=scope.max_calls,
                max_dispatches=scope.max_dispatches,
                max_cost_units=scope.max_cost_units,
                spent_calls=0,
                spent_dispatches=0,
                spent_cost_units=0,
            )
            session.add(grant)
            session.flush()
        if grant is None or grant.expires_at <= now or not self._fits(grant, binding.cost_units):
            return self._pending(session, owned, operation)
        if not self._fits(scope, binding.cost_units):
            return PolicyStop(reason_code="native_budget_exhausted")
        spent = dict(scope.budget_spent)
        for name, amount in binding.budget.items():
            limit, profile, previous = scope.budget_limits.get(name), scope.profile.get(name), spent.get(name, 0)
            if (
                type(limit) is not int
                or type(profile) is not int
                or type(previous) is not int
                or min(limit, profile, previous) < 0
                or amount > profile
                or previous + amount > limit
            ):
                return PolicyStop(reason_code="native_budget_exhausted")
            spent[name] = previous + amount
        for quota in (scope, grant):
            quota.spent_calls += 1
            quota.spent_dispatches += 1
            quota.spent_cost_units += binding.cost_units
        scope.budget_spent = spent
        permit_id = str(uuid4())
        attempt = NativeEffectAttempt(
            **_owner(owned),
            operation_id=operation.id,
            grant_id=grant.id,
            dispatch_index=1,
            permit_digest=canonical_digest(permit_id),
            expires_at=min(now + timedelta(minutes=5), grant.expires_at, scope.expires_at),
            cost_units=binding.cost_units,
            budget_reservation_ref=str(uuid4()),
            execution_id=binding.execution_id,
            invocation_digest=binding.invocation_digest,
            budget=dict(binding.budget),
            profile=dict(scope.profile),
        )
        session.add(attempt)
        operation.state = "claimed"
        session.flush()
        return DispatchPermit(
            permit_id=permit_id,
            attempt_id=attempt.id,
            operation_id=operation.id,
            expires_at=attempt.expires_at.isoformat(),
        )

    @staticmethod
    def _fits(quota, cost_units: int) -> bool:
        return (
            quota.spent_calls < quota.max_calls
            and quota.spent_dispatches < quota.max_dispatches
            and quota.spent_cost_units + cost_units <= quota.max_cost_units
        )

    def _existing(
        self, session: Session, owned: OwnedNativeTest, operation: NativeEffectOperation, now: datetime
    ) -> EffectDecision:
        if operation.state == "completed" and operation.result_ref is not None:
            return RecordedResult(operation_id=operation.id, result_ref=operation.result_ref)
        if operation.state == "claimed":
            attempt = session.scalar(
                _owned_query(NativeEffectAttempt, owned).where(NativeEffectAttempt.operation_id == operation.id)
            )
            if attempt is None:
                raise BuilderExecutionPolicyError("native_ledger_inconsistent")
            if attempt.expires_at > now:
                return PolicyStop(reason_code="dispatch_in_progress")
            self._append_outcome(session, owned, operation, attempt, "unknown", None)
            return PolicyStop(reason_code="outcome_unknown")
        return PolicyStop(
            reason_code={"unknown": "outcome_unknown", "denied": "effect_denied", "cancelled": "native_cancelled"}.get(
                operation.state, "effect_already_spent"
            )
        )

    @staticmethod
    def _pending(session: Session, owned: OwnedNativeTest, operation: NativeEffectOperation) -> ApprovalWait:
        state = session.scalar(_owned_query(NativeInvocationState, owned))
        if state is None:
            raise BuilderExecutionPolicyError("native_invocation_mismatch")
        pending = session.scalar(
            _owned_query(NativePendingAction, owned).where(
                NativePendingAction.operation_id == operation.id, NativePendingAction.generation == state.generation
            )
        )
        if pending is None:
            pending = NativePendingAction(
                **_owner(owned),
                generation=state.generation,
                category="approval",
                reference_digest=canonical_digest(operation.id),
                reference=operation.id,
                owner_ref=owned.app_id,
                owner_version=operation.scope_digest,
                operation_id=operation.id,
            )
            session.add(pending)
            session.flush()
        return ApprovalWait(pending_operation_id=pending.id)

    @overload
    def decide(
        self,
        *,
        ref: NativeTestContextRef,
        intent: EffectIntent,
        decision: Literal["approved", "denied"],
        max_calls: int,
        max_dispatches: int,
        max_cost_units: int,
        expires_at: datetime,
        pending_guard: PendingDecisionGuard,
    ) -> NativeTestPublicReceipt: ...

    @overload
    def decide(
        self,
        *,
        ref: NativeTestContextRef,
        intent: EffectIntent,
        decision: Literal["approved", "denied"],
        max_calls: int,
        max_dispatches: int,
        max_cost_units: int,
        expires_at: datetime,
        pending_guard: None = None,
    ) -> str: ...

    def decide(
        self,
        *,
        ref: NativeTestContextRef,
        intent: EffectIntent,
        decision: Literal["approved", "denied"],
        max_calls: int,
        max_dispatches: int,
        max_cost_units: int,
        expires_at: datetime,
        pending_guard: PendingDecisionGuard | None = None,
    ) -> str | NativeTestPublicReceipt:
        """Internal C04 command boundary; caller authenticates explicit user confirmation.

        Public API callers must supply pending_guard; existing trusted internal
        ledger callers omit it and retain their grant-ID return contract. Public
        commands return the first receipt saved under the serialized decision lock.
        Existing grants are immutable, including limits and expiry. Repeated
        decisions are idempotent; a retry cannot mint a new quota allowance.
        """
        self.loader.load(ref=ref)
        resolver = self._resolver(ref, intent.key)
        if pending_guard is not None:
            # Refresh actual child permission at the final public authorization
            # boundary, before SQL locking. Like root RBAC, this external check
            # cannot fence revocations that happen after the RPC returns.
            authority = validate_private_read(self.loader, ref)
            if pending_guard.resolved.ref != ref or (
                authority.app_id,
                authority.workflow_id,
                authority.graph_revision,
                authority.generation,
            ) != (
                pending_guard.invocation_app_id,
                pending_guard.invocation_workflow_id,
                pending_guard.invocation_revision,
                pending_guard.resolved.generation,
            ):
                raise BuilderExecutionPolicyError("native_pending_conflict")
        if decision not in {"approved", "denied"} or any(
            type(v) is not int or v < 0 or v > 2**63 - 1 for v in (max_calls, max_dispatches, max_cost_units)
        ):
            raise BuilderExecutionPolicyError("invalid_native_grant")
        with self.loader.locked(ref=ref) as (session, owned, scope):
            if expires_at.tzinfo is None or not self.clock() < expires_at <= scope.expires_at:
                raise BuilderExecutionPolicyError("invalid_native_grant")
            binding = self._binding(session, owned, intent, resolver=resolver)
            digest = _scope_digest(intent, binding)
            if pending_guard is not None:
                from services.dify_builder.native_testing.agent_operations import AgentOperationBindingResolver

                terms = pending_guard.terms
                if (max_calls, max_dispatches, max_cost_units, expires_at) != (
                    terms.max_calls,
                    terms.max_dispatches,
                    terms.max_cost_units,
                    datetime.fromisoformat(terms.expires_at),
                ):
                    raise BuilderExecutionPolicyError("native_pending_conflict")
                pending_guard.recheck(
                    session=session,
                    owned=owned,
                    intent=intent,
                    scope_digest=digest,
                    recovery_state=resolver.recovery if isinstance(resolver, AgentOperationBindingResolver) else None,
                )
            grant = session.scalar(
                select(NativeEffectGrant).where(
                    NativeEffectGrant.tenant_id == owned.tenant_id,
                    NativeEffectGrant.request_id == ref.request_id,
                    NativeEffectGrant.scope_digest == digest,
                )
            )
            fields = {
                "decision": decision,
                "max_calls": max_calls,
                "max_dispatches": max_dispatches,
                "max_cost_units": max_cost_units,
                "expires_at": expires_at,
            }
            if grant is not None:
                if (grant.decision, grant.max_calls, grant.max_dispatches, grant.max_cost_units, grant.expires_at) != (
                    decision,
                    max_calls,
                    max_dispatches,
                    max_cost_units,
                    expires_at,
                ):
                    raise BuilderExecutionPolicyError("native_grant_conflict")
                if pending_guard is not None:
                    return pending_guard.save(session=session, request_id=ref.request_id)
                return grant.id
            grant = NativeEffectGrant(
                tenant_id=owned.tenant_id,
                request_id=ref.request_id,
                actor_id=owned.actor_id,
                scope_digest=digest,
                spent_calls=0,
                spent_dispatches=0,
                spent_cost_units=0,
                **fields,
            )
            session.add(grant)
            session.flush()
            if pending_guard is not None:
                return pending_guard.save(session=session, request_id=ref.request_id)
            return grant.id

    def _permit_ref(self, permit: DispatchPermit) -> NativeTestContextRef:
        with self.loader.session_factory() as session:
            row = session.execute(
                select(NativeEffectAttempt, DifyBuilderExecutionRequest)
                .join(DifyBuilderExecutionRequest, DifyBuilderExecutionRequest.id == NativeEffectAttempt.request_id)
                .where(
                    NativeEffectAttempt.id == permit.attempt_id,
                    NativeEffectAttempt.operation_id == permit.operation_id,
                    NativeEffectAttempt.permit_digest == canonical_digest(permit.permit_id),
                    NativeEffectAttempt.tenant_id == self.loader.actor.tenant_id,
                    DifyBuilderExecutionRequest.tenant_id == self.loader.actor.tenant_id,
                    DifyBuilderExecutionRequest.actor_id == self.loader.actor.account_id,
                )
            ).one_or_none()
            if row is None:
                raise BuilderExecutionPolicyError("invalid_dispatch_permit")
            attempt, request = row
            if attempt.expires_at != datetime.fromisoformat(permit.expires_at):
                raise BuilderExecutionPolicyError("invalid_dispatch_permit")
            return NativeTestContextRef(
                request_id=request.id, invocation_id=attempt.invocation_id, context_digest=request.context_digest
            )

    def _attempt(self, session: Session, owned: OwnedNativeTest, permit: DispatchPermit) -> NativeEffectAttempt:
        attempt = session.scalar(
            _owned_query(NativeEffectAttempt, owned).where(
                NativeEffectAttempt.id == permit.attempt_id,
                NativeEffectAttempt.operation_id == permit.operation_id,
                NativeEffectAttempt.permit_digest == canonical_digest(permit.permit_id),
            )
        )
        if attempt is None:
            raise BuilderExecutionPolicyError("invalid_dispatch_permit")
        return attempt

    def validate_dispatch(self, *, ref: NativeTestContextRef, intent: EffectIntent, permit: DispatchPermit) -> None:
        """Revalidate a reserved permit without claiming or spending another one.

        This is NOT a physical-dispatch fence. An authenticated endpoint must
        also atomically consume its durable dispatch fence before performing I/O;
        repeated validation cannot authorize repeating the transport operation.
        """
        try:
            ref = NativeTestContextRef.model_validate(ref.model_dump(mode="json"))
            intent = EffectIntent.model_validate(intent.model_dump(mode="json"))
            permit = DispatchPermit.model_validate(permit.model_dump(mode="json"))
        except (ValidationError, AttributeError, TypeError):
            raise BuilderExecutionPolicyError("invalid_dispatch_permit") from None
        if self._permit_ref(permit) != ref:
            raise BuilderExecutionPolicyError("invalid_dispatch_permit")
        self.loader.load(ref=ref)
        resolver = self._resolver(ref, intent.key)
        with self.loader.locked(ref=ref) as (session, owned, scope):
            self._validate_dispatch_locked(
                session=session, owned=owned, scope=scope, intent=intent, permit=permit, resolver=resolver
            )

    def validate_completed_config_claim(
        self, *, ref: NativeTestContextRef, intent: EffectIntent, permit: DispatchPermit
    ) -> str:
        """Read-only ACK admission; never usable as dispatch or result-store permission.

        The caller must additionally prove its immutable typed ACK and the same
        live physical job, and repeat this check after that external read.
        """
        ref = NativeTestContextRef.model_validate(ref.model_dump(mode="json"))
        intent = EffectIntent.model_validate(intent.model_dump(mode="json"))
        permit = DispatchPermit.model_validate(permit.model_dump(mode="json"))
        if (
            intent.kind != "shell"
            or intent.key.occurrence_path[-2:-1] != ("shell_run",)
            or self._permit_ref(permit) != ref
        ):
            raise BuilderExecutionPolicyError("invalid_dispatch_permit")
        self.loader.load(ref=ref)
        resolver = self._resolver(ref, intent.key)
        with self.loader.locked(ref=ref) as (session, owned, scope):
            attempt = self._attempt(session, owned, permit)
            binding = self._binding(session, owned, intent, resolver=resolver)
            operation = session.scalar(
                _owned_query(NativeEffectOperation, owned).where(NativeEffectOperation.id == permit.operation_id)
            )
            claim = session.scalar(
                _owned_query(NativeEffectDispatchClaim, owned).where(
                    NativeEffectDispatchClaim.operation_id == permit.operation_id,
                    NativeEffectDispatchClaim.attempt_id == permit.attempt_id,
                )
            )
            outcome = session.scalar(
                _owned_query(NativeEffectOutcome, owned).where(
                    NativeEffectOutcome.operation_id == permit.operation_id,
                    NativeEffectOutcome.attempt_id == permit.attempt_id,
                )
            )
            now = self.clock()
            if (
                operation is None
                or operation.operation_digest != operation_digest(intent.key)
                or operation.intent_digest != canonical_digest(intent.model_dump(mode="json"))
                or operation.scope_digest != _scope_digest(intent, binding)
                or operation.state != "completed"
                or claim is None
                or outcome is None
                or outcome.outcome != "completed"
                or outcome.result_ref is None
                or outcome.result_ref != operation.result_ref
                or attempt.expires_at != datetime.fromisoformat(permit.expires_at)
                or now >= attempt.expires_at
                or now >= scope.expires_at
            ):
                raise BuilderExecutionPolicyError("invalid_dispatch_permit")
            require_ancestor_authority(session, ref, owned.tenant_id)
            _completed_config_read_readiness(session, owned)
            return outcome.result_ref

    def claim_dispatch(self, *, ref: NativeTestContextRef, intent: EffectIntent, permit: DispatchPermit) -> None:
        """Consume a reserved permit once, durably, before the sole owner's I/O.

        The actual owner must compare its requested resource, target, typed args
        and phase with intent before calling this boundary. Success is returned
        only after commit. A lost/uncertain response must never be retried and
        followed by I/O: the committed claim cannot be reissued or refunded.
        This method never spends another permit or performs external I/O under
        the root request lock. It does not itself execute the physical effect.
        """
        try:
            ref = NativeTestContextRef.model_validate(ref.model_dump(mode="json"))
            intent = EffectIntent.model_validate(intent.model_dump(mode="json"))
            permit = DispatchPermit.model_validate(permit.model_dump(mode="json"))
        except (BuilderExecutionPolicyError, ValidationError, AttributeError, TypeError):
            raise BuilderExecutionPolicyError("invalid_dispatch_permit") from None
        if self._permit_ref(permit) != ref:
            raise BuilderExecutionPolicyError("invalid_dispatch_permit")
        self.loader.load(ref=ref)
        resolver = self._resolver(ref, intent.key)
        # Child-app RBAC may perform RPC, so refresh before the writer lock.
        # SQL provenance and state are checked again inside the transaction.
        authority = validate_private_read(self.loader, ref)
        with self.loader.locked(ref=ref) as (session, owned, scope):
            attempt = self._validate_dispatch_locked(
                session=session, owned=owned, scope=scope, intent=intent, permit=permit, resolver=resolver
            )
            require_ancestor_authority(session, ref, owned.tenant_id)
            state = session.scalar(_owned_query(NativeInvocationState, owned))
            if (
                state is None
                or state.state not in {"running", "waiting", "queued"}
                or (state.app_id, state.workflow_id, state.context_object_id, state.generation)
                != (authority.app_id, authority.workflow_id, authority.context_object_id, authority.generation)
            ):
                raise BuilderExecutionPolicyError("invalid_dispatch_permit")
            workflow = validate_invocation_workflow(
                session, tenant_id=owned.tenant_id, app_id=authority.app_id, workflow_id=authority.workflow_id
            )
            revision_context = authority.revision_context
            revision = (
                admitted_workflow_revision(workflow, revision_context)
                if revision_context
                else native_workflow_revision(workflow)
            )
            if revision != authority.graph_revision:
                raise BuilderExecutionPolicyError("native_revision_drift")
            existing = session.scalar(
                _owned_query(NativeEffectDispatchClaim, owned).where(
                    NativeEffectDispatchClaim.operation_id == attempt.operation_id
                )
            )
            if existing is not None:
                raise BuilderExecutionPolicyError("dispatch_already_consumed")
            session.add(
                NativeEffectDispatchClaim(
                    **_owner(owned),
                    operation_id=attempt.operation_id,
                    attempt_id=attempt.id,
                    claimed_at=self.clock(),
                )
            )
            session.flush()
        # Leaving loader.locked commits before any caller can perform I/O.

    def _validate_dispatch_locked(
        self,
        *,
        session: Session,
        owned: OwnedNativeTest,
        scope: NativeTestScope,
        intent: EffectIntent,
        permit: DispatchPermit,
        resolver=None,
    ) -> NativeEffectAttempt:
        """Shared local validation inside the caller's existing root transaction.

        The durable transport consumer must call this while holding loader.locked
        and insert its unique consumption fence before committing that same lock.
        No I/O, independent transaction or quota mutation occurs in this helper.
        """
        attempt = self._attempt(session, owned, permit)
        operation = session.scalar(
            _owned_query(NativeEffectOperation, owned).where(NativeEffectOperation.id == permit.operation_id)
        )
        binding = self._binding(session, owned, intent, resolver=resolver)
        now = self.clock()
        if (
            operation is None
            or operation.operation_digest != operation_digest(intent.key)
            or operation.intent_digest != canonical_digest(intent.model_dump(mode="json"))
            or operation.scope_digest != _scope_digest(intent, binding)
            or operation.state != "claimed"
            or attempt.expires_at != datetime.fromisoformat(permit.expires_at)
            or now >= attempt.expires_at
            or now >= scope.expires_at
        ):
            raise BuilderExecutionPolicyError("invalid_dispatch_permit")
        require_ancestor_authority(session, owned.ref, owned.tenant_id)
        dispatch_readiness(session, ref=owned.ref, tenant_id=owned.tenant_id, now=now)
        return attempt

    def record_outcome(self, *, permit: DispatchPermit, outcome: Outcome, result_ref: str | None) -> EffectReceipt:
        receipt = EffectReceipt(
            operation_id=permit.operation_id, attempt_id=permit.attempt_id, outcome=outcome, result_ref=result_ref
        )
        if outcome == "completed" and result_ref is None:
            raise BuilderExecutionPolicyError("completed_result_required")
        ref = self._permit_ref(permit)
        with self.loader.locked(ref=ref, settlement=True) as (session, owned, _):
            attempt = self._attempt(session, owned, permit)
            operation = session.scalar(
                _owned_query(NativeEffectOperation, owned).where(NativeEffectOperation.id == permit.operation_id)
            )
            if operation is None:
                raise BuilderExecutionPolicyError("invalid_dispatch_permit")
            existing = session.scalar(
                _owned_query(NativeEffectOutcome, owned).where(NativeEffectOutcome.attempt_id == attempt.id)
            )
            if existing is not None:
                if existing.outcome != outcome or existing.result_ref != result_ref:
                    raise BuilderExecutionPolicyError("outcome_conflict")
                return receipt
            self._append_outcome(session, owned, operation, attempt, outcome, result_ref)
        return receipt

    @staticmethod
    def _append_outcome(session, owned, operation, attempt, outcome, result_ref):
        session.add(
            NativeEffectOutcome(
                **_owner(owned),
                operation_id=operation.id,
                attempt_id=attempt.id,
                outcome=outcome,
                result_ref=result_ref,
            )
        )
        operation.state, operation.result_ref = outcome, result_ref

    def resolve_budget_reservation(
        self, *, permit: DispatchPermit, execution_id: str, invocation_digest: str
    ) -> BudgetReservation:
        ref = self._permit_ref(permit)
        self.loader.load(ref=ref)
        with self.loader.locked(ref=ref) as (session, owned, _):
            attempt = self._attempt(session, owned, permit)
            return self._reservation(session, owned, attempt, execution_id, invocation_digest)

    def resolve_budget_reference(
        self, *, ref: NativeTestContextRef, budget_reservation_ref: str, execution_id: str, invocation_digest: str
    ) -> BudgetReservation:
        """CS6 authenticated server bridge; a supplied reference is never authority."""
        self.loader.load(ref=ref)
        with self.loader.locked(ref=ref) as (session, owned, _):
            attempt = session.scalar(
                _owned_query(NativeEffectAttempt, owned).where(
                    NativeEffectAttempt.budget_reservation_ref == budget_reservation_ref
                )
            )
            if attempt is None:
                raise BuilderExecutionPolicyError("invalid_budget_reservation")
            return self._reservation(session, owned, attempt, execution_id, invocation_digest)

    def _reservation(self, session, owned, attempt, execution_id, invocation_digest) -> BudgetReservation:
        operation = session.scalar(
            _owned_query(NativeEffectOperation, owned).where(NativeEffectOperation.id == attempt.operation_id)
        )
        if (
            attempt.execution_id != execution_id
            or attempt.invocation_digest != invocation_digest
            or attempt.expires_at <= self.clock()
            or operation is None
            or operation.state != "claimed"
        ):
            raise BuilderExecutionPolicyError("invalid_budget_reservation")
        return BudgetReservation(
            budget_reservation_ref=attempt.budget_reservation_ref,
            execution_id=attempt.execution_id,
            invocation_digest=attempt.invocation_digest,
            expires_at=attempt.expires_at,
            limits=dict(attempt.budget),
            profile=dict(attempt.profile),
        )