Merge pull request #3 from ajaynomics/docs/certified-opencode-compatibility

Document evidence-based OpenCode compatibility

.github/workflows/release.yml

@@ -0,0 +1,23 @@
+name: Push gem
+
+on:
+  push:
+    tags:
+      - "v*"
+
+jobs:
+  push:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: write
+      id-token: write
+    environment: release
+    steps:
+      - uses: actions/checkout@v5
+        with:
+          persist-credentials: false
+      - uses: ruby/setup-ruby@v1
+        with:
+          ruby-version: ruby
+          bundler-cache: true
+      - uses: rubygems/release-gem@v1

.github/workflows/test.yml

@@ -14,7 +14,7 @@ jobs:
       matrix:
         ruby: ["3.2", "3.3", "3.4"]
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v5
 
       - name: Set up Ruby ${{ matrix.ruby }}
         uses: ruby/setup-ruby@v1
@@ -30,5 +30,5 @@ jobs:
 
       - name: Verify gem loads after install
         run: |
-          gem install --local opencode-ruby-*.gem
+          gem install opencode-ruby-*.gem --no-document
           ruby -ropencode-ruby -e 'puts Opencode::VERSION'

CHANGELOG.md

@@ -1,5 +1,29 @@
 # 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
+
+- Make `Opencode::Client#stream` wait for OpenCode's initial
+  `server.connected` SSE readiness frame before submitting `prompt_async`,
+  closing the fast-response window where a turn could emit events before the
+  client was listening.
+- Keep prompt submission at-most-once across automatic SSE reconnects. A
+  reconnect now reopens only the event stream; it never posts the user prompt
+  again, and prompt transport failures remain visible to the caller.
+
 ## 0.0.1.alpha5 - 2026-07-15
 
 ### Added
@@ -76,5 +100,7 @@ First public alpha. HTTP + SSE client for OpenCode REST API.
 ### Compatibility
 
 - Ruby ≥ 3.2
-- OpenCode server ≥ 1.15 (tested against the message bus schema in `packages/opencode/src/session/message-v2.ts`)
+- OpenCode targeted the then-current 1.15 message-bus shape. This historical
+  target was not a blanket SemVer compatibility guarantee; use the README's
+  current compatibility evidence for deployment decisions.
 - Runtime dependency: `activesupport (>= 6.1)` for `blank?`/`present?`/`presence`/`truncate`/`duplicable?`/`megabytes`. ActiveSupport is *not* Rails — it's a standalone helpers gem.

README.md

@@ -67,7 +67,9 @@ child = client.create_session(
 
 Model strings use OpenCode's `provider/model` form; a preformatted model hash
 with `providerID` and `id` keys is also accepted. These configured-session
-fields require OpenCode 1.16.1 or newer.
+fields first appeared in OpenCode 1.16.1. That is an API-introduction note,
+not evidence that every later server is compatible; use the
+[certified compatibility evidence](#compatibility) for deployment choices.
 
 ### Streaming (the headline)
 
@@ -86,6 +88,11 @@ reply.reasoning_text  # => the model's hidden reasoning, if any
 reply.parts_json      # => the full ordered parts array, ready for persistence
 ```
 
+`stream` waits for OpenCode's initial `server.connected` SSE readiness frame
+before it submits the asynchronous prompt. If the event connection drops
+afterward, the client reconnects only the subscription; it never reposts the
+prompt. This prevents both missed fast responses and duplicate turns.
+
 ### Synchronous send (no streaming)
 
 ```ruby
@@ -101,8 +108,9 @@ client.update_session(session_id, permissions: permission_rules)
 
 OpenCode appends PATCHed permission rules and evaluates the last matching
 rule. Hosts should send a complete ordered policy and fingerprint it so the
-same policy is not appended on every turn. This endpoint requires OpenCode
-1.16.1 or newer; the rest of the client remains compatible with 1.15.
+same policy is not appended on every turn. This endpoint first appeared in
+OpenCode 1.16.1; that version floor describes endpoint availability, not
+compatibility with every later server.
 
 ### Lower-level event firehose
 
@@ -114,6 +122,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:
@@ -177,9 +202,29 @@ Want every OpenCode endpoint auto-generated from the OpenAPI spec? Use [`opencod
 ## Compatibility
 
 - Ruby ≥ 3.2
-- OpenCode server ≥ 1.15
 - Runtime dependency: `activesupport (>= 6.1)` — *not* Rails. ActiveSupport is a standalone helpers gem (`blank?`, `present?`, `presence`, `truncate`, etc.).
 
+OpenCode server compatibility is evidence-based, not an open-ended SemVer
+promise. OpenCode's HTTP, SSE, and runtime behavior can change independently
+of a numeric version floor, so this gem does not infer compatibility from a
+constraint such as `>= 1.15`.
+
+Use the [OpenCode compatibility corpus](https://github.com/ajaynomics/opencode-compat)
+as the source of current evidence. Its
+[image matrix](https://github.com/ajaynomics/opencode-compat/blob/main/manifests/image-matrix.json)
+records the exact gem commit and immutable OpenCode OCI digest exercised by
+the `ruby-rest-sse` profile. Its
+[runtime tuples](https://github.com/ajaynomics/opencode-compat/blob/main/manifests/runtime-tuples.json)
+add the exact consumer commit and canary evidence needed for promotion and
+rollback. A version label or image tag is provenance only, never the execution
+coordinate.
+
+A combination absent from passing evidence is **unverified**, not necessarily
+incompatible. Run the corpus against the exact gem, image digest, and consumer
+commit before adopting or promoting it. See the
+[certification policy](https://github.com/ajaynomics/opencode-compat/blob/main/docs/certification.md)
+for the full process.
+
 ## Development
 
 ```sh
@@ -187,7 +232,17 @@ bundle install
 bundle exec rake test
 ```
 
-The smoke suite covers Client end-to-end against WebMock-stubbed OpenCode endpoints.
+The smoke suite covers Client end-to-end against WebMock-stubbed OpenCode
+endpoints, including subscription-before-prompt ordering and
+reconnect-without-repost.
+
+The repository contains a tag-triggered `release.yml` workflow intended for
+RubyGems trusted publishing, but its RubyGems trusted-publisher registration is
+not configured as of `0.0.1.alpha7`; that release was published manually. A
+`v*` tag push therefore does not currently guarantee publication. Before a
+future release, verify the RubyGems registry result explicitly. Once the
+workflow is registered as a trusted publisher for the `release` environment,
+it can build, attest, and publish without a long-lived RubyGems API key.
 
 ## License
 

lib/opencode/client.rb

@@ -131,21 +131,35 @@ module Opencode
       on_activity_tick: nil,
       &block
     )
-      send_message_async(
-        session_id, text,
-        model: model, agent: agent, system: system, message_id: message_id
-      )
-
       reply = Opencode::Reply.new
       reply.add_observer(StreamBlockObserver.new(&block)) if block_given?
 
-      stream_events(
+      # Opening the event stream after prompt_async leaves a race where a fast
+      # turn can emit (and finish) before the client is subscribed. Wait for
+      # OpenCode's initial server.connected SSE frame, then submit the prompt
+      # exactly once. Reconnects invoke on_subscribed again, so mark the attempt
+      # before the POST: an ambiguous prompt response must never cause the same
+      # turn to be submitted twice.
+      prompt_attempted = false
+      on_subscribed = lambda do
+        next false if prompt_attempted
+
+        prompt_attempted = true
+        send_message_async(
+          session_id, text,
+          model: model, agent: agent, system: system, message_id: message_id
+        )
+        true
+      end
+
+      consume_event_stream(
         session_id: session_id,
         timeout: stream_timeout,
         first_event_timeout: first_event_timeout,
         idle_stream_timeout: idle_stream_timeout,
         reply: reply,
-        on_activity_tick: on_activity_tick
+        on_activity_tick: on_activity_tick,
+        on_subscribed: on_subscribed
       ) do |event|
         reply.apply(event)
       end
@@ -271,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
@@ -284,12 +305,28 @@ 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,
+        first_event_timeout: first_event_timeout,
+        idle_stream_timeout: idle_stream_timeout,
+        reply: reply,
+        on_activity_tick: on_activity_tick,
+        on_subscribed: on_subscribed,
+        &block
+      )
+    end
+
+    private def consume_event_stream(session_id:, timeout:, first_event_timeout:,
+                                     idle_stream_timeout:, reply:, on_activity_tick:,
+                                     on_subscribed: nil, &block)
       uri = build_uri("/event")
       deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout
       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)
@@ -319,6 +356,8 @@ module Opencode
         http.open_timeout = 10
         http.read_timeout = 30
 
+        subscription_callback_error = nil
+        subscription_ready = on_subscribed.nil?
         begin
           buffer = String.new
 
@@ -352,6 +391,44 @@ module Opencode
                 event = parse_sse_event(raw_event, session_id)
                 next unless event
 
+                unless subscription_ready
+                  # Every supported OpenCode server starts /event with this
+                  # frame. Receiving it proves the stream body is flowing; on
+                  # current servers the bus listener is registered eagerly,
+                  # and on older lazy-stream servers it is the strongest
+                  # available readiness handshake before prompting.
+                  next unless event[:type] == "server.connected"
+
+                  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
+                    # POST returned. Preserve those timeout semantics: the turn
+                    # and first-session-event windows begin after prompt_async
+                    # succeeds, not while establishing the readiness handshake.
+                    started_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+                    deadline = started_at + timeout
+                    first_event_deadline = started_at + first_event_timeout
+                    last_meaningful_event_at = started_at
+                  end
+                  subscription_ready = true
+                end
+
                 unless event[:type]&.start_with?("server.")
                   received_session_event = true
                   last_meaningful_event_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
@@ -368,6 +445,8 @@ module Opencode
             end
           end
         rescue *TRANSIENT_SSE_ERRORS
+          raise if subscription_callback_error
+
           # Treat transport-level SSE disconnects like clean EOF: reconnect
           # until an idle session event, the overall timeout, or first-event
           # timeout.

lib/opencode/version.rb

@@ -1,5 +1,5 @@
 # frozen_string_literal: true
 
 module Opencode
-  VERSION = "0.0.1.alpha5"
+  VERSION = "0.0.1.alpha7"
 end

test/opencode/smoke_test.rb

@@ -11,6 +11,7 @@ class SmokeTest < Minitest::Test
   BASE = "http://opencode.test"
   PASSWORD = "test-secret"
   SESSION_ID = "ses_smoke_1"
+  CONNECTED_EVENT = { type: "server.connected", properties: {} }.freeze
 
   def setup
     @client = Opencode::Client.new(
@@ -130,6 +131,7 @@ class SmokeTest < Minitest::Test
       .to_return(status: 204, body: "")
 
     sse = [
+      CONNECTED_EVENT,
       { type: "message.part.delta",
         properties: { sessionID: SESSION_ID, partID: "p1", field: "text", delta: "hello " } },
       { type: "message.part.delta",
@@ -157,11 +159,228 @@ class SmokeTest < Minitest::Test
     refute_empty parts_yielded
   end
 
+  def test_stream_waits_for_server_connected_before_posting_the_prompt
+    request_order = []
+    connection_count = 0
+    terminal_event = {
+      type: "session.status",
+      properties: { sessionID: SESSION_ID, status: { type: "idle" } }
+    }
+
+    stub_request(:get, %r{#{Regexp.escape(BASE)}/event(\?.*)?\z})
+      .to_return do
+        connection_count += 1
+        request_order << :sse_accepted
+        events = if connection_count == 1
+          [ { type: "server.heartbeat", properties: {} } ]
+        else
+          [ CONNECTED_EVENT, terminal_event ]
+        end
+        {
+          status: 200,
+          body: events.map { |event| "data: #{event.to_json}\n\n" }.join,
+          headers: { "Content-Type" => "text/event-stream" }
+        }
+      end
+
+    stub_request(:post, "#{BASE}/session/#{SESSION_ID}/prompt_async")
+      .to_return do
+        request_order << :prompt
+        { status: 204, body: "" }
+      end
+
+    stub_request(:get, "#{BASE}/session/#{SESSION_ID}/message")
+      .to_return(status: 200, body: [].to_json,
+                 headers: { "Content-Type" => "application/json" })
+
+    @client.stream(SESSION_ID, "ping", stream_timeout: 1, first_event_timeout: 1)
+
+    assert_equal [ :sse_accepted, :sse_accepted, :prompt ], request_order
+  end
+
+  def test_stream_does_not_post_when_sse_subscription_is_rejected
+    stub_request(:get, %r{#{Regexp.escape(BASE)}/event(\?.*)?\z})
+      .to_return(status: 503, body: "unavailable")
+    prompt = stub_request(:post, "#{BASE}/session/#{SESSION_ID}/prompt_async")
+      .to_return(status: 204, body: "")
+
+    error = assert_raises(Opencode::ServerError) do
+      @client.stream(SESSION_ID, "ping", stream_timeout: 1, first_event_timeout: 1)
+    end
+
+    assert_match "SSE connection failed: HTTP 503", error.message
+    assert_not_requested prompt
+  end
+
+  def test_stream_surfaces_prompt_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" })
+    prompt = stub_request(:post, "#{BASE}/session/#{SESSION_ID}/prompt_async")
+      .to_raise(Net::ReadTimeout.new("prompt timed out"))
+
+    error = assert_raises(Opencode::TimeoutError) do
+      @client.stream(SESSION_ID, "ping", stream_timeout: 1, first_event_timeout: 1)
+    end
+
+    assert_match "OpenCode timeout after 5s", error.message
+    assert_requested event_stream, times: 1
+    assert_requested prompt, times: 1
+  end
+
+  def test_stream_reconnects_without_reposting_the_prompt
+    first_connection = [
+      CONNECTED_EVENT,
+      { type: "server.heartbeat", properties: {} }
+    ]
+    second_connection = [
+      CONNECTED_EVENT,
+      {
+        type: "message.part.delta",
+        properties: { sessionID: SESSION_ID, partID: "p1", field: "text", delta: "once" }
+      },
+      {
+        type: "session.status",
+        properties: { sessionID: SESSION_ID, status: { type: "idle" } }
+      }
+    ]
+
+    event_stream = stub_request(:get, %r{#{Regexp.escape(BASE)}/event(\?.*)?\z})
+      .to_return(
+        {
+          status: 200,
+          body: first_connection.map { |event| "data: #{event.to_json}\n\n" }.join,
+          headers: { "Content-Type" => "text/event-stream" }
+        },
+        {
+          status: 200,
+          body: second_connection.map { |event| "data: #{event.to_json}\n\n" }.join,
+          headers: { "Content-Type" => "text/event-stream" }
+        }
+      )
+    prompt = stub_request(:post, "#{BASE}/session/#{SESSION_ID}/prompt_async")
+      .to_return(status: 204, body: "")
+    stub_request(:get, "#{BASE}/session/#{SESSION_ID}/message")
+      .to_return(status: 200, body: [].to_json,
+                 headers: { "Content-Type" => "application/json" })
+
+    reply = @client.stream(SESSION_ID, "ping", stream_timeout: 1, first_event_timeout: 1)
+
+    assert_equal "once", reply.full_text
+    assert_requested event_stream, times: 2
+    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 = [
+      {
+        type: "question.asked",
+        properties: { id: "que_1", sessionID: SESSION_ID, questions: [] }
+      },
+      {
+        type: "question.replied",
+        properties: { requestID: "que_1", sessionID: SESSION_ID, answers: [ [ "yes" ] ] }
+      },
+      {
+        type: "permission.asked",
+        properties: { id: "per_1", sessionID: SESSION_ID, permission: "bash" }
+      },
+      {
+        type: "permission.replied",
+        properties: { requestID: "per_1", sessionID: SESSION_ID, reply: "once" }
+      },
+      {
+        type: "session.status",
+        properties: { sessionID: SESSION_ID, status: { type: "idle" } }
+      }
+    ]
+    stub_request(:get, %r{#{Regexp.escape(BASE)}/event(\?.*)?\z})
+      .to_return(
+        status: 200,
+        body: events.map { |event| "data: #{event.to_json}\n\n" }.join,
+        headers: { "Content-Type" => "text/event-stream" }
+      )
+
+    reply = Opencode::Reply.new
+    wait_states = []
+    @client.stream_events(session_id: SESSION_ID, reply: reply) do |event|
+      reply.apply(event)
+      wait_states << reply.prompt_blocked?
+    end
+
+    assert_equal [ true, false, true, false, false ], wait_states
+  end
+
   def test_stream_block_is_optional
     stub_request(:post, "#{BASE}/session/#{SESSION_ID}/prompt_async")
       .to_return(status: 204, body: "")
 
     sse = [
+      CONNECTED_EVENT,
       { type: "message.part.delta",
         properties: { sessionID: SESSION_ID, partID: "p1", field: "text", delta: "ack" } },
       { type: "session.idle", properties: { sessionID: SESSION_ID } }
@@ -199,6 +418,7 @@ class SmokeTest < Minitest::Test
       }
     }
     sse = [
+      CONNECTED_EVENT,
       { type: "todo.updated", properties: { sessionID: SESSION_ID, todos: [] } },
       { type: "message.part.updated", properties: { sessionID: SESSION_ID, part: skill_part } },
       { type: "message.part.updated", properties: { sessionID: SESSION_ID, part: task_part } },

test/readme_contract_test.rb

@@ -0,0 +1,19 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+class ReadmeContractTest < Minitest::Test
+  README = File.read(File.expand_path("../README.md", __dir__))
+
+  def test_server_compatibility_points_to_exact_certification_evidence
+    assert_includes README, "https://github.com/ajaynomics/opencode-compat"
+    assert_includes README, "manifests/image-matrix.json"
+    assert_includes README, "manifests/runtime-tuples.json"
+    refute_match(/OpenCode server\s*(?:>=|≥)\s*\d/, README)
+  end
+
+  def test_release_guidance_does_not_claim_trusted_publishing_is_configured
+    assert_includes README, "not configured as of `0.0.1.alpha7`"
+    assert_includes README, "does not currently guarantee publication"
+  end
+end