Expose at-most-once subscription readiness

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

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:

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 = false
-                    turn_started = on_subscribed.call
+                  unless subscription_callback_attempted
-                  rescue StandardError => error
+                    # Mark the attempt before invoking the callback. A timeout
-                    # Prompt submission happens inside the open SSE response.
+                    # can mean the prompt reached OpenCode even though its HTTP
-                    # Do not mistake its transport failure for an SSE disconnect
+                    # response did not reach us; retrying would duplicate the
-                    # and hide it behind a reconnect/first-event timeout.
+                    # turn. The caller receives the original error instead.
-                    subscription_callback_error = error
+                    subscription_callback_attempted = true
-                    raise
+                    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

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

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 = [
       {