Skip to content
Kward Search API index

Class: Kward::RPC::SessionManager

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

Overview

Owns RPC-visible session lifecycle, async turn queues, and frontend events.

Server handles JSON-RPC framing/dispatch; SessionManager handles the product state behind those methods. It creates/resumes SessionStore sessions, builds agents with RPC prompt bridges, serializes turn events for clients, coordinates cancellation and follow-up queues, and integrates memory/plugin hooks for RPC sessions.

Keep JSON-RPC wire shape normalization in the RPC::*Normalizer classes, persistence in SessionStore, and model/tool behavior in Agent and ToolRegistry. This class should coordinate those pieces rather than own their low-level mechanics.

Defined Under Namespace

Classes: RpcSession, Turn

Constant Summary collapse

RECENT_EVENT_LIMIT = 
1_000
RPC_ATTACHMENT_MAX_BYTES = 
AttachmentNormalizer::MAX_BYTES
RPC_IMAGE_MIME_TYPES = 
AttachmentNormalizer::IMAGE_MIME_TYPES
STREAMING_BEHAVIORS = 
["newTurn", "followUp", "steer"].freeze
1.0
WORKER_STOP_TIMEOUT = 
2.0
WORKER_STOP = 
Object.new.freeze

Instance Method Summary collapse

Methods included from MemoryMethods

#memory_add, #memory_add_core, #memory_auto_summary_disable, #memory_auto_summary_enable, #memory_disable, #memory_enable, #memory_forget, #memory_inspect, #memory_list, #memory_manager, #memory_promote, #memory_relax, #memory_status, #memory_summarize, #memory_why

Constructor Details

#initialize(server:, client: Client.new, config_dir: ConfigFiles.config_dir, config_manager: ConfigManager.new(config_path: File.join(config_dir, "config.json")), context_usage: ContextUsage.new, session_trash: SessionTrash.new, worker_stop_timeout: WORKER_STOP_TIMEOUT, event_listener: nil) ⇒ SessionManager

Creates an object for RPC session lifecycle and turn coordination.



73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/kward/rpc/session_manager.rb', line 73

def initialize(
  server:,
  client: Client.new,
  config_dir: ConfigFiles.config_dir,
  config_manager: ConfigManager.new(config_path: File.join(config_dir, "config.json")),
  context_usage: ContextUsage.new,
  session_trash: SessionTrash.new,
  worker_stop_timeout: WORKER_STOP_TIMEOUT,
  event_listener: nil
)
  @server = server
  @client = client
  @config_dir = config_dir
  @config_manager = config_manager
  @context_usage = context_usage
  @session_metrics = SessionMetrics.new(context_usage: context_usage)
  @session_trash = session_trash
  @worker_stop_timeout = worker_stop_timeout
  @event_listeners = event_listener ? [event_listener] : []
  @sessions = {}
  @turns = {}
  @mutex = Mutex.new
end

Instance Method Details

#active_sessions ⇒ Object 



160
161
162
 
# File 'lib/kward/rpc/session_manager.rb', line 160

def active_sessions
  { sessions: @mutex.synchronize { @sessions.values.map { |rpc_session| session_payload(rpc_session) } } }
end

#answer_plugin_ui(session_id:, request_id:, value:) ⇒ Object 



498
499
500
501
502
 
# File 'lib/kward/rpc/session_manager.rb', line 498

def answer_plugin_ui(session_id:, request_id:, value:)
  rpc_session = fetch_session(session_id)
  rpc_session.prompt.answer_plugin_ui(request_id, value)
  { ok: true }
end

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



492
493
494
495
496
 
# File 'lib/kward/rpc/session_manager.rb', line 492

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

#answer_tool_approval(session_id:, approval_request_id:, approved:) ⇒ Object 



504
505
506
507
508
 
# File 'lib/kward/rpc/session_manager.rb', line 504

def answer_tool_approval(session_id:, approval_request_id:, approved:)
  rpc_session = fetch_session(session_id)
  rpc_session.prompt.answer_tool_approval(approval_request_id, approved: approved)
  { ok: true }
end

#available_models ⇒ Object 



572
573
574
575
576
577
578
 
# File 'lib/kward/rpc/session_manager.rb', line 572

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 



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

def cancel_turn(turn_id:)
  turn = fetch_turn(turn_id)
  queued, event = turn.mutex.synchronize do
    turn.cancel_requested = true
    [turn.status == "queued", append_turn_event_locked(turn, "turnCancelRequested", {})]
  end
  notify("turn/event", event)
  turn.cancellation&.cancel!
  finish_turn(turn, "canceled") if queued
  turn_payload(turn)
end

#capture_skill(session_path:) ⇒ Object 



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

def capture_skill(session_path:)
  capture = Skills::Capture.new(
    session_store: SessionStore.new(config_dir: @config_dir),
    client: @client,
    config_dir: @config_dir
  )
  draft = capture.generate(session_path)
  { content: draft.content, sourcePath: draft.source_path, name: draft.name, description: draft.description }
end

#cleanup_unused_sessions ⇒ Object

Closes idle empty sessions left behind by UI lifecycle transitions.



376
377
378
379
380
381
382
383
384
385
386
 
# File 'lib/kward/rpc/session_manager.rb', line 376

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

    remove_live_session(rpc_session)
  end
  { closed: true }
end

#clone_session(session_id:) ⇒ Object

Creates an independent copy of the current conversation branch.



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

def clone_session(session_id:)
  source = fetch_session(session_id)
  session, conversation = source.store.create_independent_from_conversation(source.conversation, parent_session: source.session)
  conversation.update_execution_profile_context!(source.execution_profile.prompt_context) if source.execution_profile
  rpc_session = build_rpc_session(source.store, session, conversation, source.workspace_root, execution_profile: source.execution_profile)
  remember_session(rpc_session)
  cleanup_other_unused_sessions(rpc_session)
  emit_footer_update(rpc_session)
  session_payload(rpc_session)
end

#close_session(session_id:) ⇒ Object

Stops workers and removes an RPC session from the live session map.



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

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

Compacts an RPC session and emits start/end events for UI progress.



241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
 
# File 'lib/kward/rpc/session_manager.rb', line 241

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, resume_last: false, execution_profile: nil) ⇒ Object

Creates a new RPC session or resumes the remembered session when allowed.

Returns the normalized session payload expected by RPC clients. The RPC session id is separate from the persisted session id so one persisted file can be closed and reopened by different client connections.



102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
 
# File 'lib/kward/rpc/session_manager.rb', line 102

def create_session(workspace_root: Dir.pwd, name: nil, resume_last: false, execution_profile: nil)
  execution_profile = validate_execution_profile(execution_profile)
  workspace_root = validate_workspace_root(workspace_root)
  store = SessionStore.new(config_dir: @config_dir, cwd: workspace_root)
  if resume_last && session_auto_resume_enabled? && name.to_s.strip.empty?
    path = store.remembered_last_session_path
    return resume_session(path: path, workspace_root: workspace_root, include_transcript: true, execution_profile: execution_profile) if path
  end

  conversation = new_conversation(workspace_root: workspace_root, execution_profile: execution_profile)
  session = store.create(provider: conversation.provider, 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, execution_profile: execution_profile)
  remember_session(rpc_session)
  cleanup_other_unused_sessions(rpc_session)
  emit_footer_update(rpc_session)
  session_payload(rpc_session)
end

#current_model ⇒ Object 



585
586
587
588
589
590
591
 
# File 'lib/kward/rpc/session_manager.rb', line 585

def current_model
  provider = @client.respond_to?(:current_provider) ? @client.current_provider : nil
  model = @client.respond_to?(:current_model) ? @client.current_model : nil
  reasoning_effort = @client.respond_to?(:current_reasoning_effort) ? @client.current_reasoning_effort : nil
  context_window = @client.respond_to?(:current_context_window) ? @client.current_context_window : nil
  ModelInfo.current_payload(provider: provider, model: model, reasoning_effort: reasoning_effort, context_window: context_window)
end

#delete_session(session_id:) ⇒ Object

Deletes the backing session file through the configured trash strategy.



360
361
362
363
364
365
366
 
# File 'lib/kward/rpc/session_manager.rb', line 360

def delete_session(session_id:)
  rpc_session = fetch_session(session_id)
  path = rpc_session.session.path
  close_rpc_session(rpc_session, delete_unused: false)
  deleted = @session_trash.delete(path)
  { deleted: deleted, path: path }
end

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

Exports the current transcript in markdown or JSON format.



350
351
352
353
354
355
356
357
 
# File 'lib/kward/rpc/session_manager.rb', line 350

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

Lists user-message entries that can be used as fork points.



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

def fork_messages(session_id:)
  rpc_session = fetch_session(session_id)
  {
    messages: session_tree_helper(rpc_session).entries.filter_map do |record|
      message = record["message"]
      next unless message.is_a?(Hash) && message_role(message) == "user"

      { entryId: record["id"], text: display_message_text(message) }
    end
  }
end

#fork_session(session_id:, entry_id:) ⇒ Object

Creates a new session from history before the selected user message.

Raises:

  • (ArgumentError) 


272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
 
# File 'lib/kward/rpc/session_manager.rb', line 272

def fork_session(session_id:, entry_id:)
  source = fetch_session(session_id)
  tree = session_tree_helper(source)
  entries = tree.entries
  resolved_entry_id = tree.resolve_entry_id(entry_id, entries: entries)
  selected_index = entries.index { |record| record["id"].to_s == resolved_entry_id.to_s }
  selected = selected_index && entries[selected_index]
  raise ArgumentError, "Unknown fork entryId: #{entry_id}" unless selected

  message = selected["message"]
  raise ArgumentError, "Entry is not forkable: #{entry_id}" unless message.is_a?(Hash) && message_role(message) == "user"

  session, conversation = source.store.create_independent_from_messages(
    entries[0...selected_index].filter_map { |record| record["message"] },
    provider: source.conversation.provider,
    model: source.conversation.model,
    reasoning_effort: source.conversation.reasoning_effort,
    parent_session: source.session
  )

  conversation.update_execution_profile_context!(source.execution_profile.prompt_context) if source.execution_profile
  rpc_session = build_rpc_session(source.store, session, conversation, source.workspace_root, execution_profile: source.execution_profile)
  remember_session(rpc_session)
  cleanup_other_unused_sessions(rpc_session)
  {
    session: session_payload(rpc_session),
    text: full_message_text(message),
    cancelled: false
  }
end

#in_flight_steer_supported? ⇒ Boolean

Returns:

  • (Boolean) 


608
609
610
 
# File 'lib/kward/rpc/session_manager.rb', line 608

def in_flight_steer_supported?
  supports_in_flight_steer?
end

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



147
148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/kward/rpc/session_manager.rb', line 147

def list_sessions(workspace_root: Dir.pwd, limit: nil, current_session_path: nil)
  root = validate_workspace_root(workspace_root)
  store = SessionStore.new(config_dir: @config_dir, cwd: root)
  requested_limit = limit.to_i if limit
  requested_limit = nil unless requested_limit&.positive?
  live_paths = @mutex.synchronize do
    @sessions.values.filter_map { |rpc_session| rpc_session.session.path if rpc_session.workspace_root == root }
  end
  live_paths << current_session_path unless current_session_path.to_s.empty?
  store.recent(limit: requested_limit, keep_empty_path: live_paths)
       .map { |info| session_info_payload(info, workspace_root: root) }
end

#list_turns(session_id: nil, active: false) ⇒ Object 



485
486
487
488
489
490
 
# File 'lib/kward/rpc/session_manager.rb', line 485

def list_turns(session_id: nil, active: false)
  turns = @mutex.synchronize { @turns.values.dup }
  turns.select! { |turn| turn.session_id == session_id.to_s } if session_id
  turns.select! { |turn| ["queued", "running"].include?(turn.status) } if active
  { turns: turns.map { |turn| turn_payload(turn) } }
end

#navigate_tree(session_id:, entry_id:, summarize: false, custom_instructions: nil) ⇒ Object

Moves the active branch to a tree entry, optionally summarizing abandoned history.

Raises:

  • (ArgumentError) 


317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
 
# File 'lib/kward/rpc/session_manager.rb', line 317

def navigate_tree(session_id:, entry_id:, summarize: false, custom_instructions: nil)
  rpc_session = fetch_session(session_id)
  tree = session_tree_helper(rpc_session)
  entries = tree.entries
  resolved_entry_id = tree.resolve_entry_id(entry_id, entries: entries)
  entry = rpc_session.store.session_entry(rpc_session.session.path, resolved_entry_id)
  raise ArgumentError, "Unknown tree entryId: #{entry_id}" unless entry

  raise ArgumentError, "Tree entry is not selectable: #{entry_id}" unless tree.selectable_entry?(entry)

  message = entry["message"]
  user_entry = tree.user_entry?(entry)
  target_leaf = user_entry ? entry["parentId"] : entry["id"]
  editor_text = user_entry ? full_message_text(message) : nil
  previous_leaf = rpc_session.session.leaf_id

  if summarize
    summary = summarize_branch(rpc_session, from_id: previous_leaf, to_id: target_leaf, custom_instructions: custom_instructions)
    target_leaf = rpc_session.session.append_branch_summary(target_leaf, from_id: previous_leaf, summary: summary, details: {})
  elsif target_leaf
    rpc_session.session.branch(target_leaf)
  end

  reload_rpc_session(rpc_session)
  {
    session: session_payload(rpc_session),
    editorText: editor_text,
    cancelled: false,
    aborted: false
  }.compact
end

#plugin_actions ⇒ Object 



555
556
557
 
# File 'lib/kward/rpc/session_manager.rb', line 555

def plugin_actions
  plugin_registry.actions
end

#plugin_commands ⇒ Object 



551
552
553
 
# File 'lib/kward/rpc/session_manager.rb', line 551

def plugin_commands
  plugin_registry.commands
end

#plugin_registry ⇒ Object

Returns the plugin registry shared by RPC sessions and plugin chats.



217
218
219
 
# File 'lib/kward/rpc/session_manager.rb', line 217

def plugin_registry
  @plugin_registry ||= PluginRegistry.load(reserved_commands: reserved_plugin_command_names).tap(&:start!)
end

#refresh_client_config ⇒ Object 



655
656
657
658
659
 
# File 'lib/kward/rpc/session_manager.rb', line 655

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

#refresh_models(provider: nil) ⇒ Object 



580
581
582
583
 
# File 'lib/kward/rpc/session_manager.rb', line 580

def refresh_models(provider: nil)
  @client.refresh_available_models(provider: provider) if @client.respond_to?(:refresh_available_models)
  available_models
end

#reload_plugins ⇒ Object 



661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
 
# File 'lib/kward/rpc/session_manager.rb', line 661

def reload_plugins
  registry = PluginRegistry.load(reserved_commands: reserved_plugin_command_names)
  @plugin_registry&.reload!
  registry.start!
  sessions = @mutex.synchronize do
    @plugin_registry = registry
    @sessions.values
  end
  sessions.each do |rpc_session|
    rpc_session.conversation.plugin_registry = registry if rpc_session.conversation.respond_to?(:plugin_registry=)
    rpc_session.conversation.refresh_system_message! if rpc_session.conversation.respond_to?(:refresh_system_message!)
    rebuild_session_tools(rpc_session)
    if registry.status?
      start_footer_worker(rpc_session)
      emit_footer_update(rpc_session)
    else
      stop_footer_worker(rpc_session)
      clear_footer_update(rpc_session)
    end
  end
end

#rename_session(session_id:, name:) ⇒ Object

Renames the persisted session attached to an RPC session id.



222
223
224
225
226
 
# File 'lib/kward/rpc/session_manager.rb', line 222

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, include_transcript: false, execution_profile: nil) ⇒ Object 



122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
 
# File 'lib/kward/rpc/session_manager.rb', line 122

def resume_session(path:, workspace_root: nil, include_transcript: false, execution_profile: nil)
  execution_profile = validate_execution_profile(execution_profile)
  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: configured_workspace(root),
    provider: current_model[:provider],
    model: current_model_id,
    reasoning_effort: current_reasoning_effort
  )
  conversation.update_execution_profile_context!(execution_profile.prompt_context) if execution_profile
  rpc_session = build_rpc_session(store, session, conversation, root, execution_profile: execution_profile)
  remember_session(rpc_session)
  cleanup_other_unused_sessions(rpc_session)
  emit_footer_update(rpc_session)
  payload = session_payload(rpc_session)
  payload[:messages] = TranscriptNormalizer.new(rpc_session.conversation.messages).normalize if include_transcript
  payload[:resumed] = true
  payload
end

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



510
511
512
513
514
515
516
517
 
# File 'lib/kward/rpc/session_manager.rb', line 510

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

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

#run_plugin_action(session_id:, id:, arguments: {}) ⇒ Object

Raises:

  • (ArgumentError) 


559
560
561
562
563
564
565
566
567
568
569
570
 
# File 'lib/kward/rpc/session_manager.rb', line 559

def run_plugin_action(session_id:, id:, arguments: {})
  rpc_session = fetch_session(session_id)
  raise ArgumentError, "Plugin actions are disabled for this session" if rpc_session.execution_profile && !rpc_session.execution_profile.plugin_commands

  action = plugin_registry.action_for(id) || raise(ArgumentError, "Unknown plugin action: #{id}")
  arguments = action.parse_arguments(arguments)
  output = []
  context = plugin_context(rpc_session, args: arguments, say_callback: lambda { |message| output << message.to_s })
  result = action.normalize_result(action.handler.call(arguments, context))
  output = rpc_session.plugin_output.shift(rpc_session.plugin_output.length) + output
  { action: action.id, output: output, result: result.to_h }
end

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

Raises:

  • (ArgumentError) 


537
538
539
540
541
542
543
544
545
546
547
548
549
 
# File 'lib/kward/rpc/session_manager.rb', line 537

def run_plugin_command(session_id:, command:, arguments: "")
  rpc_session = fetch_session(session_id)
  raise ArgumentError, "Plugin commands are disabled for this session" if rpc_session.execution_profile && !rpc_session.execution_profile.plugin_commands

  command = plugin_registry.command_for(command.to_s.delete_prefix("/")) || raise(ArgumentError, "Unknown plugin command: #{command}")
  arguments = command.parse_arguments(arguments)
  output = []
  context = plugin_context(rpc_session, args: arguments, say_callback: lambda { |message| output << message.to_s })
  result = command.normalize_result(command.handler.call(arguments, context))
  output = rpc_session.plugin_output.shift(rpc_session.plugin_output.length) + output
  serialized_result = command.typed? ? result.to_h : (result.nil? ? nil : result.to_s)
  { command: command.name, output: output, result: serialized_result }
end

#run_skill_command(session_id:, name:) ⇒ Object

Raises:

  • (ArgumentError) 


519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
 
# File 'lib/kward/rpc/session_manager.rb', line 519

def run_skill_command(session_id:, name:)
  skill_name = name.to_s.strip
  raise ArgumentError, "Missing skill name" if skill_name.empty?

  rpc_session = fetch_session(session_id)
  tool_call = {
    "id" => "skill_#{skill_name.gsub(/[^a-zA-Z0-9_-]/, "_")}",
    "type" => "function",
    "function" => {
      "name" => "read_skill",
      "arguments" => JSON.dump({ name: skill_name })
    }
  }
  rpc_session.conversation.append_assistant("role" => "assistant", "content" => nil, "tool_calls" => [tool_call])
  result = rpc_session.tool_registry.dispatch(tool_call, rpc_session.conversation)
  { command: "skill:#{skill_name}", output: [result.start_with?("Error:") ? result : "Activated skill: #{skill_name}"], result: result }
end

#runtime_state(session_id:) ⇒ Object 



612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
 
# File 'lib/kward/rpc/session_manager.rb', line 612

def runtime_state(session_id:)
  rpc_session = fetch_session(session_id)
  model = session_model(rpc_session)
  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)
  RuntimePayloads.state(
    session: session,
    model: model,
    streaming: streaming?(rpc_session),
    steering_supported: supports_in_flight_steer?,
    auto_compaction_reserve_tokens: auto_compaction_reserve_tokens,
    active_persona_label: active_persona_label(rpc_session),
    message_count: @session_metrics.message_count(rpc_session.conversation),
    pending_count: pending_turn_count(rpc_session.id),
    compaction_enabled: compaction_settings.enabled,
    workspace_guardrails_enabled: workspace_guardrails_enabled?
  )
end

#runtime_stats(session_id:) ⇒ Object 



635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
 
# File 'lib/kward/rpc/session_manager.rb', line 635

def runtime_stats(session_id:)
  rpc_session = fetch_session(session_id)
  session = session_payload(rpc_session)
  counts = @session_metrics.message_stats(rpc_session.conversation)
  model = session_model(rpc_session)
  compaction_settings = self.compaction_settings
  auto_compaction_reserve_tokens = compaction_reserve_tokens(
    context_window: model[:contextWindow],
    compaction_settings: compaction_settings
  )
  RuntimePayloads.stats(
    session: session,
    counts: counts,
    model: model,
    auto_compaction_reserve_tokens: auto_compaction_reserve_tokens,
    context_usage: @session_metrics.context_usage(rpc_session, model, client: @client),
    compaction_enabled: compaction_settings.enabled
  )
end

#save_captured_skill(content:, overwrite: false) ⇒ Object 



186
187
188
189
190
191
192
193
194
 
# File 'lib/kward/rpc/session_manager.rb', line 186

def save_captured_skill(content:, overwrite: false)
  capture = Skills::Capture.new(
    session_store: SessionStore.new(config_dir: @config_dir),
    client: @client,
    config_dir: @config_dir
  )
  draft = capture.save(content, overwrite: overwrite)
  { path: capture.skill_path(draft.name), name: draft.name, description: draft.description }
end

#session_model(rpc_session) ⇒ Object 



593
594
595
596
597
598
599
600
601
602
603
604
605
606
 
# File 'lib/kward/rpc/session_manager.rb', line 593

def session_model(rpc_session)
  current = current_model
  provider = rpc_session.conversation.provider || current[:provider]
  model = rpc_session.conversation.model || current[:id]
  reasoning_effort = rpc_session.conversation.reasoning_effort || current_reasoning_effort
  reasoning_effort = nil unless ModelInfo.reasoning_supported?(provider, model)
  context_window = context_window_for(provider, model)
  ModelInfo.current_payload(
    provider: provider,
    model: model,
    reasoning_effort: reasoning_effort,
    context_window: context_window
  )
end

#session_modified_at(session) ⇒ Object 



691
692
693
694
695
 
# File 'lib/kward/rpc/session_manager.rb', line 691

def session_modified_at(session)
  return nil unless File.exist?(session.path)

  session.respond_to?(:modified_at) ? session.modified_at : nil
end

#session_payload(rpc_session) ⇒ Object 



683
684
685
686
687
688
689
 
# File 'lib/kward/rpc/session_manager.rb', line 683

def session_payload(rpc_session)
  RuntimePayloads.session(
    rpc_session,
    modified_at: session_modified_at(rpc_session.session),
    active_persona_label: active_persona_label(rpc_session)
  )
end

#session_tree(session_id:) ⇒ Object

Returns the flattened session tree rows consumed by RPC clients.



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

def session_tree(session_id:)
  rpc_session = fetch_session(session_id)
  { items: flatten_session_tree(rpc_session) }
end

#set_tree_label(session_id:, entry_id:, label: nil) ⇒ Object

Persists a label override for one tree entry.



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

def set_tree_label(session_id:, entry_id:, label: nil)
  rpc_session = fetch_session(session_id)
  rpc_session.session.append_label_change(entry_id, label)
  { ok: true }
end

#shutdown_plugins ⇒ Object 



395
396
397
398
 
# File 'lib/kward/rpc/session_manager.rb', line 395

def shutdown_plugins
  @plugin_registry&.shutdown!
  { closed: true }
end

#shutdown_sessions ⇒ Object

Stops all live RPC session workers during server shutdown.



389
390
391
392
393
 
# File 'lib/kward/rpc/session_manager.rb', line 389

def shutdown_sessions
  rpc_sessions = @mutex.synchronize { @sessions.values.dup }
  rpc_sessions.reverse_each { |rpc_session| close_rpc_session(rpc_session) }
  { closed: true }
end

#skill_capture_sessions ⇒ Object 



164
165
166
167
168
169
170
171
172
173
174
 
# File 'lib/kward/rpc/session_manager.rb', line 164

def skill_capture_sessions
  SessionStore.new(config_dir: @config_dir).capture_candidates.map do |session|
    {
      path: session.path,
      workspaceRoot: session.cwd,
      name: session.name,
      firstMessage: session.first_message,
      modifiedAt: session.modified_at.utc.iso8601(3)
    }
  end
end

#start_turn(session_id:, input:, streaming_behavior: nil, attachments: [], options: {}, context: nil, execution_profile: nil) ⇒ Object

Queues or starts an async model turn for an RPC session.

streaming_behavior controls busy-session behavior: create a new turn, queue a follow-up, or steer the running turn when the active provider supports native steering. The returned turn id is used for status, cancellation, and event replay.



412
413
414
415
416
417
418
419
420
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
448
449
450
451
452
453
454
455
456
 
# File 'lib/kward/rpc/session_manager.rb', line 412

def start_turn(session_id:, input:, streaming_behavior: nil, attachments: [], options: {}, context: nil, execution_profile: nil)
  rpc_session = fetch_session(session_id)
  execution_profile = validate_execution_profile(execution_profile)
  normalized_options = enforce_execution_profile_options(normalize_turn_options(options), execution_profile)
  normalized_context = TurnContext.normalize(context)
  normalized_attachments = normalize_attachments(attachments)
  if execution_profile && !execution_profile.attachments && !normalized_attachments.empty?
    raise ArgumentError, "transport execution profile does not allow attachments"
  end
  plugin_command, plugin_arguments = if execution_profile && !execution_profile.plugin_commands
                                        [nil, ""]
                                      else
                                        plugin_command_turn(input, normalized_attachments)
                                      end
  display_input = input.to_s if input.is_a?(String)
  content = plugin_command ? input.to_s : user_turn_content(turn_input_with_context(expand_prompt_input(input), normalized_context), normalized_attachments)
  streaming_behavior = validate_streaming_behavior(default_streaming_behavior(rpc_session, streaming_behavior), rpc_session: rpc_session)
  if streaming_behavior == "steer"
    return steer_running_turn(rpc_session, content)
  end
  turn = Turn.new(
    id: SecureRandom.uuid,
    session_id: rpc_session.id,
    input: content,
    display_input: display_input,
    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,
    options: normalized_options,
    tool_registry: scoped_tool_registry(rpc_session, normalized_options),
    execution_profile: execution_profile,
    mutex: Mutex.new
  )
  @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

#subscribe_events(&listener) ⇒ Object

Subscribes to frontend-neutral runtime notifications.

Subscribers are isolated from session execution: an exception is logged and does not interrupt the RPC notification or active turn.

Raises:

  • (ArgumentError) 


209
210
211
212
213
214
 
# File 'lib/kward/rpc/session_manager.rb', line 209

def subscribe_events(&listener)
  raise ArgumentError, "event subscription requires a block" unless listener

  @mutex.synchronize { @event_listeners << listener }
  listener
end

#tool_schemas(session_id: nil) ⇒ Object 



196
197
198
199
200
201
202
203
 
# File 'lib/kward/rpc/session_manager.rb', line 196

def tool_schemas(session_id: nil)
  registry = if session_id
               fetch_session(session_id).tool_registry
             else
               build_tool_registry(Dir.pwd, nil)
             end
  { tools: registry.schemas }
end

#transcript(session_id:) ⇒ Object

Returns the normalized transcript for the active RPC session.



401
402
403
404
 
# File 'lib/kward/rpc/session_manager.rb', line 401

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 



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

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

#turn_status(turn_id:) ⇒ Object 



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

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

#validate_workspace_root(root) ⇒ Object 



697
698
699
700
701
702
 
# File 'lib/kward/rpc/session_manager.rb', line 697

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