mlua-swarm-dsl 0.12.3

--- 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 = {}

--- `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=, 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.
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

-- 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

--- `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).
---
--- `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()`.
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

  local stages = {}
  for i, rec in ipairs(spec) do
    stages[i] = 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

  -- 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 = build_step(rec, outs, chain_default)
    local rest = build_from(idx + 1, rest_else)

    local 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))
      children[#children + 1] = F.loop_({
        counter = F.p(counter_path),
        cond = loop_cond,
        max = max + 1,
        body = F.seq({ fix_step, step_node }),
      })
    end

    -- bafe47d4: opt-in gate decision. 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.
    local wants_gate
    if rec.gate == false then
      wants_gate = false
    elseif rec.gate == true then
      wants_gate = true
    elseif rec.retry ~= nil then
      wants_gate = true
    elseif rec.halt_on ~= nil then
      wants_gate = true
    elseif gate_default == "auto" and #halt_on > 0 then
      wants_gate = true
    else
      wants_gate = false
    end

    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