Class: AgUi::Server::SSE::Stream

Inherits:
Protocol::HTTP::Body::Writable
  • Object
show all
Defined in:
lib/ag_ui/server/sse/stream.rb

Overview

Async-native SSE body built on ::Protocol::HTTP::Body::Writable, ported from a2a's Server::SSE::Stream with the AG-UI vocabulary.

Falcon's protocol-rack passes Readable subclasses through untouched, giving true async streaming with backpressure. write() pushes frames, read() pops them (the HTTP server does this), close_write signals EOF.

One typed emitter per AG-UI event, generated from the schema bundle:

stream = AgUi::Server::SSE::Stream.new(thread_id: "t1", run_id: "r1")
stream.run_started
stream.text_message_start(message_id: "m1", role: "assistant")
stream.text_message_content(message_id: "m1", delta: "Hello")
stream.text_message_end(message_id: "m1")
stream.run_finished
stream.finish

threadId / runId are injected automatically wherever the event schema has those properties (RUN_STARTED, RUN_FINISHED, ...); kwargs override. Every event is schema-validated before hitting the wire (validate: false to skip) — a wire-contract bug should raise, not stream.

Constant Summary collapse

SSE_HEADERS =

Response headers for the /run SSE response. Matches the Node runtime (Cache-Control: no-cache, keep-alive) plus x-accel-buffering to defeat proxy buffering.

{
  "content-type"      => "text/event-stream",
  "cache-control"     => "no-cache",
  "x-accel-buffering" => "no",
  "connection"        => "keep-alive",
}.freeze

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(thread_id:, run_id:, validate: true, on_event: nil, **options) ⇒ Stream

on_event: optional tap receiving every wire payload — the run store records through it for /connect replay.



46
47
48
49
50
51
52
53
 
# File 'lib/ag_ui/server/sse/stream.rb', line 46

def initialize(thread_id:, run_id:, validate: true, on_event: nil, **options)
  @thread_id = thread_id
  @run_id    = run_id
  @validate  = validate
  @on_event  = on_event
  @encoder   = EventEncoder.new
  super(**options)
end

Instance Attribute Details

#run_id ⇒ Object (readonly)

Returns the value of attribute run_id.



42
43
44
 
# File 'lib/ag_ui/server/sse/stream.rb', line 42

def run_id
  @run_id
end

#thread_id ⇒ Object (readonly)

Returns the value of attribute thread_id.



42
43
44
 
# File 'lib/ag_ui/server/sse/stream.rb', line 42

def thread_id
  @thread_id
end

Class Method Details

.headers ⇒ Object

A fresh mutable copy — upstream middleware mutates response headers in place.



69
70
71
 
# File 'lib/ag_ui/server/sse/stream.rb', line 69

def self.headers
  SSE_HEADERS.dup
end

Instance Method Details

#event(payload) ⇒ Object

Emit a pre-built event payload (Hash with camelCase keys).



56
57
58
59
 
# File 'lib/ag_ui/server/sse/stream.rb', line 56

def event(payload)
  @on_event&.call(payload)
  write(@encoder.encode(payload))
end

#finish ⇒ Object

Signal end-of-stream; the reader receives nil and closes the SSE connection.



63
64
65
 
# File 'lib/ag_ui/server/sse/stream.rb', line 63

def finish
  close_write
end