From 2e866a618b039a05120fce5f62736d66ceb81d2f Mon Sep 17 00:00:00 2001 From: Ajay Krishnan Date: Sun, 12 Jul 2026 01:53:59 -0700 Subject: [PATCH] Handle current OpenCode stream completion --- CHANGELOG.md | 11 ++++++ README.md | 4 +- lib/opencode/client.rb | 65 ++++++++++++++++++++++++++++----- lib/opencode/response_parser.rb | 4 +- lib/opencode/version.rb | 2 +- test/opencode/smoke_test.rb | 55 +++++++++++++++++++++++++++- 6 files changed, 125 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7bf44ff..3248d10 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index e8e938a..f20fc15 100644 --- a/README.md +++ b/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 diff --git a/lib/opencode/client.rb b/lib/opencode/client.rb index 6a26732..479819f 100644 --- a/lib/opencode/client.rb +++ b/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 } ] { diff --git a/lib/opencode/response_parser.rb b/lib/opencode/response_parser.rb index 4948eab..e1fc0cf 100644 --- a/lib/opencode/response_parser.rb +++ b/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 diff --git a/lib/opencode/version.rb b/lib/opencode/version.rb index e9ec7c5..d4e1dfa 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.alpha3" + VERSION = "0.0.1.alpha4" end diff --git a/test/opencode/smoke_test.rb b/test/opencode/smoke_test.rb index 335b03e..9b26cef 100644 --- a/test/opencode/smoke_test.rb +++ b/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)