Raxol.Agent.Conversation.Log (Raxol Agent v2.6.0)

Copy Markdown View Source

Append-only conversation item log with live-tail-no-replay streaming.

Wraps a Raxol.Agent.Conversation.Store (durability) with an in-process subscriber fan-out (liveness). The stream itself keeps NO replay buffer -- all history lives in the store. This is the omnigent model: durability in the store, the stream is ephemeral pub/sub.

Exact snapshot/live-tail partition

subscribe/3 returns a snapshot AND registers the caller as a subscriber in a single serialized GenServer.call. Because append/3 is also a serialized call, the two never interleave: a subscribe processed before an append excludes those items from its snapshot but receives them on the live tail; a subscribe processed after an append includes them in the snapshot and is NOT re-sent them. So a reconnecting client sees every item exactly once with no gap and no dupe -- belt-and-suspenders deduping by Item.id is then trivial.

Live items arrive as {:conversation_item, conversation_id, item} messages. Subscribers are monitored; their subscriptions are dropped automatically when they exit.

Usage

{:ok, log} = Log.start_link(name: MyLog)
{:ok, %{snapshot: history, last_seq: cursor}} = Log.subscribe(log, "conv-1")
{:ok, [item]} = Log.append(log, "conv-1", %{type: :message, data: %{content: "hi"}})
# the subscriber's mailbox now holds {:conversation_item, "conv-1", item}

Reconnect by passing the last seq seen: Log.subscribe(log, conv, after: cursor).

Summary

Functions

Append one item (a map) or a list of items; broadcasts to subscribers.

Returns a specification to start this module under a supervisor.

Conversation metadata (%{id, last_seq, item_count}) or {:error, :not_found}.

Page through stored items (passthrough to the store, no subscription).

List known conversation ids.

Subscribe the caller to a conversation; returns its snapshot + last seq.

Stop receiving live items for a conversation.

Types

conversation_id()

@type conversation_id() :: binary()

server()

@type server() :: GenServer.server()

Functions

append(server, conversation_id, item)

Append one item (a map) or a list of items; broadcasts to subscribers.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

get_conversation(server, conversation_id)

@spec get_conversation(server(), conversation_id()) ::
  {:ok, map()} | {:error, :not_found}

Conversation metadata (%{id, last_seq, item_count}) or {:error, :not_found}.

items(server, conversation_id, opts \\ [])

@spec items(server(), conversation_id(), keyword()) ::
  {:ok, [Raxol.Agent.Conversation.Item.t()]}

Page through stored items (passthrough to the store, no subscription).

list_conversations(server)

@spec list_conversations(server()) :: {:ok, [conversation_id()]}

List known conversation ids.

start_link(opts \\ [])

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

subscribe(server, conversation_id, opts \\ [])

@spec subscribe(server(), conversation_id(), keyword()) ::
  {:ok,
   %{snapshot: [Raxol.Agent.Conversation.Item.t()], last_seq: integer() | nil}}

Subscribe the caller to a conversation; returns its snapshot + last seq.

Options: :after -- only snapshot items after this seq (for reconnect).

unsubscribe(server, conversation_id)

@spec unsubscribe(server(), conversation_id()) :: :ok

Stop receiving live items for a conversation.