Engine¶

async def coordinate(
    self,
    context: CoordinationContext,
) -> CoordinationResultWithAttribution:
    """Delegate to the multi-agent coordinator."""
    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 run(  # noqa: PLR0913, C901
    self,
    *,
    identity: AgentIdentity,
    task: Task,
    completion_config: CompletionConfig | None = None,
    max_turns: int = DEFAULT_MAX_TURNS,
    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."""
    agent_id = str(identity.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):
        start = self._clock.monotonic()
        ctx: AgentContext | None = None
        system_prompt: SystemPrompt | None = None
        provider: CompletionProvider = self._provider
        _project_budget: float = 0.0
        try:
            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,
            )

            if self._budget_enforcer:
                preflight = await self._budget_enforcer.check_can_execute(
                    agent_id,
                    provider_name=identity.model.provider,
                )
                provider, identity = self._apply_degradation(
                    preflight,
                    identity,
                    provider,
                )
                identity = await self._budget_enforcer.resolve_model(
                    identity,
                )

            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:
                logger.warning(
                    EXECUTION_PROJECT_VALIDATION_FAILED,
                    agent_id=agent_id,
                    task_id=task_id,
                    project_id=task.project,
                    reason="project_repo_not_configured",
                )

            replay_ctx: AgentContext | None = None
            if resume_execution_id is not None and self._event_reader is not None:
                from synthorg.engine.session import Session  # noqa: PLC0415

                replay_result = await Session.replay(
                    execution_id=resume_execution_id,
                    event_reader=self._event_reader,
                    identity=identity,
                    task=task,
                    max_turns=max_turns,
                )
                if (
                    replay_result.replay_completeness
                    < _REPLAY_LOW_COMPLETENESS_THRESHOLD
                ):
                    logger.warning(
                        SESSION_REPLAY_LOW_COMPLETENESS,
                        execution_id=resume_execution_id,
                        replay_completeness=replay_result.replay_completeness,
                    )
                replay_ctx = replay_result.context

            tool_invoker = self._make_tool_invoker(
                identity,
                task_id=task_id,
                effective_autonomy=effective_autonomy,
            )
            ctx, system_prompt = await self._prepare_context(
                identity=identity,
                task=task,
                agent_id=agent_id,
                task_id=task_id,
                max_turns=max_turns,
                memory_messages=memory_messages,
                tool_invoker=tool_invoker,
                effective_autonomy=effective_autonomy,
            )
            if replay_ctx is not None:
                ctx = ctx.model_copy(
                    update={
                        "execution_id": replay_ctx.execution_id,
                        "started_at": replay_ctx.started_at,
                        "conversation": (
                            *ctx.conversation,
                            *replay_ctx.conversation,
                        ),
                        "accumulated_cost": replay_ctx.accumulated_cost,
                        "turn_count": replay_ctx.turn_count,
                        "task_execution": (
                            replay_ctx.task_execution or ctx.task_execution
                        ),
                    },
                )
            return await self._execute(
                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:
            logger.exception(
                EXECUTION_ENGINE_ERROR,
                agent_id=agent_id,
                task_id=task_id,
                error="non-recoverable error in run()",
            )
            raise
        except ProjectNotFoundError, ProjectAgentNotMemberError:
            raise
        except BudgetExhaustedError as exc:
            return 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,
            )
        except Exception as exc:
            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,
) -> 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).

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

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

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

    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

    while ctx.has_turns_remaining:
        shutdown_result = check_shutdown(ctx, shutdown_checker, turns)
        if shutdown_result is not None:
            return shutdown_result

        budget_result = check_budget(ctx, budget_checker, turns)
        if budget_result is not None:
            return budget_result

        # 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
        response = await call_provider(
            ctx,
            provider,
            model_id,
            tool_defs,
            config,
            turn_number,
            turns,
        )
        if isinstance(response, ExecutionResult):
            return response

        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,
            turns,
            tool_invoker,
            shutdown_checker,
        )
        if isinstance(result, ExecutionResult):
            return result
        ctx = result

        # Stagnation detection after successful turn processing
        stag_outcome = await check_stagnation(
            ctx,
            self._stagnation_detector,
            turns,
            corrections_injected,
            execution_id=ctx.execution_id,
        )
        if isinstance(stag_outcome, ExecutionResult):
            return stag_outcome
        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

    logger.info(
        EXECUTION_LOOP_TERMINATED,
        execution_id=ctx.execution_id,
        reason=TerminationReason.MAX_TURNS.value,
        turns=len(turns),
    )
    return build_result(ctx, TerminationReason.MAX_TURNS, turns)

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

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,
) -> ExecutionResult:
    """Run the Plan-and-Execute 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.

    Returns:
        Execution result with final context and termination info.

    Raises:
        MemoryError: Re-raised unconditionally (non-recoverable).
        RecursionError: Re-raised unconditionally (non-recoverable).
    """
    logger.info(
        EXECUTION_LOOP_START,
        execution_id=context.execution_id,
        loop_type=self.get_loop_type(),
        max_turns=context.max_turns,
    )

    ctx = context
    default_model = ctx.identity.model.model_id
    planner_model = self._config.planner_model or default_model
    executor_model = self._config.executor_model or default_model
    default_config = completion_config or CompletionConfig(
        temperature=ctx.identity.model.temperature,
        max_tokens=ctx.identity.model.max_tokens,
    )
    tool_defs = get_tool_definitions(tool_invoker, ctx.loaded_tools)
    turns: list[TurnRecord] = []
    all_plans: list[ExecutionPlan] = []
    replans_used = 0

    # Planning.
    plan_result = await self._run_planning_phase(
        ctx,
        provider,
        planner_model,
        default_config,
        turns,
        shutdown_checker,
        budget_checker,
    )
    if isinstance(plan_result, ExecutionResult):
        return self._finalize(plan_result, all_plans, replans_used)
    ctx, plan = plan_result
    all_plans.append(plan)

    # Execute steps.
    return await self._run_steps(
        ctx,
        provider,
        executor_model,
        default_config,
        tool_defs,
        tool_invoker,
        plan,
        turns,
        all_plans,
        replans_used,
        planner_model,
        budget_checker,
        shutdown_checker,
    )