Class: Kward::RPC::SessionManager

Inherits:
Object
  • Object
show all
Defined in:
lib/kward/rpc/session_manager.rb

Defined Under Namespace

Classes: RpcSession, Turn

Constant Summary collapse

RECENT_EVENT_LIMIT = 
1_000
RPC_ATTACHMENT_MAX_BYTES = 
10 * 1024 * 1024
RPC_IMAGE_MIME_TYPES = 
["image/png", "image/jpeg", "image/gif", "image/webp"].freeze
STREAMING_BEHAVIORS = 
["newTurn", "followUp", "steer"].freeze
WORKER_STOP = 
Object.new.freeze

Instance Method Summary collapse

Constructor Details

#initialize(server:, client: Client.new, config_dir: ConfigFiles.config_dir, context_usage: ContextUsage.new) ⇒ SessionManager

Returns a new instance of SessionManager.



41
42
43
44
45
46
47
48
49
 
# File 'lib/kward/rpc/session_manager.rb', line 41

def initialize(server:, client: Client.new, config_dir: ConfigFiles.config_dir, context_usage: ContextUsage.new)
  @server = server
  @client = client
  @config_dir = config_dir
  @context_usage = context_usage
  @sessions = {}
  @turns = {}
  @mutex = Mutex.new
end

Instance Method Details

#answer_question(session_id:, question_request_id:, answers:) ⇒ Object 



255
256
257
258
259
 
# File 'lib/kward/rpc/session_manager.rb', line 255

def answer_question(session_id:, question_request_id:, answers:)
  rpc_session = fetch_session(session_id)
  rpc_session.prompt.answer(question_request_id, answers)
  { ok: true }
end

#available_models ⇒ Object 



357
358
359
360
361
362
363
 
# File 'lib/kward/rpc/session_manager.rb', line 357

def available_models
  models = @client.respond_to?(:available_models) ? Array(@client.available_models) : []
  normalized = models.map { |model| normalize_model(model) }
  current = current_model
  normalized << current if normalized.none? { |model| model[:provider] == current[:provider] && model[:id] == current[:id] }
  normalized
end

#cancel_turn(turn_id:) ⇒ Object 



231
232
233
234
235
236
237
238
239
240
 
# File 'lib/kward/rpc/session_manager.rb', line 231

def cancel_turn(turn_id:)
  turn = fetch_turn(turn_id)
  turn.cancel_requested = true
  turn.cancellation&.cancel!
  emit_turn_event(turn, "turnCancelRequested", {})
  if turn.status == "queued"
    finish_turn(turn, "canceled")
  end
  turn_payload(turn)
end

#cleanup_unused_sessions ⇒ Object 



183
184
185
186
187
188
189
190
191
 
# File 'lib/kward/rpc/session_manager.rb', line 183

def cleanup_unused_sessions
  rpc_sessions = @mutex.synchronize { @sessions.values.dup }
  rpc_sessions.reverse_each do |rpc_session|
    next unless session_idle?(rpc_session)

    close_rpc_session(rpc_session)
  end
  { closed: true }
end

#clone_session(session_id:) ⇒ Object 



98
99
100
101
102
103
104
105
 
# File 'lib/kward/rpc/session_manager.rb', line 98

def clone_session(session_id:)
  source = fetch_session(session_id)
  session, conversation = source.store.create_independent_from_conversation(source.conversation, parent_session: source.session)
  rpc_session = build_rpc_session(source.store, session, conversation, source.workspace_root)
  remember_session(rpc_session)
  cleanup_other_unused_sessions(rpc_session)
  session_payload(rpc_session)
end

#close_session(session_id:) ⇒ Object 



177
178
179
180
181
 
# File 'lib/kward/rpc/session_manager.rb', line 177

def close_session(session_id:)
  rpc_session = fetch_session(session_id)
  close_rpc_session(rpc_session)
  { closed: true }
end

#compact_session(session_id:, custom_instructions: "") ⇒ Object 



107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
 
# File 'lib/kward/rpc/session_manager.rb', line 107

def compact_session(session_id:, custom_instructions: "")
  rpc_session = fetch_session(session_id)
  emit_session_event(rpc_session, "compactionStart", {})
  result = Compactor.new(conversation: rpc_session.conversation, client: @client, settings: compaction_settings).compact(custom_instructions: custom_instructions)
  payload = {
    summary: result.summary,
    firstKeptEntryId: result.first_kept_entry_id,
    tokensBefore: result.tokens_before,
    details: result.details
  }.compact
  emit_session_event(rpc_session, "compactionEnd", { result: payload, aborted: false, willRetry: false, errorMessage: nil })
  payload
rescue StandardError => e
  emit_session_event(rpc_session, "compactionEnd", { result: nil, aborted: true, willRetry: false, errorMessage: e.message }) if rpc_session
  raise e
end

#create_session(workspace_root: Dir.pwd, name: nil) ⇒ Object 



51
52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/kward/rpc/session_manager.rb', line 51

def create_session(workspace_root: Dir.pwd, name: nil)
  workspace_root = validate_workspace_root(workspace_root)
  store = SessionStore.new(config_dir: @config_dir, cwd: workspace_root)
  conversation = new_conversation(workspace_root: workspace_root)
  session = store.create(model: conversation.model, reasoning_effort: conversation.reasoning_effort)
  session.rename(name) unless name.to_s.strip.empty?
  session.attach(conversation)
  rpc_session = build_rpc_session(store, session, conversation, workspace_root)
  remember_session(rpc_session)
  cleanup_other_unused_sessions(rpc_session)
  session_payload(rpc_session)
end

#current_model ⇒ Object 



370
371
372
373
374
375
 
# File 'lib/kward/rpc/session_manager.rb', line 370

def current_model
  provider = @client.respond_to?(:current_provider) ? @client.current_provider : nil
  model = @client.respond_to?(:current_model) ? @client.current_model : nil
  context_window = @client.respond_to?(:current_context_window) ? @client.current_context_window : nil
  normalize_model(provider: provider, id: model, model: model, contextWindow: context_window, current: true)
end

#delete_session(session_id:) ⇒ Object 



169
170
171
172
173
174
175
 
# File 'lib/kward/rpc/session_manager.rb', line 169

def delete_session(session_id:)
  rpc_session = fetch_session(session_id)
  path = rpc_session.session.path
  close_session(session_id: session_id)
  File.delete(path) if File.exist?(path)
  { deleted: true, path: path }
end

#export_session(session_id:, path: nil, format: nil) ⇒ Object 



160
161
162
163
164
165
166
167
 
# File 'lib/kward/rpc/session_manager.rb', line 160

def export_session(session_id:, path: nil, format: nil)
  rpc_session = fetch_session(session_id)
  format = export_format(format)
  path = export_path(rpc_session, path, format)
  content = export_content(rpc_session.conversation, format)
  File.write(path, content)
  { path: path, format: format }
end

#fork_messages(session_id:) ⇒ Object 



124
125
126
127
128
129
130
131
132
133
 
# File 'lib/kward/rpc/session_manager.rb', line 124

def fork_messages(session_id:)
  rpc_session = fetch_session(session_id)
  {
    messages: entry_messages(rpc_session.conversation).each_with_index.filter_map do |message, index|
      next unless message_role(message) == "user"

      { entryId: entry_id(index), text: display_message_text(message) }
    end
  }
end

#fork_session(session_id:, entry_id:) ⇒ Object

Raises:

  • (ArgumentError) 


135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/kward/rpc/session_manager.rb', line 135

def fork_session(session_id:, entry_id:)
  source = fetch_session(session_id)
  index = entry_index(entry_id)
  messages = entry_messages(source.conversation)
  selected = messages[index] || raise(ArgumentError, "Unknown fork entryId: #{entry_id}")
  raise ArgumentError, "Entry is not forkable: #{entry_id}" unless message_role(selected) == "user"

  session, conversation = source.store.create_independent_from_messages(
    messages[0...index],
    model: source.conversation.model,
    reasoning_effort: source.conversation.reasoning_effort,
    parent_session: source.session
  )

  # ensure forked sessions retain the original persona context
  rpc_session = build_rpc_session(source.store, session, conversation, source.workspace_root)
  remember_session(rpc_session)
  cleanup_other_unused_sessions(rpc_session)
  {
    session: session_payload(rpc_session),
    text: full_message_text(selected),
    cancelled: false
  }
end

#in_flight_steer_supported? ⇒ Boolean

Returns:

  • (Boolean) 


377
378
379
 
# File 'lib/kward/rpc/session_manager.rb', line 377

def in_flight_steer_supported?
  supports_in_flight_steer?
end

#list_sessions(workspace_root: Dir.pwd, limit: 20) ⇒ Object 



82
83
84
85
86
87
88
89
90
 
# File 'lib/kward/rpc/session_manager.rb', line 82

def list_sessions(workspace_root: Dir.pwd, limit: 20)
  root = validate_workspace_root(workspace_root)
  store = SessionStore.new(config_dir: @config_dir, cwd: root)
  limit = limit.to_i <= 0 ? 20 : limit.to_i
  store.recent_tree(limit: limit + active_session_count(root))
       .reject { |info| active_empty_unnamed_session_info?(info, root) }
       .first(limit)
       .map { |info| session_info_payload(info, workspace_root: root) }
end

#memory_add(text:, scope: nil, tags: []) ⇒ Object 



302
303
304
 
# File 'lib/kward/rpc/session_manager.rb', line 302

def memory_add(text:, scope: nil, tags: [])
  { memory: memory_manager.add_soft(text, scope: scope || "global", tags: tags) }
end

#memory_add_core(text:, scope: nil, tags: []) ⇒ Object 



306
307
308
 
# File 'lib/kward/rpc/session_manager.rb', line 306

def memory_add_core(text:, scope: nil, tags: [])
  { memory: memory_manager.add_core(text, scope: scope || "global", tags: tags) }
end

#memory_auto_summary_disable ⇒ Object 



293
294
295
296
 
# File 'lib/kward/rpc/session_manager.rb', line 293

def memory_auto_summary_disable
  memory_manager.auto_summary_disable
  { autoSummary: false }
end

#memory_auto_summary_enable ⇒ Object 



288
289
290
291
 
# File 'lib/kward/rpc/session_manager.rb', line 288

def memory_auto_summary_enable
  memory_manager.auto_summary_enable
  { autoSummary: true }
end

#memory_disable ⇒ Object 



283
284
285
286
 
# File 'lib/kward/rpc/session_manager.rb', line 283

def memory_disable
  memory_manager.disable
  { enabled: false }
end

#memory_enable ⇒ Object 



278
279
280
281
 
# File 'lib/kward/rpc/session_manager.rb', line 278

def memory_enable
  memory_manager.enable
  { enabled: true }
end

#memory_forget(id:) ⇒ Object 



310
311
312
 
# File 'lib/kward/rpc/session_manager.rb', line 310

def memory_forget(id:)
  { forgotten: memory_manager.forget_memory(id) }
end

#memory_inspect ⇒ Object 



318
319
320
 
# File 'lib/kward/rpc/session_manager.rb', line 318

def memory_inspect
  memory_manager.inspect_memory
end

#memory_list(include_inactive: false) ⇒ Object 



298
299
300
 
# File 'lib/kward/rpc/session_manager.rb', line 298

def memory_list(include_inactive: false)
  memory_manager.list(include_inactive: include_inactive)
end

#memory_manager ⇒ Object 



269
270
271
 
# File 'lib/kward/rpc/session_manager.rb', line 269

def memory_manager
  Memory::Manager.for_config_dir(@config_dir)
end

#memory_promote(id:) ⇒ Object 



314
315
316
 
# File 'lib/kward/rpc/session_manager.rb', line 314

def memory_promote(id:)
  { memory: memory_manager.promote_soft_to_core(id) }
end

#memory_status ⇒ Object 



273
274
275
276
 
# File 'lib/kward/rpc/session_manager.rb', line 273

def memory_status
  manager = memory_manager
  { enabled: manager.enabled?, autoSummary: manager.auto_summary_enabled?, paths: manager.paths }
end

#memory_summarize(session_id:) ⇒ Object 



330
331
332
333
334
335
 
# File 'lib/kward/rpc/session_manager.rb', line 330

def memory_summarize(session_id:)
  rpc_session = fetch_session(session_id)
  records = memory_manager.summarize_conversation(rpc_session.conversation, client: @client)
  persist_memory_state(rpc_session)
  { memories: records }
end

#memory_why(session_id: nil) ⇒ Object 



322
323
324
325
326
327
328
 
# File 'lib/kward/rpc/session_manager.rb', line 322

def memory_why(session_id: nil)
  if session_id
    rpc_session = fetch_session(session_id)
    return rpc_session.conversation.last_memory_retrieval || memory_manager.explain_retrieval
  end
  memory_manager.explain_retrieval
end

#openrouter_catalog ⇒ Object 



365
366
367
368
 
# File 'lib/kward/rpc/session_manager.rb', line 365

def openrouter_catalog
  models = @client.respond_to?(:openrouter_catalog) ? Array(@client.openrouter_catalog) : []
  models.map { |model| normalize_model(model) }
end

#plugin_commands ⇒ Object 



353
354
355
 
# File 'lib/kward/rpc/session_manager.rb', line 353

def plugin_commands
  plugin_registry.commands
end

#refresh_client_config ⇒ Object 



449
450
451
452
453
 
# File 'lib/kward/rpc/session_manager.rb', line 449

def refresh_client_config
  @client.reload_config if @client.respond_to?(:reload_config)
  refresh_session_runtime_contexts
  refresh_session_tool_registries
end

#rename_session(session_id:, name:) ⇒ Object 



92
93
94
95
96
 
# File 'lib/kward/rpc/session_manager.rb', line 92

def rename_session(session_id:, name:)
  rpc_session = fetch_session(session_id)
  rpc_session.session.rename(name)
  session_payload(rpc_session)
end

#resume_session(path:, workspace_root: nil) ⇒ Object 



64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/kward/rpc/session_manager.rb', line 64

def resume_session(path:, workspace_root: nil)
  root = validate_workspace_root(workspace_root || Dir.pwd)
  store = SessionStore.new(config_dir: @config_dir, cwd: root)
  location = store.session_location(path)
  root = validate_workspace_root(location[:cwd])
  store = SessionStore.new(config_dir: @config_dir, cwd: root)
  session, conversation = store.load(
    location[:path],
    workspace: Workspace.new(root: root),
    model: current_model_id,
    reasoning_effort: current_reasoning_effort
  )
  rpc_session = build_rpc_session(store, session, conversation, root)
  remember_session(rpc_session)
  cleanup_other_unused_sessions(rpc_session)
  session_payload(rpc_session)
end

#run_command(session_id:, command:, arguments: "") ⇒ Object 



261
262
263
264
265
266
267
 
# File 'lib/kward/rpc/session_manager.rb', line 261

def run_command(session_id:, command:, arguments: "")
  name = command.to_s.delete_prefix("/")
  return { ok: false, error: "unsupported", reason: "notImplemented" } if name == "crew"
  return { ok: false, error: "unsupported", reason: "clientClipboardOwnedByUi" } if name == "copy"

  run_plugin_command(session_id: session_id, command: name, arguments: arguments)
end

#run_plugin_command(session_id:, command:, arguments: "") ⇒ Object 



337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
 
# File 'lib/kward/rpc/session_manager.rb', line 337

def run_plugin_command(session_id:, command:, arguments: "")
  rpc_session = fetch_session(session_id)
  command = plugin_registry.command_for(command.to_s.delete_prefix("/")) || raise(ArgumentError, "Unknown plugin command: #{command}")
  output = []
  context = PluginRegistry::Context.new(
    conversation: rpc_session.conversation,
    args: arguments.to_s,
    session: rpc_session.session,
    workspace_root: rpc_session.workspace_root,
    say_callback: lambda { |message| output << message.to_s }
  )
  result = command.handler.call(arguments.to_s, context)
  output = rpc_session.plugin_output.shift(rpc_session.plugin_output.length) + output
  { command: command.name, output: output, result: result.nil? ? nil : result.to_s }
end

#runtime_state(session_id:) ⇒ Object 



381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
 
# File 'lib/kward/rpc/session_manager.rb', line 381

def runtime_state(session_id:)
  rpc_session = fetch_session(session_id)
  model = current_model
  compaction_settings = self.compaction_settings
  auto_compaction_reserve_tokens = compaction_reserve_tokens(
    context_window: model[:contextWindow],
    compaction_settings: compaction_settings
  )
  session = session_payload(rpc_session)
  pending_count = pending_turn_count(rpc_session.id)
  {
    model: model,
    thinkingLevel: model[:reasoningEffort],
    isStreaming: streaming?(rpc_session),
    isCompacting: false,
    steeringMode: supports_in_flight_steer? ? "in-flight" : "one-at-a-time",
    followUpMode: "one-at-a-time",
    sessionFile: session[:path],
    sessionId: session[:persistentId],
    rpcSessionId: session[:id],
    persistentSessionId: session[:persistentId],
    sessionName: session[:name],
    autoCompactionEnabled: compaction_settings.enabled,
    autoCompactionReserveTokens: auto_compaction_reserve_tokens,
    autoRetryEnabled: false,
    defaultProvider: model[:provider],
    defaultModel: default_model_label(model),
    defaultThinkingLevel: model[:reasoningEffort],
    hideThinkingBlock: false,
    quietStartup: false,
    transport: "kward-rpc",
    imageAutoResize: false,
    blockImages: false,
    enabledModels: [],
    enableSkillCommands: true,
    messageCount: message_count(rpc_session.conversation),
    pendingMessageCount: pending_count
  }.compact
end

#runtime_stats(session_id:) ⇒ Object 



421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
 
# File 'lib/kward/rpc/session_manager.rb', line 421

def runtime_stats(session_id:)
  rpc_session = fetch_session(session_id)
  session = session_payload(rpc_session)
  counts = message_stats(rpc_session.conversation)
  model = current_model
  compaction_settings = self.compaction_settings
  auto_compaction_reserve_tokens = compaction_reserve_tokens(
    context_window: model[:contextWindow],
    compaction_settings: compaction_settings
  )
  {
    sessionFile: session[:path],
    sessionId: session[:persistentId],
    rpcSessionId: session[:id],
    persistentSessionId: session[:persistentId],
    sessionName: session[:name],
    userMessages: counts[:userMessages],
    assistantMessages: counts[:assistantMessages],
    toolCalls: counts[:toolCalls],
    toolResults: counts[:toolResults],
    totalMessages: counts[:totalMessages],
    usingSubscription: model[:provider] == "Codex",
    autoCompactionEnabled: compaction_settings.enabled,
    autoCompactionReserveTokens: auto_compaction_reserve_tokens,
    contextUsage: context_usage(rpc_session, model)
  }.compact
end

#session_modified_at(session) ⇒ Object 



470
471
472
 
# File 'lib/kward/rpc/session_manager.rb', line 470

def session_modified_at(session)
  File.exist?(session.path) ? File.mtime(session.path) : nil
end

#session_payload(rpc_session) ⇒ Object 



455
456
457
458
459
460
461
462
463
464
465
466
467
468
 
# File 'lib/kward/rpc/session_manager.rb', line 455

def session_payload(rpc_session)
  {
    id: rpc_session.id,
    persistentId: rpc_session.session.id,
    path: rpc_session.session.path,
    workspaceRoot: rpc_session.workspace_root,
    cwd: rpc_session.session.cwd.to_s.empty? ? rpc_session.workspace_root : rpc_session.session.cwd,
    name: rpc_session.session.name,
    createdAt: rpc_session.session.created_at&.utc&.iso8601(3),
    modifiedAt: session_modified_at(rpc_session.session)&.utc&.iso8601(3),
    parentId: rpc_session.session.parent_id,
    parentPath: rpc_session.session.parent_path
  }
end

#start_turn(session_id:, input:, streaming_behavior: nil, attachments: []) ⇒ Object 



198
199
200
201
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
227
228
229
 
# File 'lib/kward/rpc/session_manager.rb', line 198

def start_turn(session_id:, input:, streaming_behavior: nil, attachments: [])
  rpc_session = fetch_session(session_id)
  normalized_attachments = normalize_attachments(attachments)
  plugin_command, plugin_arguments = plugin_command_turn(input, normalized_attachments)
  content = plugin_command ? input.to_s : user_turn_content(expand_prompt_input(input), normalized_attachments)
  streaming_behavior = validate_streaming_behavior(default_streaming_behavior(rpc_session, streaming_behavior), rpc_session: rpc_session)
  if streaming_behavior == "steer"
    steered_turn = steer_running_turn(rpc_session, content)
    return steered_turn if steered_turn

    streaming_behavior = "followUp"
  end
  turn = Turn.new(
    id: SecureRandom.uuid,
    session_id: rpc_session.id,
    input: content,
    status: "queued",
    cancel_requested: false,
    cancellation: Cancellation.new,
    created_at: now,
    events: [],
    next_sequence: 1,
    streaming_behavior: streaming_behavior,
    plugin_command_name: plugin_command&.name,
    plugin_arguments: plugin_arguments
  )
  @mutex.synchronize { @turns[turn.id] = turn }
  rpc_session.queue << turn.id
  ensure_worker(rpc_session)
  emit_turn_event(turn, "turnQueued", { status: "queued" })
  turn_payload(turn)
end

#transcript(session_id:) ⇒ Object 



193
194
195
196
 
# File 'lib/kward/rpc/session_manager.rb', line 193

def transcript(session_id:)
  rpc_session = fetch_session(session_id)
  { session: session_payload(rpc_session), messages: TranscriptNormalizer.new(rpc_session.conversation.messages).normalize }
end

#turn_events(turn_id:, after_sequence: 0) ⇒ Object 



246
247
248
249
250
251
252
253
 
# File 'lib/kward/rpc/session_manager.rb', line 246

def turn_events(turn_id:, after_sequence: 0)
  turn = fetch_turn(turn_id)
  after_sequence = after_sequence.to_i
  {
    turn: turn_payload(turn),
    events: turn.events.select { |event| event[:sequence].to_i > after_sequence }
  }
end

#turn_status(turn_id:) ⇒ Object 



242
243
244
 
# File 'lib/kward/rpc/session_manager.rb', line 242

def turn_status(turn_id:)
  turn_payload(fetch_turn(turn_id))
end

#validate_workspace_root(root) ⇒ Object 



474
475
476
477
478
479
 
# File 'lib/kward/rpc/session_manager.rb', line 474

def validate_workspace_root(root)
  expanded = File.expand_path(root.to_s.empty? ? Dir.pwd : root.to_s)
  raise "Workspace root is not an existing directory: #{expanded}" unless File.directory?(expanded)

  File.realpath(expanded)
end