Skip to content
Kward Search API index

Class: Kward::RPC::PluginChatManager

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

Overview

Coordinates optional RPC access to plugin-owned conversational runtimes. Unlike SessionManager, this manager never creates workspace sessions, agents, or tool registries: each plugin driver owns its chat and storage.

Defined Under Namespace

Classes: Chat, Turn

Constant Summary collapse

EVENT_LIMIT = 
1_000
WORKER_STOP = 
Object.new.freeze

Instance Method Summary collapse

Constructor Details

#initialize(server:, client: Client.new) ⇒ PluginChatManager

Returns a new instance of PluginChatManager.



26
27
28
29
30
31
32
33
 
# File 'lib/kward/rpc/plugin_chat_manager.rb', line 26

def initialize(server:, client: Client.new)
  @server = server
  @client = client
  @chats = {}
  @turns = {}
  @subscriptions = {}
  @mutex = Mutex.new
end

Instance Method Details

#cancel_turn(turn_id:) ⇒ Object 



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

def cancel_turn(turn_id:)
  turn = fetch_turn(turn_id)
  queued = turn.mutex.synchronize do
    turn.cancellation.cancel!
    turn.status == "queued"
  end
  emit_event(turn, "turnCancelRequested", {})
  finish_turn(turn, "canceled") if queued
  turn_payload(turn)
end

#list ⇒ Object 



39
40
41
 
# File 'lib/kward/rpc/plugin_chat_manager.rb', line 39

def list
  { chats: supported_types.map { |type| type_payload(type) } }
end

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



117
118
119
120
121
122
 
# File 'lib/kward/rpc/plugin_chat_manager.rb', line 117

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

#open(type_id:) ⇒ Object 



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

def open(type_id:)
  type = supported_types.find { |entry| entry.id == type_id.to_s } || raise(ArgumentError, "Unknown RPC plugin chat: #{type_id}")
  chat = chat_for(type)
  return chat_payload(chat) if chat.driver.respond_to?(:transcript_page)

  chat_payload(chat).merge(messages: TranscriptNormalizer.new(chat.driver.messages).normalize)
end

#shutdown ⇒ Object 



124
125
126
127
128
129
130
 
# File 'lib/kward/rpc/plugin_chat_manager.rb', line 124

def shutdown
  chats = @mutex.synchronize { @chats.values.dup }
  chats.each do |chat|
    chat.queue << WORKER_STOP if chat.worker&.alive?
    chat.worker&.join(0.2)
  end
end

#start_turn(chat_id:, input:, attachments: []) ⇒ Object 



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

def start_turn(chat_id:, input:, attachments: [])
  chat = fetch_chat(chat_id)
  normalized_attachments = AttachmentNormalizer.new.normalize(attachments)
  turn = Turn.new(
    id: SecureRandom.uuid,
    chat_id: chat.id,
    input: input_with_attachments(input, normalized_attachments),
    display_input: input.to_s,
    status: "queued",
    cancellation: Cancellation.new,
    created_at: Time.now.utc.iso8601(3),
    events: [],
    next_sequence: 1,
    mutex: Mutex.new
  )
  @mutex.synchronize { @turns[turn.id] = turn }
  chat.queue << turn.id
  ensure_worker(chat)
  emit_event(turn, "turnQueued", { status: "queued" })
  turn_payload(turn)
end

#subscribe(chat_id:) ⇒ Object 



62
63
64
65
66
 
# File 'lib/kward/rpc/plugin_chat_manager.rb', line 62

def subscribe(chat_id:)
  chat = fetch_chat(chat_id)
  @mutex.synchronize { @subscriptions[chat.id] = true }
  { chat: chat_payload(chat), subscribed: true }
end

#supported_types ⇒ Object 



35
36
37
 
# File 'lib/kward/rpc/plugin_chat_manager.rb', line 35

def supported_types
  plugin_registry.tab_types.select(&:rpc)
end

#transcript(chat_id:, limit: nil, before: nil) ⇒ Object 



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

def transcript(chat_id:, limit: nil, before: nil)
  chat = fetch_chat(chat_id)
  page = transcript_page(chat.driver, limit: limit, before: before)
  {
    chat: chat_payload(chat),
    messages: TranscriptNormalizer.new(page.fetch(:messages)).normalize,
    hasMore: page.fetch(:has_more),
    nextBefore: page[:next_before]
  }.compact
end

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



111
112
113
114
115
 
# File 'lib/kward/rpc/plugin_chat_manager.rb', line 111

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

#turn_status(turn_id:) ⇒ Object 



107
108
109
 
# File 'lib/kward/rpc/plugin_chat_manager.rb', line 107

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

#unsubscribe(chat_id:) ⇒ Object 



68
69
70
71
72
 
# File 'lib/kward/rpc/plugin_chat_manager.rb', line 68

def unsubscribe(chat_id:)
  chat = fetch_chat(chat_id)
  @mutex.synchronize { @subscriptions.delete(chat.id) }
  { chat: chat_payload(chat), subscribed: false }
end