ALLM.StreamCollector (allm v0.6.1)

Reduce a stream of ALLM.Event values into a collected %ALLM.Response{}, %ALLM.StepResult{}, or %ALLM.ChatResult{}.

StreamCollector is the shared reducer that every non-streaming wrapper (ALLM.generate/3, ALLM.step/3, ALLM.chat/3) builds on. Callers fold a stream with the stdlib idiom:

Enum.reduce(stream, ALLM.StreamCollector.new, fn e, s ->
  ALLM.StreamCollector.apply_event(s, e)
end)

then extract the result with to_response/1 (thread-less) or to_step_result/1 / to_chat_result/1 (thread-backed).

Constructors

new/0 builds a thread-less collector used by ALLM.stream_generate/3.
new/1 additionally accepts nil (thread-less) alongside a %ALLM.Thread{}.
to_response/1 returns the accumulated %Response{} and is the canonical output for thread-less collection.

Per-step + per-chat reset

The :step_completed fold appends a %StepResult{} (built from the PRE-RESET collector state) to :steps and resets the per-step sub-state (:current_text, :current_tool_calls, :tool_call_order, :tool_results, :halt, :finish_reason, :raw_finish_reason, :last_message). :error and :metadata are NOT reset — adapter mid-stream errors and accumulated metadata persist across step boundaries so that to_chat_result/1's fallback path can preserve them.

The :chat_completed fold stores the event's :result verbatim in :chat_result and sets :done? = true. to_chat_result/1 short- circuits to the stored result when present; otherwise it falls back to a computed %ChatResult{} whose halted_reason is :cancelled (consumer halted early) or :error (mid-stream adapter error).

Fold semantics

apply_event/2 ships explicit clauses for the nine adapter-emitted tags (:message_started, :text_delta, :text_completed, :tool_call_started, :tool_call_delta, :tool_call_completed, :message_completed, :raw_chunk, :error), three orchestration tags (:tool_result_encoded, :tool_halt, :ask_user_requested), and two chat-loop tags (:step_completed, :chat_completed). Every other tag — :tool_execution_started, :tool_execution_completed, plus any malformed event — falls through a single catch-all apply_event(state, _), do: state.

Tag	Fold	Rationale
`:tool_result_encoded`	append `%Message{role: :tool, tool_call_id: id, content: content}` to `state.tool_results`.	`to_step_result/1` reads `:tool_results` from the struct.
`:tool_halt`	when `state.halt == nil`, set `state.halt = {:halt, reason, id, result}` AND append the encoded sentinel `%Message{role: :tool, ...}` (from payload `:content`) to `state.tool_results`. Subsequent halts are no-ops (first-halt-wins).	Halt metadata lives separately from `:finish_reason` so `done?/1` can combine both signals.
`:ask_user_requested`	when `state.halt == nil`, set `state.halt = {:ask_user, :ask_user, id, q, o}` AND append a `<awaiting user response>` sentinel message to `state.tool_results`. First-halt-wins.	Same channel as `:tool_halt`.
`:step_completed`	append `%StepResult{}` (PRE-RESET state) to `state.steps`; set `state.thread = thread`; reset per-step sub-state.	Multi-step `ALLM.stream/3` reductions need a clean per-step boundary. `:error` / `:metadata` deliberately persist.
`:chat_completed`	set `state.chat_result = result`; set `state.done? = true`.	The chat-layer terminal event; `to_chat_result/1` short-circuits to the stored value when present.

Totality guarantee

apply_event/2 never raises on a well-shaped event in the 16-tag closed union. A :raw_chunk with payload {:usage, map} allows a KeyError from struct!(ALLM.Usage, map) to propagate — that's an adapter bug (emitting keys not in %Usage{}), not a collector bug.

Adapter-side contract for `:raw_chunk {:usage, _}` (Usage fold)

The {:raw_chunk, {:usage, map}} fold applies struct!(ALLM.Usage, map), which raises KeyError on any map key outside %ALLM.Usage{}'s field set. Adapters emitting {:raw_chunk, {:usage, _}} events MUST pre-map their provider's wire-level usage fields (e.g., OpenAI's prompt_tokens / completion_tokens; Anthropic's input_tokens / output_tokens / cache_read_input_tokens) to %ALLM.Usage{} field names before emitting. An adapter that forwards the raw provider payload will KeyError the reducer mid-stream, destroying the accumulated collector state. See lib/allm/usage.ex for the canonical %Usage{} field set. This is a load-bearing invariant for any real provider adapter implementing ALLM.StreamAdapter.

The recommended payload is Map.from_struct/1 of the adapter's non-streaming usage decoder: it carries exactly the %Usage{} key set (cache counters and extra included), and it makes a streamed response's usage equal the non-streamed one by construction. Every bundled chat adapter builds its payload this way. The fold replaces state.usage wholesale, so an adapter that sees usage in several wire events merges them itself and emits the result once, or emits cumulative values where the last one wins.

Mid-stream errors

A terminal {:error, struct} event folds into the collector's :error field and sets :finish_reason to :error. to_response/1 still returns %Response{} — a mid-stream error is a success at the fold layer: the caller reads response.finish_reason == :error and response.metadata.error to detect it. to_response/1 never returns {:error, _}.

Summary

Types

halt_state()

Terminal halt signal derived from :tool_halt / :ask_user_requested events. nil until the first halt event is observed; set once and never updated (first-halt-wins — see Phase 6 design Invariant 7).

state()

Functions

apply_event(state, arg2)

Fold a single ALLM.Event into the collector state. Total over the 16-tag closed union; unknown tags or malformed payloads are no-ops (state unchanged).

new()

Build a thread-less collector. Equivalent to new(nil).

new(thread)

Build a collector either thread-less (nil) or seeded with a %ALLM.Thread{}. The thread is required by to_step_result/1 and to_chat_result/1; use new/0 or new(nil) when only to_response/1 will be consumed.

to_chat_result(state)

Build a %ALLM.ChatResult{} from the collector state.

to_response(state)

Build a %ALLM.Response{} from the collector state.

to_step_result(state)

Build a %ALLM.StepResult{} from the collector state. Requires a non-nil thread (from new/1 with a %Thread{}); raises ArgumentError otherwise.

Types

halt_state()

@type halt_state() ::
  nil
  | {:halt, reason :: atom(), tool_call_id :: String.t(), result :: term()}
  | {:ask_user, :ask_user, tool_call_id :: String.t(), question :: String.t(),
     opts :: keyword()}

The :halt shape carries the handler's raw result term as the fourth element so merge_halt_metadata/2 can project it onto :halt_result, matching the non-streaming ToolRunner.run_tool_calls/3 halt_metadata shape (Phase 7.6 cleanup — chat-equivalence blocker B2).

state()

@type state() :: %ALLM.StreamCollector{
  chat_result: ALLM.ChatResult.t() | nil,
  current_text: String.t(),
  current_tool_calls: %{required(String.t()) => ALLM.ToolCall.t()},
  done?: boolean(),
  error:
    ALLM.Error.AdapterError.t() | ALLM.Error.StreamError.t() | term() | nil,
  finish_reason: ALLM.Response.finish_reason() | nil,
  halt: halt_state(),
  last_message: ALLM.Message.t() | nil,
  last_response: ALLM.Response.t() | nil,
  metadata: map(),
  raw_finish_reason: String.t() | nil,
  steps: [ALLM.StepResult.t()],
  thread: ALLM.Thread.t() | nil,
  tool_call_order: [String.t()],
  tool_results: [ALLM.Message.t()],
  usage: ALLM.Usage.t()
}

Functions

apply_event(state, arg2)

@spec apply_event(state(), ALLM.Event.t()) :: state()

Fold a single ALLM.Event into the collector state. Total over the 16-tag closed union; unknown tags or malformed payloads are no-ops (state unchanged).

See the module doc's "Fold semantics" section for the per-tag state transitions.

Examples

iex> s = ALLM.StreamCollector.new()
iex> s = ALLM.StreamCollector.apply_event(s, {:text_delta, %{id: nil, delta: "hel"}})
iex> s = ALLM.StreamCollector.apply_event(s, {:text_delta, %{id: nil, delta: "lo"}})
iex> s.current_text
"hello"

new()

@spec new() :: state()

Build a thread-less collector. Equivalent to new(nil).

Use this when reducing ALLM.stream_generate/3's stream into a %Response{} via to_response/1. Calling to_step_result/1 or to_chat_result/1 on a thread-less collector raises ArgumentError.

Examples

iex> s = ALLM.StreamCollector.new()
iex> s.thread
nil
iex> s.current_text
""

new(thread)

@spec new(ALLM.Thread.t() | nil) :: state()

Examples

iex> thread = ALLM.Thread.new()
iex> s = ALLM.StreamCollector.new(thread)
iex> s.thread
%ALLM.Thread{messages: [], metadata: %{}}

iex> ALLM.StreamCollector.new(nil).thread
nil

to_chat_result(state)

@spec to_chat_result(state()) :: ALLM.ChatResult.t()

Build a %ALLM.ChatResult{} from the collector state.

Two branches:

Stored. When state.chat_result is set (from a :chat_completed fold), return it verbatim — even when state.thread is nil. The stored result is authoritative; the orchestrator already constructed the canonical ChatResult.
Fallback (computed). When state.chat_result is nil and state.thread is set, build a %ChatResult{} from collector state. :halted_reason is :error when state.error != nil (mid-stream adapter error); otherwise ALWAYS :cancelled (the consumer halted the stream early; non-empty state.steps does NOT promote to :completed). :final_response is the last step's response when state.steps != [], or to_response(state) when no steps were observed.

Raises ArgumentError when both state.chat_result and state.thread are nil.

to_response(state)

@spec to_response(state()) :: ALLM.Response.t()

Build a %ALLM.Response{} from the collector state.

Works on any collector (thread-less or not). A mid-stream error surfaces as finish_reason: :error with the error struct under metadata.error; this function never returns {:error, _}.

Examples

iex> s = ALLM.StreamCollector.new()
iex> s = ALLM.StreamCollector.apply_event(s, {:text_delta, %{id: nil, delta: "hi"}})
iex> s = ALLM.StreamCollector.apply_event(s, {:message_completed, %{message: %ALLM.Message{role: :assistant, content: "hi"}, finish_reason: :stop}})
iex> resp = ALLM.StreamCollector.to_response(s)
iex> {resp.output_text, resp.finish_reason}
{"hi", :stop}

to_step_result(state)

@spec to_step_result(state()) :: ALLM.StepResult.t()

Build a %ALLM.StepResult{} from the collector state. Requires a non-nil thread (from new/1 with a %Thread{}); raises ArgumentError otherwise.

:done? is true when state.halt != nil (a halt event was folded in) or when finish_reason in [:stop, :length, :content_filter, :error]; false otherwise (for :tool_calls or nil with no halt).

:tool_results is populated from the :tool_result_encoded fold clause AND from the :tool_halt / :ask_user_requested sentinel appends.

:metadata merges halt metadata when state.halt != nil:

{:halt, reason, id, result} → %{halted_reason: reason, halt_tool_call_id: id, halt_result: result}.
{:ask_user, :ask_user, id, q, o} → %{halted_reason: :ask_user, pending_tool_call_id: id, pending_question: q, ask_user_opts: o}.

ALLM.StreamCollector (allm v0.6.1)

Constructors

Per-step + per-chat reset

Fold semantics

Totality guarantee

Adapter-side contract for :raw_chunk {:usage, _} (Usage fold)

Mid-stream errors

Summary

Types

Functions

Types

halt_state()

state()

Functions

apply_event(state, arg2)

Examples

new()

Examples

new(thread)

Examples

to_chat_result(state)

to_response(state)

Examples

to_step_result(state)

Adapter-side contract for `:raw_chunk {:usage, _}` (Usage fold)