Module: Legion::Extensions::Exec::Runners::Shell

Extended by:
Definitions, Shell
Included in:
Shell
Defined in:
lib/legion/extensions/exec/runners/shell.rb

Constant Summary

Constants included from Definitions

Definitions::DEFAULTS

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Definitions

definition, definition_for, definitions

Class Method Details

.trigger_words ⇒ Object 



14
15
16
 
# File 'lib/legion/extensions/exec/runners/shell.rb', line 14

def self.trigger_words
  %w[shell command exec execute terminal bash run]
end

Instance Method Details

#audit(limit: 50) ⇒ Object 



87
88
89
 
# File 'lib/legion/extensions/exec/runners/shell.rb', line 87

def audit(limit: 50, **)
  { success: true, entries: audit_log.entries(limit: limit), stats: audit_log.stats }
end

#execute(command:, cwd: Dir.pwd, timeout: Helpers::Constants::DEFAULT_TIMEOUT, env: {}) ⇒ Object 



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
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
 
# File 'lib/legion/extensions/exec/runners/shell.rb', line 30

def execute(command:, cwd: Dir.pwd, timeout: Helpers::Constants::DEFAULT_TIMEOUT, env: {}, **)
  check = default_sandbox.allowed?(command)
  return { success: false, error: :blocked, reason: check[:reason] } unless check[:allowed]

  # Rewrite bare `python3` / `python` / `pip3` / `pip` invocations to use
  # the Legion-managed venv interpreter so scripts always run inside the
  # correct environment with pre-installed packages (python-pptx, etc.).
  command = rewrite_python_command(command)

  start_time = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC)
  timeout_secs = [timeout, Helpers::Constants::MAX_TIMEOUT].min / 1000.0

  begin
    stdout, stderr, status = Timeout.timeout(timeout_secs) do
      Open3.capture3(env, command, chdir: cwd)
    end
  rescue Timeout::Error => _e
    return { success: false, error: :timeout, timeout_ms: timeout }
  rescue ArgumentError => e
    return { success: false, error: e.message }
  end

  duration_ms = ((::Process.clock_gettime(::Process::CLOCK_MONOTONIC) - start_time) * 1000).round
  exit_code   = status.exitstatus.to_i
  truncated   = false

  if stdout.bytesize > Helpers::Constants::MAX_OUTPUT_BYTES
    stdout    = stdout.byteslice(0, Helpers::Constants::MAX_OUTPUT_BYTES)
    truncated = true
  end

  stderr = stderr.byteslice(0, Helpers::Constants::MAX_OUTPUT_BYTES) if stderr.bytesize > Helpers::Constants::MAX_OUTPUT_BYTES

  audit_log.record(command: command, cwd: cwd, exit_code: exit_code,
                   duration_ms: duration_ms, truncated: truncated)

  Legion::Logging.debug("[lex-exec] exit=#{exit_code} duration=#{duration_ms}ms cmd=#{command}") # rubocop:disable Legion/HelperMigration/DirectLogging

  {
    success:     exit_code.zero?,
    stdout:      stdout,
    stderr:      stderr,
    exit_code:   exit_code,
    duration_ms: duration_ms,
    truncated:   truncated
  }
end