Class: Harnex::Adapters::Pi

Inherits:
Base
  • Object
show all
Defined in:
lib/harnex/adapters/pi.rb

Overview

Pi RPC adapter — JSONL command/event protocol over stdio.

Protocol docs: pi --mode rpc (strict LF-delimited JSON lines).

Constant Summary collapse

STOP_TERM_GRACE_SECONDS = 
0.5
STOP_KILL_GRACE_SECONDS = 
1.0
REQUEST_TIMEOUT_SECONDS = 
30.0
STDERR_TAIL_BYTES = 
16 * 1024
MIN_RPC_VERSION = 
Gem::Version.new("0.80.4")
DIALOG_UI_METHODS = 
%w[select confirm input editor].freeze
THINKING_LEVELS = 
%w[off minimal low medium high xhigh max].freeze

Constants inherited from Base

Base::AGENT_VERSION_TIMEOUT_SECONDS, Base::PROMPT_PREFIXES

Instance Attribute Summary collapse

Attributes inherited from Base

#key

Instance Method Summary collapse

Methods inherited from Base

#agent_version, #infer_repo_path, #parse_session_summary, #send_wait_seconds, #usage_input_includes_cached?, #wait_for_sendable, #wait_for_sendable_state?

Constructor Details

#initialize(extra_args = []) ⇒ Pi

Returns a new instance of Pi.



22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/harnex/adapters/pi.rb', line 22

def initialize(extra_args = [])
  super("pi", extra_args)
  @initial_prompt = extract_initial_prompt(extra_args)
  @notification_handler = nil
  @disconnect_handler = nil
  @read_io = nil
  @write_io = nil
  @stderr_io = nil
  @pid = nil
  @wait_thr = nil
  @reader_thread = nil
  @stderr_thread = nil
  @stderr_mutex = Mutex.new
  @stderr_tail = +""
  @stderr_tail.force_encoding(Encoding::BINARY)
  @closed = false
  @disconnect_signaled = false
  @state = :disconnected
  @next_id = 1
  @pending = {}
  @id_mutex = Mutex.new
  @write_mutex = Mutex.new
  @summary_mutex = Mutex.new
  @session_summary = {}
  @context_telemetry = ContextTelemetry.new(
    status: "observed",
    source: context_telemetry_source
  )
  @model = nil
  @provider = nil
  @startup_model = nil
  @startup_effort = nil
  @session_stats_requested = false
  @last_completed_at = nil
end

Instance Attribute Details

#initial_prompt ⇒ Object (readonly)

Returns the value of attribute initial_prompt.



20
21
22
 
# File 'lib/harnex/adapters/pi.rb', line 20

def initial_prompt
  @initial_prompt
end

#last_completed_at ⇒ Object (readonly)

Returns the value of attribute last_completed_at.



20
21
22
 
# File 'lib/harnex/adapters/pi.rb', line 20

def last_completed_at
  @last_completed_at
end

Instance Method Details

#base_command ⇒ Object 



82
83
84
 
# File 'lib/harnex/adapters/pi.rb', line 82

def base_command
  ["pi", "--mode", "rpc"]
end

#build_command ⇒ Object 



86
87
88
89
90
91
 
# File 'lib/harnex/adapters/pi.rb', line 86

def build_command
  args = cli_extra_args
  args += ["--model", @startup_model] if @startup_model
  args += ["--thinking", @startup_effort] if @startup_effort
  base_command + args
end

#build_send_payload(text:, submit:, enter_only:, screen_text:, force: false) ⇒ Object

Raises:

  • (ArgumentError) 


149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
 
# File 'lib/harnex/adapters/pi.rb', line 149

def build_send_payload(text:, submit:, enter_only:, screen_text:, force: false)
  state = input_state(nil)
  if !force && submit && !enter_only && state[:input_ready] != true
    raise ArgumentError, blocked_message(state, enter_only: enter_only)
  end

  raise ArgumentError, "Pi RPC cannot stage input without submitting it" unless submit || enter_only
  raise ArgumentError, "Pi RPC does not support submit-only input" if enter_only

  dispatch = { prompt: text.to_s }
  dispatch[:streaming_behavior] = "steer" if force && state[:state] == "busy"
  {
    dispatch: dispatch,
    input_state: state,
    force: force
  }
end

#close ⇒ Object 



265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
 
# File 'lib/harnex/adapters/pi.rb', line 265

def close
  return if @closed

  @closed = true
  fail_pending_requests(StandardError.new("pi rpc client closed"))

  begin
    @write_io.close unless @write_io&.closed?
    @read_io.close if !@pid && !@wait_thr && @read_io && !@read_io.closed?
  rescue IOError
    nil
  end

  @reader_thread&.join(2)
ensure
  terminate_subprocess(
    term_grace_seconds: STOP_TERM_GRACE_SECONDS,
    kill_grace_seconds: STOP_KILL_GRACE_SECONDS
  )
  @reader_thread&.join(1)
  @stderr_thread&.join(1)
  [@read_io, @stderr_io].compact.each do |io|
    io.close unless io.closed?
  rescue IOError
    nil
  end
end

#collect_session_summary ⇒ Object 



244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
 
# File 'lib/harnex/adapters/pi.rb', line 244

def collect_session_summary
  attempt_live_summary_refresh if connected?

  @summary_mutex.synchronize do
    {
      input_tokens: @session_summary[:input_tokens],
      output_tokens: @session_summary[:output_tokens],
      reasoning_tokens: nil,
      cached_tokens: @session_summary[:cached_tokens],
      total_tokens: @session_summary[:total_tokens],
      agent_session_id: @session_summary[:agent_session_id],
      tool_calls: @session_summary[:tool_calls],
      cost_usd: @session_summary[:cost_usd],
      cost_source: @session_summary[:cost_source],
      model: @session_summary[:model],
      agent_provider: @session_summary[:agent_provider],
      context: @context_telemetry.snapshot
    }
  end
end

#configure_startup(model: nil, effort: nil) ⇒ Object 



93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/harnex/adapters/pi.rb', line 93

def configure_startup(model: nil, effort: nil)
  requested_model = model.to_s.strip
  requested_effort = effort.to_s.strip
  unless requested_effort.empty? || THINKING_LEVELS.include?(requested_effort)
    raise ArgumentError,
          "unsupported Pi thinking level #{requested_effort.inspect}; expected one of #{THINKING_LEVELS.join(', ')}"
  end

  @startup_model = requested_model.empty? ? nil : requested_model
  @startup_effort = requested_effort.empty? ? nil : requested_effort
  self
end

#context_telemetry_source ⇒ Object 



78
79
80
 
# File 'lib/harnex/adapters/pi.rb', line 78

def context_telemetry_source
  "pi_get_session_stats"
end

#context_telemetry_supported? ⇒ Boolean

Returns:

  • (Boolean) 


74
75
76
 
# File 'lib/harnex/adapters/pi.rb', line 74

def context_telemetry_supported?
  true
end

#current_model ⇒ Object 



66
67
68
 
# File 'lib/harnex/adapters/pi.rb', line 66

def current_model
  @model
end

#describe ⇒ Object 



123
124
125
126
127
128
129
130
131
132
133
134
135
136
 
# File 'lib/harnex/adapters/pi.rb', line 123

def describe
  {
    transport: transport,
    protocol: "jsonl",
    minimum_version: MIN_RPC_VERSION.to_s,
    events: %w[
      agent_start agent_end agent_settled turn_start turn_end message_start message_update message_end
      tool_execution_start tool_execution_update tool_execution_end queue_update
      compaction_start compaction_end auto_retry_start auto_retry_end
      summarization_retry_scheduled summarization_retry_attempt_start summarization_retry_finished
      bash_execution_update extension_error extension_ui_request
    ]
  }
end

#disconnect_diagnostic ⇒ Object 



337
338
339
340
341
342
 
# File 'lib/harnex/adapters/pi.rb', line 337

def disconnect_diagnostic
  tail = stderr_tail.strip
  return "pi rpc disconnected" if tail.empty?

  "pi rpc disconnected; stderr: #{tail}"
end

#dispatch(prompt:, model: nil, effort: nil, streaming_behavior: nil) ⇒ Object 



198
199
200
201
202
203
204
205
206
 
# File 'lib/harnex/adapters/pi.rb', line 198

def dispatch(prompt:, model: nil, effort: nil, streaming_behavior: nil)
  ensure_open!
  apply_dispatch_overrides(model: model, effort: effort) unless @state == :busy
  payload = { "type" => "prompt", "message" => prompt.to_s }
  payload["streamingBehavior"] = streaming_behavior if streaming_behavior
  request(payload)
  @state = :busy
  nil
end

#inject_exit(_writer, **_kwargs) ⇒ Object 



167
168
169
 
# File 'lib/harnex/adapters/pi.rb', line 167

def inject_exit(_writer, **_kwargs)
  nil
end

#input_state(_screen_text = nil) ⇒ Object 



142
143
144
145
146
147
 
# File 'lib/harnex/adapters/pi.rb', line 142

def input_state(_screen_text = nil)
  {
    state: @state.to_s,
    input_ready: @state == :prompt
  }
end

#interrupt(turn_id: nil) ⇒ Object 



208
209
210
211
212
213
 
# File 'lib/harnex/adapters/pi.rb', line 208

def interrupt(turn_id: nil)
  ensure_open!
  request({ "type" => "abort" })
rescue StandardError
  nil
end

#on_disconnect(&block) ⇒ Object 



175
176
177
 
# File 'lib/harnex/adapters/pi.rb', line 175

def on_disconnect(&block)
  @disconnect_handler = block
end

#on_notification(&block) ⇒ Object 



171
172
173
 
# File 'lib/harnex/adapters/pi.rb', line 171

def on_notification(&block)
  @notification_handler = block
end

#parsed_agent_version ⇒ Object 



116
117
118
119
120
121
 
# File 'lib/harnex/adapters/pi.rb', line 116

def parsed_agent_version
  match = agent_version.to_s.match(/(\d+\.\d+\.\d+)/)
  match ? Gem::Version.new(match[1]) : nil
rescue ArgumentError
  nil
end

#pid ⇒ Object 



313
314
315
 
# File 'lib/harnex/adapters/pi.rb', line 313

def pid
  @pid
end

#provider ⇒ Object 



62
63
64
 
# File 'lib/harnex/adapters/pi.rb', line 62

def provider
  @provider
end

#request_session_stats_async ⇒ Object 



215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
 
# File 'lib/harnex/adapters/pi.rb', line 215

def request_session_stats_async
  return if @closed

  should_request = @summary_mutex.synchronize do
    next false if @session_stats_requested

    @session_stats_requested = true
  end
  return unless should_request

  write_line("type" => "get_session_stats")
rescue StandardError
  clear_session_stats_request
  nil
end

#respond_extension_ui_cancel(request_id:, method:) ⇒ Object 



231
232
233
234
235
236
237
238
239
240
241
242
 
# File 'lib/harnex/adapters/pi.rb', line 231

def respond_extension_ui_cancel(request_id:, method:)
  return false unless DIALOG_UI_METHODS.include?(method.to_s)

  write_line(
    "type" => "extension_ui_response",
    "id" => request_id,
    "cancelled" => true
  )
  true
rescue StandardError
  false
end

#start_rpc(env: nil, cwd: nil, read_io: nil, write_io: nil, pid: nil) ⇒ Object 



179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
 
# File 'lib/harnex/adapters/pi.rb', line 179

def start_rpc(env: nil, cwd: nil, read_io: nil, write_io: nil, pid: nil)
  if read_io && write_io
    @read_io = read_io
    @write_io = write_io
    @pid = pid
  else
    validate_runtime!
    @pid, @write_io, @read_io, @stderr_io, @wait_thr = spawn_subprocess(env, cwd)
  end

  @closed = false
  @disconnect_signaled = false
  @state = :prompt
  @stderr_thread = Thread.new { drain_stderr } if @stderr_io
  @reader_thread = Thread.new { read_loop }
  request_state_async
  self
end

#state ⇒ Object 



138
139
140
 
# File 'lib/harnex/adapters/pi.rb', line 138

def state
  @state
end

#stderr_tail ⇒ Object 



333
334
335
 
# File 'lib/harnex/adapters/pi.rb', line 333

def stderr_tail
  @stderr_mutex.synchronize { @stderr_tail.dup.force_encoding(Encoding::UTF_8).scrub("") }
end

#terminate_subprocess(term_grace_seconds: STOP_TERM_GRACE_SECONDS, kill_grace_seconds: STOP_KILL_GRACE_SECONDS) ⇒ Object 



293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
 
# File 'lib/harnex/adapters/pi.rb', line 293

def terminate_subprocess(term_grace_seconds: STOP_TERM_GRACE_SECONDS, kill_grace_seconds: STOP_KILL_GRACE_SECONDS)
  return false unless @pid

  begin
    Process.kill("TERM", @pid)
  rescue Errno::ESRCH
    return true
  end

  return true if wait_for_process_exit(@pid, term_grace_seconds)

  begin
    Process.kill("KILL", @pid)
  rescue Errno::ESRCH
    return true
  end

  wait_for_process_exit(@pid, kill_grace_seconds)
end

#transport ⇒ Object 



58
59
60
 
# File 'lib/harnex/adapters/pi.rb', line 58

def transport
  :stdio_jsonl_rpc
end

#usage_telemetry_supported? ⇒ Boolean

Returns:

  • (Boolean) 


70
71
72
 
# File 'lib/harnex/adapters/pi.rb', line 70

def usage_telemetry_supported?
  true
end

#validate_runtime! ⇒ Object 



106
107
108
109
110
111
112
113
114
 
# File 'lib/harnex/adapters/pi.rb', line 106

def validate_runtime!
  version = parsed_agent_version
  if version.nil?
    raise "could not determine Pi version from `pi --version`; Pi RPC requires >= #{MIN_RPC_VERSION}"
  end
  return true if version >= MIN_RPC_VERSION

  raise "Pi #{version} is too old for reliable RPC settlement; harnex requires Pi >= #{MIN_RPC_VERSION} (run `pi update`)"
end

#wait_for_exit ⇒ Object 



317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
 
# File 'lib/harnex/adapters/pi.rb', line 317

def wait_for_exit
  return nil unless @wait_thr || @pid

  status = if @wait_thr
             @wait_thr.value
           else
             _waited_pid, process_status = Process.wait2(@pid)
             process_status
           end
  @pid = nil
  status
rescue Errno::ECHILD
  @pid = nil
  nil
end