mlua-swarm-dsl 0.23.0

--- bp_dsl.lua — pure-Lua internal DSL for the Blueprint vocabulary.
---
--- `B = require("bp_dsl")`. Depends on `flow_dsl` (`F`) for Expr / Node
--- construction; `flow_dsl` does NOT depend on `bp_dsl` (one-way
--- dependency).
---
--- `B.pipeline{}` is the authoring sugar this module exists for: default
--- `in` / `out` wiring, automatic per-stage verdict-gate insertion, and a
--- 3-part retry-loop expansion. `B.stage` is a curried 2-arg constructor
--- (`B.stage "id" { ... }`) that returns a plain "stage record" — NOT yet
--- an AST Node; `B.pipeline` is what turns stage records into flow.ir
--- Nodes.

local F = require("flow_dsl")

local M = {}

-- Authoring-time warnings accumulated across every `M.pipeline` call in
-- one script run. The Rust host drains them via
-- `M.take_authoring_warnings()` after the script returns (a fresh VM per
-- build means no cross-script leakage).
M._authoring_warnings = {}

--- Return the accumulated authoring warnings and clear the buffer.
function M.take_authoring_warnings()
  local out = M._authoring_warnings
  M._authoring_warnings = {}
  return out
end

--- `B.bp{ id=, agents=, flow=, ... }` — the whole Blueprint table.
--- bp_dsl does not gate-keep field names: anything beyond `id` / `flow`
--- (e.g. `agents`, `operators`, `strategy`, `metadata`) passes through
--- verbatim — the Blueprint schema itself is the source of truth for
--- what's valid there.
function M.bp(t)
  return t
end

--- `B.agent{ md=, verdict=, ... }` — an `$agent_md` file-ref `AgentDef`
--- entry: `{ ["$agent_md"] = md, verdict = verdict, ... }`. Every sibling
--- field besides `md` passes through verbatim, mirroring the loader's own
--- shallow-merge-onto-`$agent_md` semantics (see the guide's `$agent_md`
--- file-ref expansion section).
function M.agent(t)
  local out = {}
  for k, v in pairs(t) do
    if k ~= "md" then
      out[k] = v
    end
  end
  out["$agent_md"] = t.md
  return out
end

-- ── Stage records + B.from placeholders ──────────────────────────────────

local Placeholder = {}
Placeholder.__index = Placeholder

local function is_placeholder(v)
  return type(v) == "table" and getmetatable(v) == Placeholder
end

--- `B.from "stage_id"` — an unresolved reference to another stage's `out`
--- path. Resolved by `B.pipeline` once every stage's `out` is known (so
--- forward references — a stage referencing one declared later in the
--- same pipeline — work too); referencing an undefined stage id is an
--- `error()` at `B.pipeline` time.
function M.from(stage_id)
  return setmetatable({ stage_id = stage_id }, Placeholder)
end

--- `B.stage "id" { agent=, input=, out=, gate=, halt_on=, skip_on=, retry= }` —
--- curried 2-arg stage constructor. Returns a plain "stage record" table
--- (NOT an AST Node yet — `B.pipeline` expands stage records into
--- flow.ir Nodes, applying the default-wiring + gate + retry rules
--- below). `halt_on` here overrides `B.pipeline`'s pipeline-wide default
--- for this stage only. `skip_on = { "SKIP", ... }` (GH #76 DSL sugar) wraps
--- the stage's own body in a pre-emptive `Branch` guard against the
--- stage's INPUT verdict path (`<input>.parts["verdict"]`): if that
--- upstream verdict is in the skip list, the stage's body is not
--- executed (the enclosing pipeline continues to the next stage), else
--- the body runs unchanged. `skip_on` may coexist with `halt_on` /
--- `retry` / `gate` — the skip guard nests OUTSIDE the retry loop but
--- INSIDE the enclosing gate/rest chain so a skipped stage does not
--- prevent later stages from running. `fanout = { ... }` replaces `agent`
--- (the two are mutually exclusive) and expands the stage's slot into an
--- `F.fanout` node instead of a `step` — see `B.pipeline`'s
--- "## Fanout stages" section for the sub-record's fields and how the
--- other stage options compose with it.
function M.stage(id)
  return function(t)
    t.id = id
    return t
  end
end

-- ── B.pipeline: default wiring + gate + retry expansion ─────────────────

-- A stage's default `in` path (used when the stage record's own `input`
-- is nil and it isn't a `B.from` placeholder either).
local function default_input_path(stage_id)
  return "$.d." .. stage_id
end

-- A stage's default `out` path.
local function default_out_path(stage_id)
  return "$." .. stage_id
end

-- A stage's default retry-loop counter path (used when the stage's own
-- `retry.counter` is nil).
local function default_counter_path(stage_id)
  return "$." .. stage_id .. "_n"
end

-- The verdict-gate condition for one stage: `eq(<out>.parts["verdict"],
-- lit(v))` for a single halt_on value, or an N-ary `or` of that shape
-- across every halt_on value when there's more than one.
local function gate_cond(out_path, halt_on_values)
  local verdict_path = out_path .. '.parts["verdict"]'
  if #halt_on_values == 1 then
    return F.p(verdict_path):eq(halt_on_values[1])
  end
  local eqs = {}
  for i, v in ipairs(halt_on_values) do
    eqs[i] = F.p(verdict_path):eq(v)
  end
  return F.any(eqs)
end

-- The skip-guard condition for one stage: `in(<verdict>, lit(skip_on))`
-- (GH #76 DSL sugar). `verdict_path` is the upstream verdict address this
-- stage is inspecting — by convention `<input_path>.parts["verdict"]`
-- (the stage's own INPUT path, so a chained pipeline reads the previous
-- stage's verdict), which in the non-chained R1-default case yields a
-- lookup against a config-time input where a `.parts.verdict` key is
-- typically absent (missing needle -> `in` false -> guard never fires,
-- safe by construction). Emits `{op="in", needle=path(verdict_path),
-- haystack=lit([v1, v2, ...])}` — the reverse-argument shape of the
-- `Expr:contains` builder (haystack is `self`), which is why this
-- helper builds the wrapper via `F.lit(...):contains(F.p(...))`.
local function skip_cond(verdict_path, skip_on_values)
  return F.lit(skip_on_values):contains(F.p(verdict_path))
end

-- Resolve a stage's `input` field to a path string: `B.from "x"`
-- placeholders resolve against `outs` (populated for every stage, and
-- every retry `fix` stage, before any Node is built — so both forward and
-- backward references work); `nil` falls back to `chain_default` when the
-- caller supplies one (see `B.pipeline`'s `chain` option), otherwise to
-- the R1 default; any other value is assumed to already be a path string.
local function resolve_input_path(input, stage_id, outs, chain_default)
  if input == nil then
    return chain_default or default_input_path(stage_id)
  end
  if is_placeholder(input) then
    local target = outs[input.stage_id]
    if target == nil then
      error(
        'bp_dsl: B.from("' .. tostring(input.stage_id) .. '") references an undefined stage',
        0
      )
    end
    return target
  end
  return input
end

-- Build the `step` Node for one stage record. `outs` must already carry
-- this stage's own resolved `out` path (`rec._out`, set by the
-- register-outs pass below) so R6 `B.from` references to THIS stage
-- resolve correctly even from earlier stages in the list. `chain_default`
-- (nilable) is the `input` fallback for a stage whose `input` is `nil` —
-- see `B.pipeline`'s `chain` option.
local function build_step(rec, outs, chain_default)
  local input_path = resolve_input_path(rec.input, rec.id, outs, chain_default)
  return F.step({
    id = rec.id,
    agent = rec.agent,
    input = F.p(input_path),
    out = F.p(rec._out),
  })
end

-- ── Fanout stages ───────────────────────────────────────────────────────

-- The `join` modes flow.ir's `fanout` node accepts (see
-- `mse://guides/blueprint-authoring` § "Flow node kinds"). The list keeps a
-- stable order for the error message; the set is the membership check.
local FANOUT_JOIN_MODE_LIST = { "all", "any", "race", "all_settled" }
local FANOUT_JOIN_MODES = {}
for _, mode in ipairs(FANOUT_JOIN_MODE_LIST) do
  FANOUT_JOIN_MODES[mode] = true
end

-- The default write target each item is bound to inside a lane's own ctx.
local DEFAULT_BIND_PATH = "$.item"

-- The default `join` mode: every lane runs, results gather into an array.
local DEFAULT_JOIN = "all"

-- The lane body's default `out` for the homogeneous shape (one agent, N
-- items) — the depth-1 path the bundled `mse://blueprints/samples/10-fanout`
-- uses. Only one agent writes it, so the single address is unambiguous.
local DEFAULT_LANE_OUT_PATH = "$.branch_out"

-- The lane body's default `out` for ONE heterogeneous lane. Nested under a
-- shared `$.lane` root so the N lanes land on N distinct addresses: lanes
-- are disjoint ctx copies, so N lanes sharing one depth-1 path would be an
-- alias the reader cannot tell apart in the joined result.
local function default_lane_out_path(lane_name)
  return "$.lane." .. lane_name
end

-- Normalize `fanout.lanes` into an ordered array of
-- `{ lane =, agent =, input =, out = }` records. Accepts a bare string
-- shorthand (lane name and agent name are the same) or the full table form.
-- Rejects a keyed table: `pairs` order is undefined, so a map would emit a
-- non-deterministic lane order (and therefore non-deterministic JSON).
local function normalize_lanes(stage_id, lanes)
  local where = 'bp_dsl: stage "' .. tostring(stage_id) .. '": '
  if type(lanes) ~= "table" then
    error(where .. "fanout.lanes must be an ordered array, got " .. type(lanes), 0)
  end
  local n = #lanes
  local count = 0
  for _ in pairs(lanes) do
    count = count + 1
  end
  if count == 0 then
    error(
      where
        .. "fanout.lanes is empty — declare at least one lane, or use"
        .. " fanout.agent for the homogeneous (one agent, N items) shape.",
      0
    )
  end
  if n ~= count then
    error(
      where
        .. 'fanout.lanes must be an ordered array ({ "danger", { lane ='
        .. ' "leak", agent = "gate-leak" } }), not a keyed table — `pairs`'
        .. " order is undefined, so a keyed table would emit a"
        .. " non-deterministic lane order.",
      0
    )
  end
  local out = {}
  for i = 1, n do
    local raw = lanes[i]
    if type(raw) == "string" then
      out[i] = { lane = raw, agent = raw }
    elseif type(raw) == "table" then
      local name = raw.lane or raw.agent
      if name == nil then
        error(
          where
            .. "fanout.lanes["
            .. i
            .. "] needs a `lane` name or an `agent` (a bare string is"
            .. " shorthand for both).",
          0
        )
      end
      out[i] = {
        lane = name,
        agent = raw.agent or name,
        input = raw.input,
        out = raw.out,
      }
    else
      error(
        where
          .. "fanout.lanes["
          .. i
          .. "] must be a lane-name string or a { lane =, agent =, input =,"
          .. " out = } table, got "
          .. type(raw),
        0
      )
    end
  end
  return out
end

-- Hard validation of one stage record's `fanout` sub-record, run before any
-- Node is built so a malformed record fails loud instead of emitting a
-- half-formed fanout. Caches the normalized lane list on the record
-- (`_fanout_lanes`, same convention as `_out`) for the build pass.
local function validate_fanout(rec)
  local fo = rec.fanout
  if fo == nil then
    return
  end
  local where = 'bp_dsl: stage "' .. tostring(rec.id) .. '": '
  if type(fo) ~= "table" then
    error(where .. "fanout must be a table, got " .. type(fo), 0)
  end
  if rec.agent ~= nil then
    error(
      where
        .. "`agent` and `fanout` are mutually exclusive — a fanout stage"
        .. " names its agent inside the fanout record (fanout.agent for one"
        .. " agent over N items, fanout.lanes for one agent per lane).",
      0
    )
  end
  if rec.retry ~= nil then
    error(
      where
        .. "`retry` is not supported on a fanout stage — the retry loop's"
        .. " cond reads `<out>.parts[\"verdict\"]`, but a fanout's `out`"
        .. " holds the join result rather than one agent's verdict, and a"
        .. " `retry.fix` step has no lane ctx to write back into. Put the"
        .. " retry on the aggregate stage that reduces this stage's `out`"
        .. " to a scalar verdict.",
      0
    )
  end
  if fo.agent ~= nil and fo.lanes ~= nil then
    error(
      where
        .. "fanout.agent and fanout.lanes are mutually exclusive — agent is"
        .. " the homogeneous shape (one agent over N items), lanes is the"
        .. " heterogeneous shape (one agent per lane).",
      0
    )
  end
  if fo.agent == nil and fo.lanes == nil then
    error(
      where
        .. "fanout needs either agent = \"<name>\" (one agent over N items)"
        .. " or lanes = { ... } (one agent per lane).",
      0
    )
  end
  local join = fo.join or DEFAULT_JOIN
  if not FANOUT_JOIN_MODES[join] then
    error(
      where
        .. "fanout.join must be one of "
        .. table.concat(FANOUT_JOIN_MODE_LIST, " / ")
        .. ", got "
        .. tostring(join),
      0
    )
  end
  if fo.lanes ~= nil then
    rec._fanout_lanes = normalize_lanes(rec.id, fo.lanes)
  end
end

-- Resolve a `fanout.items` field to an Expr. A `B.from "stage"` placeholder
-- becomes a `path` Expr against that stage's `out` — the case that must NOT
-- collapse into a `lit` (a literal placeholder table would emit the
-- unresolved record as data). Everything else follows flow_dsl's usual
-- Expr-or-raw-value convention (`F.p"$.x"` stays a path, a raw Lua value
-- auto-`lit`s), and `nil` means "no explicit items" so the caller applies
-- its own default.
local function resolve_items(items, outs)
  if is_placeholder(items) then
    local target = outs[items.stage_id]
    if target == nil then
      error(
        'bp_dsl: B.from("' .. tostring(items.stage_id) .. '") references an undefined stage',
        0
      )
    end
    return F.p(target)
  end
  return items
end

-- Build the heterogeneous lane body: a branch cascade on the bound item,
-- one lane per arm, the last lane the terminal `else` (the item set is
-- closed by construction — it is the literal lane-name array this stage
-- also emits as `items`). A single lane degenerates to a bare step. Each
-- lane's `input` goes through the ordinary stage-input resolution
-- (`$.d.<lane>` by default, `B.from` accepted), and its `out` defaults to
-- `$.lane.<lane>`.
local function build_lane_body(lanes, idx, bind_path, outs)
  local lane = lanes[idx]
  local step = F.step({
    agent = lane.agent,
    input = F.p(resolve_input_path(lane.input, lane.lane, outs, nil)),
    out = F.p(lane.out or default_lane_out_path(lane.lane)),
  })
  if idx >= #lanes then
    return step
  end
  return F.branch({
    cond = F.p(bind_path):eq(lane.lane),
    on_true = step,
    on_false = build_lane_body(lanes, idx + 1, bind_path, outs),
  })
end

-- Build the `fanout` Node for one stage record — `build_step`'s sibling for
-- a stage that declares `fanout = { ... }`. The stage's `out` is unchanged
-- (`$.<stage_id>` by default), so a downstream aggregate stage reads the
-- join result via `B.from "<stage_id>"` or `chain = true` exactly as it
-- would read an ordinary stage's output.
local function build_fanout(rec, outs, chain_default)
  local fo = rec.fanout
  local bind_path = fo.bind or DEFAULT_BIND_PATH
  local items = resolve_items(fo.items, outs)
  local body
  if rec._fanout_lanes ~= nil then
    local lanes = rec._fanout_lanes
    body = build_lane_body(lanes, 1, bind_path, outs)
    if items == nil then
      -- Heterogeneous default: the literal lane-name array the cascade
      -- branches on.
      local names = {}
      for i, lane in ipairs(lanes) do
        names[i] = lane.lane
      end
      items = F.lit(names)
    end
  else
    body = F.step({
      agent = fo.agent,
      input = F.p(bind_path),
      out = F.p(fo.lane_out or DEFAULT_LANE_OUT_PATH),
    })
    if items == nil then
      -- Homogeneous default: the stage's own resolved `input` is where the
      -- item array comes from (so the R1 `$.d.<id>` default, `chain`, and an
      -- explicit `input` / `B.from` all carry over unchanged).
      items = F.p(resolve_input_path(rec.input, rec.id, outs, chain_default))
    end
  end
  return F.fanout({
    items = items,
    bind = F.p(bind_path),
    join = fo.join or DEFAULT_JOIN,
    out = F.p(rec._out),
    body = body,
  })
end

--- `B.pipeline{ stage..., halt_on={"BLOCKED"}, halted_at="$.halted_at",
--- done="$.xxx" }` — the default-wiring authoring sugar. Positional
--- entries are stage records (`B.stage "id" {...}`); `halt_on` /
--- `halted_at` / `done` are pipeline-wide options. Returns a `seq` Node
--- (a raw flow.ir table).
---
--- ## Default in/out
---
--- A stage's `input` defaults to `$.d.{stage_id}`; its `out` defaults to
--- `$.{stage_id}`. An explicit `input` / `out` on the stage record
--- overrides the default (per-stage `input` may also be a `B.from`
--- placeholder — see R6 below).
---
--- ## Chained pipelines
---
--- `chain = true` at the top level of the pipeline spec changes the
--- `input` fallback for stage N (N ≥ 2) from `$.d.{stage_id}` to
--- `$.{stage[N-1]_id}` — i.e. each stage reads the previous stage's own
--- `out` path. Stage 1's default is unchanged (still `$.d.{stage_1_id}`).
--- An explicit `input` on a stage record (whether a path string or a
--- `B.from` placeholder) still overrides the chained default. Retry
--- `fix` stages are not chained: they retain the R1 default so the
--- fixer's own input can be seeded independently of the review stage's
--- output. Omitting `chain` (or setting it `false`) preserves the R1
--- default in every position.
---
--- ## Opt-in verdict gate (bafe47d4)
---
--- A stage emits a verdict gate iff it opts in explicitly. The default
--- is NO gate (fix for the pre-bafe47d4 dead-branch pattern where every
--- stage in a pipeline with pipeline-level `halt_on` got a gate whose
--- `cond` compared against a verdict the stage never emitted).
---
--- Opt-in rules (any one triggers gate emission):
---   - `gate = true` explicit on the stage record.
---   - `halt_on = {...}` set on the stage record (declares halt values,
---     implies the stage means to gate).
---   - `retry = {...}` set on the stage record (the retry loop reads
---     verdict; the post-retry gate makes sense).
---   - `gate_default = "auto"` at the pipeline level restores the old
---     cascade — pipeline-level `halt_on` is inherited by every stage
---     whose own `gate` / `halt_on` / `retry` are all unset, and they
---     emit a gate. This is an escape hatch for pre-fix bp.lua sources
---     that want their existing shape preserved; new code should not
---     use it. Default is `gate_default = "explicit"` (the new,
---     bug-fixed behavior).
---
--- A pipeline that declares pipeline-level `halt_on` but has no stage
--- opting in at all compiles to a flow that can never halt. That state is
--- reported (never an `error()`) as an authoring warning: one line pushed
--- into `M._authoring_warnings`, drained by the Rust host via
--- `M.take_authoring_warnings()` and surfaced as `dsl warn:` on the CLI /
--- `authoring_warnings` in the `bp_build` MCP response.
---
--- `gate = false` overrides all four (opts out even with retry / stage
--- halt_on / auto cascade). When set the stage's step spliced directly
--- into the enclosing `seq` with no `branch`, and the rest of the
--- pipeline continues unconditionally (NOT nested under an `else`).
---
--- When a gate emits, its shape:
---   - `cond`: `eq(path(<out>.parts["verdict"]), lit(halt_on_value))`
---     — `or`-combined across every value when there is more than one.
---     Stage-level `halt_on` supersedes pipeline-level for the cond
---     value list; the pipeline-level list stays a shared default for
---     opted-in stages that do not name their own values.
---   - `then` (halt): `assign{at=halted_at, value=lit(stage_id)}`.
---     Every remaining stage is skipped.
---   - `else`: the rest of the pipeline (next stage's step + its own
---     gate if any). The innermost `else` (after the last stage) is
---     `assign{at=done, value=lit(true)}` when `done` was given, or an
---     empty `seq{}` otherwise.
---
--- ## Retry
---
--- `retry = { max = N, fix = <stage record>, counter = "$.path" }` on a
--- stage record expands to 3 parts, in order: (1) the stage's own `step`
--- Node; (2) `loop_{counter = <counter path>, cond = <lt(counter, max)
--- AND <the gate cond above>>, max = max + 1, body = seq{fix step, stage
--- step re-run}}`; (3) the ordinary verdict gate (evaluated once more,
--- after the loop settles — or spliced in directly if `gate = false`).
--- `counter` is optional; when omitted the loop counter path defaults to
--- `"$.{stage_id}_n"`. The `fix` stage record goes through the same
--- default in/out wiring as any other stage.
---
--- ## `B.from`
---
--- Resolved against every stage's `out` path (including retry `fix`
--- stages) before any Node is assembled, so forward references work; an
--- unresolved reference is an `error()`.
---
--- ## Fanout stages
---
--- `fanout = { ... }` on a stage record (mutually exclusive with `agent`)
--- expands that stage's slot into an `F.fanout` Node instead of a `step`,
--- so a pipeline needing parallel lanes no longer has to drop out of
--- `B.pipeline` into a hand-built `F.seq{F.fanout{...}}`. Two shapes:
---
---   - homogeneous — `fanout = { agent = "check" }`: one agent dispatched
---     once per item. `items` defaults to the stage's own resolved `input`
---     (`$.d.{stage_id}`, or the `chain` / explicit / `B.from` override),
---     and the lane body is one `step` reading the bound item and writing
---     `lane_out` (default `"$.branch_out"`).
---   - heterogeneous — `fanout = { lanes = { "danger", { lane = "leak",
---     agent = "gate-leak" } } }`: one agent per lane. `lanes` must be an
---     ORDERED array (a keyed table is an `error()`: `pairs` order is
---     undefined, so the emitted lane order would be non-deterministic);
---     each entry is a lane-name string (lane name = agent name) or a
---     `{ lane=, agent=, input=, out= }` table. `items` defaults to the
---     literal lane-name array, and the body is a branch cascade on the
---     bound item — one arm per lane, the last lane the terminal `else`.
---     A lane's `input` defaults to `$.d.{lane}` (and accepts `B.from`);
---     its `out` defaults to `$.lane.{lane}`.
---
--- Other `fanout` fields: `bind` (default `"$.item"`), `join` (default
--- `"all"`; `all` / `any` / `race` / `all_settled`, anything else is an
--- `error()`), and `items` — an explicit `items` accepts an Expr
--- (`F.p"$.d.targets"`), a `B.from "stage"` placeholder (resolved to that
--- stage's `out` path), or a raw Lua value (auto-`lit`, so a bare string is
--- a literal, NOT a path — unlike a stage's `input`).
---
--- Composition with the other stage options: the stage's `out` is
--- unchanged (`$.{stage_id}` by default), so the aggregate stage reads the
--- join result via `B.from "{stage_id}"` or `chain = true`; `skip_on` and
--- `chain` work unchanged; a verdict gate (`gate = true` or a stage-level
--- `halt_on`) emits verbatim but is reported as an authoring warning — the
--- gate belongs on the aggregate stage that reduces the join result to a
--- scalar verdict, since a fanout's `out` is the join result and not one
--- agent's verdict; `gate_default = "auto"` skips fanout stages for the
--- same reason; and `retry` on a fanout stage is an `error()` (the loop
--- cond would compare that same join result, and `retry.fix` has no lane
--- ctx to write back into).
---
--- ## `skip_on` (GH #76 DSL sugar)
---
--- `skip_on = { "SKIP", "NOT_APPLICABLE", ... }` on a stage record wraps
--- the stage's own body (`step` + optional retry `loop`) with a
--- pre-emptive `Branch` whose `cond` is `in(<input>.parts["verdict"],
--- <skip_on_list>)`. When the check hits, the stage body is elided
--- (`then` = empty `Seq{}`); when it misses, the body runs unchanged
--- (`else` = the original stage body). The gate + rest chain sits
--- OUTSIDE the skip guard so a skipped stage does not prevent later
--- stages from running (the follow-on gate reads the current stage's
--- own `out`, which stays absent when skipped — the `eq` cond against
--- `halt_on` therefore evaluates false and cleanly threads through to
--- `rest`). Sibling to `halt_on`: `skip_on` skips just this stage but
--- continues, whereas `halt_on` halts the whole pipeline; the two may
--- coexist on the same stage. `skip_on = {}` is a no-op (equivalent to
--- omitting the option). The runtime-path Skip (agent submits
--- `--verdict=skip`) coexists with this DSL sugar — both paths land on
--- `DispatchOutcome::Skip`; the sugar is the *pre-emptive* path (skip
--- BEFORE dispatch based on an upstream verdict) and the runtime path
--- is the *self-declared* path (agent runs, then declares its own
--- output non-applicable).
function M.pipeline(spec)
  local halt_on = spec.halt_on or {}
  -- Default `halted_at` so a pipeline without an explicit halt-site knob
  -- still compiles: the per-stage gate always emits
  -- `assign{at=F.p(halted_at), value=lit(stage_id)}` on its `then` branch,
  -- and a nil target passes through as an empty `path{}` node that fails
  -- shape validation downstream. `"$.halted_at"` matches the bundled
  -- sample's convention (see mse://blueprints/samples/07-dsl-pipeline);
  -- authors who care can still override via `halted_at = "$.custom"`.
  local halted_at = spec.halted_at or "$.halted_at"
  local done = spec.done
  local chain = spec.chain == true
  -- bafe47d4: gate emission is opt-in by default (`"explicit"`). The
  -- old cascade behavior — pipeline-level `halt_on` triggers a gate
  -- on every stage — is available via `gate_default = "auto"` for
  -- pre-fix bp.lua sources that want their existing shape preserved.
  local gate_default = spec.gate_default or "explicit"
  if gate_default ~= "explicit" and gate_default ~= "auto" then
    error(
      'bp_dsl: gate_default must be "explicit" (default) or "auto", got '
        .. tostring(gate_default),
      0
    )
  end

  -- bafe47d4: opt-in gate decision for one stage record. Order matters —
  -- `gate = false` must win against every other opt-in signal (retry /
  -- stage halt_on / auto cascade), and `gate = true` must beat the
  -- default-off / auto-cascade split so an author can force a gate in a
  -- pipeline that otherwise wouldn't emit one. Shared by the per-stage
  -- build below and the pipeline-wide dead-halt lint.
  local function stage_wants_gate(rec)
    if rec.gate == false then
      return false
    elseif rec.gate == true then
      return true
    elseif rec.retry ~= nil then
      return true
    elseif rec.halt_on ~= nil then
      return true
    elseif gate_default == "auto" and #halt_on > 0 and rec.fanout == nil then
      -- A fanout stage is outside the auto cascade: the cascade's gate cond
      -- reads a single agent's verdict, which a join result is not (see the
      -- fanout × gate lint below). Explicit opt-in still wins above.
      return true
    else
      return false
    end
  end

  local stages = {}
  for i, rec in ipairs(spec) do
    stages[i] = rec
  end

  -- Pass 0: hard validation of every `fanout` stage record (mutually
  -- exclusive fields, join mode, lane shape). Runs before anything is
  -- registered or built so a malformed record fails loud.
  for _, rec in ipairs(stages) do
    validate_fanout(rec)
  end

  -- Pass 1: resolve every stage's (and retry fix stage's) `out` path
  -- up front, so B.from() references work regardless of declaration
  -- order.
  local outs = {}
  local function register_out(rec)
    rec._out = rec.out or default_out_path(rec.id)
    outs[rec.id] = rec._out
  end
  for _, rec in ipairs(stages) do
    register_out(rec)
    if rec.retry ~= nil then
      register_out(rec.retry.fix)
    end
  end

  -- Dead-halt lint (issue: opt-in flip drift, sibling of bp_doctor's
  -- verdict_contract_never_read). Pipeline-level `halt_on` with zero
  -- gate-emitting stages compiles to a flow that can never halt — the
  -- halt machinery is decorative. `done` is deliberately NOT part of the
  -- trigger (without gates the final assign still runs unconditionally),
  -- and an explicit `halted_at` alone is not flagged either (it is a
  -- target path, not halt intent).
  --
  -- The same walk carries the fanout × verdict-gate lint: a gate on a
  -- fanout stage compares the stage's `out` — the join result — against the
  -- halt values, so it can never see a single agent's verdict. That is
  -- reported, never an `error()`: the gate is still emitted exactly as
  -- written (same posture as the dead-halt lint).
  local any_gate = false
  local stage_ids = {}
  for i, rec in ipairs(stages) do
    stage_ids[i] = tostring(rec.id)
    local wants_gate = stage_wants_gate(rec)
    if wants_gate then
      any_gate = true
    end
    if wants_gate and rec.fanout ~= nil then
      M._authoring_warnings[#M._authoring_warnings + 1] = 'B.pipeline stage "'
        .. tostring(rec.id)
        .. '": a verdict gate on a fanout stage compares '
        .. tostring(rec._out)
        .. '.parts["verdict"], but '
        .. tostring(rec._out)
        .. " holds the fanout's join result — not one agent's verdict — so"
        .. " the gate can never fire. Put the gate on the aggregate stage"
        .. ' that reads this one (B.stage "aggregate" { agent = ..., input ='
        .. ' B.from "'
        .. tostring(rec.id)
        .. '", gate = true }) and reduces the join result to a scalar'
        .. " verdict. The gate is emitted as written."
    end
  end

  if #halt_on > 0 then
    if not any_gate then
      M._authoring_warnings[#M._authoring_warnings + 1] = "B.pipeline stages ["
        .. table.concat(stage_ids, ", ")
        .. "]: halt_on = {"
        .. table.concat(halt_on, ", ")
        .. "} is declared at the pipeline level but no stage emits a verdict"
        .. " gate — gates are opt-in since bafe47d4, so this pipeline can"
        .. " never halt. Add gate = true (or a stage-level halt_on / retry)"
        .. " to at least one stage, or set gate_default = \"auto\" to restore"
        .. " the pre-flip cascade for this source."
    end
  end

  -- Pass 2: build, from the first stage forward, the step (+ retry loop)
  -- + gate chain, threading `rest_else` (the tail of the pipeline) into
  -- each gate's `else`.
  local function build_from(idx, rest_else)
    if idx > #stages then
      return rest_else
    end
    local rec = stages[idx]
    local this_halt_on = rec.halt_on or halt_on
    local chain_default
    if chain and idx >= 2 then
      chain_default = stages[idx - 1]._out
    end
    local step_node
    if rec.fanout ~= nil then
      step_node = build_fanout(rec, outs, chain_default)
    else
      step_node = build_step(rec, outs, chain_default)
    end
    local rest = build_from(idx + 1, rest_else)

    -- GH #76 DSL sugar: `skip_on` wraps the stage's OWN body (step + optional
    -- retry loop) with a pre-emptive Branch. Empty / nil list is a
    -- no-op (unset case). The verdict path defaults to
    -- `<input_path>.parts["verdict"]`, which in a chained pipeline
    -- (chain=true, or explicit `input = B.from "prev"`) resolves to
    -- the previous stage's own verdict; in the R1-default case the
    -- input is `$.d.<stage_id>` where `.parts.verdict` is typically
    -- absent -> `in` false -> guard never fires (safe default).
    local skip_on = rec.skip_on
    local wants_skip_guard = skip_on ~= nil and #skip_on > 0
    local skip_verdict_path
    if wants_skip_guard then
      local input_path =
        resolve_input_path(rec.input, rec.id, outs, chain_default)
      skip_verdict_path = input_path .. '.parts["verdict"]'
    end

    local body_children = { step_node }

    if rec.retry ~= nil then
      local fix_step = build_step(rec.retry.fix, outs)
      local max = rec.retry.max
      local counter_path = rec.retry.counter or default_counter_path(rec.id)
      local loop_cond = F.p(counter_path):lt(max):And(gate_cond(rec._out, this_halt_on))
      body_children[#body_children + 1] = F.loop_({
        counter = F.p(counter_path),
        cond = loop_cond,
        max = max + 1,
        body = F.seq({ fix_step, step_node }),
      })
    end

    local children
    if wants_skip_guard then
      children = {
        F.branch({
          cond = skip_cond(skip_verdict_path, skip_on),
          on_true = F.seq({}),
          on_false = F.seq(body_children),
        }),
      }
    else
      children = body_children
    end

    -- bafe47d4: opt-in gate decision (see `stage_wants_gate` above).
    local wants_gate = stage_wants_gate(rec)

    if not wants_gate then
      children[#children + 1] = rest
      return F.seq(children)
    end

    children[#children + 1] = F.branch({
      cond = gate_cond(rec._out, this_halt_on),
      on_true = F.assign({ at = F.p(halted_at), value = F.lit(rec.id) }),
      on_false = rest,
    })
    return F.seq(children)
  end

  local final_else
  if done ~= nil then
    final_else = F.assign({ at = F.p(done), value = F.lit(true) })
  else
    final_else = F.seq({})
  end

  return build_from(1, final_else)
end

return M