Stream Events¶

Kedi exposes completed semantic agent messages as an adapter-neutral event stream. This lets a terminal, editor, web UI, or orchestrator show useful progress while the authoritative adapter call continues normally.

The stream is deliberately not a token stream. Provider deltas, mutable snapshots, and hidden reasoning stay private. Kedi emits a message only after the adapter has identified a complete assistant message and classified it as commentary or final output.

Event Lifecycle¶

Every observed run emits one start event and one terminal event:

started
commentary  Inspecting the package manifest.
commentary  The dependency graph is consistent.
final       Validation passed with no blocking findings.
completed

The terminal state is exactly one of completed, failed, or cancelled. Failed and cancelled runs do not publish unfinished private text as a final message. The adapter's return value or exception remains authoritative; events are an observation side channel.

Commentary¶

phase="commentary" is a complete progress message that is not the run's final answer. Kedi can identify commentary when:

• a provider or harness marks a completed message as commentary;
• a tool-call boundary seals completed assistant text produced before the tool runs;
• a later completed assistant message makes an earlier message intermediate.

Kedi does not expose chain-of-thought or infer commentary from partial token chunks. If a backend provides only a final response, the run may contain no commentary events.

Final¶

phase="final" is the completed natural-language response returned by a successful raw invoke() call. At most one final message is authoritative for one run. Structured produce() results remain typed data and are not mirrored as invented natural-language final messages.

Tool calls are not AgentMessageEvent values. Use Kedi's tool telemetry or the backend's native event surface when a UI also needs tool-call cards. A tool boundary only determines that preceding completed text is commentary.

Observe With a Callback¶

The public types and observer are exported from kedi:

import asyncio

from kedi import AgentMessageEvent, AgentRunStateEvent, observe_agent_events
from kedi.agent_adapter import PydanticAdapter


def render(event: AgentMessageEvent | AgentRunStateEvent) -> None:
    if isinstance(event, AgentMessageEvent):
        print(f"{event.phase}: {event.content}")
    else:
        print(f"run {event.run_id}: {event.state}")


adapter = PydanticAdapter("openai:gpt-4o-mini")

with observe_agent_events(render):
    result = asyncio.run(adapter.invoke(prompt="Inspect the parser."))

print("result:", result)

The callback runs on Kedi's dispatcher thread. Keep it short and thread-safe. For UI frameworks or asyncio applications, bridge through their event loop instead of mutating UI state directly from the callback.

Event observation does not modify prompts, schemas, tools, message history, usage accounting, return values, or exception behavior.

Consume From Async Code¶

AsyncAgentEventQueue is a thread-safe sink that forwards dispatcher events to an asyncio.Queue:

import asyncio

from kedi import AgentRunStateEvent, AsyncAgentEventQueue, observe_agent_events
from kedi.agent_adapter import PydanticAdapter


async def main() -> str:
    adapter = PydanticAdapter("openai:gpt-4o-mini")
    events = AsyncAgentEventQueue()
    root_run_id = None

    with observe_agent_events(events):
        task = asyncio.create_task(adapter.invoke(prompt="Inspect the parser."))

        while True:
            event = await events.queue.get()
            if (
                isinstance(event, AgentRunStateEvent)
                and event.state == "started"
                and event.parent_run_id is None
            ):
                root_run_id = event.run_id

            if (
                isinstance(event, AgentRunStateEvent)
                and event.run_id == root_run_id
                and event.state in {"completed", "failed", "cancelled"}
            ):
                break

        return await task


print(asyncio.run(main()))

Construct the queue inside the target event loop. Pass loop= explicitly only when construction and consumption occur in different setup code.

Event Fields¶

AgentMessageEvent carries:

AgentRunStateEvent carries run_id, parent_run_id, adapter, sequence, and state. Sequence numbers are scoped per run; use (run_id, sequence) when merging concurrently emitted parent and subagent events.

Subagents and Concurrent Runs¶

Subagent events use the coordinator-assigned child run_id and identify the calling run through parent_run_id. Do not assume events from concurrent runs are globally ordered. Maintain independent render state for each run_id, then use the parent relationship to nest child activity.

A root run has parent_run_id=None. A terminal event closes only its own run; a child completion must not be mistaken for root completion.

Delivery and Backpressure¶

Kedi dispatches events away from provider and SDK reader threads so a slow observer does not directly block model execution. Control and final events are kept separate from the bounded commentary queue. Under sustained commentary overflow, Kedi drops the oldest pending commentary and logs a warning rather than allowing unbounded memory growth.

Observer failures are isolated from the model call. Repeated callback failures detach that observer, and a callback that cannot drain during shutdown is detached after a bounded wait. Observation is therefore suitable for progress reporting, not durable audit storage. Persist authoritative traces through Telemetry.

Adapter Support¶

Semantic message boundaries are capability-driven:

LazyAdapter reports the resolved adapter's capability. Unsupported adapters set supports_stream_events=False; Kedi does not fabricate trustworthy message boundaries from raw token chunks.

See the generated Capability Matrix for the current built-in adapter metadata.