Handle current OpenCode stream completion

CHANGELOG.md

@@ -1,5 +1,16 @@
 # Changelog
 
+## 0.0.1.alpha4 - 2026-07-12
+
+### Fixed
+
+- End SSE streams on current OpenCode `session.status` idle events while
+  retaining compatibility with legacy `session.idle` events.
+- Reconcile every assistant message in the current user turn after multi-step
+  tool loops, preserving stream-only parts without duplicating final text.
+- Parse terminal tool parts in standalone Ruby clients without relying on the
+  Rails-loaded `Object#in?` extension.
+
 ## 0.0.1.alpha3 - 2026-07-10
 
 ### Added

README.md

@@ -89,7 +89,7 @@ If you need raw SSE events (every server tick, todo update, prompt asked/replied
 
 ```ruby
 client.stream_events(session_id: session_id) do |event|
-  puts event[:type] # "message.part.delta", "todo.updated", "session.idle", ...
+  puts event[:type] # "message.part.delta", "todo.updated", "session.status", ...
 end
 ```
 
@@ -112,7 +112,7 @@ begin
 rescue Opencode::ConnectionError      # server unreachable
 rescue Opencode::TimeoutError         # client-side timeout
 rescue Opencode::SessionNotFoundError # 404 on a session
-rescue Opencode::StaleSessionError    # session.idle never arrived
+rescue Opencode::StaleSessionError    # no session event arrived after the prompt
 rescue Opencode::IdleStreamError      # mid-turn SSE wedge
 rescue Opencode::ServerError          # 5xx
 rescue Opencode::BadRequestError      # 4xx other than 404

lib/opencode/client.rb

@@ -237,8 +237,10 @@ module Opencode
     ].freeze
 
     # Opens SSE connection to GET /event, yields parsed events filtered by session_id.
-    # Blocks until session goes idle or timeout, reconnecting across dropped
-    # event-stream connections.
+    # Blocks until the session reports idle or timeout, reconnecting across
+    # dropped event-stream connections. Current OpenCode emits
+    # `session.status` with `status.type == "idle"`; older versions emitted the
+    # standalone `session.idle` event, so both remain terminal.
     #
     # first_event_timeout: seconds to wait for a session-specific event before
     # declaring the session stale. Server heartbeats don't count — they're global
@@ -345,13 +347,14 @@ module Opencode
                 # the reaper doesn't kill it mid-wait.
                 on_activity_tick&.call(event)
                 block.call(event)
-                return if event[:type] == "session.idle"
+                return if terminal_session_event?(event)
               end
             end
           end
         rescue *TRANSIENT_SSE_ERRORS
           # Treat transport-level SSE disconnects like clean EOF: reconnect
-          # until session.idle, the overall timeout, or first-event timeout.
+          # until an idle session event, the overall timeout, or first-event
+          # timeout.
         ensure
           begin
             http&.finish if http&.started?
@@ -385,13 +388,16 @@ module Opencode
     # the caller's reply is still a usable Result either way.
     def merge_final_exchange(session_id, reply)
       exchange = get_messages(session_id)
-      last_assistant = Array(exchange).reverse_each.find do |message|
-        message.dig(:info, :role) == "assistant"
-      end
-      return unless last_assistant
+      polled = current_turn_parts(exchange)
+      return if polled.empty?
 
-      polled = Opencode::ResponseParser.extract_interleaved_parts(last_assistant)
-      reply.sync_recovered_parts(polled) if polled.any?
+      merged = merge_stream_only_parts(reply.result.parts_json, polled)
+      reply.sync_recovered_parts(merged)
+      # sync_recovered_parts intentionally never deletes live parts because it
+      # is also used during mid-stream recovery. This is the terminal poll, so
+      # the wire snapshot is authoritative: remove any replayed trailing wire
+      # part after observers have seen recovered additions/updates.
+      reply.replace_parts(merged) unless reply.result.parts_json == merged
     rescue Opencode::Error
       # Stream's result is still complete; the merge was a polish, not a
       # requirement.
@@ -408,6 +414,45 @@ module Opencode
       deadline
     end
 
+    def terminal_session_event?(event)
+      return true if event[:type] == "session.idle"
+      return false unless event[:type] == "session.status"
+
+      status = event.dig(:properties, :status)
+      status = status[:type] || status["type"] if status.is_a?(Hash)
+      status == "idle"
+    end
+
+    # OpenCode persists one assistant message per model step. A tool loop can
+    # therefore produce several assistant messages for one user turn (for
+    # example skill -> task -> final text). Reconcile the complete current turn
+    # instead of aligning the live parts array with only the last assistant
+    # message, which corrupts tool parts and duplicates final text.
+    def current_turn_parts(exchange)
+      messages = Array(exchange)
+      last_user_index = messages.rindex { |message| message.dig(:info, :role) == "user" }
+      current_turn = last_user_index ? messages.drop(last_user_index + 1) : messages
+
+      current_turn
+        .select { |message| message.dig(:info, :role) == "assistant" }
+        .flat_map { |message| Opencode::ResponseParser.extract_interleaved_parts(message) }
+    end
+
+    def merge_stream_only_parts(stream_parts, wire_parts)
+      remaining_wire = Array(wire_parts).dup
+      merged = []
+
+      Array(stream_parts).each do |part|
+        if Opencode::PartSource.stream_only?(part)
+          merged << part
+        elsif remaining_wire.any?
+          merged << remaining_wire.shift
+        end
+      end
+
+      merged.concat(remaining_wire)
+    end
+
     def prompt_payload(text, parts:, model:, agent:, system:, message_id:, no_reply:, tools:, format:, variant:)
       message_parts = parts || [ { type: "text", text: text } ]
       {

lib/opencode/response_parser.rb

@@ -27,7 +27,7 @@ module Opencode
     def self.extract_tool_summary(response_body)
       parts = response_body[:parts] || []
       parts
-        .select { |p| p[:type] == "tool" && p.dig(:state, :status).in?(TERMINAL_STATUSES) }
+        .select { |p| p[:type] == "tool" && TERMINAL_STATUSES.include?(p.dig(:state, :status)) }
         .map { |p| build_tool_summary(p) }
     end
 
@@ -42,7 +42,7 @@ module Opencode
           { "type" => "reasoning", "content" => part[:text] }
         when "tool"
           status = part.dig(:state, :status)
-          next unless status.in?(TERMINAL_STATUSES)
+          next unless TERMINAL_STATUSES.include?(status)
 
           build_tool_summary(part)
         else

lib/opencode/version.rb

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

test/opencode/smoke_test.rb

@@ -88,7 +88,7 @@ class SmokeTest < Minitest::Test
         properties: { sessionID: SESSION_ID, partID: "p1", field: "text", delta: "hello " } },
       { type: "message.part.delta",
         properties: { sessionID: SESSION_ID, partID: "p1", field: "text", delta: "world" } },
-      { type: "session.idle", properties: { sessionID: SESSION_ID } }
+      { type: "session.status", properties: { sessionID: SESSION_ID, status: { type: "idle" } } }
     ].map { |e| "data: #{e.to_json}\n\n" }.join
 
     stub_request(:get, %r{#{Regexp.escape(BASE)}/event(\?.*)?\z})
@@ -133,6 +133,59 @@ class SmokeTest < Minitest::Test
     assert_equal "ack", reply.full_text
   end
 
+  def test_stream_merges_a_multi_assistant_tool_loop_without_duplicate_text
+    stub_request(:post, "#{BASE}/session/#{SESSION_ID}/prompt_async")
+      .to_return(status: 204, body: "")
+
+    skill_part = {
+      id: "p_skill", sessionID: SESSION_ID, messageID: "m_skill",
+      type: "tool", tool: "skill", callID: "call_skill",
+      state: { status: "completed", input: { name: "travelwolf-itinerary" }, output: "loaded" }
+    }
+    task_part = {
+      id: "p_task", sessionID: SESSION_ID, messageID: "m_task",
+      type: "tool", tool: "task", callID: "call_task",
+      state: {
+        status: "completed",
+        input: { subagent_type: "itinerary-planner" },
+        output: "{\"days\":[]}",
+        metadata: { sessionId: "ses_child" }
+      }
+    }
+    sse = [
+      { 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 } },
+      { type: "message.part.delta",
+        properties: { sessionID: SESSION_ID, partID: "p_text", field: "text", delta: "SUBAGENT_OK" } },
+      { type: "message.part.delta",
+        properties: { sessionID: SESSION_ID, partID: "p_text_replay", field: "text", delta: "SUBAGENT_OK" } },
+      { type: "session.status", properties: { sessionID: SESSION_ID, status: { type: "idle" } } }
+    ].map { |event| "data: #{event.to_json}\n\n" }.join
+
+    stub_request(:get, %r{#{Regexp.escape(BASE)}/event(\?.*)?\z})
+      .to_return(status: 200, body: sse,
+                 headers: { "Content-Type" => "text/event-stream" })
+
+    exchange = [
+      { info: { role: "user" }, parts: [ { type: "text", text: "previous" } ] },
+      { info: { role: "assistant" }, parts: [ { type: "text", text: "previous answer" } ] },
+      { info: { role: "user" }, parts: [ { type: "text", text: "plan" } ] },
+      { info: { role: "assistant" }, parts: [ skill_part ] },
+      { info: { role: "assistant" }, parts: [ task_part ] },
+      { info: { role: "assistant" }, parts: [ { type: "text", text: "SUBAGENT_OK" } ] }
+    ]
+    stub_request(:get, "#{BASE}/session/#{SESSION_ID}/message")
+      .to_return(status: 200, body: exchange.to_json,
+                 headers: { "Content-Type" => "application/json" })
+
+    reply = @client.stream(SESSION_ID, "plan", stream_timeout: 1)
+
+    assert_equal "SUBAGENT_OK", reply.full_text
+    assert_equal %w[todowrite skill task], reply.tool_parts.map { |part| part.fetch("tool") }
+    assert_equal "ses_child", reply.tool_parts.last.dig("metadata", "sessionId")
+  end
+
   def test_connection_refused_raises_ConnectionError
     stub_request(:get, "http://opencode.dead/global/health")
       .to_raise(Errno::ECONNREFUSED)