Class: Ask::AppServer::Server

Inherits:
Object
  • Object
show all
Defined in:
lib/ask/app_server/server.rb

Overview

JSON-RPC protocol engine implementing the canonical Ask::SessionProtocol. The session host: owns agent sessions behind one versioned contract that every client (terminal TUI, web console, bots, IDE) speaks.

The engine is transport-agnostic: messages arrive per-connection (#dispatch) and responses/notifications are written back to the requesting connection. Event delivery is cursor-based — each connection tracks the last delivered seq per subscribed session (Connection), so any number of clients can attach to the same sessions. Two transports use the engine:

Server#start     — stdio mode (one Connection over stdin/stdout)
SocketServer     — unix socket mode (a Connection per client)

Client → host methods (see Ask::SessionProtocol::Methods):

ping, initialize, session/create, session/list, session/resume,
session/subscribe, session/events, session/send, session/abort,
session/close, session/artifacts, session/artifact/get,
interaction/list, interaction/approve, interaction/reject,
interaction/approve-all, interaction/reject-all,
interaction/respond, plan/approve, plan/reject, workspace/readState

Host → client notifications:

session/event — carries a canonical event envelope {type, seq, payload}

The engine also handles incoming responses to its outgoing reverse requests (interaction/requestPermission, interaction/requestUserInput — the app-server interop surface).

Constant Summary collapse

HOST_CAPABILITIES =

Host capabilities advertised in initialize. The host emits every canonical event except file.changed today.

(Ask::SessionProtocol::Methods::CAPABILITIES - %w[fileEvents]).freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(session_manager: nil) ⇒ Server

Returns a new instance of Server.



43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/ask/app_server/server.rb', line 43

def initialize(session_manager: nil)
  @session_manager = session_manager || SessionManager.new
  @running = false
  @started_at = nil
  @connections = []
  @connections_mutex = Mutex.new
  @response_handlers = {}  # outgoing request_id => Proc
  @outgoing_id = 0
  @logger = Logger.new($stdout, level: ENV["DEBUG"] ? Logger::DEBUG : Logger::WARN)

  # Register the protocol method handlers
  @handlers = {}
  register_default_handlers
end

Instance Attribute Details

#session_manager ⇒ Object (readonly)

Returns the value of attribute session_manager.



58
59
60
 
# File 'lib/ask/app_server/server.rb', line 58

def session_manager
  @session_manager
end

Instance Method Details

#add_connection(connection) ⇒ Object

Register a connection for dispatch and event delivery.



68
69
70
71
 
# File 'lib/ask/app_server/server.rb', line 68

def add_connection(connection)
  @connections_mutex.synchronize { @connections << connection }
  connection
end

#connections ⇒ Object

All live connections (snapshot).



63
64
65
 
# File 'lib/ask/app_server/server.rb', line 63

def connections
  @connections_mutex.synchronize { @connections.dup }
end

#dispatch(msg, connection) ⇒ Object

Process one JSON-RPC message from a connection. Requests are validated against the contract before dispatch; responses and errors are written back to connection.

Parameters:

  • msg (Hash) —

    parsed JSON-RPC message

  • connection (Connection) —

    the sending connection



118
119
120
121
122
123
124
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/ask/app_server/server.rb', line 118

def dispatch(msg, connection)
  id = msg["id"] || msg[:id]

  # Check if this is a response to an outgoing request.
  # A response has an id and a result (or error), but no method.
  if id && !msg.key?("method") && !msg.key?(:method)
    if msg.key?("result") || msg.key?(:result)
      handle_incoming_response(id, msg["result"] || msg[:result])
      return
    elsif msg.key?("error") || msg.key?(:error)
      handle_incoming_response(id, nil, msg["error"] || msg[:error])
      return
    end
  end

  method = msg["method"] || msg[:method]
  params = (msg["params"] || msg[:params] || {})
  params = params.transform_keys(&:to_s) if params.is_a?(Hash)

  unless method
    send_error(id, -32600, "Method not specified", connection) if id
    return
  end

  handler_block = @handlers[method]
  unless handler_block
    send_error(id, -32601, "Method not found: #{method}", connection) if id
    return
  end

  begin
    validate_request!(method, params)
    result = handler_block.call(params, id, connection)
    send_result(id, result, connection) if id
  rescue Ask::AppServer::SessionNotFound => e
    send_error(id, -32004, e.message, connection) if id
  rescue Ask::AppServer::SessionAlreadyExists => e
    send_error(id, -32005, e.message, connection) if id
  rescue Ask::AppServer::InteractionNotFound => e
    send_error(id, -32006, e.message, connection) if id
  rescue Ask::AppServer::PlanNotFound => e
    send_error(id, -32007, e.message, connection) if id
  rescue Ask::AppServer::InvalidRequest => e
    send_error(id, -32602, e.message, connection) if id
  rescue ArgumentError => e
    # Contract validation failures (invalid params)
    send_error(id, -32602, e.message, connection) if id
  rescue => e
    @logger.error("Handler error for #{method}: #{e.message}")
    send_error(id, -32603, "Internal error: #{e.message}", connection) if id
  end
end

#push_pending ⇒ Object

One delivery pass: for every subscribed connection, deliver the events its sessions have produced since the connection's cursor, then advance the cursor. Cursor-based, so each client receives exactly the events after its own seq (replay on subscribe).



191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
 
# File 'lib/ask/app_server/server.rb', line 191

def push_pending
  connections.each do |connection|
    connection.subscriptions.keys.each do |session_id|
      adapter = @session_manager.get(session_id)
      next unless adapter

      events = adapter.events_after(connection.cursor(session_id))
      next if events.empty?

      events.each do |ev|
        connection.write({ method: "session/event", params: { event: ev.to_h } })
      end
      connection.advance(session_id, events.last.seq)
    end
  end
rescue => e
  @logger.debug("Push error: #{e.message}") if ENV["DEBUG"]
end

#remove_connection(connection) ⇒ Object

Deregister a connection (client disconnected).



74
75
76
77
 
# File 'lib/ask/app_server/server.rb', line 74

def remove_connection(connection)
  @connections_mutex.synchronize { @connections.delete(connection) }
  connection.close
end

#running? ⇒ Boolean

Whether the server is running.

Returns:

  • (Boolean) 


106
107
108
 
# File 'lib/ask/app_server/server.rb', line 106

def running?
  @running
end

#send_error(id, code, message, connection = nil) ⇒ Object 



216
217
218
219
 
# File 'lib/ask/app_server/server.rb', line 216

def send_error(id, code, message, connection = nil)
  response = { id: id, error: { code: code, message: message } }
  connection ? connection.write(response) : write_line(response)
end

#send_notification(method, params, connection = nil) ⇒ Object 



221
222
223
224
 
# File 'lib/ask/app_server/server.rb', line 221

def send_notification(method, params, connection = nil)
  msg = { method: method, params: params }
  connection ? connection.write(msg) : write_line(msg)
end

#send_request(method, params, connection = nil, &block) ⇒ Object

Send an outgoing JSON-RPC request to a client (reverse request). If a block is given, it will be called with (result, error) when the client responds. Defaults to the stdio transport ($stdout).



174
175
176
177
178
179
180
181
182
183
 
# File 'lib/ask/app_server/server.rb', line 174

def send_request(method, params, connection = nil, &block)
  id = next_outgoing_id
  @response_handlers[id] = block if block
  if connection
    connection.write({ id: id, method: method, params: params })
  else
    write_line({ id: id, method: method, params: params })
  end
  id
end

#send_result(id, result, connection = nil) ⇒ Object

── Response/notification helpers ──────────────────────────────────



212
213
214
 
# File 'lib/ask/app_server/server.rb', line 212

def send_result(id, result, connection = nil)
  connection ? connection.write({ id: id, result: result }) : write_line({ id: id, result: result })
end

#start ⇒ Object

Start the stdio transport: one Connection over stdin/stdout, a reader thread dispatching messages, and a pusher delivering events. Blocks until the input closes or #stop is called.



84
85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/ask/app_server/server.rb', line 84

def start
  @running = true
  @started_at = Time.now
  connection = add_connection(Connection.new($stdin, $stdout))

  @reader = Thread.new do
    reader_loop(connection)
  end
  @pusher = Thread.new { pusher_loop }

  @reader.join
  stop
end

#stop ⇒ Object

Stop the server.



99
100
101
102
103
 
# File 'lib/ask/app_server/server.rb', line 99

def stop
  @running = false
  @reader&.kill rescue nil
  @pusher&.kill rescue nil
end

#write_line(msg) ⇒ Object 



226
227
228
229
 
# File 'lib/ask/app_server/server.rb', line 226

def write_line(msg)
  $stdout.puts(JSON.generate(msg))
  $stdout.flush
end