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 fails becomes an in-band ToolResult with an explicit error fact. 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]"
OUTCOME_UNKNOWN = 
"[interrupted: this tool call's outcome is unavailable; the tool may have " \
"executed, so verify its effects before retrying]"
VERIFY_BEFORE_RETRY = 
"The tool may have completed partially; verify its effects before " \
"retrying."

Class Method Summary collapse

Class Method Details

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



35
36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/mistri/tool_executor.rb', line 35

def call(calls, tools_by_name, signal: nil, max_concurrency: 4, session: nil, emit: nil,
         app: nil)
  call_with_outcomes(
    calls,
    tools_by_name,
    signal:,
    max_concurrency:,
    session:,
    emit:,
    app:,
    prepared_arguments: false
  ).map { |call, result, seconds, _committed| [call, result, seconds] }
end

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

Agent prepares model arguments before approval and execution. This lower-level form preserves that boundary and exposes commitment so hooks cannot run for queued calls that an abort prevented from starting.



52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/mistri/tool_executor.rb', line 52

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

  delivery = EventDelivery.wrap(emit, passthrough: [InvocationTimeout])
  context_class = prepared_arguments ? PreparedContext : ToolContext
  context = context_class.new(session: session, signal: signal,
                              emit: thread_safe(delivery, signal), app: app)
  results = Array.new(calls.length)
  queue = Queue.new
  errors = 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, errors) }.each(&:join)
  unless errors.empty?
    error = errors.pop
    raise EventDelivery.unwrap(error, delivery)
  end

  calls.each_with_index.map do |call, index|
    entry = results[index]
    entry = [failure(OUTCOME_UNKNOWN), nil, true] if entry.equal?(COMMITTED)
    value, seconds, committed = entry || [failure(INTERRUPTED), nil, false]
    [call, value, seconds, committed]
  end
end

.commit(call, context) ⇒ Object 



135
136
137
138
139
140
 
# File 'lib/mistri/tool_executor.rb', line 135

def commit(call, context)
  return false if context.signal&.aborted?
  return true unless context.emit

  context.emit.call(Event.new(type: :tool_started, tool_call: call))
end

.elapsed(started) ⇒ Object 



144
145
146
 
# File 'lib/mistri/tool_executor.rb', line 144

def elapsed(started)
  started && (Process.clock_gettime(Process::CLOCK_MONOTONIC) - started)
end

.failure(content) ⇒ Object 



142
 
# File 'lib/mistri/tool_executor.rb', line 142

def failure(content) = ToolResult.new(content:, error: true)

.invoke(tool, call, context) ⇒ Object 



127
128
129
130
131
132
133
 
# File 'lib/mistri/tool_executor.rb', line 127

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

  Timeout.timeout(tool.timeout, InvocationTimeout) do
    tool.call(call.arguments, context)
  end
end

.invoke_one(tool, call, context) ⇒ Object 



111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
 
# File 'lib/mistri/tool_executor.rb', line 111

def invoke_one(tool, call, context)
  started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
  value = with_rails_executor { invoke(tool, call, context) }
  [value, elapsed(started)]
rescue EventDelivery::Failure
  raise
rescue InvocationTimeout
  content = "Error running tool #{call.name.inspect}: timed out after #{tool.timeout}s. " \
            "#{VERIFY_BEFORE_RETRY}"
  [failure(content), elapsed(started)]
rescue StandardError => e
  [failure("Error running tool #{call.name.inspect}: #{e.class}: #{e.message}. " \
           "#{VERIFY_BEFORE_RETRY}"),
   elapsed(started)]
end

.run_one(call, index, results, tools_by_name, context) ⇒ Object 



101
102
103
104
105
106
107
108
109
 
# File 'lib/mistri/tool_executor.rb', line 101

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

  return [failure(INTERRUPTED), nil, false] unless commit(call, context)

  results[index] = COMMITTED
  [*invoke_one(tool, call, context), true]
end

.thread_safe(delivery, signal) ⇒ Object

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



150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
 
# File 'lib/mistri/tool_executor.rb', line 150

def thread_safe(delivery, signal)
  return nil unless delivery

  mutex = Mutex.new
  # Already-committed workers may report after a sibling fails delivery;
  # give all of them the first failure without calling the broken sink.
  failure = nil
  lambda do |event|
    mutex.synchronize do
      next false if event.type == :tool_started && (failure || signal&.aborted?)
      raise failure if failure

      result = begin
        delivery.call(event)
      rescue InvocationTimeout
        raise
      rescue EventDelivery::Failure => e
        failure = e
        raise
      end
      event.type == :tool_started ? true : result
    end
  end
end

.with_rails_executor ⇒ Object 



175
176
177
178
 
# File 'lib/mistri/tool_executor.rb', line 175

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, errors) ⇒ Object 



79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
 
# File 'lib/mistri/tool_executor.rb', line 79

def worker(queue, results, tools_by_name, context, errors)
  Thread.new do
    loop do
      break unless errors.empty?

      call, index = begin
        queue.pop(true)
      rescue ThreadError
        break
      end
      if context.signal&.aborted?
        results[index] = [failure(INTERRUPTED), nil, false]
        next
      end

      results[index] = run_one(call, index, results, tools_by_name, context)
    end
  rescue StandardError => e
    errors << e
  end
end