Merge pull request #2 from ajaynomics/fix/turn-subscribe-before-prompt-alpha7

Submit Rails turns after SSE readiness

CHANGELOG.md

@@ -1,13 +1,31 @@
 # Changelog
 
+## 0.0.1.alpha7 - 2026-07-18
+
+### Fixed
+
+- Make `Opencode::Turn` submit `prompt_async` through the transport's
+  at-most-once `on_subscribed` callback, after `server.connected` proves the
+  SSE listener is ready. Reconnects reopen only SSE and never replay the user
+  prompt.
+- Fail the turn directly when subscription setup or the prompt POST fails
+  before a turn is confirmed started. The recovery path no longer risks
+  finalizing stale exchange text after a pre-turn failure.
+- Add a gem-level behavioral regression for the cross-gem ordering contract,
+  including reconnect and ambiguous prompt timeout cases.
+
+### Bumped
+
+- Runtime dependency `opencode-ruby` pinned to `= 0.0.1.alpha7`.
+
 ## 0.0.1.alpha6 - 2026-07-18
 
 ### Bumped
 
-- Runtime dependency `opencode-ruby` pinned to `= 0.0.1.alpha6`. Rails turns
-  now establish and validate the OpenCode SSE subscription before submitting
-  `prompt_async`, and automatic SSE reconnects reopen only the subscription
-  without replaying the user prompt.
+- Runtime dependency `opencode-ruby` pinned to `= 0.0.1.alpha6`. This exposed
+  subscribe-before-prompt through `Client#stream`, but the lower-level
+  `Opencode::Turn` path still sent before `stream_events`; that orchestration
+  gap is fixed in alpha7.
 
 ## 0.0.1.alpha5 - 2026-07-15
 

Gemfile

@@ -2,6 +2,6 @@ source "https://rubygems.org"
 
 gem "opencode-ruby",
   git: "https://github.com/ajaynomics/opencode-ruby.git",
-  ref: "b16292723e6385064064fdd61698c9618f6cb156"
+  ref: "78b6f9c9e9c7d58b699af1c3c17764acd33de798"
 
 gemspec

README.md

@@ -29,7 +29,7 @@ bundle install
 
 Runtime deps: `activerecord`, `activestorage`, `activesupport` (>= 7.1). Depends on `opencode-ruby` for the underlying HTTP/SSE primitives.
 
-During the alpha series both gems are pinned in lockstep. Version 0.0.1.alpha6
+During the alpha series both gems are pinned in lockstep. Version 0.0.1.alpha7
 uses a subscribe-ready-before-prompt transport contract and reconnects an
 accepted turn without posting its prompt again.
 

lib/opencode/rails_version.rb

@@ -11,5 +11,5 @@
 # We can't reuse the same constant from a second gem, so we use a
 # distinct, non-namespaced constant.
 module Opencode
-  RAILS_VERSION = "0.0.1.alpha6"
+  RAILS_VERSION = "0.0.1.alpha7"
 end

lib/opencode/turn.rb

@@ -213,12 +213,6 @@ module Opencode
       emit_session_created_if_new
       validate_session!(session_id)
 
-      @client.send_message_async(
-        session_id, @query_text,
-        agent: @agent_name.call(@subject),
-        system: @system_context.call(@subject)
-      )
-
       stream_result = stream_response(session_id)
       exchange = fetch_current_exchange(session_id)
       stream_result, exchange = wait_for_final_exchange_result(session_id, stream_result, exchange)
@@ -260,6 +254,22 @@ module Opencode
       last_activity_touch_at = stream_started_at
       first_token_at = nil
       event_count = 0
+      prompt_attempted = false
+      prompt_succeeded = false
+      on_subscribed = lambda do
+        # stream_events guarantees at-most-once invocation, but keep this
+        # guard here as a second line of defense because an ambiguous prompt
+        # response must never become a duplicate model turn.
+        next false if prompt_attempted
+
+        prompt_attempted = true
+        @client.send_message_async(
+          session_id, @query_text,
+          agent: @agent_name.call(@subject),
+          system: @system_context.call(@subject)
+        )
+        prompt_succeeded = true
+      end
 
       begin
         release_active_record_connections
@@ -293,7 +303,8 @@ module Opencode
         @client.stream_events(
           session_id: session_id,
           reply: reply,
-          on_activity_tick: activity_tick
+          on_activity_tick: activity_tick,
+          on_subscribed: on_subscribed
         ) do |event|
           event_count += 1
           reply.apply(event)
@@ -308,6 +319,13 @@ module Opencode
       rescue Opencode::SessionNotFoundError
         raise
       rescue StandardError => e
+        # Subscription rejection or prompt transport failure happened before
+        # a turn was confirmed started. Recovering from the pre-turn exchange
+        # could finalize stale text, and retrying an ambiguous POST could
+        # duplicate spend, so surface the original failure to the outer error
+        # path without reconnect/recovery.
+        raise unless prompt_succeeded
+
         Opencode::ErrorReporter.report(e, handled: true, severity: :warning,
           context: { feature: @error_feature, error_class: e.class.name })
         emit("stream.interrupted",

opencode-rails.gemspec

@@ -37,7 +37,7 @@ Gem::Specification.new do |spec|
   # this gem builds on. During alpha both gems evolve in lockstep — we pin
   # exactly (= not ~>) so that consumers always pick the version this gem
   # was tested against.
-  spec.add_runtime_dependency "opencode-ruby", "= 0.0.1.alpha6"
+  spec.add_runtime_dependency "opencode-ruby", "= 0.0.1.alpha7"
 
   # Rails sub-libraries used at runtime. Depending on these individually
   # (instead of the `rails` umbrella) avoids forcing host apps to load

test/opencode/turn_test.rb

@@ -3,11 +3,107 @@
 require "test_helper"
 
 # Contract smoke for Opencode::Turn (the orchestrator) and its inner
-# Result value object. Behavioral coverage (the full send -> stream ->
-# recover -> finalize loop) lives in the host application — Turn needs
-# an Opencode::Client, an AR Message, a subject record, etc., which are
-# all integration-level concerns.
+# Result value object. Most ActiveRecord behavior lives in host applications,
+# but the subscribe-before-prompt ordering is a cross-gem transport contract
+# and belongs here so a host cannot silently bypass opencode-ruby's guarantee.
 class Opencode::TurnTest < Minitest::Test
+  SESSION_ID = "ses_turn_test"
+
+  class FakeMessage
+    attr_reader :id, :finalized, :error_content
+    attr_accessor :cost, :input_tokens, :output_tokens, :tool_calls_json
+
+    def initialize
+      @id = 12
+    end
+
+    def reload = self
+    def cancelled? = false
+
+    def finalize!(**attrs)
+      @finalized = attrs
+      @cost = attrs[:cost]
+      @input_tokens = attrs[:input_tokens]
+      @output_tokens = attrs[:output_tokens]
+      @tool_calls_json = attrs[:tool_calls_json]
+      true
+    end
+
+    def error!(content)
+      @error_content = content
+    end
+  end
+
+  FakeSubject = Struct.new(:id, :opencode_session_id, keyword_init: true)
+
+  class FakeSession
+    def ensure!(_client) = SESSION_ID
+    def just_created? = false
+  end
+
+  class FakeObserver
+    def watch(_reply); end
+  end
+
+  class OrderedClient
+    attr_reader :order, :prompt_count, :message_reads
+
+    def initialize(prompt_error: nil)
+      @order = []
+      @prompt_count = 0
+      @message_reads = 0
+      @prompt_error = prompt_error
+    end
+
+    def get_messages(_session_id)
+      @message_reads += 1
+      @order << (@prompt_count.zero? ? :messages_before : :messages_after)
+      return [] if @prompt_count.zero?
+
+      [
+        { info: { role: "user" }, parts: [ { type: "text", text: "ping" } ] },
+        {
+          info: {
+            role: "assistant", finish: "stop",
+            time: { created: 1, completed: 2 },
+            cost: 0.01,
+            tokens: { input: 2, output: 1 }
+          },
+          parts: [ { type: "text", text: "pong" } ]
+        }
+      ]
+    end
+
+    def send_message_async(session_id, text, agent:, system:)
+      @prompt_count += 1
+      @order << :prompt
+      raise @prompt_error if @prompt_error
+
+      raise "wrong prompt" unless session_id == SESSION_ID && text == "ping"
+      raise "wrong routing" unless agent == "test-agent" && system == "test-system"
+
+      {}
+    end
+
+    def stream_events(session_id:, reply:, on_activity_tick:, on_subscribed:)
+      raise "wrong session" unless session_id == SESSION_ID
+      raise "missing reply" unless reply.is_a?(Opencode::Reply)
+      raise "missing activity callback" unless on_activity_tick.respond_to?(:call)
+
+      @order << :sse_ready
+      on_subscribed.call
+      @order << :sse_reconnected
+      on_subscribed.call
+      yield(
+        type: "message.part.delta",
+        properties: { sessionID: SESSION_ID, partID: "p1", field: "text", delta: "pong" }
+      )
+      yield(
+        type: "session.status",
+        properties: { sessionID: SESSION_ID, status: { type: "idle" } }
+      )
+    end
+  end
   REQUIRED_INIT_KEYS = %i[
     message subject query_text client session_for observer_factory
     system_context agent_name tracer
@@ -59,4 +155,56 @@ class Opencode::TurnTest < Minitest::Test
     assert_equal 100,   result.input_tokens
     assert_equal 50,    result.output_tokens
   end
+
+  def test_turn_subscribes_before_prompt_and_never_reprompts_on_reconnect
+    client = OrderedClient.new
+    message = FakeMessage.new
+    results = []
+
+    build_turn(client:, message:, results:).call
+
+    assert_equal 1, client.prompt_count
+    assert_equal(
+      [ :messages_before, :sse_ready, :prompt, :sse_reconnected, :messages_after ],
+      client.order
+    )
+    assert_equal "pong", message.finalized.fetch(:content)
+    assert_nil message.error_content
+    assert results.last.completed?
+  end
+
+  def test_turn_does_not_recover_or_retry_an_ambiguous_prompt_failure
+    client = OrderedClient.new(prompt_error: Net::ReadTimeout.new("prompt timed out"))
+    message = FakeMessage.new
+    results = []
+
+    build_turn(client:, message:, results:).call
+
+    assert_equal 1, client.prompt_count
+    assert_equal 1, client.message_reads
+    assert_nil message.finalized
+    assert_equal Opencode::Turn::ERROR_FALLBACK_CONTENT, message.error_content
+    assert results.last.failed?
+    assert_instance_of Net::ReadTimeout, results.last.error
+  end
+
+  private
+
+  def build_turn(client:, message:, results:)
+    Opencode::Turn.new(
+      message: message,
+      subject: FakeSubject.new(id: 34, opencode_session_id: SESSION_ID),
+      query_text: "ping",
+      client: client,
+      session_for: FakeSession.new,
+      observer_factory: ->(_message) { FakeObserver.new },
+      system_context: ->(_subject) { "test-system" },
+      agent_name: ->(_subject) { "test-agent" },
+      tracer: ->(_name, **_payload) {},
+      on_turn_finished: ->(result) { results << result },
+      empty_stream_retry_delay: 0,
+      final_exchange_timeout: 0,
+      final_exchange_retry_delay: 0
+    )
+  end
 end