Class: SwarmSDK::Workflow::TransformerExecutor

Inherits:
Object
  • Object
show all
Defined in:
lib/swarm_sdk/workflow/transformer_executor.rb

Overview

Executes bash command transformers for node input/output transformation

Transformers are shell commands that receive NodeContext data on STDIN as JSON and produce transformed content on STDOUT.

Exit Code Behavior

  • Exit 0: Transform success

    • Use STDOUT as the transformed content
    • Node execution proceeds with transformed content

  • Exit 1: Skip node execution (pass-through)

    • STDOUT is IGNORED
    • Input transformer: Use current_input unchanged (no transformation)
    • Output transformer: Use result.content unchanged (no transformation)
    • For input transformer: Also skips the node's LLM execution

  • Exit 2: Halt entire workflow

    • STDOUT is IGNORED
    • STDERR is shown as error message
    • Workflow stops immediately with error

JSON Input Format (STDIN)

Input transformer receives:

{
  "event": "input",
  "node": "implementation",
  "original_prompt": "Build auth API",
  "content": "PLAN: Create endpoints...",
  "all_results": {
    "planning": {
      "content": "Create endpoints...",
      "agent": "planner",
      "duration": 2.5,
      "success": true
    }
  },
  "dependencies": ["planning"]
}

Output transformer receives:

{
  "event": "output",
  "node": "implementation",
  "original_prompt": "Build auth API",
  "content": "Implementation complete",
  "all_results": {
    "planning": {...},
    "implementation": {...}
  }
}

Examples:

Input transformer that validates

# validate.sh
#!/bin/bash
INPUT=$(cat)
CONTENT=$(echo "$INPUT" | jq -r '.content')

if [ ${#CONTENT} -gt 10000 ]; then
  echo "Content too long" >&2
  exit 2  # Halt workflow
fi

echo "$CONTENT"
exit 0

Input transformer that caches (skip execution)

# cache_check.sh
#!/bin/bash
INPUT=$(cat)
CONTENT=$(echo "$INPUT" | jq -r '.content')

if cached "$CONTENT"; then
  exit 1  # Skip node execution, pass through unchanged
fi

echo "$CONTENT"
exit 0

Defined Under Namespace

Classes: TransformerResult

Class Method Summary collapse

Class Method Details

.execute(command:, context:, event:, node_name:, fallback_content:, timeout: nil) ⇒ TransformerResult

Execute a transformer shell command

Parameters:

  • command (String) —

    Shell command to execute

  • context (NodeContext) —

    Node context for building JSON input

  • event (String) —

    Event type ("input" or "output")

  • node_name (Symbol) —

    Current node name

  • fallback_content (String) —

    Content to use if skip (exit 1)

  • timeout (Integer) (defaults to: nil) —

    Timeout in seconds (default: 60)

Returns:



116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
 
# File 'lib/swarm_sdk/workflow/transformer_executor.rb', line 116

def execute(command:, context:, event:, node_name:, fallback_content:, timeout: nil)
  timeout ||= SwarmSDK.config.transformer_command_timeout

  # Build JSON input for transformer
  input_json = build_transformer_input(context, event, node_name)

  # Build environment variables
  env = build_environment(node_name: node_name)

  # Execute command with JSON stdin and timeout
  stdout, stderr, status = Timeout.timeout(timeout) do
    Open3.capture3(
      env,
      command,
      stdin_data: JSON.generate(input_json),
    )
  end

  # Handle exit code
  # Exit 0: Transform success, use STDOUT
  # Exit 1: Skip node execution, use fallback_content (IGNORE STDOUT)
  # Exit 2: Halt workflow with error (IGNORE STDOUT)
  case status.exitstatus
  when 0
    # Success: use STDOUT as transformed content (strip trailing newline)
    TransformerResult.new(
      success: true,
      content: stdout.chomp, # Remove trailing newline from echo
      skip_execution: false,
      halt: false,
      error_message: nil,
    )
  when 1
    # Skip node execution: use fallback_content unchanged (IGNORE STDOUT)
    # For input transformer: skip_execution = true (skip LLM call)
    # For output transformer: skip_execution = false (just pass through)
    TransformerResult.new(
      success: true,
      content: fallback_content,
      skip_execution: (event == "input"), # Only skip LLM for input transformers
      halt: false,
      error_message: nil,
    )
  when 2
    # Halt workflow: return error (IGNORE STDOUT)
    error_msg = stderr.strip.empty? ? "Transformer halted workflow (exit 2)" : stderr.strip
    TransformerResult.new(
      success: false,
      content: nil,
      skip_execution: false,
      halt: true,
      error_message: error_msg,
    )
  else
    # Unknown exit code: treat as error (halt)
    error_msg = "Transformer exited with code #{status.exitstatus}\nSTDERR: #{stderr}"
    TransformerResult.new(
      success: false,
      content: nil,
      skip_execution: false,
      halt: true,
      error_message: error_msg,
    )
  end
rescue Timeout::Error
  # Timeout: halt workflow
  TransformerResult.new(
    success: false,
    content: nil,
    skip_execution: false,
    halt: true,
    error_message: "Transformer command timed out after #{timeout}s",
  )
rescue StandardError => e
  # Execution error: halt workflow
  TransformerResult.new(
    success: false,
    content: nil,
    skip_execution: false,
    halt: true,
    error_message: "Transformer command failed: #{e.message}",
  )
end