A2aEngine.Codec.SSE (a2a_engine v0.1.0)

Copy Markdown View Source

Server-Sent Events codec with Last-Event-ID resumption.

Event id scheme: "<task_id>:<seq>". seq is a monotonic integer, so clients can use Last-Event-ID to request replay from a known point without ambiguity across tasks.

Wire format per the HTML Living Standard:

id: <task_id>:<seq>
event: <name>
data: <serialized JSON>
<blank line>

Multiple data: lines are supported (spec says consumer concatenates with \n). This encoder emits a single line and JSON-encodes the payload verbatim, which is the common case and unambiguous.

Comment frames (: keep-alive) are emitted via keep_alive/0 to prevent intermediary timeouts on long-idle streams.

Summary

Functions

decode_event(block)

Decode a full SSE event block (lines separated by \n, terminated by blank line). Returns the parsed {id, event, data} tuple or an error.

encode_event(task_id, event_name, seq, payload)

Encode one SSE event.

event_id(task_id, seq)

Build the canonical SSE event id for a (task_id, seq) pair.

keep_alive()

Emit a keep-alive comment frame to reset intermediary timeouts.

parse_event_id(id)

Parse an event id back into {task_id, seq}.

resume_from(id)

Interpret a Last-Event-ID header value for stream resumption.

Functions

decode_event(block)

@spec decode_event(String.t()) ::
  {:ok, %{id: String.t() | nil, event: String.t() | nil, data: String.t()}}
  | {:error, term()}

Decode a full SSE event block (lines separated by \n, terminated by blank line). Returns the parsed {id, event, data} tuple or an error.

Does NOT parse the data as JSON — callers decide what to do with it.

encode_event(task_id, event_name, seq, payload)

@spec encode_event(String.t(), String.t(), integer(), map() | String.t()) :: iodata()

Encode one SSE event.

payload may be a map (will be JSON-encoded) or an already-encoded string.

event_id(task_id, seq)

@spec event_id(String.t(), integer()) :: String.t()

Build the canonical SSE event id for a (task_id, seq) pair.

keep_alive()

@spec keep_alive() :: iodata()

Emit a keep-alive comment frame to reset intermediary timeouts.

parse_event_id(id)

@spec parse_event_id(String.t()) :: {:ok, {String.t(), integer()}} | {:error, term()}

Parse an event id back into {task_id, seq}.

iex> A2aEngine.Codec.SSE.parse_event_id("abc123:7")
{:ok, {"abc123", 7}}
iex> A2aEngine.Codec.SSE.parse_event_id("bogus")
{:error, :invalid_event_id}

resume_from(id)

@spec resume_from(String.t() | nil) ::
  {:ok, {String.t(), integer()}} | {:error, term()}

Interpret a Last-Event-ID header value for stream resumption.

Returns {:ok, {task_id, seq}} when the header is a well-formed A2A event id, or {:error, _} when the client has never connected before (header absent or malformed).