-module(claude@streaming). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]). -define(FILEPATH, "src/claude/streaming.gleam"). -export([parse_event/2, parse_sse/1]). -export_type([delta/0, delta_usage/0, stream_event/0]). -if(?OTP_RELEASE >= 27). -define(MODULEDOC(Str), -moduledoc(Str)). -define(DOC(Str), -doc(Str)). -else. -define(MODULEDOC(Str), -compile([])). -define(DOC(Str), -compile([])). -endif. -type delta() :: {text_delta, binary()} | {input_json_delta, binary()} | {thinking_delta, binary()}. -type delta_usage() :: {delta_usage, integer()}. -type stream_event() :: {message_start, claude@types@message:message()} | {content_block_start, integer(), claude@types@content:content_block()} | {content_block_delta, integer(), delta()} | {content_block_stop, integer()} | {message_delta, gleam@option:option(claude@types@message:stop_reason()), delta_usage()} | message_stop | ping | {unknown_event, binary(), binary()}. -file("src/claude/streaming.gleam", 65). ?DOC(" Extract the value after a field prefix (e.g. \"event:\" or \"data:\") from SSE lines.\n"). -spec find_field(list(binary()), binary()) -> {ok, binary()} | {error, nil}. find_field(Lines, Prefix) -> gleam@list:find_map( Lines, fun(Line) -> Trimmed = gleam@string:trim(Line), case gleam_stdlib:string_starts_with(Trimmed, Prefix) of true -> Value = gleam@string:drop_start( Trimmed, string:length(Prefix) ), {ok, gleam@string:trim(Value)}; false -> {error, nil} end end ). -file("src/claude/streaming.gleam", 94). -spec parse_message_start(binary()) -> stream_event(). parse_message_start(Data) -> Decoder = begin gleam@dynamic@decode:field( <<"message"/utf8>>, claude@json@decode:message_decoder(), fun(Message) -> gleam@dynamic@decode:success({message_start, Message}) end ) end, case gleam@json:parse(Data, Decoder) of {ok, Event} -> Event; {error, _} -> {unknown_event, <<"message_start"/utf8>>, Data} end. -file("src/claude/streaming.gleam", 129). -spec parse_content_block_stop(binary()) -> stream_event(). parse_content_block_stop(Data) -> Decoder = begin gleam@dynamic@decode:field( <<"index"/utf8>>, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun(Index) -> gleam@dynamic@decode:success({content_block_stop, Index}) end ) end, case gleam@json:parse(Data, Decoder) of {ok, Event} -> Event; {error, _} -> {unknown_event, <<"content_block_stop"/utf8>>, Data} end. -file("src/claude/streaming.gleam", 140). -spec parse_message_delta(binary()) -> stream_event(). parse_message_delta(Data) -> Decoder = begin gleam@dynamic@decode:subfield( [<<"delta"/utf8>>, <<"stop_reason"/utf8>>], gleam@dynamic@decode:optional( claude@json@decode:stop_reason_decoder() ), fun(Stop_reason) -> gleam@dynamic@decode:subfield( [<<"usage"/utf8>>, <<"output_tokens"/utf8>>], {decoder, fun gleam@dynamic@decode:decode_int/1}, fun(Output_tokens) -> gleam@dynamic@decode:success( {message_delta, Stop_reason, {delta_usage, Output_tokens}} ) end ) end ) end, case gleam@json:parse(Data, Decoder) of {ok, Event} -> Event; {error, _} -> {unknown_event, <<"message_delta"/utf8>>, Data} end. -file("src/claude/streaming.gleam", 165). ?DOC( " Decoder for a content block in content_block_start events.\n" " Handles \"text\" and \"tool_use\" types.\n" ). -spec content_block_start_decoder() -> gleam@dynamic@decode:decoder(claude@types@content:content_block()). content_block_start_decoder() -> gleam@dynamic@decode:field( <<"type"/utf8>>, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Block_type) -> case Block_type of <<"text"/utf8>> -> gleam@dynamic@decode:field( <<"text"/utf8>>, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Text) -> gleam@dynamic@decode:success({text, Text, []}) end ); <<"tool_use"/utf8>> -> gleam@dynamic@decode:field( <<"id"/utf8>>, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Id) -> gleam@dynamic@decode:field( <<"name"/utf8>>, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Name) -> gleam@dynamic@decode:success( {tool_use, Id, Name, <<""/utf8>>} ) end ) end ); _ -> gleam@dynamic@decode:success({text, <<""/utf8>>, []}) end end ). -file("src/claude/streaming.gleam", 105). -spec parse_content_block_start(binary()) -> stream_event(). parse_content_block_start(Data) -> Decoder = begin gleam@dynamic@decode:field( <<"index"/utf8>>, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun(Index) -> gleam@dynamic@decode:field( <<"content_block"/utf8>>, content_block_start_decoder(), fun(Block) -> gleam@dynamic@decode:success( {content_block_start, Index, Block} ) end ) end ) end, case gleam@json:parse(Data, Decoder) of {ok, Event} -> Event; {error, _} -> {unknown_event, <<"content_block_start"/utf8>>, Data} end. -file("src/claude/streaming.gleam", 183). ?DOC(" Decoder for the delta object in content_block_delta events.\n"). -spec delta_decoder() -> gleam@dynamic@decode:decoder(delta()). delta_decoder() -> gleam@dynamic@decode:field( <<"type"/utf8>>, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Delta_type) -> case Delta_type of <<"text_delta"/utf8>> -> gleam@dynamic@decode:field( <<"text"/utf8>>, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Text) -> gleam@dynamic@decode:success({text_delta, Text}) end ); <<"input_json_delta"/utf8>> -> gleam@dynamic@decode:field( <<"partial_json"/utf8>>, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Partial_json) -> gleam@dynamic@decode:success( {input_json_delta, Partial_json} ) end ); <<"thinking_delta"/utf8>> -> gleam@dynamic@decode:field( <<"thinking"/utf8>>, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Thinking) -> gleam@dynamic@decode:success( {thinking_delta, Thinking} ) end ); _ -> gleam@dynamic@decode:success({text_delta, <<""/utf8>>}) end end ). -file("src/claude/streaming.gleam", 117). -spec parse_content_block_delta(binary()) -> stream_event(). parse_content_block_delta(Data) -> Decoder = begin gleam@dynamic@decode:field( <<"index"/utf8>>, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun(Index) -> gleam@dynamic@decode:field( <<"delta"/utf8>>, delta_decoder(), fun(Delta) -> gleam@dynamic@decode:success( {content_block_delta, Index, Delta} ) end ) end ) end, case gleam@json:parse(Data, Decoder) of {ok, Event} -> Event; {error, _} -> {unknown_event, <<"content_block_delta"/utf8>>, Data} end. -file("src/claude/streaming.gleam", 79). ?DOC(" Parse a single event given its type and JSON data string.\n"). -spec parse_event(binary(), binary()) -> stream_event(). parse_event(Event_type, Data) -> case Event_type of <<"message_start"/utf8>> -> parse_message_start(Data); <<"content_block_start"/utf8>> -> parse_content_block_start(Data); <<"content_block_delta"/utf8>> -> parse_content_block_delta(Data); <<"content_block_stop"/utf8>> -> parse_content_block_stop(Data); <<"message_delta"/utf8>> -> parse_message_delta(Data); <<"message_stop"/utf8>> -> message_stop; <<"ping"/utf8>> -> ping; _ -> {unknown_event, Event_type, Data} end. -file("src/claude/streaming.gleam", 48). ?DOC( " Parse a single SSE block into a StreamEvent.\n" " Returns Error(Nil) if the block cannot be parsed.\n" ). -spec parse_block(binary()) -> {ok, stream_event()} | {error, nil}. parse_block(Block) -> Lines = gleam@string:split(Block, <<"\n"/utf8>>), Event_type = find_field(Lines, <<"event:"/utf8>>), Data = find_field(Lines, <<"data:"/utf8>>), case {Event_type, Data} of {{ok, Evt}, {ok, D}} -> {ok, parse_event(Evt, D)}; {{ok, Evt@1}, {error, _}} -> case Evt@1 of <<"ping"/utf8>> -> {ok, ping}; _ -> {ok, {unknown_event, Evt@1, <<""/utf8>>}} end; {_, _} -> {error, nil} end. -file("src/claude/streaming.gleam", 39). ?DOC( " Parse a full SSE text payload into a list of StreamEvents.\n" "\n" " SSE events are separated by blank lines (\"\\n\\n\"). Each event block\n" " contains an `event:` line and a `data:` line.\n" ). -spec parse_sse(binary()) -> list(stream_event()). parse_sse(Text) -> _pipe = Text, _pipe@1 = gleam@string:split(_pipe, <<"\n\n"/utf8>>), _pipe@2 = gleam@list:filter( _pipe@1, fun(Block) -> not gleam@string:is_empty(gleam@string:trim(Block)) end ), gleam@list:filter_map(_pipe@2, fun parse_block/1).