can1357/oh-my-pi · error · TypeError

pipeline() stages must be callables

Error message

pipeline() stages must be callables

What it means

`pipeline(items, *stages)` runs each stage concurrently over the item list, feeding each stage's output into the next. Each stage must be a callable (Proc/lambda responding to `#call`); a non-callable stage raises TypeError before any work is dispatched.

Source

Thrown at packages/coding-agent/src/eval/rb/prelude.rb:495

      threads.each { |t| (t.kill rescue nil) }
      raise
    end
    raise errors[errors.keys.min] unless errors.empty?
    results
  end

  def parallel(thunks)
    list = thunks.to_a
    list.each do |t|
      raise TypeError, "parallel() expects an iterable of zero-arg callables" unless t.respond_to?(:call)
    end
    __omp_pool_map(list) { |t| t.call }
  end

  def pipeline(items, *stages)
    current = items.to_a
    stages.each do |stage|
      raise TypeError, "pipeline() stages must be callables" unless stage.respond_to?(:call)
      current = __omp_pool_map(current) { |item| stage.call(item) }
    end
    current
  end

  # -------------------------------------------------------------------------
  # Progress + budget
  # -------------------------------------------------------------------------

  def log(message)
    __omp_emit_status("log", "message" => message.to_s)
    nil
  end

  def phase(title)
    $__omp_current_phase = title.to_s
    __omp_emit_status("phase", "title" => title.to_s)
    nil

View on GitHub (pinned to 9690622007)

Solutions

1. Convert named methods to procs: `pipeline(rows, method(:normalize))`.
2. Use a lambda per stage: `pipeline(rows, ->(row) { transform(row) })`.
3. For Symbol shorthand behavior, wrap explicitly: `pipeline(rows, ->(x) { x.to_json })` instead of `:to_json`.

Example fix

// before
pipeline(rows, :normalize)
// after
pipeline(rows, method(:normalize))   # or ->(row) { normalize(row) }

raise TypeError, 'stages must be callables' unless stages.all? { |s| s.respond_to?(:call) }

def stage_callable?(stage)
  stage.respond_to?(:call)
end

begin
  out = pipeline(items, *stages)
rescue TypeError => e
  raise unless e.message.include?('stages must be callables')
  out = pipeline(items, *stages.map { |s| s.is_a?(Symbol) ? ->(x) { x.public_send(s) } : s })
end

When it happens

Trigger: Calling `pipeline(items, tool_transform)` where `tool_transform` is a Symbol, String method name, or class rather than a Proc — e.g. `pipeline(rows, :to_json)` or `pipeline(rows, "normalize")`.

Common situations: Passing method names as Symbols (Rails habit) instead of procs; passing a Hash of stage config instead of a lambda; forgetting `&:symbol` won't work here because the value itself must respond to `#call`.

Related errors

• parallel() expects an iterable of zero-arg callables
• query cannot be combined with offset/limit
• Output #{output_id} is not valid JSON: #{e.message}
• Offset #{start_line} is beyond end of output (#{total_lines}
• tool bridge is unavailable in this kernel

AI-assisted analysis of can1357/oh-my-pi@9690622007 (2026-08-31). Data as JSON: /api/errors/c3f793b5356207db. Report an issue: GitHub.