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
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.



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

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 = []
  @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 



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

def answer_transport_interaction(session_id:, request_id:, answer:)
  if answer == true || answer == false
    @session_manager.answer_tool_approval(
      session_id: session_id,
      approval_request_id: request_id,
      approved: answer
    )
  else
    answers = answer.is_a?(Array) ? answer : [{ question: request_id, answer: answer.to_s }]
    @session_manager.answer_question(
      session_id: session_id,
      question_request_id: request_id,
      answers: answers
    )
  end
end

#cancel_transport_turn(turn_id:) ⇒ Object 



71
72
73
 
# File 'lib/kward/transport/gateway.rb', line 71

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) 


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

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 



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

def shutdown
  subscriptions = @mutex.synchronize do
    current = @subscriptions
    @subscriptions = []
    @interaction_subscribers = []
    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 



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

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) 


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

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 



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

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 



58
59
60
 
# File 'lib/kward/transport/gateway.rb', line 58

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

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



62
63
64
65
 
# File 'lib/kward/transport/gateway.rb', line 62

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 



67
68
69
 
# File 'lib/kward/transport/gateway.rb', line 67

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