From edad9c7018bb8b71bf15237c537865289a29cc26 Mon Sep 17 00:00:00 2001 From: Ajay Krishnan Date: Sat, 18 Jul 2026 14:26:22 -0700 Subject: [PATCH] Expose at-most-once subscription readiness --- CHANGELOG.md | 12 +++++++ README.md | 17 ++++++++++ lib/opencode/client.rb | 35 ++++++++++++++------ lib/opencode/version.rb | 2 +- test/opencode/smoke_test.rb | 64 +++++++++++++++++++++++++++++++++++++ 5 files changed, 120 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c789f62..d12b325 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,17 @@ # Changelog +## 0.0.1.alpha7 - 2026-07-18 + +### Fixed + +- Add an at-most-once `on_subscribed` hook to the lower-level + `Opencode::Client#stream_events` API. Higher-level orchestrators can now + wait for `server.connected` before submitting `prompt_async` without + abandoning their own Reply observers, persistence, or recovery pipeline. +- Propagate subscription-hook failures directly and never invoke the hook + again on SSE reconnect. Ambiguous prompt transport failures therefore + cannot silently become duplicate turns. + ## 0.0.1.alpha6 - 2026-07-18 ### Fixed diff --git a/README.md b/README.md index c75ef56..ce4038c 100644 --- a/README.md +++ b/README.md @@ -119,6 +119,23 @@ client.stream_events(session_id: session_id) do |event| end ``` +Orchestrators that submit their own async prompt must do so through +`on_subscribed`; the callback runs only after the first `server.connected` +frame and at most once across automatic reconnects: + +```ruby +client.stream_events( + session_id: session_id, + on_subscribed: -> { client.send_message_async(session_id, prompt) } +) do |event| + reply.apply(event) +end +``` + +If the callback raises, `stream_events` propagates that error and does not +retry it. This is intentional: a timed-out prompt response is ambiguous, so +reposting could duplicate the model turn and its cost. + ### Interactive prompts When the agent uses the `question` or `permission` tools, opencode emits `question.asked` / `permission.asked` events. Answer them via: diff --git a/lib/opencode/client.rb b/lib/opencode/client.rb index 33780c6..9308ff3 100644 --- a/lib/opencode/client.rb +++ b/lib/opencode/client.rb @@ -285,6 +285,13 @@ module Opencode # without nuking real reasoning. Callers that know their agent is # short-prompt + fast can pass a lower value. # + # on_subscribed: optional callable invoked at most once, after the first + # `server.connected` frame proves the SSE response body is flowing. This + # is the safe place for higher-level orchestrators to submit prompt_async; + # reconnects wait for their own connected frame but never invoke it again. + # A raised callback error is propagated directly and is never treated as + # a reconnectable SSE transport failure. + # # idle_stream_timeout: seconds to wait BETWEEN meaningful events once # the session has started producing them. Default nil = no check # (preserves the overall `timeout` ceiling behavior). Opt-in heartbeat @@ -298,7 +305,7 @@ module Opencode # translate into a user-visible error / retry affordance. def stream_events(session_id:, timeout: 600, first_event_timeout: 120, idle_stream_timeout: nil, - reply: nil, on_activity_tick: nil, &block) + reply: nil, on_activity_tick: nil, on_subscribed: nil, &block) consume_event_stream( session_id: session_id, timeout: timeout, @@ -306,6 +313,7 @@ module Opencode idle_stream_timeout: idle_stream_timeout, reply: reply, on_activity_tick: on_activity_tick, + on_subscribed: on_subscribed, &block ) end @@ -318,6 +326,7 @@ module Opencode first_event_deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + first_event_timeout received_session_event = false last_meaningful_event_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) + subscription_callback_attempted = on_subscribed.nil? loop do now = Process.clock_gettime(Process::CLOCK_MONOTONIC) @@ -390,14 +399,22 @@ module Opencode # available readiness handshake before prompting. next unless event[:type] == "server.connected" - begin - turn_started = on_subscribed.call - rescue StandardError => error - # Prompt submission happens inside the open SSE response. - # Do not mistake its transport failure for an SSE disconnect - # and hide it behind a reconnect/first-event timeout. - subscription_callback_error = error - raise + turn_started = false + unless subscription_callback_attempted + # Mark the attempt before invoking the callback. A timeout + # can mean the prompt reached OpenCode even though its HTTP + # response did not reach us; retrying would duplicate the + # turn. The caller receives the original error instead. + subscription_callback_attempted = true + begin + turn_started = on_subscribed.call + rescue StandardError => error + # Prompt submission happens inside the open SSE response. + # Do not mistake its transport failure for an SSE disconnect + # and hide it behind a reconnect/first-event timeout. + subscription_callback_error = error + raise + end end if turn_started # Before this fix stream_events began only after the prompt diff --git a/lib/opencode/version.rb b/lib/opencode/version.rb index e309a3c..bee0247 100644 --- a/lib/opencode/version.rb +++ b/lib/opencode/version.rb @@ -1,5 +1,5 @@ # frozen_string_literal: true module Opencode - VERSION = "0.0.1.alpha6" + VERSION = "0.0.1.alpha7" end diff --git a/test/opencode/smoke_test.rb b/test/opencode/smoke_test.rb index 1890385..b911962 100644 --- a/test/opencode/smoke_test.rb +++ b/test/opencode/smoke_test.rb @@ -271,6 +271,70 @@ class SmokeTest < Minitest::Test assert_requested prompt, times: 1 end + def test_stream_events_invokes_on_subscribed_once_after_connected_across_reconnects + order = [] + connections = [ + [ CONNECTED_EVENT, { type: "server.heartbeat", properties: {} } ], + [ + CONNECTED_EVENT, + { + type: "session.status", + properties: { sessionID: SESSION_ID, status: { type: "idle" } } + } + ] + ] + + event_stream = stub_request(:get, %r{#{Regexp.escape(BASE)}/event(\?.*)?\z}) + .to_return do + events = connections.shift or raise "unexpected third SSE connection" + order << :sse_accepted + { + status: 200, + body: events.map { |event| "data: #{event.to_json}\n\n" }.join, + headers: { "Content-Type" => "text/event-stream" } + } + end + + subscribed_calls = 0 + @client.stream_events( + session_id: SESSION_ID, + timeout: 1, + first_event_timeout: 1, + on_subscribed: -> { + subscribed_calls += 1 + order << :prompt + true + } + ) { |_event| } + + assert_equal 1, subscribed_calls + assert_equal [ :sse_accepted, :prompt, :sse_accepted ], order + assert_requested event_stream, times: 2 + end + + def test_stream_events_surfaces_on_subscribed_timeout_without_reconnecting + event_stream = stub_request(:get, %r{#{Regexp.escape(BASE)}/event(\?.*)?\z}) + .to_return(status: 200, body: "data: #{CONNECTED_EVENT.to_json}\n\n", + headers: { "Content-Type" => "text/event-stream" }) + + calls = 0 + error = assert_raises(Net::ReadTimeout) do + @client.stream_events( + session_id: SESSION_ID, + timeout: 1, + first_event_timeout: 1, + on_subscribed: -> { + calls += 1 + raise Net::ReadTimeout, "ambiguous prompt response" + } + ) { |_event| } + end + + assert_equal "Net::ReadTimeout with \"ambiguous prompt response\"", error.message + assert_equal 1, calls + assert_requested event_stream, times: 1 + end + def test_stream_events_preserves_question_and_permission_wait_state events = [ {