Class: PiAgent::Client

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

Overview

High-level client. Owns a Transport, correlates request/response by id, and fans notifications out to subscribers.

client = PiAgent::Client.new.start
client.subscribe { |msg| ... }              # all server-pushed messages
future = client.request("get_commands")     # request/response
future.value!(timeout: 5)                   # blocks for response
client.notify("set_thinking", level: "off") # fire-and-forget (no id)
client.close

By default the client spawns pi --mode rpc as a local subprocess. Pass transport_factory: — a callable (on_message:, on_stderr:) -> transport — to run pi somewhere else (e.g. inside a remote sandbox). A factory may additionally accept on_close: to report transport death; the keyword is passed only when the factory accepts it, so older two-keyword factories keep working. See Transport for the transport contract.

Since pi 0.79.0 project-local inputs (.pi/settings.json, project extensions, resources, packages) are trust-gated, and in RPC mode pi silently ignores them unless the project was already trusted. Pass approve: true to trust the project (--approve), or approve: false to explicitly ignore project inputs (--no-approve).

Constant Summary collapse

DEFAULT_BIN = 
"pi"
DEFAULT_ARGS = 
["--mode", "rpc"].freeze
TRANSPORT_CLOSED_TYPE =

Synthetic message fanned out to subscribers when the transport dies, so event streams wake promptly instead of waiting out their timeouts. Never sent by pi (the "pi_agent/" prefix keeps it out of upstream's event namespace); carries the death reason under "reason".

"pi_agent/transport_closed"

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(bin: nil, args: DEFAULT_ARGS, env: {}, cwd: nil, approve: nil, extension_ui: nil, on_extension_ui_error: nil, transport_factory: nil) ⇒ Client

Returns a new instance of Client.



63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/pi_agent/client.rb', line 63

def initialize(bin: nil, args: DEFAULT_ARGS, env: {}, cwd: nil, approve: nil,
               extension_ui: nil, on_extension_ui_error: nil, transport_factory: nil)
  @extension_ui_handler = extension_ui
  @extension_ui_error_handler = on_extension_ui_error
  args = [*args, approve ? "--approve" : "--no-approve"] unless approve.nil?
  @transport_factory = transport_factory || build_subprocess_factory(bin, args, env, cwd)
  @pending = {}
  @pending_mutex = Mutex.new
  @next_id = 0
  @subscribers = []
  @subscribers_mutex = Mutex.new
  @transport = nil
  @extension_ui = nil
  # Close/death bookkeeping, guarded by @pending_mutex except
  # @close_broadcast (@subscribers_mutex).
  @caller_closed = false
  @close_reason = nil
  @close_broadcast = nil
end

Instance Attribute Details

#bin ⇒ Object (readonly)

Returns the value of attribute bin.



37
38
39
 
# File 'lib/pi_agent/client.rb', line 37

def bin
  @bin
end

Class Method Details

.resolve_bin(override = nil) ⇒ Object

Raises:



39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/pi_agent/client.rb', line 39

def self.resolve_bin(override = nil)
  candidate = override || ENV["PI_BIN"] || DEFAULT_BIN
  path = which(candidate)
  return path if path

  raise BinaryNotFoundError, <<~MSG
    Could not find the `pi` binary on PATH (looked for #{candidate.inspect}).

    Install with: npm install -g @earendil-works/pi-coding-agent@#{PiAgent::SUPPORTED_PI_VERSION}
    Or set PI_BIN to an explicit path.
  MSG
end

.which(cmd) ⇒ Object 



52
53
54
55
56
57
58
59
60
61
 
# File 'lib/pi_agent/client.rb', line 52

def self.which(cmd)
  exts = ENV["PATHEXT"] ? ENV["PATHEXT"].split(";") : [""]
  ENV["PATH"].to_s.split(File::PATH_SEPARATOR).each do |dir|
    exts.each do |ext|
      candidate = File.join(dir, "#{cmd}#{ext}")
      return candidate if File.executable?(candidate) && !File.directory?(candidate)
    end
  end
  nil
end

Instance Method Details

#alive? ⇒ Boolean

Returns:

  • (Boolean) 


154
155
156
 
# File 'lib/pi_agent/client.rb', line 154

def alive?
  @transport&.alive? || false
end

#close ⇒ Object 



142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/pi_agent/client.rb', line 142

def close
  # A caller-initiated close is a clean shutdown: mark it first so the
  # transport teardown's own death notification is ignored and never
  # surfaces as a TransportClosedError.
  @pending_mutex.synchronize { @caller_closed = true }
  # Drain extension UI handler threads while the transport is still
  # open so their responses can still be written.
  @extension_ui&.shutdown
  @transport&.close
  reject_pending(ProtocolError.new("Transport closed before response"))
end

#notify(type, params = {}) ⇒ Object 



111
112
113
114
115
116
117
118
119
120
121
 
# File 'lib/pi_agent/client.rb', line 111

def notify(type, params = {})
  @pending_mutex.synchronize do
    raise TransportClosedError, @close_reason if @close_reason
  end
  payload = { type: type }.merge(params)
  begin
    @transport.write(payload)
  rescue StandardError => e
    raise close_error_or(e)
  end
end

#request(type, params = {}) ⇒ Object 



94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
 
# File 'lib/pi_agent/client.rb', line 94

def request(type, params = {})
  id = next_id
  future = Future.new
  @pending_mutex.synchronize do
    raise TransportClosedError, @close_reason if @close_reason

    @pending[id] = future
  end
  payload = { id: id, type: type }.merge(params)
  begin
    @transport.write(payload)
  rescue StandardError => e
    raise abandon_request(id, e)
  end
  future
end

#start ⇒ Object 



83
84
85
86
87
88
89
90
91
92
 
# File 'lib/pi_agent/client.rb', line 83

def start
  @transport = @transport_factory.call(**transport_callbacks)
  @extension_ui = ExtensionUI.new(
    writer: @transport,
    handler: @extension_ui_handler,
    on_error: @extension_ui_error_handler
  )
  @transport.start
  self
end

#subscribe(&block) ⇒ Object

Subscribers registered after a transport death still learn about it: the synthetic TRANSPORT_CLOSED_TYPE message is replayed to them once, outside any lock. (The death fanout snapshots subscribers atomically with recording the broadcast, so nobody sees it twice.)

Raises:

  • (ArgumentError) 


127
128
129
130
131
132
133
134
135
136
 
# File 'lib/pi_agent/client.rb', line 127

def subscribe(&block)
  raise ArgumentError, "subscribe requires a block" unless block

  replay = @subscribers_mutex.synchronize do
    @subscribers << block
    @close_broadcast
  end
  deliver([block], replay) if replay
  block
end

#unsubscribe(handle) ⇒ Object 



138
139
140
 
# File 'lib/pi_agent/client.rb', line 138

def unsubscribe(handle)
  @subscribers_mutex.synchronize { @subscribers.delete(handle) }
end