LemonAgent.EventStream (lemon_agent v0.1.0)

View Source

Async event stream for streaming Agent responses with OTP-compliant lifecycle management.

This module provides a producer/consumer pattern for streaming events from agent execution with the following BEAM/OTP guarantees:

Features

  • Owner Monitoring: Streams are linked to an owner process and automatically cancel when the owner dies.
  • Task Linking: Streaming tasks can be attached and are properly shutdown when the stream is canceled.
  • Bounded Queues: Configurable queue limits prevent unbounded memory growth.

  • Backpressure: push/2 returns :ok | {:error, :overflow} for flow control.

  • Cancellation: Explicit cancel/2 API for clean stream termination.
  • Timeouts: Configurable stream timeout with automatic cancellation.

Usage

# Start a stream with options
{:ok, stream} = EventStream.start_link(
  owner: self(),
  max_queue: 1000,
  timeout: 300_000
)

# Producer pushes events (with backpressure)
case EventStream.push(stream, {:tool_call, %{name: "read_file"}}) do
  :ok -> :continue
  {:error, :overflow} -> :stop_producing
end

# Or use async push for fire-and-forget
EventStream.push_async(stream, {:agent_start})

# Consumer reads events
stream
|> EventStream.events()
|> Enum.each(fn event -> IO.inspect(event) end)

# Get the final result
{:ok, messages} = EventStream.result(stream)

# Or cancel explicitly
EventStream.cancel(stream, :user_requested)

Terminal Events

The stream terminates when one of these events is received:

  • {:agent_end, messages} - Successful completion with final message list
  • {:error, reason, partial_state} - Error with reason and any partial state
  • {:canceled, reason} - Stream was canceled

Summary

Types

drop_strategy()
event()

Agent events that can be pushed to the stream.

option()
t()

Functions

attach_task(stream, task_pid)

Attach a task to this event stream.

cancel(stream, reason \\ :canceled)

Cancel the stream with a reason.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

complete(stream, messages)

Complete the stream with a final list of messages.

error(stream, reason, partial_state \\ nil)

Signal an error on the stream.

events(stream)

Get a lazy enumerable of events from the stream.

push(stream, event)

Push an event to the stream (synchronous with backpressure).

push_async(stream, event)

Push an event to the stream (asynchronous, fire-and-forget).

result(stream, timeout \\ :infinity)

Get the final result of the stream, blocking until complete.

start(opts \\ [])

Start a new event stream without linking to the caller.

start_link(opts \\ [])

Start a new event stream.

stats(stream)

Get current queue statistics.

Types

drop_strategy()

@type drop_strategy() :: :drop_oldest | :drop_newest | :error

event()

@type event() ::
  {:agent_start, map()}
  | {:agent_end, list()}
  | {:tool_call, map()}
  | {:tool_result, map()}
  | {:thinking, String.t()}
  | {:text_delta, String.t()}
  | {:error, term(), term()}
  | {:canceled, term()}
  | term()

Agent events that can be pushed to the stream.

Terminal events:

  • {:agent_end, messages} - Successful completion
  • {:error, reason, partial_state} - Error state
  • {:canceled, reason} - Stream was canceled

Non-terminal events can be any term representing agent activity.

option()

@type option() ::
  {:owner, pid()}
  | {:runner, pid()}
  | {:max_queue, pos_integer()}
  | {:drop_strategy, drop_strategy()}
  | {:timeout, timeout()}

t()

@type t() :: pid()

Functions

attach_task(stream, task_pid)

@spec attach_task(t(), pid()) :: :ok

Attach a task to this event stream.

When the stream is canceled or the owner dies, the attached task will be shutdown. Only one task can be attached at a time.

cancel(stream, reason \\ :canceled)

@spec cancel(t(), term()) :: :ok

Cancel the stream with a reason.

This will:

  1. Mark the stream as canceled
  2. Shutdown any attached task
  3. Wake up all waiters with a terminal event
  4. Stop the GenServer

Examples

:ok = EventStream.cancel(stream, :user_requested)

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

complete(stream, messages)

@spec complete(t(), list()) :: :ok

Complete the stream with a final list of messages.

This pushes an {:agent_end, messages} terminal event and marks the stream as done. After completion, result/1 will return {:ok, messages}.

Examples

:ok = EventStream.complete(stream, [user_msg, assistant_msg])

error(stream, reason, partial_state \\ nil)

@spec error(t(), term(), term()) :: :ok

Signal an error on the stream.

This pushes an {:error, reason, partial_state} terminal event and marks the stream as done. After an error, result/1 will return {:error, reason, partial_state}.

Examples

:ok = EventStream.error(stream, :timeout, %{messages: partial_messages})

events(stream)

@spec events(t()) :: Enumerable.t()

Get a lazy enumerable of events from the stream.

This returns a Stream that will block when no events are available and complete when a terminal event is received. Terminal events ({:agent_end, _}, {:error, _, _}, or {:canceled, _}) are included in the stream before it halts.

Examples

stream
|> EventStream.events()
|> Enum.each(fn
  {:agent_end, messages} -> IO.puts("Done with #{length(messages)} messages")
  {:tool_call, call} -> IO.puts("Calling tool: #{call.name}")
  event -> IO.inspect(event)
end)

push(stream, event)

@spec push(t(), event()) :: :ok | {:error, :overflow | :canceled}

Push an event to the stream (synchronous with backpressure).

Returns :ok on success or {:error, :overflow} if the queue is full (when using :error drop strategy) or {:error, :canceled} if the stream has been canceled.

Use push_async/2 if you don't need backpressure feedback.

Examples

case EventStream.push(stream, {:tool_call, %{name: "read_file"}}) do
  :ok -> :continue
  {:error, :overflow} -> :stop_producing
end

push_async(stream, event)

@spec push_async(t(), event()) :: :ok

Push an event to the stream (asynchronous, fire-and-forget).

This is a non-blocking push that ignores backpressure. Use push/2 if you need confirmation that the event was accepted.

If using :drop_oldest or :drop_newest strategies, events will be dropped silently on overflow. With :error strategy, overflow events are still dropped but a warning is logged.

Examples

:ok = EventStream.push_async(stream, {:agent_start})

result(stream, timeout \\ :infinity)

@spec result(t(), timeout()) ::
  {:ok, list()} | {:error, term(), term()} | {:error, term()}

Get the final result of the stream, blocking until complete.

Returns {:ok, messages} on successful completion or {:error, reason, partial_state} on error/cancellation.

Options

  • timeout - How long to wait for completion (default: :infinity)

Examples

{:ok, messages} = EventStream.result(stream)
{:ok, messages} = EventStream.result(stream, 5000)

start(opts \\ [])

@spec start([option()]) :: GenServer.on_start()

Start a new event stream without linking to the caller.

This is useful when the stream should outlive the calling process, such as when the caller is a short-lived runner process.

See start_link/1 for options.

start_link(opts \\ [])

@spec start_link([option()]) :: GenServer.on_start()

Start a new event stream.

Options

  • :owner - Process to monitor. Stream cancels if owner dies. Default: self()
  • :runner - Process producing events (e.g., CLI runner). Stream errors if runner dies. This ensures consumers are woken up if the producer crashes.
  • :max_queue - Maximum events to buffer. Default: 10000
  • :drop_strategy - What to do on overflow: :drop_oldest, :drop_newest, or :error. Default: :error
  • :timeout - Stream timeout in milliseconds. Default: 300000ms. Set to :infinity to disable timeout.

Examples

{:ok, stream} = LemonAgent.EventStream.start_link(
  owner: self(),
  max_queue: 5000,
  timeout: 60_000
)

stats(stream)

@spec stats(t()) :: %{
  queue_size: non_neg_integer(),
  max_queue: pos_integer(),
  dropped: non_neg_integer()
}

Get current queue statistics.

Returns a map with:

  • :queue_size - Current number of buffered events
  • :max_queue - Maximum queue size
  • :dropped - Number of events dropped due to overflow

Examples

%{queue_size: 42, max_queue: 10000, dropped: 0} = EventStream.stats(stream)