Class: Ask::SessionProtocol::Client

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

Overview

A thin client of the canonical Ask::SessionProtocol. Speaks NDJSON over an input/output pair (the spawned host's stdio, or a unix socket) and knows nothing else: no runtime, no sessions — the host owns all of it.

A reader thread routes incoming messages: responses are matched to their pending request by id and delivered with a ConditionVariable; everything else (session/event notifications, reverse requests) is pushed to a notification queue for the caller to drain.

Defined Under Namespace

Classes: ConnectionClosed, ProtocolError, RequestTimeout

Constant Summary collapse

CONNECTION_CLOSED =

Sentinel notification pushed when the host closes the connection, so waiters on #wait_notification unblock.

{ "method" => "connection.closed" }.freeze

Instance Method Summary collapse

Constructor Details

#initialize(input:, output:) ⇒ Client

Returns a new instance of Client.

Parameters:

  • input (IO) —

    NDJSON source (host stdout / socket)

  • output (IO) —

    NDJSON sink (host stdin / socket)



41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/ask/session_protocol/client.rb', line 41

def initialize(input:, output:)
  @input = input
  @output = output
  @pending = {}
  @next_id = 1
  @mutex = Mutex.new
  @condition = ConditionVariable.new
  @notifications = Queue.new
  @closed = false
  @reader = Thread.new { read_loop }
end

Instance Method Details

#abort(session_id) ⇒ Object 



153
154
155
 
# File 'lib/ask/session_protocol/client.rb', line 153

def abort(session_id)
  request("session/abort", { sessionId: session_id })
end

#approve(session_id, interaction_id) ⇒ Object 



165
166
167
 
# File 'lib/ask/session_protocol/client.rb', line 165

def approve(session_id, interaction_id)
  request("interaction/approve", { sessionId: session_id, interactionId: interaction_id })
end

#approve_all(session_id) ⇒ Object 



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

def approve_all(session_id)
  request("interaction/approve-all", { sessionId: session_id })
end

#close ⇒ Object

Close the client: close the sink (which ends the host's stdin, for spawned hosts) and stop the reader thread.



121
122
123
124
 
# File 'lib/ask/session_protocol/client.rb', line 121

def close
  @output.close rescue nil
  @reader&.kill rescue nil
end

#close_session(session_id) ⇒ Object 



157
158
159
 
# File 'lib/ask/session_protocol/client.rb', line 157

def close_session(session_id)
  request("session/close", { sessionId: session_id })
end

#closed? ⇒ Boolean

True when the host connection has closed.

Returns:

  • (Boolean) 


115
116
117
 
# File 'lib/ask/session_protocol/client.rb', line 115

def closed?
  @closed
end

#create_session(workspace_path: nil, mode: nil, model: nil, tools: nil, system_prompt: nil) ⇒ Object 



132
133
134
135
136
137
138
139
140
141
 
# File 'lib/ask/session_protocol/client.rb', line 132

def create_session(workspace_path: nil, mode: nil, model: nil, tools: nil, system_prompt: nil)
  params = {
    workspace: { workspacePath: workspace_path },
    mode: mode,
    model: model
  }.compact
  params[:tools] = tools if tools
  params[:systemPrompt] = system_prompt if system_prompt
  request("session/create", params)
end

#initialize!(client: { name: "ask-session-protocol", version: Ask::SessionProtocol::VERSION }) ⇒ Object

── Convenience wrappers over the canonical surface ────────────────



128
129
130
 
# File 'lib/ask/session_protocol/client.rb', line 128

def initialize!(client: { name: "ask-session-protocol", version: Ask::SessionProtocol::VERSION })
  request("initialize", { client: client })
end

#list_interactions(session_id) ⇒ Object 



161
162
163
 
# File 'lib/ask/session_protocol/client.rb', line 161

def list_interactions(session_id)
  request("interaction/list", { sessionId: session_id })
end

#plan_approve(session_id) ⇒ Object 



181
182
183
 
# File 'lib/ask/session_protocol/client.rb', line 181

def plan_approve(session_id)
  request("plan/approve", { sessionId: session_id })
end

#plan_reject(session_id) ⇒ Object 



185
186
187
 
# File 'lib/ask/session_protocol/client.rb', line 185

def plan_reject(session_id)
  request("plan/reject", { sessionId: session_id })
end

#read_workspace_state(session_id = nil) ⇒ Object 



189
190
191
192
 
# File 'lib/ask/session_protocol/client.rb', line 189

def read_workspace_state(session_id = nil)
  params = session_id ? { sessionId: session_id } : {}
  request("workspace/readState", params)
end

#reject(session_id, interaction_id) ⇒ Object 



169
170
171
 
# File 'lib/ask/session_protocol/client.rb', line 169

def reject(session_id, interaction_id)
  request("interaction/reject", { sessionId: session_id, interactionId: interaction_id })
end

#reject_all(session_id) ⇒ Object 



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

def reject_all(session_id)
  request("interaction/reject-all", { sessionId: session_id })
end

#request(method, params = {}, timeout: nil) ⇒ Hash

Send a request and wait for its response.

Parameters:

  • method (String) —

    canonical client → host method

  • params (Hash) (defaults to: {})
  • timeout (Numeric, nil) (defaults to: nil) —

    seconds to wait; nil waits forever

Returns:

  • (Hash) —

    the response result

Raises:



62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
 
# File 'lib/ask/session_protocol/client.rb', line 62

def request(method, params = {}, timeout: nil)
  raise ConnectionClosed, "connection is closed" if @closed

  id = next_id
  entry = { done: false, result: nil, error: nil }
  @mutex.synchronize { @pending[id] = entry }

  begin
    @output.puts(JSON.generate({ id: id, method: method, params: params }))
    @output.flush
  rescue Errno::EPIPE, IOError
    raise ConnectionClosed, "host closed the connection"
  end

  deadline = timeout && (Time.now + timeout)
  @mutex.synchronize do
    until entry[:done]
      remaining = deadline && (deadline - Time.now)
      raise RequestTimeout, "#{method} timed out after #{timeout}s" if remaining && remaining <= 0

      @condition.wait(@mutex, remaining || 1)
    end
  end
  raise entry[:error] if entry[:error]

  entry[:result]
ensure
  @mutex.synchronize { @pending.delete(id) } if id
end

#send(session_id, content, expected_turn_id: nil) ⇒ Object 



147
148
149
150
151
 
# File 'lib/ask/session_protocol/client.rb', line 147

def send(session_id, content, expected_turn_id: nil)
  params = { sessionId: session_id, content: content }
  params[:expectedTurnId] = expected_turn_id if expected_turn_id
  request("session/send", params)
end

#subscribe(session_id, after_seq: 0) ⇒ Object 



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

def subscribe(session_id, after_seq: 0)
  request("session/subscribe", { sessionId: session_id, afterSeq: after_seq })
end

#wait_notification(timeout: nil) ⇒ Object

Wait for the next notification (blocking, or with a timeout). Returns a Hash { "method" =>, "params" => }, or the CONNECTION_CLOSED sentinel when the host disconnected.

The sentinel is delivered to every waiter: several watchers may share one connection (the board watches every task of a project over a single client), and a single queued sentinel would wake only one of them — the rest would wait on an empty queue forever, keeping their runs' leases fresh so recovery never touches them.



102
103
104
105
106
107
108
109
110
111
112
 
# File 'lib/ask/session_protocol/client.rb', line 102

def wait_notification(timeout: nil)
  if @closed && @notifications.empty?
    return CONNECTION_CLOSED
  end

  return @notifications.pop unless timeout

  Timeout.timeout(timeout) { @notifications.pop }
rescue Timeout::Error
  nil
end