Module: Mistri::ToolExecutor

Defined in:
lib/mistri/tool_executor.rb

Overview

Runs a turn's tool calls and returns their results in the order the model emitted them, regardless of completion order. Independent calls run concurrently up to max_concurrency; each runs inside the Rails executor when Rails is present, so ActiveRecord connections return to the pool.

A tool that raises becomes an in-band error string the model can read. An abort never starts a not-yet-started call: it gets an interrupted result instead, so the turn always pairs and the session replays cleanly.

Constant Summary collapse

INTERRUPTED = 
"[interrupted: this tool call never ran]"

Class Method Summary collapse

Class Method Details

.call(calls, tools_by_name, signal: nil, max_concurrency: 4, session: nil, emit: nil, app: nil) ⇒ Object 



19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/mistri/tool_executor.rb', line 19

def call(calls, tools_by_name, signal: nil, max_concurrency: 4, session: nil, emit: nil,
         app: nil)
  return [] if calls.empty?

  context = ToolContext.new(session: session, signal: signal, emit: thread_safe(emit),
                            app: app)
  results = Array.new(calls.length)
  queue = Queue.new
  calls.each_with_index { |call, index| queue << [call, index] }
  workers = max_concurrency.clamp(1, calls.length)
  Array.new(workers) { worker(queue, results, tools_by_name, context) }.each(&:join)
  calls.zip(results).map do |call, entry|
    value, seconds = entry || [INTERRUPTED, nil]
    [call, value, seconds]
  end
end

.invoke(tool, call, context) ⇒ Object

A tool with a timeout answers in band when it stalls, so one hung handler cannot stall the whole run.



67
68
69
70
71
72
73
 
# File 'lib/mistri/tool_executor.rb', line 67

def invoke(tool, call, context)
  return tool.call(call.arguments, context) unless tool.timeout

  Timeout.timeout(tool.timeout) { tool.call(call.arguments, context) }
rescue Timeout::Error
  "Error running tool #{call.name.inspect}: timed out after #{tool.timeout}s"
end

.run_one(call, tools_by_name, context) ⇒ Object 



56
57
58
59
60
61
62
63
 
# File 'lib/mistri/tool_executor.rb', line 56

def run_one(call, tools_by_name, context)
  tool = tools_by_name[call.name]
  return "Error: unknown tool #{call.name.inspect}" unless tool

  with_rails_executor { invoke(tool, call, context) }
rescue StandardError => e
  "Error running tool #{call.name.inspect}: #{e.class}: #{e.message}"
end

.thread_safe(emit) ⇒ Object

Concurrent tools share the caller's sink; sinks are not required to be thread-safe, so forwarded events serialize here.



77
78
79
80
81
82
 
# File 'lib/mistri/tool_executor.rb', line 77

def thread_safe(emit)
  return nil unless emit

  mutex = Mutex.new
  ->(event) { mutex.synchronize { emit.call(event) } }
end

.with_rails_executor ⇒ Object 



84
85
86
87
 
# File 'lib/mistri/tool_executor.rb', line 84

def with_rails_executor(&)
  executor = defined?(Rails) && Rails.respond_to?(:application) && Rails.application&.executor
  executor ? executor.wrap(&) : yield
end

.worker(queue, results, tools_by_name, context) ⇒ Object 



36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/mistri/tool_executor.rb', line 36

def worker(queue, results, tools_by_name, context)
  Thread.new do
    loop do
      call, index = begin
        queue.pop(true)
      rescue ThreadError
        break
      end
      if context.signal&.aborted?
        results[index] = [INTERRUPTED, nil]
        next
      end

      started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
      value = run_one(call, tools_by_name, context)
      results[index] = [value, Process.clock_gettime(Process::CLOCK_MONOTONIC) - started]
    end
  end
end