Skip to content
Kward Search API index

Class: Kward::Transport::Gateway

Inherits:
Object
  • Object
show all
Defined in:
lib/kward/transport/gateway.rb

Overview

Adapts the existing session manager to the transport host contract.

The adapter deliberately uses the manager's public session and turn methods. It can be replaced by a frontend-neutral runtime gateway later without changing transport plugins.

Constant Summary collapse

POLL_INTERVAL = 
0.05
INTERACTION_ROUTE_LIMIT = 
1_000
TERMINAL_STATUSES = 
%w[completed canceled failed].freeze

Instance Method Summary collapse

Constructor Details

#initialize(session_manager:, transport_id:, storage: nil, poll_interval: POLL_INTERVAL) ⇒ Gateway

Returns a new instance of Gateway.



19
20
21
22
23
24
25
26
27
28
29
 
# File 'lib/kward/transport/gateway.rb', line 19

def initialize(session_manager:, transport_id:, storage: nil, poll_interval: POLL_INTERVAL)
  @session_manager = session_manager
  @transport_id = transport_id.to_s
  @storage = storage || Store.new(@transport_id)
  @poll_interval = poll_interval
  @subscriptions = []
  @interaction_subscribers = []
  @interaction_routes = {}
  @mutex = Mutex.new
  @session_manager.subscribe_events { |method, payload| handle_runtime_event(method, payload) } if @session_manager.respond_to?(:subscribe_events)
end

Instance Method Details

#answer_transport_interaction(session_id:, request_id:, answer:) ⇒ Object 



77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/kward/transport/gateway.rb', line 77

def answer_transport_interaction(session_id:, request_id:, answer:)
  route = @mutex.synchronize { @interaction_routes.delete(request_id.to_s) }
  case route
  when :plugin_ui
    @session_manager.answer_plugin_ui(session_id: session_id, request_id: request_id, value: answer)
  when :tool_approval
    @session_manager.answer_tool_approval(session_id: session_id, approval_request_id: request_id, approved: answer == true)
  when :question
    answer_transport_question(session_id, request_id, answer)
  else
    answer == true || answer == false ? answer_transport_approval(session_id, request_id, answer) : answer_transport_question(session_id, request_id, answer)
  end
end

#cancel_transport_turn(turn_id:) ⇒ Object 



73
74
75
 
# File 'lib/kward/transport/gateway.rb', line 73

def cancel_transport_turn(turn_id:)
  @session_manager.cancel_turn(turn_id: turn_id)
end

#resolve_transport_session(transport_id:, conversation:, actor:, workspace_root: nil, name: nil, execution_profile: nil) ⇒ Object

Raises:

  • (ArgumentError) 


31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/kward/transport/gateway.rb', line 31

def resolve_transport_session(transport_id:, conversation:, actor:, workspace_root: nil, name: nil, execution_profile: nil)
  raise ArgumentError, "transport id does not match gateway" unless transport_id.to_s == @transport_id
  if conversation.respond_to?(:transport_id) && conversation.transport_id.to_s != @transport_id
    raise ArgumentError, "conversation transport does not match gateway"
  end

  binding_key = binding_key_for(conversation)
  binding = @storage.get(binding_key)
  session = if binding
               resume_bound_session(binding, execution_profile: execution_profile)
             else
               @session_manager.create_session(workspace_root: workspace_root || Dir.pwd, name: name, **execution_profile_arguments(execution_profile))
             end
  persist_binding(binding_key, session)
  session_handle(session)
end

#shutdown ⇒ Object 



119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/kward/transport/gateway.rb', line 119

def shutdown
  subscriptions = @mutex.synchronize do
    current = @subscriptions
    @subscriptions = []
    @interaction_subscribers = []
    @interaction_routes = {}
    current
  end
  subscriptions.each(&:kill)
  subscriptions.each(&:join)
  nil
end

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



48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/kward/transport/gateway.rb', line 48

def start_transport_turn(session_id:, input:, attachments: [], options: {}, streaming_behavior: nil, execution_profile: nil)
  payload = @session_manager.start_turn(
    session_id: session_id,
    input: input,
    attachments: normalize_attachments(attachments, execution_profile: execution_profile),
    options: profile_options(options, execution_profile),
    streaming_behavior: streaming_behavior,
    execution_profile: execution_profile
  )
  { id: payload.fetch(:id), session_id: payload.fetch(:sessionId) }
end

#subscribe_transport_interactions(&block) ⇒ Object

Raises:

  • (ArgumentError) 


91
92
93
94
95
96
 
# File 'lib/kward/transport/gateway.rb', line 91

def subscribe_transport_interactions(&block)
  raise ArgumentError, "interaction subscription requires a block" unless block

  @mutex.synchronize { @interaction_subscribers << block }
  block
end

#subscribe_transport_turn(turn_id:, after: nil) ⇒ Object 



98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/kward/transport/gateway.rb', line 98

def subscribe_transport_turn(turn_id:, after: nil)
  cursor = after.to_i
  thread = Thread.new do
    loop do
      events = transport_turn_events(turn_id: turn_id, after: cursor)
      events.each do |event|
        cursor = event.sequence
        yield event
      end
      status = transport_turn_status(turn_id: turn_id)
      break if TERMINAL_STATUSES.include?(status[:status].to_s)

      sleep @poll_interval
    end
  rescue StandardError
    nil
  end
  @mutex.synchronize { @subscriptions << thread }
  thread
end

#transport_transcript(session_id:) ⇒ Object 



60
61
62
 
# File 'lib/kward/transport/gateway.rb', line 60

def transport_transcript(session_id:)
  @session_manager.transcript(session_id: session_id)
end

#transport_turn_events(turn_id:, after: nil) ⇒ Object 



64
65
66
67
 
# File 'lib/kward/transport/gateway.rb', line 64

def transport_turn_events(turn_id:, after: nil)
  payload = @session_manager.turn_events(turn_id: turn_id, after_sequence: after.to_i)
  Array(payload[:events]).map { |event| normalize_event(event) }
end

#transport_turn_status(turn_id:) ⇒ Object 



69
70
71
 
# File 'lib/kward/transport/gateway.rb', line 69

def transport_turn_status(turn_id:)
  @session_manager.turn_status(turn_id: turn_id)
end