Class: Ask::ACP::Client

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

Overview

ACP client that connects to a coding agent over stdio using JSON-RPC 2.0.

Spawns the agent CLI as a subprocess and communicates via stdin/stdout. Supports all ACP methods: initialize, session/new, session/prompt, etc.

Examples:

client = Ask::ACP::Client.new(command: ["codex", "acp"])
client.start
result = client.initialize!(name: "askoda", version: "0.1.0")
session = client.session_new(cwd: "/tmp")
client.session_prompt(session[:id], "Hello") { |event| puts event }
client.stop

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(command:, request_timeout: 30.0) ⇒ Client

Returns a new instance of Client.



23
24
25
26
27
28
29
30
31
32
33
 
# File 'lib/ask/acp/client.rb', line 23

def initialize(command:, request_timeout: 30.0)
  @command = command
  @request_timeout = request_timeout
  @stdin = nil
  @next_id = 1
  @pending = {}
  @event_handlers = []
  @running = false
  @mutex = Mutex.new
  @initialized = false
end

Instance Attribute Details

#command ⇒ Object (readonly)

Returns the value of attribute command.



21
22
23
 
# File 'lib/ask/acp/client.rb', line 21

def command
  @command
end

#running ⇒ Object (readonly)

Returns the value of attribute running.



21
22
23
 
# File 'lib/ask/acp/client.rb', line 21

def running
  @running
end

Instance Method Details

#authenticate(token:, scheme: "bearer") ⇒ Object

Authenticate with the agent (if required).



95
96
97
98
99
 
# File 'lib/ask/acp/client.rb', line 95

def authenticate(token:, scheme: "bearer")
  request(Protocol::AGENT_METHODS[:authenticate], {
    credentials: { scheme: scheme, token: token }
  })
end

#initialize!(client_name:, client_version:, capabilities: {}) ⇒ Object

Initialize handshake. Must be called first.



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

def initialize!(client_name:, client_version:, capabilities: {})
  result = request(Protocol::AGENT_METHODS[:initialize], {
    protocolVersion: Protocol::PROTOCOL_VERSION,
    clientInfo: { name: client_name, version: client_version },
    capabilities: capabilities
  })
  @initialized = true
  result
end

#on_notification(&handler) ⇒ Object

Register a handler for incoming notifications (client methods from agent).



77
78
79
 
# File 'lib/ask/acp/client.rb', line 77

def on_notification(&handler)
  @mutex.synchronize { @event_handlers << handler }
end

#request(method, params = nil, timeout: nil) ⇒ Object

Send a JSON-RPC request and wait for the response.



202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
 
# File 'lib/ask/acp/client.rb', line 202

def request(method, params = nil, timeout: nil)
  ensure_running!
  future = { done: false, result: nil, error: nil, condition: ConditionVariable.new }

  @mutex.synchronize do
    id = @next_id
    @next_id += 1
    @pending[id] = future
    write(Protocol.build_request(method, params, id: id))
  end

  timeout ||= @request_timeout
  deadline = Time.now + timeout

  @mutex.synchronize do
    until future[:done]
      remaining = deadline - Time.now
      raise TimeoutError, "Request timed out after #{timeout}s" if remaining <= 0
      future[:condition].wait(@mutex, remaining)
    end
  end

  raise future[:error] if future[:error]
  future[:result]
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


72
73
74
 
# File 'lib/ask/acp/client.rb', line 72

def running?
  @running && @wait_thr&.alive?
end

#session_cancel(session_id) ⇒ Object

Cancel the current prompt execution.



172
173
174
175
 
# File 'lib/ask/acp/client.rb', line 172

def session_cancel(session_id)
  ensure_initialized!
  request(Protocol::AGENT_METHODS[:session_cancel], { sessionId: session_id })
end

#session_close(session_id) ⇒ Object

Close a session.



142
143
144
145
 
# File 'lib/ask/acp/client.rb', line 142

def session_close(session_id)
  ensure_initialized!
  request(Protocol::AGENT_METHODS[:session_close], { sessionId: session_id })
end

#session_fork(session_id) ⇒ Object

Fork a session from an existing one.



128
129
130
131
132
 
# File 'lib/ask/acp/client.rb', line 128

def session_fork(session_id)
  ensure_initialized!
  result = request(Protocol::AGENT_METHODS[:session_fork], { sessionId: session_id })
  normalize_session(result)
end

#session_list(cwd: nil) ⇒ Object

List sessions.



119
120
121
122
123
124
125
 
# File 'lib/ask/acp/client.rb', line 119

def session_list(cwd: nil)
  ensure_initialized!
  params = {}
  params[:cwd] = cwd if cwd
  result = request(Protocol::AGENT_METHODS[:session_list], params)
  (result["sessions"] || result[:sessions] || []).map { |s| normalize_session(s) }
end

#session_load(session_id) ⇒ Object

Load an existing session by ID.



112
113
114
115
116
 
# File 'lib/ask/acp/client.rb', line 112

def session_load(session_id)
  ensure_initialized!
  result = request(Protocol::AGENT_METHODS[:session_load], { sessionId: session_id })
  normalize_session(result)
end

#session_new(cwd: ".", model: nil, tools: nil) ⇒ Object

Create a new session in the given working directory. Returns { id:, status:, created_at: }.



103
104
105
106
107
108
109
 
# File 'lib/ask/acp/client.rb', line 103

def session_new(cwd: ".", model: nil, tools: nil)
  ensure_initialized!
  params = { cwd: cwd, mcpServers: [] }
  params[:model] = model if model
  result = request(Protocol::AGENT_METHODS[:session_new], params)
  normalize_session(result)
end

#session_prompt(session_id, prompt, timeout: nil) {|Hash| ... } ⇒ Hash

Send a prompt to a session and stream events via the block. The prompt is automatically wrapped in a ContentBlock array if given as a plain string.

Yields:

  • (Hash) —

    event with :method and :params

Returns:

  • (Hash) —

    the final session/prompt response



153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
 
# File 'lib/ask/acp/client.rb', line 153

def session_prompt(session_id, prompt, timeout: nil, &block)
  ensure_initialized!
  prompt_blocks = prompt.is_a?(Array) ? prompt : [{ type: "text", text: prompt.to_s }]
  params = { sessionId: session_id, prompt: prompt_blocks }
  if block
    # Register a temporary handler for streaming events
    handler = ->(msg) { block.call(method: msg["method"], params: msg["params"] || {}) if msg["method"] }
    on_notification(&handler)
    begin
      request(Protocol::AGENT_METHODS[:session_prompt], params, timeout: timeout)
    ensure
      @mutex.synchronize { @event_handlers.delete(handler) }
    end
  else
    request(Protocol::AGENT_METHODS[:session_prompt], params, timeout: timeout)
  end
end

#session_resume(session_id) ⇒ Object

Resume a session.



135
136
137
138
139
 
# File 'lib/ask/acp/client.rb', line 135

def session_resume(session_id)
  ensure_initialized!
  result = request(Protocol::AGENT_METHODS[:session_resume], { sessionId: session_id })
  normalize_session(result)
end

#session_set_config_option(session_id, key, value) ⇒ Object

Set a config option on a session.



178
179
180
181
182
183
 
# File 'lib/ask/acp/client.rb', line 178

def session_set_config_option(session_id, key, value)
  ensure_initialized!
  request(Protocol::AGENT_METHODS[:session_set_config_option], {
    sessionId: session_id, key: key, value: value
  })
end

#session_set_mode(session_id, mode) ⇒ Object

Set the mode on a session.



186
187
188
189
190
191
 
# File 'lib/ask/acp/client.rb', line 186

def session_set_mode(session_id, mode)
  ensure_initialized!
  request(Protocol::AGENT_METHODS[:session_set_mode], {
    sessionId: session_id, mode: mode
  })
end

#session_set_model(session_id, provider:, model:) ⇒ Object

Set the model on a session.



194
195
196
197
198
199
 
# File 'lib/ask/acp/client.rb', line 194

def session_set_model(session_id, provider:, model:)
  ensure_initialized!
  request(Protocol::AGENT_METHODS[:session_set_model], {
    sessionId: session_id, provider: provider, model: model
  })
end

#start ⇒ Object

Start the agent subprocess and begin reading stdout.



36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/ask/acp/client.rb', line 36

def start
  @mutex.synchronize do
    return if @running
    @stdin, stdout, stderr, @wait_thr = Open3.popen3(*@command)

    @stdout_thread = Thread.new(stdout) do |io|
      io.each_line do |line|
        line = line.strip
        next if line.empty?
        handle_message(Protocol.parse(line))
      end
    end

    # Drain stderr to prevent deadlocks
    Thread.new(stderr) { |io| io.each_line { |_| } }

    @running = true
  end
end

#stop ⇒ Object

Stop the agent subprocess.



57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/ask/acp/client.rb', line 57

def stop
  @mutex.synchronize do
    return unless @running
    @running = false
    @initialized = false
    @stdin&.close rescue nil
    @stdout_thread&.join(3) rescue nil
    Process.kill("TERM", @wait_thr.pid) rescue nil
    @wait_thr&.join(5) rescue nil
    @stdin = nil
    @pending.each_value { |f| f[:error] = Error.new("process exited"); f[:done] = true; f[:condition].signal }
    @pending.clear
  end
end