Lazy Stream combinators over a ClaudeWrapper.DuplexSession turn.
DuplexSession broadcasts a turn's progress to subscribers as
{:claude, event} messages (see its module doc for the vocabulary).
This module wraps one turn as an Elixir Enumerable.t/0 so the
events compose with the standard Stream/Enum toolkit, and adds a
few decode-and-project combinators on top. It is a pure ergonomic
layer: no new runtime dependencies, no new transport.
One turn per enumeration
stream/3 returns a lazy stream. Like any Stream, it is a recipe:
each time it is enumerated it triggers a fresh turn. Pick one
transform or terminal per turn. If you need both the text and the
ClaudeWrapper.Result.t/0, use collect/1 (it returns both) rather
than enumerating twice.
session
|> ClaudeWrapper.Stream.stream("Write a haiku about Elixir.")
|> ClaudeWrapper.Stream.text_deltas()
|> Enum.each(&IO.write/1)
%{text: text, result: %ClaudeWrapper.Result{}} =
session
|> ClaudeWrapper.Stream.stream("Summarize this file.")
|> ClaudeWrapper.Stream.collect()Elements
The foundational stream/3 yields the same event payloads a
subscriber receives (the second element of {:claude, event}):
{:system_init, session_id}, {:assistant, msg}, {:stream_event, raw}, {:user, msg}, and the terminal {:result, %Result{}}. A
failed turn (including :turn_in_flight) yields a terminal
{:error, %ClaudeWrapper.Error{}} instead of {:result, _}.
Early halt
Halting early (e.g. Stream.take/2) stops consumption and unsubscribes,
but does not cancel the turn server-side -- the claude subprocess
runs the turn to completion. Use ClaudeWrapper.DuplexSession.interrupt/1
for a real mid-turn cancel.
Summary
Functions
Run the turn to completion, returning both the text and the result.
Keep only the events for which fun returns truthy.
Run the turn to completion and return its ClaudeWrapper.Result.t/0.
Run the turn to completion and return the accumulated answer text.
Wrap one DuplexSession turn as a lazy stream of event/0 values.
Run fun on each event for its side effect, passing the event through.
Project a turn stream to its text deltas (the streamed answer tokens).
Project a turn stream to its extended-thinking deltas.
Project a turn stream to its tool-use starts, as %{id, name} maps.
Types
@type event() :: {:system_init, String.t()} | {:assistant, map()} | {:stream_event, map()} | {:user, map()} | {:result, ClaudeWrapper.Result.t()} | {:error, ClaudeWrapper.Error.t()}
A decoded turn event, as yielded by stream/3.
@type option() :: {:timeout, timeout()}
Options for stream/3.
Functions
@spec collect(Enumerable.t()) :: %{ text: String.t(), result: ClaudeWrapper.Result.t() | nil }
Run the turn to completion, returning both the text and the result.
%{text: text, result: %ClaudeWrapper.Result{}} =
ClaudeWrapper.Stream.collect(stream)result is nil if the turn ended without a result event. Terminal:
enumerates the whole turn once (the right call when you need both the
text and the result from a single turn).
@spec filter(Enumerable.t(), (event() -> as_boolean(term()))) :: Enumerable.t()
Keep only the events for which fun returns truthy.
A thin pass-through to Stream.filter/2, here for discoverability
alongside the projections.
@spec final_result(Enumerable.t()) :: ClaudeWrapper.Result.t() | nil
Run the turn to completion and return its ClaudeWrapper.Result.t/0.
Returns nil if the turn yielded no result event (e.g. it ended in an
{:error, _}). Terminal: enumerates the whole turn.
@spec final_text(Enumerable.t()) :: String.t()
Run the turn to completion and return the accumulated answer text.
Concatenates every text delta. Terminal: enumerates the whole turn.
@spec stream(GenServer.server(), String.t(), [option()]) :: Enumerable.t()
Wrap one DuplexSession turn as a lazy stream of event/0 values.
Enumerating subscribes the calling process, sends prompt, and yields
each broadcast event in arrival order, ending with the terminal
{:result, %Result{}} (or {:error, %Error{}}).
Options:
:timeout-- per-turn timeout forwarded toClaudeWrapper.DuplexSession.send/3(default120_000).
@spec tap(Enumerable.t(), (event() -> term())) :: Enumerable.t()
Run fun on each event for its side effect, passing the event through.
A thin pass-through to Stream.each/2.
@spec text_deltas(Enumerable.t()) :: Enumerable.t()
Project a turn stream to its text deltas (the streamed answer tokens).
session
|> ClaudeWrapper.Stream.stream("Say hi.")
|> ClaudeWrapper.Stream.text_deltas()
|> Enum.join()@spec thinking_deltas(Enumerable.t()) :: Enumerable.t()
Project a turn stream to its extended-thinking deltas.
@spec tool_uses(Enumerable.t()) :: Enumerable.t()
Project a turn stream to its tool-use starts, as %{id, name} maps.
Emits one element per tool-use content block the assistant opens. The
tool input arrives as later input_json deltas; this surfaces the
identity of each tool call.