# `ClaudeWrapper.Stream`
[🔗](https://github.com/genagent/claude_wrapper_ex/blob/main/lib/claude_wrapper/stream.ex#L1)

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 `t: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
`t: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.

# `event`

```elixir
@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`.

# `option`

```elixir
@type option() :: {:timeout, timeout()}
```

Options for `stream/3`.

# `collect`

```elixir
@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).

# `filter`

```elixir
@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.

# `final_result`

```elixir
@spec final_result(Enumerable.t()) :: ClaudeWrapper.Result.t() | nil
```

Run the turn to completion and return its `t: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.

# `final_text`

```elixir
@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.

# `stream`

```elixir
@spec stream(GenServer.server(), String.t(), [option()]) :: Enumerable.t()
```

Wrap one `DuplexSession` turn as a lazy stream of `t: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 to
    `ClaudeWrapper.DuplexSession.send/3` (default `120_000`).

# `tap`

```elixir
@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`.

# `text_deltas`

```elixir
@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()

# `thinking_deltas`

```elixir
@spec thinking_deltas(Enumerable.t()) :: Enumerable.t()
```

Project a turn stream to its extended-thinking deltas.

# `tool_uses`

```elixir
@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.

---

*Consult [api-reference.md](api-reference.md) for complete listing*