Module: Legion::LLM::Tools::Special

Extended by:
Legion::Logging::Helper
Defined in:
lib/legion/llm/tools/special.rb

Constant Summary collapse

LIST_SPECIAL_TOOLS_NAME = 
'legion_list_special_tools'
DEFAULT_TIMEOUT_MS = 
120_000
MAX_TIMEOUT_MS = 
600_000
PYTHON_PACKAGES = 
%w[
  python-pptx
  python-docx
  openpyxl
  pandas
  pillow
  requests
  lxml
  PyYAML
  tabulate
  markdown
].freeze

Class Method Summary collapse

Class Method Details

.array_args(args) ⇒ Object 



254
255
256
257
258
259
260
261
262
263
264
 
# File 'lib/legion/llm/tools/special.rb', line 254

def array_args(args)
  raw = args[:args] || args['args']
  case raw
  when Array
    raw.map(&:to_s)
  when String
    Shellwords.split(raw)
  else
    []
  end
end

.dispatch(tool_name, **args) ⇒ Object 



42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/legion/llm/tools/special.rb', line 42

def dispatch(tool_name, **args)
  case normalize_tool_name(tool_name)
  when LIST_SPECIAL_TOOLS_NAME
    { status: :success, result: Legion::JSON.dump(inventory) }
  when 'ruby'
    dispatch_runtime('ruby', ruby_path, **args)
  when 'python', 'python3'
    dispatch_runtime('python', python_path, **args)
  when 'pip', 'pip3'
    dispatch_runtime('pip', pip_path, **args)
  else
    { status: :error, result: "Unknown Legion special tool: #{tool_name}" }
  end
rescue StandardError => e
  handle_exception(e, level: :warn, handled: true, operation: 'llm.tools.special.dispatch', tool_name: tool_name)
  { status: :error, result: e.message }
end

.dispatch_runtime(runtime_name, executable, **args) ⇒ Object 



232
233
234
235
236
237
238
239
240
241
242
243
 
# File 'lib/legion/llm/tools/special.rb', line 232

def dispatch_runtime(runtime_name, executable, **args)
  return { status: :error, result: "#{runtime_name} runtime is unavailable." } unless executable_file?(executable)

  argv = runtime_argv(runtime_name, **args)
  return { status: :error, result: "#{runtime_name} tool requires `code`, `command`, or `args`." } if argv.empty?

  output, status = run_process(executable, argv, **args)
  result = "command=#{Shellwords.join([executable, *argv])}\nexit=#{status.exitstatus}\n#{output}"
  { status: status.success? ? :success : :error, result: result }
rescue Timeout::Error
  { status: :error, result: "#{runtime_name} tool timed out after #{timeout_ms(args)}ms." }
end

.executable_file?(path) ⇒ Boolean

Returns:

  • (Boolean) 


315
316
317
 
# File 'lib/legion/llm/tools/special.rb', line 315

def executable_file?(path)
  !path.to_s.empty? && File.file?(path.to_s) && File.executable?(path.to_s)
end

.executable_from_path(command) ⇒ Object 



308
309
310
311
312
313
 
# File 'lib/legion/llm/tools/special.rb', line 308

def executable_from_path(command)
  ENV.fetch('PATH', '').split(File::PATH_SEPARATOR).filter_map do |dir|
    path = File.join(dir, command)
    path if executable_file?(path)
  end.first
end

.inventory ⇒ Object 



60
61
62
63
64
65
66
 
# File 'lib/legion/llm/tools/special.rb', line 60

def inventory
  {
    special_tools:             special_tool_summaries,
    settings_extensions_tools: settings_extensions_tools,
    runtime:                   runtime_inventory
  }
end

.legionio_packaged_ruby?(path) ⇒ Boolean

Returns:

  • (Boolean) 


303
304
305
306
 
# File 'lib/legion/llm/tools/special.rb', line 303

def legionio_packaged_ruby?(path)
  normalized = path.to_s
  normalized.include?('/Cellar/legionio/') || normalized.include?('/libexec/bin/ruby')
end

.normalize_inventory_entry(entry) ⇒ Object 



191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
 
# File 'lib/legion/llm/tools/special.rb', line 191

def normalize_inventory_entry(entry)
  return unless entry.respond_to?(:transform_keys)

  normalized = entry.transform_keys { |key| key.respond_to?(:to_sym) ? key.to_sym : key }
  name = normalized[:name].to_s
  return if name.empty?

  {
    name:          name,
    description:   normalized[:description].to_s,
    deferred:      normalized[:deferred] == true,
    extension:     normalized[:extension],
    runner:        normalized[:runner],
    function:      normalized[:function],
    trigger_words: Array(normalized[:trigger_words]).map(&:to_s),
    parameters:    normalized[:input_schema] || normalized[:parameters] || {},
    source:        normalized[:tool_class] ? 'registry' : 'extension'
  }.compact
end

.normalize_tool_name(tool_name) ⇒ Object 



319
320
321
 
# File 'lib/legion/llm/tools/special.rb', line 319

def normalize_tool_name(tool_name)
  tool_name.to_s.tr('.', '_')
end

.path_ruby ⇒ Object 



299
300
301
 
# File 'lib/legion/llm/tools/special.rb', line 299

def path_ruby
  executable_from_path('ruby')
end

.pinned_definitions ⇒ Object 



36
37
38
39
40
 
# File 'lib/legion/llm/tools/special.rb', line 36

def pinned_definitions
  definitions = [special_tools_definition, ruby_definition]
  definitions.concat(python_definitions) if python_available?
  definitions
end

.pip_candidates_for(bin_dir) ⇒ Object 



289
290
291
292
293
 
# File 'lib/legion/llm/tools/special.rb', line 289

def pip_candidates_for(bin_dir)
  return [] if bin_dir.to_s.empty?

  [File.join(bin_dir, 'pip'), File.join(bin_dir, 'pip3')]
end

.pip_path ⇒ Object 



88
89
90
91
92
93
94
95
 
# File 'lib/legion/llm/tools/special.rb', line 88

def pip_path
  @pip_path ||= begin
    candidates = []
    candidates.concat(pip_candidates_for(File.dirname(python_path))) if python_available?
    candidates.concat(pip_candidates_for(File.join(python_venv_dir, 'bin')))
    candidates.compact.uniq.find { |path| executable_file?(path) }
  end
end

.pip_schema ⇒ Object 



158
159
160
161
162
163
164
165
166
167
168
169
 
# File 'lib/legion/llm/tools/special.rb', line 158

def pip_schema
  {
    type:       'object',
    properties: {
      command: { type: 'string', description: 'pip arguments, such as `install pandas` or `list`.' },
      args:    { type: 'array', items: { type: 'string' }, description: 'pip argv entries.' },
      cwd:     { type: 'string', description: 'Working directory. Defaults to the current process directory.' },
      timeout: { type: 'integer', description: 'Timeout in milliseconds.' },
      stdin:   { type: 'string', description: 'Optional stdin content.' }
    }
  }
end

.process_cwd(args) ⇒ Object 



272
273
274
275
 
# File 'lib/legion/llm/tools/special.rb', line 272

def process_cwd(args)
  cwd = args[:cwd] || args['cwd']
  cwd.to_s.empty? ? Dir.pwd : cwd.to_s
end

.process_ruby_path ⇒ Object 



295
296
297
 
# File 'lib/legion/llm/tools/special.rb', line 295

def process_ruby_path
  RbConfig.ruby
end

.process_stdin(args) ⇒ Object 



277
278
279
280
 
# File 'lib/legion/llm/tools/special.rb', line 277

def process_stdin(args)
  stdin = args[:stdin] || args['stdin']
  stdin.nil? ? '' : stdin.to_s
end

.python_available? ⇒ Boolean

Returns:

  • (Boolean) 


68
69
70
 
# File 'lib/legion/llm/tools/special.rb', line 68

def python_available?
  !python_path.to_s.empty?
end

.python_definitions ⇒ Object 



127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
 
# File 'lib/legion/llm/tools/special.rb', line 127

def python_definitions
  [
    Types::ToolDefinition.build(
      name:        'python',
      description: "Run Python with the Legion-managed Python environment from `legionio setup python`. Current path: #{python_path}.",
      parameters:  runtime_schema('Python'),
      source:      { type: :special, handler: :python_runtime, pinned: true, executable: python_path }
    ),
    Types::ToolDefinition.build(
      name:        'pip',
      description: "Run pip inside the Legion-managed Python environment from `legionio setup python`. Current path: #{pip_path || 'unavailable'}.",
      parameters:  pip_schema,
      source:      { type: :special, handler: :pip_runtime, pinned: true, executable: pip_path }
    )
  ]
end

.python_path ⇒ Object 



79
80
81
82
83
84
85
86
 
# File 'lib/legion/llm/tools/special.rb', line 79

def python_path
  @python_path ||= begin
    candidates = []
    candidates << ENV.fetch('LEGION_PYTHON', nil)
    candidates << File.join(python_venv_dir, 'bin', 'python3')
    candidates.compact.find { |path| executable_file?(path) }
  end
end

.python_venv_dir ⇒ Object 



102
103
104
 
# File 'lib/legion/llm/tools/special.rb', line 102

def python_venv_dir
  ENV['LEGION_PYTHON_VENV'] || File.expand_path('~/.legionio/python')
end

.reset_runtime_cache! ⇒ Object 



97
98
99
100
 
# File 'lib/legion/llm/tools/special.rb', line 97

def reset_runtime_cache!
  @python_path = nil
  @pip_path = nil
end

.ruby_definition ⇒ Object 



118
119
120
121
122
123
124
125
 
# File 'lib/legion/llm/tools/special.rb', line 118

def ruby_definition
  Types::ToolDefinition.build(
    name:        'ruby',
    description: "Run Ruby with the current Legion Ruby environment. Current path: #{ruby_path}.",
    parameters:  runtime_schema('Ruby'),
    source:      { type: :special, handler: :ruby_runtime, pinned: true, executable: ruby_path }
  )
end

.ruby_path ⇒ Object 



72
73
74
75
76
77
 
# File 'lib/legion/llm/tools/special.rb', line 72

def ruby_path
  process_path = process_ruby_path
  return process_path if legionio_packaged_ruby?(process_path) && executable_file?(process_path)

  path_ruby || process_path
end

.run_process(executable, argv, **args) ⇒ Object 



266
267
268
269
270
 
# File 'lib/legion/llm/tools/special.rb', line 266

def run_process(executable, argv, **args)
  Timeout.timeout(timeout_ms(args) / 1000.0) do
    Open3.capture2e(executable, *argv, chdir: process_cwd(args), stdin_data: process_stdin(args))
  end
end

.runtime_argv(runtime_name, **args) ⇒ Object 



245
246
247
248
249
250
251
252
 
# File 'lib/legion/llm/tools/special.rb', line 245

def runtime_argv(runtime_name, **args)
  code = args[:code] || args['code']
  return [runtime_name == 'python' ? '-c' : '-e', code.to_s, *array_args(args)] unless code.to_s.empty?

  command = args[:command] || args['command']
  command_args = command.to_s.empty? ? [] : Shellwords.split(command.to_s)
  command_args + array_args(args)
end

.runtime_inventory ⇒ Object 



211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
 
# File 'lib/legion/llm/tools/special.rb', line 211

def runtime_inventory
  {
    ruby:   {
      path:            ruby_path,
      path_ruby:       path_ruby,
      process_ruby:    process_ruby_path,
      description:     RUBY_DESCRIPTION,
      bundler_version: defined?(Bundler) ? Bundler::VERSION : nil,
      bundle_gemfile:  ENV.fetch('BUNDLE_GEMFILE', nil),
      bundle_bin_path: ENV.fetch('BUNDLE_BIN_PATH', nil)
    }.compact,
    python: {
      available:        python_available?,
      path:             python_path,
      venv_dir:         python_venv_dir,
      pip:              pip_path,
      default_packages: PYTHON_PACKAGES
    }.compact
  }
end

.runtime_schema(language) ⇒ Object 



144
145
146
147
148
149
150
151
152
153
154
155
156
 
# File 'lib/legion/llm/tools/special.rb', line 144

def runtime_schema(language)
  {
    type:       'object',
    properties: {
      code:    { type: 'string', description: "#{language} code to execute with -e/-c." },
      command: { type: 'string', description: "#{language} command arguments, such as a script path plus args." },
      args:    { type: 'array', items: { type: 'string' }, description: 'Additional argv entries.' },
      cwd:     { type: 'string', description: 'Working directory. Defaults to the current process directory.' },
      timeout: { type: 'integer', description: 'Timeout in milliseconds.' },
      stdin:   { type: 'string', description: 'Optional stdin content.' }
    }
  }
end

.settings_extensions_tools ⇒ Object 



182
183
184
185
186
187
188
189
 
# File 'lib/legion/llm/tools/special.rb', line 182

def settings_extensions_tools
  return [] unless defined?(Legion::Settings::Extensions) && Legion::Settings::Extensions.respond_to?(:tools)

  Array(Legion::Settings::Extensions.tools).filter_map { |entry| normalize_inventory_entry(entry) }
rescue StandardError => e
  handle_exception(e, level: :warn, handled: true, operation: 'llm.tools.special.settings_extensions_inventory')
  []
end

.special_tool_summaries ⇒ Object 



171
172
173
174
175
176
177
178
179
180
 
# File 'lib/legion/llm/tools/special.rb', line 171

def special_tool_summaries
  pinned_definitions.map do |definition|
    {
      name:        definition.name,
      description: definition.description,
      parameters:  definition.parameters,
      source:      'legion-special'
    }
  end
end

.special_tools_definition ⇒ Object 



106
107
108
109
110
111
112
113
114
115
116
 
# File 'lib/legion/llm/tools/special.rb', line 106

def special_tools_definition
  Types::ToolDefinition.build(
    name:        LIST_SPECIAL_TOOLS_NAME,
    description: 'Show all Legion special tools available to this LLM from the current Legion::Settings::Extensions inventory.',
    parameters:  {
      type:       'object',
      properties: {}
    },
    source:      { type: :special, handler: :settings_extensions_inventory, pinned: true }
  )
end

.timeout_ms(args) ⇒ Object 



282
283
284
285
286
287
 
# File 'lib/legion/llm/tools/special.rb', line 282

def timeout_ms(args)
  requested = (args[:timeout] || args['timeout'] || DEFAULT_TIMEOUT_MS).to_i
  return DEFAULT_TIMEOUT_MS unless requested.positive?

  [requested, MAX_TIMEOUT_MS].min
end