LemonRouter.StreamCoalescer (lemon_router v0.1.0)

View Source

Coalesces streaming deltas for efficient channel output.

The coalescer buffers incoming deltas and flushes them to channels based on configurable thresholds:

  • min_chars: 48 - Minimum characters before flushing
  • idle_ms: 400 - Flush after this idle time
  • max_latency_ms: 1200 - Maximum time before forced flush

Output

Produces semantic LemonCore.DeliveryIntent snapshots/finalization and hands them to LemonChannels.Dispatcher. Channel-specific presentation lives entirely in lemon_channels.

Summary

Functions

Returns a specification to start this module under a supervisor.

Commit the current turn: flush pending buffer, clear PresentationState for the :answer surface so the next delta creates a fresh message, and reset internal text accumulators. No-op when full_text is empty or run_id doesn't match.

Finalize a run for a session/channel.

Force flush the coalescer buffer.

Finalize the current answer turn, move its message tracking to another surface, and reset the answer coalescer so the next delta creates a fresh message.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

commit_turn(session_key, channel_id, run_id)

@spec commit_turn(session_key :: binary(), channel_id :: binary(), run_id :: binary()) ::
  :ok

Commit the current turn: flush pending buffer, clear PresentationState for the :answer surface so the next delta creates a fresh message, and reset internal text accumulators. No-op when full_text is empty or run_id doesn't match.

Used at turn boundaries (model text → tool action) in multi-turn runs so each intermediate answer becomes its own Telegram message instead of overwriting the previous one.

current_text(session_key, channel_id, run_id)

@spec current_text(binary(), binary(), binary()) :: binary() | nil

finalize_run(session_key, channel_id, run_id, opts \\ [])

@spec finalize_run(
  session_key :: binary(),
  channel_id :: binary(),
  run_id :: binary(),
  opts :: keyword()
) :: :ok | {:error, term()}

Finalize a run for a session/channel.

Delegates semantic finalization to the channels dispatcher.

flush(session_key, channel_id)

@spec flush(session_key :: binary(), channel_id :: binary()) :: :ok

Force flush the coalescer buffer.

handoff_turn(session_key, channel_id, run_id, target_surface)

@spec handoff_turn(
  session_key :: binary(),
  channel_id :: binary(),
  run_id :: binary(),
  target_surface :: term()
) :: {:ok, binary()} | :noop

Finalize the current answer turn, move its message tracking to another surface, and reset the answer coalescer so the next delta creates a fresh message.

ingest_delta(session_key, channel_id, run_id, seq, text, opts \\ [])

@spec ingest_delta(
  session_key :: binary(),
  channel_id :: binary(),
  run_id :: binary(),
  seq :: non_neg_integer(),
  text :: binary(),
  opts :: keyword()
) :: :ok

Ingest a delta into the coalescer.

Options

  • :meta - Optional metadata map, can include :answer_msg_id for edit mode

start_link(opts)