# `SuperPi.AI.StreamNormalizer`
[🔗](https://github.com/oleksandr/super-pi-beam/blob/v0.1.0/lib/super_pi/ai/stream_normalizer.ex#L1)

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.

# `t`

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

# `feed_chunk`

```elixir
@spec feed_chunk(t(), binary()) :: {[tuple()], t()}
```

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`

```elixir
@spec flush(t()) :: {[tuple()], t()}
```

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

# `new`

Initializes a new stream normalizer state for `provider` and `model`.

# `parse_partial_json`

```elixir
@spec parse_partial_json(String.t()) :: map()
```

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`

```elixir
@spec sanitize_messages_for_target([map()], atom(), String.t()) :: [map()]
```

Sanitizes conversation history messages when switching across LLM providers or models (`spec.md` §5):
- Strips provider-proprietary `thinkingSignature` / `redacted_thinking` fields or downgrades
  unrecognized thinking blocks to plain text / omitted metadata when moving between providers
  (e.g. `:anthropic` -> `:openai` or `:google`).
- Preserves `thinkingSignature` intact when the target `(provider, model)` matches the block's origin.

---

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