ALLM.Pipeline.Dsl.Runtime (allm_pipeline v0.1.0)

The interpreter the generated run/1 calls.

Everything use ALLM.Pipeline generates is three functions — run/1, __pipeline__/1 and nothing else. All of the behaviour lives here, as ordinary code against __pipeline__(:stages), so the macro stays a declaration reader and this module stays steppable in iex.

It calls what a hand-written orchestrator calls

Executor.create_pipeline_run/3, Executor.run_step/5, Executor.log_section/3, Executor.log_summary/4, Executor.borrowed_run/1, Metrics.record/3 and PipelineRun.borrow/1, plus — through ALLM.Pipeline.Lifecycle, which owns the terminal write — PipelineRun.complete/2 and Executor.fail_pipeline_run/2. Re-derive the set rather than trusting this list:

grep -oEn '(Executor|PipelineRun|Metrics|FanOut|Lifecycle).[a-z_]+[!?]?('       lib/allm/pipeline/dsl/runtime.ex lib/allm/pipeline/lifecycle.ex       | sed 's/^[0-9]*://' | sort -u

The load-bearing claim is the stronger negative one: this module reaches a step_logs / pipeline_runs / pipeline_metrics row only through those functions — no Repo., no repo(), no changeset of its own. That is the whole reason a ported pipeline's step-log tree can be structurally identical to the hand-written one rather than merely equivalent.

Ownership

The owning handle is confined to the four private functions that create or settle a run — run_owned/3, execute/4, execute_dry/4 and settle/5 — and is passed to no hook and to no return value. (run_borrowed/4 never sees one: on that path the umbrella owns the run and this pipeline terminates nothing.) Re-derive rather than trusting the list: grep -n 'owning' lib/allm/pipeline/dsl/runtime.ex. Every Context a stage body, over:, input:, skip_when: or summarize hook receives is built from PipelineRun.borrow/1's projection, so a body reading ctx.pipeline_run cannot complete its own parent run. returns: :run hands back the completed run borrowed for the same reason: Repo.update carries virtual fields through, so the struct PipelineRun.complete/2 returns still holds the token.

State threaded through the stages

Key	Meaning
`acc`	the accumulator; written only by a body returning `{item_result, acc}`
`prev`	the previous stage's output — a `fan_out`'s is `[Dsl.Item.t()]`
`parent`	the lineage parent: the last successfully executed step's log id
`carry`	values captured by a `carry: [...]` declaration
`resources`	framework-managed handles, acquired once per run by a `resource` declaration

Summary

Types

acc_update()

Whether a body wrote the accumulator.

item_result()

The value a FanOut.reduce/5 fold function or escape-hatch stage body returns, once normalized.

Functions

run(module, opts)

Execute module's declaration under a fresh PipelineRun.

Types

acc_update()

@type acc_update() :: :keep | {:update, term()}

Whether a body wrote the accumulator.

item_result()

@type item_result() :: ALLM.Pipeline.FanOut.item_result()

The value a FanOut.reduce/5 fold function or escape-hatch stage body returns, once normalized.

Functions

run(module, opts)

@spec run(
  module(),
  keyword()
) :: {:ok, term()} | {:error, term()}

Execute module's declaration under a fresh PipelineRun.

The generated run/1 is a one-line delegation to this. Returns {:ok, summary} — or {:ok, completed_run} under returns: :run — and {:error, reason} when the run could not be created or a stage failed with on_error: :fail_run.

The run is written terminal on every exit path: success, a named stage failure, a raise, an exit and a throw. A raise/exit/throw is re-raised unchanged after the run is failed, so a caller's own error handling is unaffected — the DSL adds the terminal write those paths were missing, and nothing else. That guard is ALLM.Pipeline.Lifecycle, shared with the affordance a hand-written entry point calls.

Two declarations divert before any of that:

borrowed_run: true and a lent :pipeline_run in opts — the stages run under the umbrella's handle and this function creates and terminates nothing.
dry_run: and a truthy :dry_run in opts — see ALLM.Pipeline's moduledoc, ## --dry-run, which is where that contract is stated.

They cannot both fire: declaring borrowed_run: true and dry_run: is a compile error (Dsl.__validate__!/2) — see the same section's "mutually exclusive" paragraph for why.