Class: Opencode::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/opencode/client.rb

Overview

HTTP client for OpenCode REST API. Thread safety: Each instance creates its own Net::HTTP connection. Do NOT share instances across threads. Create per-job.

Constant Summary collapse

MAX_SSE_BUFFER =

1 MB — safety valve against pathological server responses

1_048_576
SSE_EVENT_BOUNDARY =

Keep CRLF atomic so it cannot backtrack into CR + LF and look like a blank line. SSE also permits bare CR and LF line endings.

/(?>\r\n|[\r\n]){2}/
UTF8_BOM = 
"\xEF\xBB\xBF".b.freeze
SSE_RECONNECT_DELAY = 
0.1
TRANSIENT_SSE_ERRORS = 
[
  EOFError,
  IOError,
  Net::OpenTimeout,
  Net::ReadTimeout,
  Errno::ECONNREFUSED,
  Errno::ECONNRESET,
  Errno::EPIPE
].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(base_url: ENV["OPENCODE_BASE_URL"] || "http://localhost:4096", password: ENV["OPENCODE_SERVER_PASSWORD"], timeout: (ENV["OPENCODE_TIMEOUT"] || 120).to_i, directory: nil, workspace: nil) ⇒ Client

Returns a new instance of Client.



14
15
16
17
18
19
20
21
22
23
24
25
26
 
# File 'lib/opencode/client.rb', line 14

def initialize(
  base_url: ENV["OPENCODE_BASE_URL"] || "http://localhost:4096",
  password: ENV["OPENCODE_SERVER_PASSWORD"],
  timeout: (ENV["OPENCODE_TIMEOUT"] || 120).to_i,
  directory: nil,
  workspace: nil
)
  @uri = URI.parse(base_url)
  @password = password
  @timeout = timeout || 120
  @directory = directory
  @workspace = workspace
end

Instance Attribute Details

#directory ⇒ Object (readonly)

Returns the value of attribute directory.



12
13
14
 
# File 'lib/opencode/client.rb', line 12

def directory
  @directory
end

Instance Method Details

#abort_session(session_id) ⇒ Object 



205
206
207
 
# File 'lib/opencode/client.rb', line 205

def abort_session(session_id)
  post("/session/#{session_id}/abort", {})
end

#children(session_id) ⇒ Object 



181
182
183
184
185
 
# File 'lib/opencode/client.rb', line 181

def children(session_id)
  uri = build_uri("/session/#{session_id}/children")
  request = Net::HTTP::Get.new(uri)
  execute(request)
end

#close ⇒ Object 



479
480
481
482
483
 
# File 'lib/opencode/client.rb', line 479

def close
  @http&.finish if @http&.started?
rescue IOError
  # already closed
end

#create_session(title: nil, permissions: nil, parent_id: nil, agent: nil, model: nil, metadata: nil, workspace_id: nil) ⇒ Object 



28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/opencode/client.rb', line 28

def create_session(
  title: nil,
  permissions: nil,
  parent_id: nil,
  agent: nil,
  model: nil,
  metadata: nil,
  workspace_id: nil
)
  body = {
    title: title,
    permission: permissions,
    parentID: parent_id,
    agent: agent,
    model: format_session_model(model),
    metadata: metadata,
    workspaceID: workspace_id
  }.compact
  post("/session", body)
end

#delete_session(session_id) ⇒ Object 



187
188
189
190
191
 
# File 'lib/opencode/client.rb', line 187

def delete_session(session_id)
  uri = build_uri("/session/#{session_id}")
  request = Net::HTTP::Delete.new(uri)
  execute(request)
end

#get_messages(session_id) ⇒ Object 



199
200
201
202
203
 
# File 'lib/opencode/client.rb', line 199

def get_messages(session_id)
  uri = build_uri("/session/#{session_id}/message")
  request = Net::HTTP::Get.new(uri)
  execute(request)
end

#health ⇒ Object 



251
252
253
254
255
 
# File 'lib/opencode/client.rb', line 251

def health
  uri = build_uri("/global/health", scoped: false)
  request = Net::HTTP::Get.new(uri)
  execute(request)
end

#list_questions ⇒ Object

Returns pending question requests as an Array of Hashes with SYMBOL keys, consistent with every other endpoint that flows through handle_response (e.g., health, list_sessions, get_messages). Callers that compare against persisted JSON column data should symbolize their side, not desymbolize this side.



228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
 
# File 'lib/opencode/client.rb', line 228

def list_questions
  uri = build_uri("/question")
  request = Net::HTTP::Get.new(uri)
  add_auth_header(request)

  response = Opencode::Instrumentation.instrument("opencode.request", method: request.method, path: request.path) do
    http_client.request(request)
  end

  unless response.code.to_i.between?(200, 299)
    raise ServerError, "list_questions failed: HTTP #{response.code} — #{response.body.to_s[0, 200]}"
  end

  return [] if response.body.blank?
  JSON.parse(response.body, symbolize_names: true)
rescue JSON::ParserError => e
  raise ServerError, "list_questions returned invalid JSON: #{e.message}"
rescue Net::OpenTimeout, Net::ReadTimeout, Net::WriteTimeout => e
  raise TimeoutError, "OpenCode timeout after #{@timeout}s: #{e.message}"
rescue Errno::ECONNREFUSED, SocketError => e
  raise ConnectionError, "OpenCode unreachable: #{e.message}"
end

#list_sessions ⇒ Object 



171
172
173
174
175
 
# File 'lib/opencode/client.rb', line 171

def list_sessions
  uri = build_uri("/session")
  request = Net::HTTP::Get.new(uri)
  execute(request)
end

#reject_question(request_id:) ⇒ Object 



213
214
215
 
# File 'lib/opencode/client.rb', line 213

def reject_question(request_id:)
  post("/question/#{request_id}/reject", {})
end

#reply_permission(request_id:, reply:, message: nil) ⇒ Object 



217
218
219
220
221
 
# File 'lib/opencode/client.rb', line 217

def reply_permission(request_id:, reply:, message: nil)
  body = { reply: reply }
  body[:message] = message if message.present?
  post("/permission/#{request_id}/reply", body)
end

#reply_question(request_id:, answers:) ⇒ Object 



209
210
211
 
# File 'lib/opencode/client.rb', line 209

def reply_question(request_id:, answers:)
  post("/question/#{request_id}/reply", { answers: answers })
end

#send_message(session_id, text, parts: nil, model: nil, agent: nil, system: nil, message_id: nil, no_reply: nil, tools: nil, format: nil, variant: nil) ⇒ Object 



49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/opencode/client.rb', line 49

def send_message(
  session_id, text,
  parts: nil,
  model: nil,
  agent: nil,
  system: nil,
  message_id: nil,
  no_reply: nil,
  tools: nil,
  format: nil,
  variant: nil
)
  body = prompt_payload(
    text,
    parts: parts,
    model: model,
    agent: agent,
    system: system,
    message_id: message_id,
    no_reply: no_reply,
    tools: tools,
    format: format,
    variant: variant
  )
  post("/session/#{session_id}/message", body)
end

#send_message_async(session_id, text, parts: nil, model: nil, agent: nil, system: nil, message_id: nil, no_reply: nil, tools: nil, format: nil, variant: nil) ⇒ Object 



76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/opencode/client.rb', line 76

def send_message_async(
  session_id, text,
  parts: nil,
  model: nil,
  agent: nil,
  system: nil,
  message_id: nil,
  no_reply: nil,
  tools: nil,
  format: nil,
  variant: nil
)
  body = prompt_payload(
    text,
    parts: parts,
    model: model,
    agent: agent,
    system: system,
    message_id: message_id,
    no_reply: no_reply,
    tools: tools,
    format: format,
    variant: variant
  )
  post("/session/#{session_id}/prompt_async", body)
end

#session_status ⇒ Object 



193
194
195
196
197
 
# File 'lib/opencode/client.rb', line 193

def session_status
  uri = build_uri("/session/status")
  request = Net::HTTP::Get.new(uri)
  execute(request)
end

#stream(session_id, text, model: nil, agent: nil, system: nil, message_id: nil, stream_timeout: 600, first_event_timeout: 120, idle_stream_timeout: nil, on_activity_tick: nil, &block) ⇒ Object

Block-form streaming — the headline API for callers who want the full async-prompt + SSE-loop + final-exchange-merge flow in one call. Returns the final Opencode::Reply::Result value object once the agent finishes.

reply = client.stream(session_id, "Explain monads") do |part|
print part["content"] if part["type"] == "text"
end
reply.full_text   # => the final accumulated text
reply.tool_parts  # => array of terminal tool parts

The block is invoked every time a part is added, grows, finalizes, or (for tool parts) advances state — i.e., whenever a user-visible change happens. The block receives the current part hash (string keys: "type", "content", "tool", "status", "input", ...).

If you need raw events (every server.* tick, todo.updated, prompt asked/replied, etc.), use #stream_events instead.

Optional kwargs are forwarded to send_message_async — model, agent, system prompt override, and the SSE pacing knobs supported by stream_events.



125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
 
# File 'lib/opencode/client.rb', line 125

def stream(
  session_id, text,
  model: nil, agent: nil, system: nil, message_id: nil,
  stream_timeout: 600,
  first_event_timeout: 120,
  idle_stream_timeout: nil,
  on_activity_tick: nil,
  &block
)
  reply = Opencode::Reply.new
  reply.add_observer(StreamBlockObserver.new(&block)) if block_given?

  # 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_subscribed: on_subscribed
  ) do |event|
    reply.apply(event)
  end

  merge_final_exchange(session_id, reply)
  reply.result
end

#stream_events(session_id:, timeout: 600, first_event_timeout: 120, idle_stream_timeout: nil, reply: nil, on_activity_tick: nil, on_subscribed: nil, &block) ⇒ Object

Opens SSE connection to GET /event, yields parsed events filtered by session_id. 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 keep-alives that flow regardless of session state.

Default 120s rather than the more aggressive 30s used originally: slow-thinking reasoning models (Kimi K2, GPT-5 with extended thinking, etc.) routinely spend 30-90s of pure reasoning before emitting their first message.part.* event, especially on cold sessions with long system prompts. 30s false-positive trips on legitimate first turns and converts them to StaleSessionError; 120s catches genuine zombies 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 watchdog for callers whose user-facing surface needs to fail fast rather than sit forever when an upstream LLM stream wedges mid-turn. Distinct from first_event_timeout (which only protects cold-start) and from the overall timeout ceiling of 600s (which is forgiving — a hung stream holding a thread for 10 minutes is already a bad UX). When the window is exceeded the call raises Opencode::IdleStreamError, which the caller is expected to catch and translate into a user-visible error / retry affordance.



310
311
312
313
314
315
316
317
318
319
320
321
322
323
 
# File 'lib/opencode/client.rb', line 310

def stream_events(session_id:, timeout: 600, first_event_timeout: 120,
                   idle_stream_timeout: nil,
                   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

#update_session(session_id, permissions:) ⇒ Object 



177
178
179
 
# File 'lib/opencode/client.rb', line 177

def update_session(session_id, permissions:)
  patch("/session/#{session_id}", { permission: permissions })
end