SuperPi.AI.StreamNormalizer (super_pi v0.1.0)

Unified multi-provider LLM SSE stream normalizer and cross-provider message sanitizer (REQ-PBT-04).

Normalizes raw Server-Sent Event (SSE) byte streams from:

:anthropic (anthropic-messages)
:openai (openai-responses / openai-completions)
:google (google-generative-ai)

into a canonical sequence of events:

{:start, %{model: String.t(), provider: atom()}}
{:text_delta, %{index: non_neg_integer(), delta: String.t()}}
{:thinking_delta, %{index: non_neg_integer(), delta: String.t(), signature: String.t() | nil}}
{:toolcall_delta, %{index: non_neg_integer(), id: String.t(), name: String.t(), raw_json_delta: String.t(), partial_args: map()}}
{:done, %{stop_reason: String.t(), content: list(map()), usage: map()}}
{:error, %{reason: term()}}

Guarantees chunk-boundary invariance (REQ-PBT-04): arbitrary byte-level chunk splitting of an SSE stream produces the exact same final :done message and accumulated tool-call arguments.

Summary

Functions

feed_chunk(state, chunk)

Feeds an arbitrary binary chunk into the SSE normalizer and returns {normalized_events, next_state}. Handles arbitrary mid-line and mid-JSON chunk boundaries without losing state.

flush(state)

Flushes any remaining complete SSE data in the buffer and returns {events, final_state}.

new(provider \\ :anthropic, model \\ "claude-sonnet-4-5")

Initializes a new stream normalizer state for provider and model.

parse_partial_json(raw)

Best-effort incremental JSON object parser for streaming toolcall_delta arguments. Automatically closes unclosed strings, arrays, and objects so callers can inspect partial_args before the tool call block finishes streaming.

sanitize_messages_for_target(messages, target_provider, target_model)

Sanitizes conversation history messages when switching across LLM providers or models (spec.md §5)

Types

t()

@type t() :: %SuperPi.AI.StreamNormalizer{
  active_block_idx: non_neg_integer() | nil,
  block_counter: non_neg_integer(),
  blocks: map(),
  buffer: String.t(),
  data_lines: [String.t()],
  done?: boolean(),
  last_was_cr?: boolean(),
  model: String.t(),
  provider: :anthropic | :openai | :google,
  started?: boolean(),
  stop_reason: String.t(),
  usage: map()
}