Class: AgUi::Server::Middleware::StreamBuilder

Inherits:
Object
  • Object
show all
Defined in:
lib/ag_ui/server/middleware/sse_stream.rb

Overview

Factory that creates the SSE stream and runs the caller's block inside Async with automatic finish on exit.

Created by SSEStream middleware — not intended for direct use.

Instance Method Summary collapse

Constructor Details

#initialize(env) ⇒ StreamBuilder

Returns a new instance of StreamBuilder.



52
53
54
 
# File 'lib/ag_ui/server/middleware/sse_stream.rb', line 52

def initialize(env)
  @env = env
end

Instance Method Details

#open(thread_id:, run_id:, validate: true, on_event: nil, on_finish: nil, on_task: nil, &block) ⇒ Object

Create and open the SSE stream for the current run.

The block runs inside an Async fiber; the stream is finished when the block exits, even if an exception is raised. The run handler owns terminal-event semantics (RUN_FINISHED / RUN_ERROR) — this layer only guarantees the connection closes.

on_event / on_finish / on_task are the run-store taps (set by the Server's recording wrapper): every payload, end-of-run, and the Async task handle for /stop cancellation.



66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/ag_ui/server/middleware/sse_stream.rb', line 66

def open(thread_id:, run_id:, validate: true,
         on_event: nil, on_finish: nil, on_task: nil, &block)
  stream = SSE::Stream.new(
    thread_id: thread_id,
    run_id: run_id,
    validate: validate,
    on_event: on_event,
  )

  @env["ag_ui.stream"] = stream

  task = Async do
    block.call(stream)
  ensure
    stream.finish
    on_finish&.call
  end
  on_task&.call(task)

  nil
end