Class: Mistri::Sinks::Coalesced

Inherits:
Object
  • Object
show all
Defined in:
lib/mistri/sinks/coalesced.rb

Overview

Wraps any sink and merges bursts of streaming deltas, so a transport broadcasts at UI speed instead of token speed. Deltas buffer per content block and flush merged when the interval elapses or any other event arrives, so ordering is preserved and a turn always ends flushed.

sink = Mistri::Sinks::Coalesced.new(
Mistri::Sinks::ActionCable.new("agent_1"), interval: 0.1,
)

Constant Summary collapse

DELTAS = 
%i[text_delta thinking_delta toolcall_delta].freeze

Instance Method Summary collapse

Constructor Details

#initialize(sink, interval: 0.05) ⇒ Coalesced

Returns a new instance of Coalesced.



16
17
18
19
20
21
 
# File 'lib/mistri/sinks/coalesced.rb', line 16

def initialize(sink, interval: 0.05)
  @sink = sink
  @interval = interval
  @buffer = nil
  @flushed_at = now
end

Instance Method Details

#call(event) ⇒ Object 



23
24
25
26
27
28
29
30
31
 
# File 'lib/mistri/sinks/coalesced.rb', line 23

def call(event)
  unless DELTAS.include?(event.type)
    flush
    return @sink.call(event)
  end

  merge(event)
  flush if now - @flushed_at >= @interval
end

#to_proc ⇒ Object 



33
 
# File 'lib/mistri/sinks/coalesced.rb', line 33

def to_proc = method(:call).to_proc