Engine

Engine is the execution kernel that owns the agent loop: it calls your AgentModule hooks in sequence, executes actions, manages tracing, and enforces stop criteria.

Advanced/compatible programmatic examples. Fragments illustrate mechanisms; HostEnv or workspace does not provide isolation. New projects start with Quickstart, Session and explicit resource configuration.

Engine is the single execution kernel for all QitOS agent workflows. It runs the step loop, coordinates your AgentModule hooks, dispatches tool calls to the ToolRegistry for execution, applies critics (post-step evaluators that can approve, stop, or retry the step), checks stop conditions, and writes trace artifacts (persistent records like steps.jsonl and events.jsonl that capture what happened during the run). You interact with it directly only when you need control beyond what agent.run() provides.

QitOS enforces a single-kernel rule: there is exactly one Engine per run. Extensions like parsers, critics, memory adapters, and toolkits attach to this pipeline — they do not introduce a second execution loop.

How the loop works

Each step follows a fixed sequence:

prepare → decide → act → reduce → critics → check_stop → trace

1. prepare — agent.prepare(state) formats state into the model-ready prompt text.
2. decide — agent.decide(state, observation) is checked first; if it returns None, the Engine calls the LLM via the configured parser.
3. act — Tool calls in the decision’s actions list are executed against the ToolRegistry.
4. reduce — agent.reduce(state, observation, decision) updates state with the new observation.
5. critics — Any registered Critic instances (post-step evaluators that can approve, stop, or retry the step) evaluate the step; they can trigger a stop or retry.
6. check_stop — Budget exhaustion, FinalResultCriteria, agent.should_stop(), and any custom StopCriteria are evaluated.
7. trace — The step record and events are written to the TraceWriter.

Constructor

Illustrative fragment (not a standalone program; use the complete example linked on this page).

from qitos import Engine

engine = Engine(
    agent=agent,
    budget=RuntimeBudget(max_steps=20),
    parser=ReActTextParser(),
    critics=[my_critic],
    stop_criteria=[FinalResultCriteria()],
    env=host_env,
    trace_writer=trace_writer,
    hooks=[my_hook],
)

Parameter	Type	Description
agent	AgentModule	Required. The policy module to execute.
budget	RuntimeBudget | None	Step/time/token limits. Defaults to RuntimeBudget(max_steps=10).
parser	Parser | None	Parses raw model output into a Decision. Must match your prompt format.
critics	list[Critic] | None	Post-step evaluators that can stop or retry a step.
stop_criteria	list[StopCriteria] | None	Stop conditions. Defaults to [FinalResultCriteria()].
env	Env | None	Environment providing reset/observe/step/is_terminal/close.
trace_writer	TraceWriter | None	Writes manifest.json, events.jsonl, steps.jsonl for this run.
hooks	list[EngineHook] | None	Hook instances called on lifecycle events.
render_hooks	list | None	Render hooks (e.g., terminal UI). Appended to hooks internally.
history_policy	HistoryPolicy | None	Controls how the in-run conversation history is managed.
recovery_policy	RecoveryPolicy | None	Controls how the Engine responds to step failures.

Prefer calling agent.run() for single-run workflows. Use Engine directly when you need to reuse an Engine across multiple runs or configure hooks dynamically between runs.

Docker environment paths and variables

DockerEnv(container_env={...}) passes an explicit environment mapping when it creates a container. Relative filesystem paths resolve under the configured container workdir; absolute paths are interpreted as absolute paths inside the container. QitOS does not copy benchmark- or task-specific host environment variables implicitly.

Protocol resolution

Engine resolves the model protocol once per run using the following precedence:

1. The protocol passed directly to Engine.
2. agent.model_protocol.
3. A protocol inferred from the Engine or Agent parser.
4. agent.llm.qitos_protocol, then agent.llm.qitos_harness_metadata["protocol"].
5. A protocol inferred from the model name.
6. The framework default, react_text_v1.

The model-level declarations in step 4 are attached by build_model_for_preset(...). This means direct construction such as Engine(agent=agent) preserves an explicitly selected family preset even when the provider model name is an alias that QitOS cannot infer. Unknown model-level protocol identifiers are ignored so the existing model-name and framework fallbacks remain available. Traces record model_qitos_protocol or model_harness_metadata as the resolution source when either declaration is selected.

Empty model responses

After a model call, the Engine records the normalized response and gives AgentModule.interpret_model_response() the first opportunity to turn it into a decision. If the hook returns None and the response contains neither non-whitespace text nor tool calls, the Engine classifies the result as a model_error before it reaches the parser. The default recovery path retries once; a second consecutive empty response stops the run with unrecoverable_error instead of consuming the remaining step budget as parser wait decisions.

The model_output event and StepRecord.model_response still retain the response’s finish_reason, usage, model, and provider fields. This guard does not change native tool-call responses, explicit Decision.wait values, or parser feedback for non-empty malformed model output.

Engine.run(task)

Illustrative fragment (not a standalone program; use the complete example linked on this page).

result = engine.run(task)

Accepts a plain string objective or a structured Task object. Returns an EngineResult.

When you pass a Task, the Engine extracts the budget from task.budget and overrides the Engine’s own budget for that run. It also orchestrates resource staging and environment lifecycle (reset, observe, close) automatically.

EngineResult

Illustrative fragment (not a standalone program; use the complete example linked on this page).

@dataclass
class EngineResult(Generic[StateT]):
    state: StateT
    records: List[StepRecord]
    events: List[RuntimeEvent]
    step_count: int
    task_result: Optional[TaskResult]

Field	Description
state	Final typed state after the run. Check state.final_result and state.stop_reason.
records	One StepRecord per step, containing decision, observation, and diffs.
events	All RuntimeEvent objects emitted during the run.
step_count	Number of steps executed (len(records)).
task_result	Structured outcome including success flag and criterion results.

Typical usage:

Illustrative fragment (not a standalone program; use the complete example linked on this page).

result = engine.run("summarize the paper")

print(result.state.final_result)   # the agent's final answer
print(result.state.stop_reason)    # why the loop ended
print(result.step_count)           # how many steps ran
print(result.task_result.success)  # whether success criteria passed

Hooks

Hooks observe and react to lifecycle events without modifying Engine internals. They implement EngineHook and are called at on_before_step and on_after_step boundaries.

Illustrative fragment (not a standalone program; use the complete example linked on this page).

engine.register_hook(my_hook)
engine.unregister_hook(my_hook)
engine.clear_hooks()

You can also pass hooks at construction time via the hooks parameter, or at run time via agent.run(hooks=[...]).

Runtime exception reporting

Handled exceptions — anything the recovery layer catches during DECIDE/ACT, plus fatal init_state and env-setup failures — are reported through Engine._report_runtime_exception(): one stderr line with the full traceback, a RECOVER trace event carrying the structured payload, and (when QITOS_ERROR_LOG or QITOS_TRACE_DIR is set) an appended step_error.log. The last payload is exposed as last_error in the EngineResult run summary, so a run that ends with unrecoverable_error keeps usable diagnostics even under redirected output.

Tool-call loop detection can be disabled for long-running agents that intentionally repeat calls: ContextConfig(tool_call_loop_detection_enabled=False) (default True, with loop_max_repeats controlling the threshold).

Budget exhaustion

When the step budget, wall-clock time limit, or token budget is exceeded, the Engine sets state.stop_reason to the appropriate value and emits an END event. The run terminates gracefully and EngineResult is still returned — inspect state.stop_reason to detect this case.

Illustrative fragment (not a standalone program; use the complete example linked on this page).

from qitos.engine.states import RuntimeBudget

engine = Engine(
    agent=agent,
    budget=RuntimeBudget(
        max_steps=30,
        max_runtime_seconds=120.0,
        max_tokens=50_000,
    ),
)

Building an Engine from AgentModule

AgentModule.build_engine() is a convenience factory that creates an Engine pre-bound to the agent:

Illustrative fragment (not a standalone program; use the complete example linked on this page).

engine = agent.build_engine(
    budget=RuntimeBudget(max_steps=15),
    trace_writer=trace_writer,
)
result = engine.run(task)

This is equivalent to Engine(agent=agent, ...) and is what agent.run() calls internally.

AsyncEngine

AsyncEngine provides non-blocking execution for agent workflows. It wraps the same Engine loop but runs blocking calls in a thread pool, making it safe to use inside asyncio event loops.

Illustrative fragment (not a standalone program; use the complete example linked on this page).

from qitos import AsyncEngine

async_engine = AsyncEngine(agent=agent, budget=RuntimeBudget(max_steps=20))

AsyncEngine.arun(task)

Run the agent loop asynchronously. Returns the same EngineResult as Engine.run().

Illustrative fragment (not a standalone program; use the complete example linked on this page).

result = await async_engine.arun("analyze the data")
print(result.state.final_result)

AsyncEngine.arun_stream(task)

Run the agent loop and yield structured EngineEvent objects as they occur — ideal for real-time UI updates or streaming progress to a client.

Illustrative fragment (not a standalone program; use the complete example linked on this page).

async for event in async_engine.arun_stream("analyze the data"):
    print(event.event_type, event.step_id, event.payload)

Events are emitted at step boundaries (step_start, step_end), phase transitions (decide, act, reduce, critic, check_stop), and multi-agent events (handoff, delegate, fanout). The stream always begins with run_start and ends with run_end.

EngineEvent

Illustrative fragment (not a standalone program; use the complete example linked on this page).

@dataclass
class EngineEvent:
    event_type: EngineEventType   # step_start, step_end, decide, act, ...
    step_id: int
    agent_id: str | None          # which agent produced this step
    phase: RuntimePhase | None
    ok: bool
    payload: Dict[str, Any]
    error: str | None
    ts: str                       # ISO 8601 timestamp

EventStream

EventStream is the async queue that underpins arun_stream(). You can also use it standalone to fan out events to multiple consumers:

Illustrative fragment (not a standalone program; use the complete example linked on this page).

from qitos.engine.events import EventStream

stream = EventStream()
q1 = stream.subscribe()  # asyncio.Queue for consumer 1
q2 = stream.subscribe()  # asyncio.Queue for consumer 2

Async models

When the configured model implements acall() (from AsyncModel), AsyncEngine can invoke it without blocking the event loop. Built-in async model adapters:

Illustrative fragment (not a standalone program; use the complete example linked on this page).

from qitos.models import AsyncOpenAICompatibleModel, AsyncOpenAIModel

llm = AsyncOpenAICompatibleModel(
    model="qwen-plus",
    api_key="...",
    base_url="https://...",
)
result = await llm.acall([{"role": "user", "content": "Hello"}])

AsyncEngine.arun() works with any model — sync models are automatically dispatched to a thread pool. Use async model adapters when you need true non-blocking I/O, e.g. in high-concurrency web servers.