Engine¶

def set_autonomy_resolution(self, resolution: AutonomyResolution) -> None:
    """Bind the one resolver every dispatch path asks for autonomy.

    Called by the boot path once the worker execution service exists.
    Until it is bound, a caller that supplies no autonomy runs
    degraded, which is what a coordinated wave did permanently.

    Args:
        resolution: The single owner of "what autonomy governs this
            run", asked whenever :meth:`run` is called without one.
    """
    self._autonomy_resolution = resolution

async def coordinate(
    self,
    context: CoordinationContext,
) -> CoordinationResultWithAttribution:
    """Delegate to the multi-agent coordinator.

    Returns:
        The :class:`CoordinationResultWithAttribution` from the
        coordinator's ``coordinate()`` call.

    Raises:
        ExecutionStateError: If no coordinator was configured.
    """
    if self._coordinator is None:
        msg = "No coordinator configured for multi-agent dispatch"
        logger.warning(
            EXECUTION_ENGINE_ERROR,
            error=msg,
        )
        raise ExecutionStateError(msg)
    return await self._coordinator.coordinate(context)

async def project_background_failure(self, *, task_id: str, agent_id: str) -> None:
    """Project a terminal RUN_ERROR for a run that failed before the loop.

    A backgrounded conversational run can fail in the pipeline spine
    (project resolution, decomposition, assignment) before the execution
    loop ever runs to publish its own terminal frame, leaving a dashboard
    subscribed to the task's SSE stream hung on "Working". Called by the
    worker's background wrapper on such a failure so the operator sees the
    run end. No-op when no event-stream hub is wired.
    """
    hub = self._event_stream_hub
    if hub is None:
        return
    await publish_run_terminated(
        hub, task_id=task_id, agent_id=agent_id, reason=TerminationReason.ERROR
    )

async def run(
    self,
    *,
    identity: AgentIdentity,
    task: Task,
    completion_config: CompletionConfig | None = None,
    max_turns: int | None = None,
    memory_messages: tuple[ChatMessage, ...] = (),
    timeout_seconds: float | None = None,
    effective_autonomy: EffectiveAutonomy | None = None,
    resume_execution_id: str | None = None,
) -> AgentRunResult:
    """Execute an agent on a task.

    Returns:
        The :class:`AgentRunResult` from the loop, with cost
        tracking, post-execution transitions, and recovery /
        checkpoint resume applied.

    Raises:
        MemoryError: Re-raised after logging from the explicit
            log-and-raise critical-error path (the engine surfaces
            non-recoverable interpreter signals to the worker).
        RecursionError: Same path as ``MemoryError``.
        ProjectNotFoundError: From project validation when the
            task references a missing project.
    """
    agent_id = str(identity.id)
    task_id = str(task.id)
    if max_turns is None:
        max_turns = await self._resolve_max_turns(
            agent_id=agent_id, task_id=task_id
        )

    validate_run_inputs(
        agent_id=agent_id,
        task_id=task_id,
        max_turns=max_turns,
        timeout_seconds=timeout_seconds,
    )
    validate_agent(identity, agent_id)
    validate_task(task, agent_id, task_id)
    validate_task_metadata(task, agent_id, task_id)

    with (
        correlation_scope(
            agent_id=agent_id,
            task_id=task_id,
            project_id=task.project,
        ),
        ExitStack() as run_scopes,
    ):
        start = self._clock.monotonic()
        ctx: AgentContext | None = None
        system_prompt: SystemPrompt | None = None
        provider: CompletionProvider = self._provider
        _project_budget: float = 0.0
        try:
            # Entered here rather than in the `with` header above so a
            # capture failure lands inside the fatal-error boundary. The
            # probe deliberately propagates everything that is not storage
            # I/O, and outside the boundary that left the run with no
            # terminal projection at all: no FAILED, nothing to replan.
            run_scopes.enter_context(
                artifact_baseline_scope(
                    await capture_run_baseline(
                        self._artifact_probe,
                        project_id=task.project,
                        expected=task.artifacts_expected,
                    )
                )
            )
            # Dispatch to the provider serving this agent's own model,
            # which nothing downstream re-points; a registry miss (agent
            # pinned to an unregistered provider) fails the run here
            # rather than mis-dispatching to the engine default.
            provider = self._dispatch_client_for(identity, self._provider)
            if effective_autonomy is None:
                effective_autonomy = await self._effective_autonomy_for(
                    identity, task_id=task_id, project_id=task.project
                )
            loop_mode = (
                "auto"
                if self._auto_loop_config is not None
                else self._loop.get_loop_type()
            )
            logger.info(
                EXECUTION_ENGINE_START,
                agent_id=agent_id,
                task_id=task_id,
                loop_type=loop_mode,
                max_turns=max_turns,
            )

            provider, identity, completion_config = await self._bind_run(
                identity=identity,
                task=task,
                provider=provider,
                completion_config=completion_config,
            )

            if self._project_repo is not None:
                _project_budget = await self._validate_project(
                    task=task,
                    agent_id=agent_id,
                    task_id=task_id,
                )
            elif task.project:
                # Fail loud for a work task (aborts to the fatal-error
                # boundary, which terminates the task FAILED) rather than
                # running it unvalidated against an unconfigured repo.
                self._reject_unconfigured_project_repo(
                    task=task,
                    agent_id=agent_id,
                    task_id=task_id,
                )

            replay_ctx: AgentContext | None = None
            if resume_execution_id is not None:
                replay_ctx = await self._replay_session(
                    resume_execution_id=resume_execution_id,
                    identity=identity,
                    task=task,
                    max_turns=max_turns,
                )

            # Once for the whole task. The reconciler can replace a
            # backend between these two calls, and a task whose tools
            # came from one strategy while its context came from another
            # would recall against a backend its tools do not write to.
            memory_strategy = self._resolve_memory_strategy()
            tool_invoker = self._make_tool_invoker(
                identity,
                task_id=task_id,
                effective_autonomy=effective_autonomy,
                project_id=task.project,
                memory_strategy=memory_strategy,
            )
            ctx, system_prompt = await self._prepare_context(
                identity=identity,
                task=task,
                agent_id=agent_id,
                task_id=task_id,
                max_turns=max_turns,
                memory=MemoryContextInputs(
                    messages=memory_messages, strategy=memory_strategy
                ),
                tool_invoker=tool_invoker,
                effective_autonomy=effective_autonomy,
            )
            if replay_ctx is not None:
                ctx = self._merge_replayed(ctx, replay_ctx)
            # Bind the run identity (same execution_id flight frames
            # carry) so capture leaves tag records the receipt joins on.
            with run_identity_scope(
                execution_id=ctx.execution_id,
                task_id=task_id,
                project_id=task.project,
            ):
                return await self._execute(
                    AgentExecuteRequest(
                        identity=identity,
                        task=task,
                        agent_id=agent_id,
                        task_id=task_id,
                        completion_config=completion_config,
                        ctx=ctx,
                        system_prompt=system_prompt,
                        start=start,
                        timeout_seconds=timeout_seconds,
                        tool_invoker=tool_invoker,
                        effective_autonomy=effective_autonomy,
                        provider=provider,
                        project_budget=_project_budget,
                    )
                )
        except (MemoryError, RecursionError) as exc:
            log_exception_redacted(
                logger,
                EXECUTION_ENGINE_ERROR,
                exc,
                agent_id=agent_id,
                task_id=task_id,
                reason="non-recoverable error in run()",
            )
            raise
        except ProjectNotFoundError:
            raise
        except BudgetExhaustedError as exc:
            budget_result = await self._handle_budget_error(
                exc=exc,
                identity=identity,
                task=task,
                agent_id=agent_id,
                task_id=task_id,
                duration_seconds=self._clock.monotonic() - start,
                ctx=ctx,
                system_prompt=system_prompt,
            )
            # Project the terminal the budget handler actually selected: a
            # parked hard-ceiling crossing is PARKED (silent -- the pause
            # surfaces via the approval-interrupt projection), a plain
            # controlled stop is BUDGET_EXHAUSTED (RUN_ERROR). The inner
            # handler skips RUN_ERROR for a budget error precisely so a
            # parked run is never projected as failed.
            budget_hub = self._event_stream_hub
            if budget_hub is not None:
                await publish_run_terminated(
                    budget_hub,
                    task_id=task_id,
                    agent_id=agent_id,
                    reason=budget_result.execution_result.termination_reason,
                )
            return budget_result
        except StakesModelUnavailableError as exc:
            return await self._handle_stakes_unavailable(
                exc=exc,
                identity=identity,
                task=task,
                agent_id=agent_id,
                task_id=task_id,
                duration_seconds=self._clock.monotonic() - start,
                ctx=ctx,
                system_prompt=system_prompt,
                completion_config=completion_config,
                effective_autonomy=effective_autonomy,
                provider=provider,
            )
        except Exception as exc:  # noqa: BLE001 -- engine fatal-error boundary
            # lint-allow: swallow-ok -- fatal-error boundary returns FAILED
            return await self._handle_fatal_error(
                exc=exc,
                identity=identity,
                task=task,
                agent_id=agent_id,
                task_id=task_id,
                duration_seconds=self._clock.monotonic() - start,
                ctx=ctx,
                system_prompt=system_prompt,
                completion_config=completion_config,
                effective_autonomy=effective_autonomy,
                provider=provider,
            )

async def execute(  # noqa: PLR0913
    self,
    *,
    context: AgentContext,
    provider: CompletionProvider,
    tool_invoker: ToolInvokerProtocol | None = None,
    budget_checker: BudgetChecker | None = None,
    shutdown_checker: ShutdownChecker | None = None,
    completion_config: CompletionConfig | None = None,
    task_cancellation_checker: TaskCancellationChecker | None = None,
    turn_observer: TurnObserver | None = None,
    streaming_enabled: bool = False,
) -> ExecutionResult:
    """Run the execution loop.

    Args:
        context: Initial agent context with conversation and identity.
        provider: LLM completion provider.
        tool_invoker: Optional tool invoker for tool execution.
        budget_checker: Optional callback; returns ``True`` when
            budget is exhausted.
        shutdown_checker: Optional callback; returns ``True`` when
            a graceful shutdown has been requested.
        completion_config: Optional per-execution override for
            temperature/max_tokens (defaults to identity's model config).
        task_cancellation_checker: Optional async callback; returns
            ``True`` when the running task was cancelled or superseded
            externally, so the loop halts at the next safe boundary.
        turn_observer: Optional per-run progress callback; used to
            project live execution progress onto the AG-UI stream and to
            keep the live-activity state current. Awaited once per turn
            with a single :class:`TurnProgress`.
        streaming_enabled: When ``True``, each per-turn LLM call streams
            and is interruptible mid-flight (operator cancellation and
            steering REDIRECT); otherwise a non-streaming call is used.

    Returns:
        Execution result with final context and termination reason.
    """
    ...

def get_loop_type(self) -> str:
    """Return the loop type identifier (e.g. ``"react"``).

    Returns:
        The loop's type discriminator string.
    """
    ...

def get_loop_type(self) -> str:
    """Return the loop type identifier."""
    return "react"

async def execute(  # noqa: PLR0913
    self,
    *,
    context: AgentContext,
    provider: CompletionProvider,
    tool_invoker: ToolInvokerProtocol | None = None,
    budget_checker: BudgetChecker | None = None,
    shutdown_checker: ShutdownChecker | None = None,
    completion_config: CompletionConfig | None = None,
    task_cancellation_checker: TaskCancellationChecker | None = None,
    turn_observer: TurnObserver | None = None,
    streaming_enabled: bool = False,
) -> ExecutionResult:
    """Run the ReAct loop until termination.

    Args:
        context: Initial agent context with conversation.
        provider: LLM completion provider.
        tool_invoker: Optional tool invoker for tool execution.
        budget_checker: Optional budget exhaustion callback.
        shutdown_checker: Optional callback; returns ``True`` when
            a graceful shutdown has been requested.
        completion_config: Optional per-execution config override.
        task_cancellation_checker: Optional async callback; returns
            ``True`` when the task was cancelled/superseded externally.
        turn_observer: Optional per-run progress callback; when given,
            it takes precedence over the construction-time observer so
            a per-execution stream (e.g. AG-UI task progress) can be
            wired without rebuilding the shared loop.
        streaming_enabled: When ``True``, each per-turn LLM call streams
            and is interruptible mid-flight (operator cancellation and
            steering REDIRECT); otherwise a non-streaming call is used.

    Returns:
        Execution result with final context and termination info.

    Raises:
        MemoryError: Re-raised unconditionally (non-recoverable).
        RecursionError: Re-raised unconditionally (non-recoverable).
    """
    model_id, config, tool_defs, turns = self._prepare_loop(
        context, completion_config, tool_invoker
    )
    ctx = context
    corrections_injected = 0
    effective_observer = turn_observer or self._turn_observer

    # Bounded by the turn budget and its extensions; every iteration
    # re-checks shutdown, task cancellation and the cost budget below.
    # lint-allow: long-running-loop-kill-switch -- turn-budget bounded
    while True:
        if not ctx.has_turns_remaining:
            # The ceiling is a backstop against a pathological loop, not
            # a verdict on work that is taking longer than the estimate.
            # Carry on while there are extensions left; park only once
            # they are spent, so nothing is discarded either way.
            extended = grant_extension(ctx, turns)
            if extended is None:
                break
            ctx = extended
        shutdown_result = check_shutdown(ctx, shutdown_checker, turns)
        if shutdown_result is not None:
            return await self._attach_whole_run_signals(shutdown_result, turns)

        budget_result = check_budget(ctx, budget_checker, turns)
        if budget_result is not None:
            return await self._attach_whole_run_signals(budget_result, turns)

        cancel_result = await check_task_cancelled(
            ctx, task_cancellation_checker, turns
        )
        if cancel_result is not None:
            return await self._attach_whole_run_signals(cancel_result, turns)

        # Adopt any pending steering directives before the LLM call so
        # the operator's constraint is in context for this turn.
        steered = await check_steering(ctx, self._steering_inbox)
        if steered is not None:
            ctx = steered

        # Refresh tool defs each turn so newly loaded tools appear
        tool_defs = get_tool_definitions(tool_invoker, ctx.loaded_tools)

        turn_number = ctx.turn_count + 1
        outcome = await run_provider_turn(
            ctx,
            provider,
            model_id,
            tool_defs=tool_defs,
            config=config,
            turns=turns,
            streaming_enabled=streaming_enabled,
            watch=InterruptWatch(
                cancellation_checker=task_cancellation_checker,
                steering_inbox=self._steering_inbox,
                clock=self._clock,
            ),
        )
        if isinstance(outcome, ExecutionResult):
            return await self._attach_whole_run_signals(outcome, turns)
        if isinstance(outcome, _TurnInterrupted):
            # A steering REDIRECT aborted the in-flight call; fold the
            # partial usage and re-issue the turn so the top-of-loop
            # steering check adopts the directive into context.
            ctx = fold_interrupt_usage(ctx, outcome)
            continue
        response = outcome

        turns.append(
            make_turn_record(
                turn_number,
                response,
                call_category=classify_turn(turn_number, response, ctx),
                provider_metadata=response.provider_metadata,
            )
        )

        result = await self._process_turn_response(
            ctx,
            response,
            turn_number=turn_number,
            turns=turns,
            tool_invoker=tool_invoker,
            shutdown_checker=shutdown_checker,
        )
        if isinstance(result, ExecutionResult):
            return await self._attach_whole_run_signals(result, turns)
        ctx = result

        await self._notify_turn_observer(
            turn_number, response, effective_observer, ctx
        )

        # Before the fingerprint detector, because this signal survives
        # drifting arguments: a turn whose every tool call resolved to
        # nothing ran nothing, and a run doing only that has no way back.
        unresolved = unresolved_tools_result(ctx, turns)
        if unresolved is not None:
            return await self._attach_whole_run_signals(unresolved, turns)

        # Stagnation detection after successful turn processing
        stag_outcome = await check_stagnation(
            ctx,
            self._stagnation_detector,
            turns,
            corrections_injected,
        )
        if isinstance(stag_outcome, ExecutionResult):
            return await self._attach_whole_run_signals(stag_outcome, turns)
        if isinstance(stag_outcome, tuple):
            ctx, corrections_injected = stag_outcome

        # Context compaction at turn boundaries
        compacted = await invoke_compaction(
            ctx,
            self._compaction_callback,
            turn_number,
        )
        if compacted is not None:
            ctx = compacted

    return await self._attach_whole_run_signals(
        ceiling_result(ctx, turns),
        turns,
    )