run_stream

Convert a v2 StreamPart to a ProtocolEvent.

Central event dispatcher for the streaming infrastructure.

Owns the main event log and routes events through a transformer pipeline. StreamChannels with a name discovered in transformer projections are auto-wired so that every push() also injects a ProtocolEvent into the main log. StreamChannels without a name are local-only.

Pass is_async=True when the mux will be consumed via async iteration (handler.astream()). All StreamChannel instances discovered during registration are automatically bound to the matching mode.

A protocol event emitted by the streaming infrastructure.

Wraps a raw stream part (values, messages, custom, etc.) in a uniform envelope with a monotonic sequence number assigned by the root StreamMux. Consumers that need a total order across root events should use seq, not params.timestamp (which is wall-clock and not monotonic).

Single-consumer drainable queue for streaming events, with optional protocol auto-forwarding.

When constructed with a name, the StreamMux auto-wires every push() to also inject a ProtocolEvent into the main event stream using the channel's name as the method. When constructed without a name, the channel is local-only — items are only visible to in-process consumers that iterate the channel directly.

Items are popped off the front as the consumer advances — there is no retention beyond what's currently queued. A channel accepts exactly one subscriber; a second __iter__ / __aiter__ call raises. Use tee(n) / atee(n) for fan-out.

Starts unbound — neither __iter__ nor __aiter__ is available until the StreamMux calls _bind(is_async). After binding, only the matching iteration protocol works; the other raises TypeError.

Pump wiring (set by the run stream, not by _bind): - _request_more: sync pump callable, returns True if a new event was produced. - _arequest_more: async pump coroutine factory, same contract.

Memory is bounded by caller pace: both sync and async use caller- driven pumps, so each cursor advance produces at most one event.

Lazy-subscribe: push appends to the local buffer only when a subscriber has registered. Auto-forward via _wire_fn always fires regardless of subscription state.

Lifecycle (close / fail) is managed by the mux — transformers don't need to close their channels manually.

Payload of a lifecycle event surfaced on the lifecycle channel.

Auto-forwarded as lifecycle protocol events (no custom: prefix because LifecycleTransformer is a native transformer) so remote SDK clients receive the same data in-process consumers see via run.lifecycle.

Sync run stream with caller-driven pumping.

The caller's iteration on any projection (values, messages, raw events, or output) drives the graph forward. No background thread is used — the caller's for loop is the pump.

Projections are single-consumer — iterating run.values twice raises. Use projection.tee(n) if you genuinely need fan-out.

All transformer projections live in extensions. Native transformer projections (those with _native = True) are also set as direct attributes on this instance (e.g. run.values, run.messages).

Sync handle for a discovered subgraph (extends GraphRunStream).

Async handle for a discovered subgraph (extends AsyncGraphRunStream).

Async run stream with caller-driven pumping.

Async iteration on any projection drives the graph forward — there is no background task. Concurrent consumers share a single-flight pump via an asyncio.Lock, so each awaiting cursor contributes one event per acquisition. Backpressure comes from the logs: when a subscribed log's buffer reaches maxlen, apush awaits the subscriber to drain, which holds back the pump and paces the graph.

Projections are single-consumer — a second aiter(run.values) raises. Use projection.tee(n) for fan-out.

Use as an async context manager to guarantee clean shutdown on early exit:

async with await handler.astream(input) as run:
    async for msg in run.messages:
        ...