Skip to content

Runtime

Runtime types, events, and registry for pluggable execution engines.

RuntimeEvent

RuntimeEvent dataclass

Unified event stream from the runtime.

Types: - assistant_delta: data={"text": "..."} - status: data={"text": "..."} - tool_call_started: data={"name": "...", "correlation_id": "...", "args": {...}} - tool_call_finished: data={"name": "...", "correlation_id": "...", "ok": bool, "result_summary": "..."} - approval_required: data={"action_name": "...", "args": {...}, "allowed_decisions": [...], "interrupt_id": "..."} - user_input_requested: data={"prompt": "...", "interrupt_id": "..."} - native_notice: data={"text": "...", "metadata": {...}} - final: data={"text": "...", "new_messages": [...], "metrics": {...}, ...metadata} - error: data=RuntimeErrorData.to_dict()

text property 

text: str

Text content for assistant_delta, status, and final events.

tool_name property 

tool_name: str

Tool name for tool_call_started/finished events.

is_final property 

is_final: bool

True if this is a final event.

is_error property 

is_error: bool

True if this is an error event.

is_text property 

is_text: bool

True if this is an assistant_delta (text streaming) event.

structured_output property 

structured_output: Any

Structured output from final event, or None.

assistant_delta staticmethod 

assistant_delta(text: str) -> RuntimeEvent

Streamed text fragment.

status staticmethod 

status(text: str) -> RuntimeEvent

Status message.

approval_required staticmethod 

approval_required(action_name: str, args: dict[str, Any] | None = None, allowed_decisions: list[str] | None = None, interrupt_id: str | None = None, description: str = '') -> RuntimeEvent

Request human approval / tool review.

user_input_requested staticmethod 

user_input_requested(prompt: str, interrupt_id: str | None = None) -> RuntimeEvent

The runtime expects user/human input.

native_notice staticmethod 

native_notice(text: str, metadata: dict[str, Any] | None = None) -> RuntimeEvent

Explicit notice about native-specific semantics.

tool_call_started staticmethod 

tool_call_started(name: str, args: dict[str, Any] | None = None, correlation_id: str | None = None) -> RuntimeEvent

Start of a tool call.

tool_call_finished staticmethod 

tool_call_finished(name: str, correlation_id: str, ok: bool = True, result_summary: str = '') -> RuntimeEvent

End of a tool call.

final staticmethod 

final(text: str, new_messages: list[Message] | None = None, metrics: TurnMetrics | None = None, session_id: str | None = None, total_cost_usd: float | None = None, usage: dict[str, Any] | None = None, structured_output: Any = None, native_metadata: dict[str, Any] | None = None) -> RuntimeEvent

Final response.

error staticmethod 

error(error: RuntimeErrorData) -> RuntimeEvent

Runtime error.

to_dict 

to_dict() -> dict[str, Any]

Serialize to dict.

RuntimeConfig

RuntimeConfig dataclass

Configuration for runtime selection and parameters.

Priority: runtime_override > runtime_name > env COGNITIA_RUNTIME > default.

is_native_mode property 

is_native_mode: bool

True if the native upstream path is used.

AgentRuntime

AgentRuntime

Bases: Protocol

Agent Runtime protocol — canonical interface for all runtimes.

ISP: 4 methods (run, cleanup, cancel, context manager).

run 

run(*, messages: list[Message], system_prompt: str, active_tools: list[ToolSpec], config: RuntimeConfig | None = None, mode_hint: str | None = None) -> AsyncIterator[RuntimeEvent]

Run the agent loop, yielding streaming events.

cleanup async 

cleanup() -> None

Release resources held by the runtime.

cancel 

cancel() -> None

Request cooperative cancellation of the current operation.

__aenter__ async 

__aenter__() -> AgentRuntime

Async context manager entry.

__aexit__ async 

__aexit__(*args: Any) -> None

Async context manager exit.

RuntimeRegistry

RuntimeRegistry

Thread-safe extensible registry for runtime factories.

Each entry maps a runtime name to: - factory_fn: Callable[[RuntimeConfig, ...], AgentRuntime] - capabilities: RuntimeCapabilities | None

register 

register(name: str, factory_fn: Callable[..., Any], capabilities: RuntimeCapabilities | None = None) -> None

Register a runtime factory by name. Overwrites if exists.

unregister 

unregister(name: str) -> bool

Remove a runtime. Returns True if existed.

get 

get(name: str) -> Callable[..., Any] | None

Get factory function by name.

get_capabilities 

get_capabilities(name: str) -> RuntimeCapabilities | None

Get capabilities for registered runtime.

list_available 

list_available() -> list[str]

List all registered runtime names.

is_registered 

is_registered(name: str) -> bool

Check if a runtime is registered.

CostBudget

CostBudget dataclass

Budget limits for cost tracking.

Attributes: max_cost_usd: Maximum total cost in USD. None = no limit. max_total_tokens: Maximum total tokens (input + output). None = no limit. action_on_exceed: What to do when budget is exceeded. "error" = emit budget_exceeded error event. "warn" = report warning status but continue.

CostTracker

CostTracker

Accumulates token usage and checks budget limits.

Thread-safe for single-threaded async usage (no locks needed).

total_cost_usd property 

total_cost_usd: float

Total accumulated cost in USD.

total_tokens property 

total_tokens: int

Total accumulated tokens (input + output).

record 

record(model: str, input_tokens: int, output_tokens: int) -> None

Record token usage for a single LLM call.

check_budget 

check_budget() -> BudgetStatus

Check whether budget limits have been exceeded.

Returns: "ok" - wiThin limits (or no limits set). "exceeded" - over limit with action_on_exceed="error". "warning" - over limit with action_on_exceed="warn".

reset 

reset() -> None

Reset all accumulated usage to zero.