Raxol.Agent.SessionStreamer (Raxol Agent v2.6.0)

Copy Markdown View Source

Real-time event streaming for agent sessions.

Provides a pub/sub mechanism for observing agent activity. Subscribers receive events as they happen -- tool calls, text output, state changes, errors. Used by SessionStreamServer for HTTP/SSE delivery.

Usage

# Subscribe to events from an agent session
SessionStreamer.subscribe(session_id)

# Events arrive as messages:
receive do
  {:session_event, ^session_id, event} -> handle(event)
end

# Emit an event (called by agent infrastructure)
SessionStreamer.emit(session_id, {:text_delta, "Hello"})

# Unsubscribe
SessionStreamer.unsubscribe(session_id)

Event Types

Events mirror Raxol.Agent.Stream types:

  • {:text_delta, text} -- streaming text chunk
  • {:tool_use, %{name, arguments, id}} -- tool invocation
  • {:tool_result, %{name, result}} -- tool result
  • {:state_change, %{from, to}} -- agent status transition
  • {:turn_complete, info} -- end of one reasoning turn
  • {:done, info} -- session completed
  • {:error, reason} -- error occurred

Summary

Types

event()
session_id()
t()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

emit(session_id, event, server \\ __MODULE__)

Emit an event for a session (broadcast to all subscribers).

history(session_id, server \\ __MODULE__)

Get recent event history for a session.

list_sessions(server \\ __MODULE__)

List sessions with active subscribers.

start_link(init_opts \\ [])

Start the SessionStreamer (typically in your supervision tree).

subscribe(session_id, server \\ __MODULE__)

Subscribe the calling process to events from a session.

unsubscribe(session_id, server \\ __MODULE__)

Unsubscribe the calling process from a session.

Types

event()

@type event() ::
  {:text_delta, String.t()}
  | {:tool_use, map()}
  | {:tool_result, map()}
  | {:state_change, map()}
  | {:turn_complete, map()}
  | {:done, map()}
  | {:error, term()}

session_id()

@type session_id() :: term()

t()

@type t() :: %Raxol.Agent.SessionStreamer{
  history: %{required(session_id()) => :queue.queue()},
  max_history: pos_integer(),
  subscriptions: %{required(session_id()) => MapSet.t(pid())}
}

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

emit(session_id, event, server \\ __MODULE__)

@spec emit(session_id(), event(), GenServer.server()) :: :ok

Emit an event for a session (broadcast to all subscribers).

history(session_id, server \\ __MODULE__)

@spec history(session_id(), GenServer.server()) :: [event()]

Get recent event history for a session.

list_sessions(server \\ __MODULE__)

@spec list_sessions(GenServer.server()) :: [session_id()]

List sessions with active subscribers.

start_link(init_opts \\ [])

@spec start_link(keyword()) :: GenServer.on_start()

Start the SessionStreamer (typically in your supervision tree).

subscribe(session_id, server \\ __MODULE__)

@spec subscribe(session_id(), GenServer.server()) :: :ok

Subscribe the calling process to events from a session.

unsubscribe(session_id, server \\ __MODULE__)

@spec unsubscribe(session_id(), GenServer.server()) :: :ok

Unsubscribe the calling process from a session.