Skip to main content

Module streaming_sink

oxios_kernel

Module streaming_sink 

Source
Expand description

Streaming sink registry — per-session lookup table for the WebSocket streaming path.

§Why this exists

The agent runtime’s AgentEvent::TextChunk callback fires per text delta during run_streaming. The Web UI needs each delta as a live token chunk. Without plumbing streaming_sink: Option<...> through every layer between the gateway and the runtime callback (Supervisor trait, AgentApi, AgentTool, AgentLifecycleManager, Orchestrator, AgentRuntime), we use a session-keyed registry:

  1. The gateway creates mpsc::unbounded() + a collector task right before invoking the orchestrator. The collector holds the strong Arc<UnboundedSender<StreamDelta>> for the duration of the turn (with target_conn_id = Some(conn_id) on every partial OutgoingMessage, mirroring gateway.rs:491). The gateway also stores a Weak under session_id in this registry.
  2. The runtime callback looks up the registry by session_id (which it already has via transparency_session) and sends StreamDelta::Text directly into the channel — no plumbing through intermediate layers.
  3. When the collector task completes, the strong Arc drops. Weak::upgrade on the next runtime lookup returns None and the registry lookup cleanly misses — no explicit unregister required, no stale entries.

§Design rationale (vs threaded Option<StreamingSinkTx> params)

run_with_directive is a Supervisor trait method with concrete impls on BasicSupervisor and NoOpSupervisor, and dyn Supervisor is held by AgentLifecycleManager, AgentApi, and AgentTool. Threading the sink through there would touch every impl + every holder. The registry avoids all of that because the runtime callback already has the session_id it needs to look up the sink — no new arguments required.

§Concurrency

The registry is Send + Sync (it owns a Mutex<HashMap<...>> of Weaks, neither contains interior mutability beyond the mutex itself). Lookup is O(n) where n = active turns; in practice n ≤ a few during normal use, so a HashMap is fine. A DashMap would be a one-line swap if load grows.

Structs§

StreamingSinkRegistry
Per-session lookup table. Shared via Arc between the kernel handle, the agent runtime, and the gateway dispatch layer.

Type Aliases§

StreamingSinkSender
Strong sender side, wrapped in Arc so the runtime callback can clone cheaply on every lookup. Re-exported here so callers don’t have to know the runtime’s internal type.