What causes "pipeline() stages must be callables" in can1357/oh-my-pi?

Pass function objects, not names: pipeline(items, normalize, clean) not pipeline(items, 'normalize', 'clean')

can1357/oh-my-pi · error · TypeError

pipeline() stages must be callables

Error message

pipeline() stages must be callables

What it means

pipeline() threads items through one-arg stage callables with a barrier between stages. It validates each stage argument is callable before use and raises TypeError otherwise. This catches bad stage lists up front rather than failing inside the pool.

Source

Thrown at packages/coding-agent/src/eval/py/prelude.py:635

        thunk raised. Pool width tracks the task tool's ``task.maxConcurrency``.
        """
        thunks = list(thunks)
        for t in thunks:
            if not callable(t):
                raise TypeError("parallel() expects an iterable of zero-arg callables")
        return _pool_map(thunks, lambda t: t())

    def pipeline(items, *stages):
        """Map items left-to-right through one-arg stage callables.

        Every item clears stage N before any item enters stage N+1 (barrier per
        stage). Stage 1 receives the original item; later stages receive the
        previous stage's result. Pool width tracks ``task.maxConcurrency``.
        """
        current = _AwaitableList(items)
        for stage in stages:
            if not callable(stage):
                raise TypeError("pipeline() stages must be callables")
            current = _pool_map(current, stage)
        return current

    def log(message):
        """Emit a status ``log`` event for TUI rendering."""
        _emit_status("log", message=str(message))
        return None

    def phase(title):
        """Record the current readable phase and emit a status ``phase`` event."""
        globals()["__omp_current_phase__"] = str(title)
        _emit_status("phase", title=str(title))
        return None

    class _Budget:
        """Live view of the host Goal Mode token budget via the host bridge."""

        @property

View on GitHub (pinned to 9690622007)

Solutions

Pass function objects, not names: pipeline(items, normalize, clean) not pipeline(items, 'normalize', 'clean')
Verify each stage is imported and defined before the pipeline call
Wrap stages needing extra args: pipeline(items, lambda x: transform(x, opts))
Print/repr the stages list to find the non-callable entry

Example fix

// before
pipeline(rows, 'dedupe', enrich)
// after
pipeline(rows, dedupe, enrich)

Defensive patterns

Strategy: validation

Validate before calling

assert all(callable(s) for s in stages), "pipeline stages must be callables"

Type guard

def is_stage(s) -> bool:
    return callable(s)

Try / catch

try:
    out = pipeline(items, *stages)
except TypeError as e:
    if "stages must be callables" in str(e):
        stages = [STAGE_REGISTRY[s] if isinstance(s, str) else s for s in stages]
        out = pipeline(items, *stages)
    else:
        raise

Prevention

Keep a name->function registry if stages are configured by name
Validate stage lists at config load time
Don't confuse line-magic strings with functions

When it happens

Trigger: Calling pipeline(items, stage1, stage2) where a stage is not a callable — e.g. pipeline(rows, 'normalize', clean), passing a config object, or a value that is a method-call result instead of the method.

Common situations: Typing a stage name as a string instead of the function, forgetting to import the stage function, or passing a partially applied result that is data rather than a function.

Related errors

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