dataflow_rs/engine/

workflow_executor.rs

//! # Workflow Execution Module
//!
//! This module handles the execution of workflows and their associated tasks.
//! It provides a clean separation between workflow orchestration and task execution.

use crate::engine::error::{
    DataflowError, ErrorContextConfig, ErrorInfo, Result, from_datalogic_eval, service_error_code,
};
use crate::engine::executor::{
    ArenaContext, eval_to_owned, evaluate_condition, evaluate_condition_in_arena, with_arena,
};
use crate::engine::for_each::ForEach;
use crate::engine::functions::BoxedFunctionHandler;
use crate::engine::message::{AuditTrail, Change, Message};
use crate::engine::observer::{
    ExecutionObserver, MessageFinished, MessageStarted, TaskEvent, WorkflowFinished,
    WorkflowStarted,
};
use crate::engine::task::{HaltOn, Task};
use crate::engine::task_context::{CallStamp, TaskIdentity};
use crate::engine::task_executor::TaskExecutor;
use crate::engine::task_outcome::TaskOutcome;
use crate::engine::trace::{
    ExecutionStep, ExecutionTrace, StepStamp, StepTiming, duration_us_between,
};
use crate::engine::utils::{
    compute_path_parts, get_nested_value_parts, remove_nested_value, set_nested_value,
    set_nested_value_parts, strip_hash_prefix,
};
use crate::engine::workflow::{LoopConfig, Workflow};
use chrono::{DateTime, Utc};
use core::time::Duration;
use datalogic_rs::{Engine, Logic};
use datavalue::OwnedDataValue;
use log::{debug, error, info, warn};
use serde_json::Value;
use std::collections::HashMap;
use std::sync::Arc;

/// Result of handling a task, including possible control flow signals
enum TaskControlFlow {
    /// Continue executing the next task
    Continue,
    /// Stop executing further tasks in this workflow (filter halt)
    HaltWorkflow,
}

/// Constants shared by every task in one pass over a workflow's task list.
///
/// Tracks one workflow's observer span across however many sweeps it runs.
///
/// `started_at` doubles as "has `workflow_started` been emitted": a looping
/// workflow opens the span on its first admitted sweep and closes it once, so
/// the observer sees one pair for the whole loop rather than one per sweep.
#[derive(Default)]
struct WorkflowSpan {
    started_at: Option<DateTime<Utc>>,
    sweeps: u32,
}

/// Bundles the per-message timestamp with the loop counter so that threading
/// the counter through the task loop did not push `run_tasks_slice_in_arena`
/// and `handle_task_result` past clippy's argument-count threshold.
#[derive(Clone, Copy)]
struct PassCtx {
    /// The single `Utc::now()` read for this `process_message` call, shared by
    /// every `AuditTrail` it produces.
    now: DateTime<Utc>,
    /// Loop counter of the sweep this pass is, or `None` for a workflow
    /// without a `loop`.
    loop_counter: Option<i64>,
    /// Fan-out element being recorded, or `None` outside a `for_each`. Set
    /// per element on a copy of the pass by `run_for_each`.
    element_index: Option<usize>,
}

impl PassCtx {
    /// The single pass of a workflow without a `loop`.
    #[inline]
    fn once(now: DateTime<Utc>) -> Self {
        Self {
            now,
            loop_counter: None,
            element_index: None,
        }
    }

    /// Record the executed-step trace entry for one task.
    ///
    /// Emitted identically by both task loops — only `mapping_contexts`
    /// differs, since the sync stretch collects per-mapping snapshots for `map`
    /// tasks and the async boundary never has any. Shared so a field added to
    /// the step cannot be added to one loop and forgotten in the other.
    fn note_executed(
        self,
        trace: Option<&mut ExecutionTrace>,
        workflow_id: &str,
        task_id: &str,
        message: &Message,
        clocks: TaskClocks,
        mapping_contexts: Option<Vec<Value>>,
    ) {
        let Some(t) = trace else {
            return;
        };
        let started_at = clocks.trace_start.unwrap_or(self.now);
        t.add_executed_step(
            workflow_id,
            task_id,
            message,
            StepTiming {
                started_at,
                duration_us: duration_us_between(started_at, Utc::now()),
            },
            mapping_contexts,
            StepStamp {
                loop_counter: self.loop_counter,
                element_index: self.element_index,
            },
        );
    }
}

impl PassCtx {
    /// As [`Self::note_executed`], with the step's timing measured by the
    /// caller.
    ///
    /// A fan-out element is recorded when the calls are folded back, which
    /// may be long after the call ended — and for concurrent calls, after
    /// several others. Timing it at record time would report the fold, so
    /// `run_element` times the call itself and hands the result here.
    fn note_executed_timed(
        self,
        trace: Option<&mut ExecutionTrace>,
        workflow_id: &str,
        task_id: &str,
        message: &Message,
        timing: StepTiming,
    ) {
        if let Some(t) = trace {
            t.add_executed_step(
                workflow_id,
                task_id,
                message,
                timing,
                None,
                StepStamp {
                    loop_counter: self.loop_counter,
                    element_index: self.element_index,
                },
            );
        }
    }
}

/// The two clock reads and the error watermark a task takes before its body
/// runs.
///
/// Bundled and sampled in one place because *when* they are read is the whole
/// contract: a `trace_start` taken after the body would mistime the step, and
/// an `errors_before` taken after it would drop every error the task
/// contributed. Both task loops spelled the three out identically, where a
/// one-sided edit had nothing to catch it.
#[derive(Clone, Copy)]
struct TaskClocks {
    /// `Utc::now()` at task start, but only when a trace is live.
    trace_start: Option<DateTime<Utc>>,
    /// `trace_start`, or the observer's own clock read when only an observer is
    /// attached. `None` on the plain path, which is what keeps the documented
    /// one-`Utc::now()`-per-message invariant.
    obs_start: Option<DateTime<Utc>>,
    /// `message.errors.len()` immediately before the body ran, so the errors
    /// this task contributed are exactly the tail beyond this index.
    errors_before: usize,
}

/// The task list one pass walks, plus the workflow it belongs to.
///
/// A pass runs over either the body (`workflow.tasks`) or a loop's `setup`
/// list. Group jump targets index into whichever list is being walked, so the
/// list travels with the workflow rather than being read off it.
#[derive(Clone, Copy)]
struct PassList<'a> {
    workflow: &'a Workflow,
    tasks: &'a [Task],
}

impl<'a> PassList<'a> {
    /// The workflow's body — the whole of a non-looping workflow, and one sweep
    /// of a looping one.
    #[inline]
    fn body(workflow: &'a Workflow) -> Self {
        Self {
            workflow,
            tasks: &workflow.tasks,
        }
    }
}

/// What the group gate and the two conditions decided about one task.
enum Admission {
    /// A terminal group closed at this task — the workflow halts before it.
    Halt,
    /// A group's condition was false; resume at this absolute task index.
    Jump(usize),
    /// The task's own condition was false; move on to the next task.
    Skip,
    /// Run it.
    Run,
}

/// Run the group gate and both conditions for one task, recording the skips
/// they imply.
///
/// This is the identical opening of both task loops: close spans that ended
/// before this task (a terminal one halts), open the spans that start here (a
/// false condition jumps past the span without consulting the member tasks'
/// own conditions), then the task's own condition. Only the *evaluation*
/// differs between the loops — the owned context on the async path, the shared
/// arena view on the sync one — so it arrives as `eval`. Both flavours already
/// map a `None` condition to `true`, so `eval` takes the `Option` directly.
fn admit_task(
    list: PassList<'_>,
    task: &Task,
    abs: usize,
    gate: &mut GroupGate,
    mut trace: Option<&mut ExecutionTrace>,
    pass: PassCtx,
    mut eval: impl FnMut(Option<&Arc<Logic>>) -> Result<bool>,
) -> Result<Admission> {
    if gate.close_through(abs) {
        return Ok(Admission::Halt);
    }

    if let Some(target) = gate.enter(task, &mut eval)? {
        note_group_skip(trace.as_deref_mut(), list, abs, target, pass.loop_counter);
        return Ok(Admission::Jump(target));
    }

    if !eval(task.compiled_condition.as_ref())? {
        note_task_skip(trace, &list.workflow.id, &task.id, pass.loop_counter);
        return Ok(Admission::Skip);
    }

    Ok(Admission::Run)
}

/// The two per-*task* values `handle_task_result` needs beyond the shared
/// [`PassCtx`].
///
/// Bundled rather than passed separately because `handle_task_result` already
/// sits at clippy's `too_many_arguments` threshold, and `PassCtx` cannot carry
/// them — it is per-pass and shared by every task in a sweep.
#[derive(Clone, Copy)]
struct TaskPass {
    /// The task-level `continue_on_error` flag.
    continue_on_error: bool,
    /// The task-level `terminal` flag — halt the workflow once this task has
    /// run, whatever it returned.
    terminal: bool,
    /// The task-level `halt_on` rule — halt once this task has run *and* failed.
    /// Folded together with `terminal` by [`Self::halts_at`].
    halt_on: HaltOn,
    /// `message.errors.len()` immediately before this task ran, so the errors it
    /// contributed can be identified as the tail beyond this index.
    errors_before: usize,
}

impl TaskPass {
    /// Whether this task ends the workflow, given the status its run recorded.
    ///
    /// `terminal` is unconditional — it is about position. `halt_on` fires only
    /// on a failure, which is a status of `400` or above: the same threshold the
    /// classification in `handle_task_result` already splits on to warn (4xx) or
    /// to record `TASK_STATUS_ERROR` (5xx). Defined once here so the two notions
    /// of "failed" cannot drift apart.
    #[inline]
    fn halts_at(self, status: u16) -> bool {
        self.terminal || (self.halt_on == HaltOn::Failure && status >= 400)
    }
}

/// The per-task values every call of one fan-out shares.
///
/// Bundled so `run_element` stays under clippy's argument-count limit, and
/// `Copy` so each call's future takes its own.
#[derive(Clone, Copy)]
struct ElementCall<'a> {
    workflow: &'a Workflow,
    task: &'a Task,
    for_each: &'a ForEach,
    /// The enclosing loop sweep, stamped on each call alongside its element.
    loop_counter: Option<i64>,
    /// Whether to read the clock around each call — only when a trace or an
    /// observer will use it, keeping the one-`Utc::now()`-per-message
    /// invariant on the plain path.
    timed: bool,
}

/// What one fan-out call left behind once its copy of the message is gone.
///
/// A call's copy is reduced to this the moment the call ends, so at most
/// `max_concurrency` copies are alive at once however long the array is.
struct ElementOutcome {
    index: usize,
    /// The call's outcome. Its writes are in `changes`, not here, so an `Err`
    /// simply has none: a call that failed hard contributes no writes.
    result: Result<TaskOutcome>,
    /// Every write the call made through `TaskContext::set`, in order.
    changes: Vec<Change>,
    /// Errors the call recorded itself, through `TaskContext::add_error`.
    new_errors: Vec<ErrorInfo>,
    /// The value at `collect` in the call's copy, or `Null`.
    collected: OwnedDataValue,
    /// Measured around the call when [`ElementCall::timed`] is set.
    timing: Option<StepTiming>,
}

impl ElementOutcome {
    /// Whether this element failed: a hard `Err`, or a recorded status of 400
    /// or more — the threshold `halt_on` and the `null` result both use.
    fn failed(&self) -> bool {
        match &self.result {
            Err(_) => true,
            Ok(outcome) => outcome.audit_status().is_some_and(|s| s >= 400),
        }
    }

    /// Whether no further call may start after this one.
    ///
    /// True when the call halts the workflow, or fails the task outright — an
    /// `Err` or a `5xx` without `continue_on_error`. Calls already running
    /// still finish; this only stops new ones.
    fn stops_fan_out(&self, continue_on_error: bool) -> bool {
        match &self.result {
            Err(_) => !continue_on_error,
            Ok(TaskOutcome::Halt) => true,
            Ok(outcome) => !continue_on_error && outcome.audit_status().is_some_and(|s| s >= 500),
        }
    }
}

/// The per-element `TaskPass`. `terminal` and `halt_on` are neutralised here
/// and applied once, to the whole fan-out, by `run_for_each` — a terminal
/// task must halt after its last element, not after its first.
fn fan_out_pass(task: &Task, errors_before: usize) -> TaskPass {
    TaskPass {
        continue_on_error: task.continue_on_error,
        terminal: false,
        halt_on: HaltOn::Never,
        errors_before,
    }
}

/// The control flow a whole fan-out ends in: a halt any element or the
/// failure rule requested, or `terminal`.
fn fan_out_flow(task: &Task, halt: bool) -> TaskControlFlow {
    if halt || task.terminal {
        TaskControlFlow::HaltWorkflow
    } else {
        TaskControlFlow::Continue
    }
}

/// One slice of a workflow's task list, plus the group state that spans slices.
///
/// Bundled into a single parameter because `run_tasks_slice_in_arena` already
/// sits at clippy's `too_many_arguments` threshold, and because the three
/// travel together: an absolute task index is `offset + i`, and `gate` is the
/// only thing that has to survive from one slice to the next.
struct TaskSlice<'a, 'arena> {
    /// The tasks to run — a sub-slice of `list.tasks`.
    tasks: &'arena [Task],
    /// Index of `tasks[0]` within `list.tasks`.
    offset: usize,
    /// Group state for the whole pass, shared across every slice in it.
    gate: &'a mut GroupGate,
    /// The whole list this slice was cut from, and its workflow.
    list: PassList<'arena>,
}

/// Result of running one slice of a workflow's task list.
enum SliceOutcome {
    /// The slice ran to its end.
    Completed,
    /// A task halted the workflow — `TaskOutcome::Halt`, `Task::terminal`, or
    /// the end of a terminal group.
    Halted,
    /// A group condition was false and its span ends beyond this slice, so the
    /// caller must resume at this absolute task index.
    JumpTo(usize),
}

/// Tracks which task groups are currently open during one pass over a
/// workflow's task list.
///
/// Group spans are recorded at parse time on the task that opens them
/// (`Task::group_starts`), so the executor keeps walking a flat `&[Task]`.
/// This gate turns those spans back into control flow: evaluate a group's
/// condition **once** on entry, jump past the span when it is false, and halt
/// when a terminal group closes.
///
/// A workflow using no groups never pushes, so the gate costs one
/// `Vec::is_empty` check per task and never allocates.
#[derive(Default)]
struct GroupGate {
    /// `(end, terminal)` for each open group, outermost first.
    open: Vec<(usize, bool)>,
}

impl GroupGate {
    /// Close every open group whose span ends at or before `idx`, returning
    /// `true` if any of them was `terminal`.
    ///
    /// Driven by `end` rather than by a per-task close count because a jump can
    /// skip straight past the task that would have carried the count: with
    /// `group A { group B { t1 } }` and `B`'s condition false, nothing in `A`
    /// ever executes, yet `A` was entered and — if terminal — must still halt.
    fn close_through(&mut self, idx: usize) -> bool {
        let mut terminal = false;
        while let Some(&(end, is_terminal)) = self.open.last() {
            if end > idx {
                break;
            }
            self.open.pop();
            terminal |= is_terminal;
        }
        terminal
    }

    /// Evaluate the groups opening at `task`, outermost first. Returns
    /// `Some(end)` when one's condition is false and the cursor must jump past
    /// its span; the groups outside it stay open.
    fn enter<F>(&mut self, task: &Task, mut eval: F) -> Result<Option<usize>>
    where
        F: FnMut(Option<&Arc<Logic>>) -> Result<bool>,
    {
        for group in &task.group_starts {
            if !eval(group.compiled_condition.as_ref())? {
                return Ok(Some(group.end));
            }
            self.open.push((group.end, group.terminal));
        }
        Ok(None)
    }
}

/// Record the skip of every task in `list.tasks[from..to]` — the span of a
/// group whose condition was false.
///
/// The trace stays task-granular rather than growing a group-level step, so
/// `StepResult` and the npm wire type it mirrors are unchanged.
fn note_group_skip(
    mut trace: Option<&mut ExecutionTrace>,
    list: PassList<'_>,
    from: usize,
    to: usize,
    loop_counter: Option<i64>,
) {
    for task in &list.tasks[from..to.min(list.tasks.len())] {
        note_task_skip(
            trace.as_deref_mut(),
            &list.workflow.id,
            &task.id,
            loop_counter,
        );
    }
}

/// Result of one pass over a workflow's task list.
enum PassOutcome {
    /// The workflow condition evaluated false — no task ran.
    ConditionFalse,
    /// Every task ran (or was individually skipped) to the end of the list.
    Completed,
    /// A task returned [`TaskOutcome::Halt`].
    Halted,
}

/// Return the index of the first task at or after `start` that is *not* a
/// synchronous built-in. Used to chunk `workflow.tasks` into sync-only
/// stretches that can share a single `ArenaContext`.
fn next_async_boundary(tasks: &[Task], start: usize) -> usize {
    let mut i = start;
    while i < tasks.len() && tasks[i].function.is_sync_builtin() {
        i += 1;
    }
    i
}

/// Log and (if tracing) record a whole-workflow skip. `reason` is only for the
/// debug log — `ExecutionStep::workflow_skipped` doesn't carry one, so a
/// rollout-bucket exclusion and a false condition are indistinguishable in the
/// trace, same as before this was factored out of its four call sites.
fn note_workflow_skip(trace: Option<&mut ExecutionTrace>, workflow_id: &str, reason: &str) {
    debug!("Skipping workflow {} - {}", workflow_id, reason);
    if let Some(t) = trace {
        t.add_step(ExecutionStep::workflow_skipped(workflow_id));
    }
}

/// Log and (if tracing) record a single task's condition skip.
///
/// The async task loop and the shared-arena one both reach this point with the
/// same state, and previously spelled the block out twice — every field added
/// to the skipped step had to be added in both places, with nothing to catch a
/// one-sided edit. Companion to [`note_workflow_skip`] above.
fn note_task_skip(
    trace: Option<&mut ExecutionTrace>,
    workflow_id: &str,
    task_id: &str,
    loop_counter: Option<i64>,
) {
    debug!("Skipping task {} - condition not met", task_id);
    if let Some(t) = trace {
        t.add_step(
            ExecutionStep::task_skipped(workflow_id, task_id).with_loop_counter(loop_counter),
        );
    }
}

/// Whether `workflow` serves this message's routing bucket.
///
/// A workflow with no `rollout`, or a message with no bucket, is admitted. The
/// missing-bucket case admits deliberately: every message any existing caller
/// builds has no bucket, and the wasm entry points have no way to set one, so
/// rejecting would silently stop those workflows running.
fn rollout_admits(workflow: &Workflow, message: &Message) -> bool {
    match (workflow.rollout, message.routing_bucket()) {
        (Some(r), Some(b)) => r.accepts(b),
        _ => true,
    }
}

/// Whether `workflow` may join a shared-arena run of consecutive fully-sync
/// workflows.
///
/// A looping workflow is excluded even when every task is a sync built-in: its
/// sweeps run through `execute_inner`, which opens a fresh arena scope per
/// sweep. Bump arenas never free mid-scope, so sweeping inside one shared
/// scope would grow memory with the iteration count.
fn joins_sync_run(workflow: &Workflow) -> bool {
    workflow.fully_sync && workflow.loop_config.is_none()
}

/// Resolve one loop slot's pre-split write path, once per looping workflow.
///
/// `LogicCompiler` pre-splits every `temp_data.{…}` slot of a `loop` at build
/// time. A workflow constructed directly rather than through `Engine::builder`
/// never got that pass, so the parts are computed here instead — once, ahead
/// of the loop, rather than re-split on every iteration.
///
/// An unnamed slot resolves to an empty slice, which `set_nested_value_parts`
/// treats as a no-op: an unnamed counter still bounds the loop (the audit trail
/// carries it either way), an unnamed `as` still stops at the array's end, and
/// an unnamed `scratch` resets nothing.
fn resolve_parts(name: Option<&str>, precomputed: &Arc<[Arc<str>]>) -> Arc<[Arc<str>]> {
    match name {
        Some(name) if precomputed.is_empty() => compute_path_parts("temp_data", name),
        _ => Arc::clone(precomputed),
    }
}

/// The kind of a value, for the `loop.over` error message.
fn describe_kind(value: &OwnedDataValue) -> &'static str {
    match value {
        OwnedDataValue::Null => "null",
        OwnedDataValue::Bool(_) => "a boolean",
        OwnedDataValue::Number(_) => "a number",
        OwnedDataValue::String(_) => "a string",
        OwnedDataValue::Array(_) => "an array",
        OwnedDataValue::Object(_) => "an object",
        // The `datetime` and `tensor` variants exist only under those features.
        #[allow(unreachable_patterns)]
        _ => "a non-array value",
    }
}

/// Build a fresh `metadata.progress` object value.
fn new_progress_object(workflow_id: &str, task_id: &str, status: u16) -> OwnedDataValue {
    OwnedDataValue::Object(vec![
        (
            "workflow_id".to_string(),
            OwnedDataValue::String(workflow_id.to_string()),
        ),
        (
            "task_id".to_string(),
            OwnedDataValue::String(task_id.to_string()),
        ),
        (
            "status_code".to_string(),
            OwnedDataValue::from(u64::from(status)),
        ),
    ])
}

/// Overwrite a string slot by reusing its existing buffer where possible.
///
/// The ids written per task are drawn from a small, repeating set — in a loop
/// they are outright constant across every sweep — so the common case is
/// writing the value that is already there. Comparing first turns that case
/// into a no-op, and the mismatch case still reuses the allocation.
fn overwrite_str_in_place(slot: &mut OwnedDataValue, value: &str) {
    match slot {
        OwnedDataValue::String(existing) => {
            if existing != value {
                existing.clear();
                existing.push_str(value);
            }
        }
        _ => *slot = OwnedDataValue::String(value.to_string()),
    }
}

/// Overwrite the three fields of an existing 3-key `progress` object without
/// reallocating it. Returns `false` when the object's shape diverges from
/// `{workflow_id, task_id, status_code}`, in which case the caller replaces
/// the slot wholesale (partial overwrites here are harmless — the whole slot
/// gets replaced).
fn overwrite_progress_in_place(
    fields: &mut [(String, OwnedDataValue)],
    workflow_id: &str,
    task_id: &str,
    status: u16,
) -> bool {
    if fields.len() != 3 {
        return false;
    }
    let mut matched = 0;
    for (k, v) in fields.iter_mut() {
        match k.as_str() {
            "workflow_id" => {
                overwrite_str_in_place(v, workflow_id);
                matched += 1;
            }
            "task_id" => {
                overwrite_str_in_place(v, task_id);
                matched += 1;
            }
            "status_code" => {
                *v = OwnedDataValue::from(u64::from(status));
                matched += 1;
            }
            _ => {}
        }
    }
    matched == 3
}

/// Write `metadata.progress = {workflow_id, task_id, status_code}` with a
/// single tree walk. From the second task of a message onward the slot
/// already holds the expected 3-key object, so the three values are
/// overwritten in place, reusing the id `String` buffers — no allocation at
/// all once the shape settles. First write (or any shape divergence)
/// replaces the slot wholesale; a context whose `metadata` is missing or
/// non-Object falls back to the generic `set_nested_value` writer, which
/// creates intermediate containers as needed.
fn write_progress_metadata(
    context: &mut OwnedDataValue,
    workflow_id: &str,
    task_id: &str,
    status: u16,
) {
    if let OwnedDataValue::Object(top) = context
        && let Some((_, OwnedDataValue::Object(meta))) =
            top.iter_mut().find(|(k, _)| k == "metadata")
    {
        match meta.iter_mut().find(|(k, _)| k == "progress") {
            Some((_, slot)) => {
                if let OwnedDataValue::Object(fields) = slot
                    && overwrite_progress_in_place(fields, workflow_id, task_id, status)
                {
                    return;
                }
                *slot = new_progress_object(workflow_id, task_id, status);
            }
            None => {
                meta.push((
                    "progress".to_string(),
                    new_progress_object(workflow_id, task_id, status),
                ));
            }
        }
        return;
    }
    set_nested_value(
        context,
        "metadata.progress",
        new_progress_object(workflow_id, task_id, status),
    );
}

/// Build one context record for a failed task.
///
/// `workflow_id`, `task_id` and `status` come from the executor rather than from
/// the `ErrorInfo`: `validation` builds its entries with `ErrorInfo::simple_ref`,
/// which leaves both ids `None`, and `ErrorInfo` carries no status at all.
///
/// The error `message` and the operator-only `detail` are deliberately omitted —
/// this value lands in `Message.context`, which is serialized back to callers.
fn new_error_record(workflow_id: &str, task_id: &str, code: &str, status: u16) -> OwnedDataValue {
    OwnedDataValue::Object(vec![
        ("workflow_id".to_string(), OwnedDataValue::from(workflow_id)),
        ("task_id".to_string(), OwnedDataValue::from(task_id)),
        ("code".to_string(), OwnedDataValue::from(code)),
        (
            "status".to_string(),
            OwnedDataValue::from(u64::from(status)),
        ),
    ])
}

/// Take `node` as an `Object`, replacing whatever non-`Object` sat there.
///
/// Normalise first, then destructure — the inverse order (match, then assign in
/// the fallback arm and re-match the same binding) is NLL problem case #3 and
/// does not compile.
fn as_object_slot(node: &mut OwnedDataValue) -> &mut Vec<(String, OwnedDataValue)> {
    if !matches!(node, OwnedDataValue::Object(_)) {
        *node = OwnedDataValue::Object(Vec::new());
    }
    match node {
        OwnedDataValue::Object(fields) => fields,
        _ => unreachable!("just normalised to an Object"),
    }
}

/// Append one record per entry in `new_errors` to the configured context path,
/// keeping at most `cfg.limit` of them.
///
/// Hand-walks to the slot the way [`write_progress_metadata`] does. The generic
/// [`set_nested_value`] cannot express an append — it indexes arrays by numeric
/// segment and `Null`-pads the gap — and silently no-ops when a non-numeric
/// segment meets an `Array`. A slot holding something other than an `Array` is
/// replaced wholesale rather than skipped, so the shape a workflow author reads
/// is predictable even if a `map` task wrote over the path first.
///
/// The array is created lazily, only when there is something to push, so a
/// message whose tasks all succeed keeps the exact wire shape it had before the
/// option existed — the key is absent, not `[]`.
fn append_error_records(
    context: &mut OwnedDataValue,
    cfg: &ErrorContextConfig,
    workflow_id: &str,
    task_id: &str,
    status: u16,
    new_errors: &[ErrorInfo],
) {
    if new_errors.is_empty() {
        return;
    }
    // Walk to the parent of the final segment, creating containers as needed,
    // then take the slot itself.
    let Some((last, parents)) = cfg.path_parts.split_last() else {
        return;
    };

    let mut node = context;
    for part in parents {
        let key = strip_hash_prefix(part);
        if !matches!(node, OwnedDataValue::Object(_)) {
            // A scalar or array on the way down cannot hold a named child. The
            // host declared the engine owns this path, so resolve the conflict
            // in favour of the records rather than dropping them — but say so:
            // whatever was written here is being discarded.
            warn!(
                "error context path `{}` runs through a non-object at `{}` — replacing it",
                cfg.path, key
            );
        }
        let fields = as_object_slot(node);
        let idx = match fields.iter().position(|(k, _)| k == key) {
            Some(i) => i,
            None => {
                fields.push((key.to_string(), OwnedDataValue::Object(Vec::new())));
                fields.len() - 1
            }
        };
        node = &mut fields[idx].1;
    }

    let key = strip_hash_prefix(last);
    let fields = as_object_slot(node);
    let idx = match fields.iter().position(|(k, _)| k == key) {
        Some(i) => i,
        None => {
            fields.push((key.to_string(), OwnedDataValue::Array(Vec::new())));
            fields.len() - 1
        }
    };
    let slot = &mut fields[idx].1;
    if !matches!(slot, OwnedDataValue::Array(_)) {
        warn!(
            "error context path `{}` held a non-array — replacing it",
            cfg.path
        );
        *slot = OwnedDataValue::Array(Vec::new());
    }
    let OwnedDataValue::Array(items) = slot else {
        unreachable!("just ensured an Array");
    };

    for error in new_errors {
        items.push(new_error_record(workflow_id, task_id, &error.code, status));
    }
    // Keep-newest: a looping workflow with a failing body would otherwise grow
    // this list once per sweep, and `Message.context` is deep-cloned into every
    // trace snapshot.
    if items.len() > cfg.limit {
        items.drain(..items.len() - cfg.limit);
    }
}

/// Handles the execution of workflows and their tasks
///
/// The `WorkflowExecutor` is responsible for:
/// - Evaluating workflow conditions
/// - Orchestrating task execution within workflows
/// - Managing workflow-level error handling
/// - Recording audit trails
pub struct WorkflowExecutor {
    /// Task executor for executing individual tasks
    task_executor: Arc<TaskExecutor>,
    /// Shared datalogic engine for condition evaluation
    engine: Arc<Engine>,
    /// Optional per-task observer. `None` keeps the instrumentation — and its
    /// clock reads — entirely out of the dispatch path.
    observer: Option<Arc<dyn ExecutionObserver>>,
    /// Optional context path where per-task failure codes are mirrored. `None`
    /// keeps the whole mechanism out of the dispatch path.
    error_context: Option<Arc<ErrorContextConfig>>,
}

impl WorkflowExecutor {
    /// Create a new WorkflowExecutor
    pub fn new(task_executor: Arc<TaskExecutor>, engine: Arc<Engine>) -> Self {
        Self {
            task_executor,
            engine,
            observer: None,
            error_context: None,
        }
    }

    /// Attach an observer to an existing executor. Replaces any previous one.
    pub fn with_observer(mut self, observer: Arc<dyn ExecutionObserver>) -> Self {
        self.observer = Some(observer);
        self
    }

    /// The registered observer, if any.
    ///
    /// Used by `Engine::with_new_workflows` to carry the observer across a hot
    /// reload — without it, metrics would stop silently at the first reload.
    pub fn observer(&self) -> Option<&Arc<dyn ExecutionObserver>> {
        self.observer.as_ref()
    }

    /// Attach an error-context path to an existing executor. Replaces any
    /// previous one.
    pub(crate) fn with_error_context(mut self, cfg: Arc<ErrorContextConfig>) -> Self {
        self.error_context = Some(cfg);
        self
    }

    /// The configured error-context path, if any.
    ///
    /// Used by `Engine`'s executor rebuilds to carry the setting across a hot
    /// reload or a `with_observer` call — without it, failure codes would stop
    /// being recorded silently.
    pub(crate) fn error_context(&self) -> Option<&Arc<ErrorContextConfig>> {
        self.error_context.as_ref()
    }

    /// Emit a task event, deriving the status from the dispatch result.
    ///
    /// Called before `handle_task_result`, which takes `result` by value and
    /// whose `?` propagates on a hard failure — emitting afterwards would
    /// silently drop exactly the tasks a host most wants timed.
    #[inline]
    fn emit_task_event(
        &self,
        workflow: &Workflow,
        task: &Task,
        result: &Result<(TaskOutcome, Vec<Change>)>,
        started_at: Option<DateTime<Utc>>,
    ) {
        if self.observer.is_some() {
            let duration = started_at
                .map(|s| Duration::from_micros(duration_us_between(s, Utc::now())))
                .unwrap_or_default();
            self.emit_task_finished(workflow, task, result, duration);
        }
    }

    /// Emit a task event with a duration the caller measured. The fan-out
    /// path uses this directly, since a folded element's call ended before it
    /// is recorded.
    fn emit_task_finished(
        &self,
        workflow: &Workflow,
        task: &Task,
        result: &Result<(TaskOutcome, Vec<Change>)>,
        duration: Duration,
    ) {
        if let Some(observer) = self.observer.as_ref() {
            let status = match result {
                Ok((outcome, _)) => outcome.audit_status(),
                Err(_) => Some(500),
            };
            observer.task_finished(&TaskEvent {
                workflow_id: &workflow.id,
                task_id: &task.id,
                function: task.function.function_name(),
                status,
                duration,
            });
        }
    }

    /// Emit `workflow_started` the first time a workflow is admitted, and
    /// remember when — so a looping workflow reports one pair for the whole
    /// loop rather than one per sweep.
    fn begin_workflow(&self, span: &mut WorkflowSpan, workflow: &Workflow) {
        span.sweeps += 1;
        // Unobserved: no clock read, and `started_at` stays `None` so
        // `end_workflow` is a no-op too. The crate's documented
        // one-`Utc::now()`-per-message invariant holds unchanged.
        let Some(observer) = self.observer.as_ref() else {
            return;
        };
        if span.started_at.is_some() {
            return;
        }
        span.started_at = Some(Utc::now());
        observer.workflow_started(&WorkflowStarted {
            workflow_id: &workflow.id,
        });
    }

    /// Close a span opened by [`Self::begin_workflow`]. A no-op when the
    /// workflow was never admitted, so a skipped workflow emits nothing.
    fn end_workflow(&self, span: &WorkflowSpan, workflow: &Workflow, halted: bool) {
        let Some(observer) = self.observer.as_ref() else {
            return;
        };
        let Some(started) = span.started_at else {
            return;
        };
        observer.workflow_finished(&WorkflowFinished {
            workflow_id: &workflow.id,
            duration: Duration::from_micros(duration_us_between(started, Utc::now())),
            sweeps: span.sweeps,
            halted,
        });
    }

    /// Clock read for the observer, only when one is attached.
    ///
    /// Gated so that `process_message`'s documented "one `Utc::now()` per
    /// message" holds for every caller that has not opted in.
    #[inline]
    fn observer_clock(&self) -> Option<DateTime<Utc>> {
        self.observer.as_ref().map(|_| Utc::now())
    }

    /// Sample the per-task clocks and error watermark. See [`TaskClocks`] for
    /// why the three are taken together, and here rather than in each loop.
    #[inline]
    fn open_task_clocks(&self, tracing: bool, message: &Message) -> TaskClocks {
        // Clock reads only when a trace is live or an observer is attached, so
        // the plain path keeps its documented one-`Utc::now()`-per-message
        // invariant.
        let trace_start = if tracing { Some(Utc::now()) } else { None };
        TaskClocks {
            trace_start,
            obs_start: trace_start.or_else(|| self.observer_clock()),
            // Sampled before the body runs: `validation` and
            // `TaskContext::add_error` both push during it, so the tail beyond
            // this index is exactly what this task contributed.
            errors_before: message.errors.len(),
        }
    }

    /// Get a clone of the task_functions Arc for reuse in new engines
    pub fn task_functions(&self) -> Arc<HashMap<String, BoxedFunctionHandler>> {
        self.task_executor.task_functions()
    }

    /// Borrow the handler registry, for enumeration keyed to this executor's
    /// lifetime. See `TaskExecutor::registry`.
    pub fn registry(&self) -> &HashMap<String, BoxedFunctionHandler> {
        self.task_executor.registry()
    }

    /// Execute a workflow if its condition is met
    ///
    /// This method:
    /// 1. Evaluates the workflow condition
    /// 2. Executes tasks sequentially if condition is met
    /// 3. Handles error recovery based on workflow configuration
    /// 4. Updates message metadata and audit trail
    ///
    /// # Arguments
    /// * `workflow` - The workflow to execute
    /// * `message` - The message being processed
    ///
    /// # Returns
    /// * `Result<bool>` - Ok(true) if workflow was executed, Ok(false) if skipped, Err on failure
    pub async fn execute(
        &self,
        workflow: &Workflow,
        message: &mut Message,
        now: DateTime<Utc>,
    ) -> Result<bool> {
        self.execute_inner(workflow, message, None, now).await
    }

    /// Execute a workflow with step-by-step tracing
    ///
    /// Similar to `execute` but records execution steps for debugging.
    pub async fn execute_with_trace(
        &self,
        workflow: &Workflow,
        message: &mut Message,
        trace: &mut ExecutionTrace,
        now: DateTime<Utc>,
    ) -> Result<bool> {
        self.execute_inner(workflow, message, Some(trace), now)
            .await
    }

    /// Run `workflow` against `message`: the rollout gate, then either a single
    /// pass over the task list or — for a workflow carrying a `loop` — a
    /// bounded sweep loop.
    ///
    /// `trace` is `None` for the production path and `Some(&mut trace)` for the
    /// debug path; stepping is the only behavioural difference between them.
    async fn execute_inner(
        &self,
        workflow: &Workflow,
        message: &mut Message,
        mut trace: Option<&mut ExecutionTrace>,
        now: DateTime<Utc>,
    ) -> Result<bool> {
        // Traffic-split gate, ahead of any arena work so an excluded workflow
        // costs no `ArenaContext::from_owned` walk. Reuses the existing skipped
        // path verbatim, so an excluded workflow is indistinguishable from a
        // false condition.
        if !rollout_admits(workflow, message) {
            note_workflow_skip(trace.as_deref_mut(), &workflow.id, "outside rollout bucket");
            return Ok(false);
        }

        if let Some(loop_config) = workflow.loop_config.as_ref() {
            return self
                .execute_loop(workflow, loop_config, message, trace, now)
                .await;
        }

        // Opened inside `execute_pass` the moment the condition admits, so a
        // skipped workflow leaves it closed and emits nothing.
        let mut span = WorkflowSpan::default();
        let outcome = self
            .execute_pass(
                PassList::body(workflow),
                message,
                trace.as_deref_mut(),
                PassCtx::once(now),
                &mut span,
            )
            .await;
        self.end_workflow(&span, workflow, matches!(outcome, Ok(PassOutcome::Halted)));

        match outcome {
            Ok(PassOutcome::ConditionFalse) => {
                // Last use of `trace` on this path — no reborrow needed.
                note_workflow_skip(trace, &workflow.id, "condition not met");
                Ok(false)
            }
            Ok(_) => {
                info!("Successfully completed workflow: {}", workflow.id);
                Ok(true)
            }
            Err(e) => {
                // Single-channel contract: every error appears in
                // `message.errors`. The `Result::Err` return only signals to
                // the caller that we stopped before processing further
                // workflows. The workflow-level wrapper records workflow
                // context that the underlying task error doesn't carry.
                if self.record_workflow_error(workflow, message, &e) {
                    Err(e)
                } else {
                    Ok(true)
                }
            }
        }
    }

    /// Drive a looping workflow.
    ///
    /// A loop with `setup` or `over` first runs the setup list once (the
    /// workflow condition gates it, folded in by `execute_pass`) and then
    /// evaluates `over`. A counter-only loop skips that entirely and enters its
    /// first sweep exactly as it always has — no extra condition evaluation.
    ///
    /// Per iteration — write counter, check `max`, check the element list,
    /// write the element, reset scratch, check the condition, run the body,
    /// advance — is the documented contract. The counter is in `temp_data`
    /// before the condition is evaluated, so a condition that indexes by it
    /// works on the first sweep. None of the engine-owned writes needs an arena
    /// refresh: `execute_pass` builds its `ArenaContext` from `message.context`
    /// after them.
    ///
    /// Returns `Ok(false)` only when the workflow was never admitted.
    async fn execute_loop(
        &self,
        workflow: &Workflow,
        config: &LoopConfig,
        message: &mut Message,
        mut trace: Option<&mut ExecutionTrace>,
        now: DateTime<Utc>,
    ) -> Result<bool> {
        let counter_parts = resolve_parts(config.counter.as_deref(), &config.counter_parts);
        let item_parts = resolve_parts(config.item.as_deref(), &config.item_parts);
        let scratch_parts = resolve_parts(config.scratch.as_deref(), &config.scratch_parts);
        let body = PassList::body(workflow);
        // One span for the whole loop: per-sweep events would explode
        // cardinality, so the sweep count goes on the single finished event.
        let mut span = WorkflowSpan::default();
        let mut halted = false;

        // ---- Setup, then `over` ----
        // `admitted` records that the condition already admitted this workflow
        // before setup, so a later false condition is a loop exit rather than
        // a whole-workflow skip.
        let mut admitted = false;
        let mut items: Option<Vec<OwnedDataValue>> = None;
        if !config.setup.is_empty() || config.over.is_some() {
            let setup = PassList {
                workflow,
                tasks: &config.setup,
            };
            match self
                .execute_pass(
                    setup,
                    message,
                    trace.as_deref_mut(),
                    PassCtx::once(now),
                    &mut span,
                )
                .await
            {
                Ok(PassOutcome::ConditionFalse) => {
                    note_workflow_skip(trace, &workflow.id, "condition not met");
                    return Ok(false);
                }
                Ok(PassOutcome::Halted) => {
                    // `terminal`, `halt_on` or `TaskOutcome::Halt` in setup
                    // ends the workflow before its first iteration.
                    span.sweeps = 0;
                    self.end_workflow(&span, workflow, true);
                    info!(
                        "Successfully completed workflow: {} (halted in loop setup)",
                        workflow.id
                    );
                    return Ok(true);
                }
                Ok(PassOutcome::Completed) => {}
                Err(e) => {
                    // No first iteration without a completed setup. The error
                    // is recorded either way; `continue_on_error` decides only
                    // whether the *message* goes on to the next workflow.
                    span.sweeps = 0;
                    let stop = self.record_workflow_error(workflow, message, &e);
                    self.end_workflow(&span, workflow, false);
                    return if stop { Err(e) } else { Ok(true) };
                }
            }
            admitted = true;
            // Setup is not a sweep: `WorkflowFinished::sweeps` counts
            // iterations, and `begin_workflow` counted the setup pass.
            span.sweeps = 0;

            if config.over.is_some() {
                match self.evaluate_over(workflow, config, message) {
                    Ok(list) => items = Some(list),
                    Err(e) => {
                        let stop = self.record_workflow_error(workflow, message, &e);
                        self.end_workflow(&span, workflow, false);
                        return if stop { Err(e) } else { Ok(true) };
                    }
                }
            }
        }

        let mut counter = config.init;
        let mut sweeps_run: u32 = 0;

        loop {
            // Written before the bound and condition checks so a condition
            // indexing by the counter — the per-item pattern — resolves on the
            // very first sweep.
            set_nested_value_parts(
                &mut message.context,
                &counter_parts,
                OwnedDataValue::from_i64(counter),
            );

            // `>=`, not `>`, and that is load-bearing for termination rather
            // than a style choice. `increment >= 1` is validated at build time
            // and the advance below saturates, so the counter strictly
            // increases until it pins at `i64::MAX` — which satisfies
            // `>= config.max` for every representable `max`. With `>` a loop
            // whose counter saturates would spin forever.
            if counter >= config.max {
                // Normal completion: `max` is always author-supplied, so
                // reaching it is the stated bound rather than a runaway. A
                // condition that was still true wanted to keep going, which is
                // worth a log line but not an error.
                if workflow.compiled_condition.is_some() {
                    warn!(
                        "Workflow {} stopped at its loop bound (max {}) with the condition \
                         still true after {} sweep(s)",
                        workflow.id, config.max, sweeps_run
                    );
                }
                if let Some(items) = &items
                    && let Ok(next) = usize::try_from(counter)
                    && next < items.len()
                {
                    warn!(
                        "Workflow {} stopped at its loop bound (max {}) with {} of {} \
                         element(s) of loop.over unvisited",
                        workflow.id,
                        config.max,
                        items.len() - next,
                        items.len()
                    );
                }
                break;
            }

            // The second bound of an `over` loop: the element list. Running
            // out of elements is normal completion, like reaching `max`.
            if let Some(items) = items.as_mut() {
                let Some(slot) = usize::try_from(counter).ok().and_then(|i| items.get_mut(i))
                else {
                    break;
                };
                // Moved, not cloned: the list is this loop's own evaluation
                // result, and `increment >= 1` means no index is visited twice.
                let element = std::mem::take(slot);
                set_nested_value_parts(&mut message.context, &item_parts, element);
            }

            // A fresh scratch object for this iteration, so nothing written for
            // the previous element survives into this one. Before the
            // condition, so the whole iteration — condition included — sees the
            // reset. An unnamed scratch resolves to empty parts: a no-op.
            set_nested_value_parts(
                &mut message.context,
                &scratch_parts,
                OwnedDataValue::Object(Vec::new()),
            );

            let pass = PassCtx {
                now,
                loop_counter: Some(counter),
                element_index: None,
            };

            match self
                .execute_pass(body, message, trace.as_deref_mut(), pass, &mut span)
                .await
            {
                Ok(PassOutcome::ConditionFalse) => {
                    if sweeps_run == 0 && !admitted {
                        // Never entered: indistinguishable from a plain
                        // condition-skipped workflow, and reported as one.
                        note_workflow_skip(trace.as_deref_mut(), &workflow.id, "condition not met");
                    } else {
                        debug!(
                            "Workflow {} loop exited at counter {} - condition no longer met",
                            workflow.id, counter
                        );
                    }
                    break;
                }
                Ok(PassOutcome::Halted) => {
                    sweeps_run += 1;
                    debug!(
                        "Workflow {} loop halted at counter {}",
                        workflow.id, counter
                    );
                    halted = true;
                    break;
                }
                Ok(PassOutcome::Completed) => {
                    sweeps_run += 1;
                }
                Err(e) => {
                    sweeps_run += 1;
                    // Same single-channel contract as the non-looping path. On
                    // `continue_on_error` the loop advances past the failing
                    // sweep rather than abandoning the rest — the per-item case
                    // wants item 8 processed after item 7 failed.
                    if self.record_workflow_error(workflow, message, &e) {
                        // Closed on the error path too: an observer measuring
                        // engine overhead must not lose the span of the
                        // workflow that actually failed.
                        self.end_workflow(&span, workflow, halted);
                        return Err(e);
                    }
                }
            }

            counter = counter.saturating_add(config.increment);
        }

        self.end_workflow(&span, workflow, halted);

        if sweeps_run > 0 {
            info!(
                "Successfully completed workflow: {} ({} loop sweep(s))",
                workflow.id, sweeps_run
            );
        }
        Ok(sweeps_run > 0 || admitted)
    }

    /// Run one task's fan-out: evaluate `over`, run a call per element, then
    /// fold the calls back into `message` in element order.
    ///
    /// See [`ForEach`] for the contract. The shape here is what makes it hold:
    /// every call clones the message from the same untouched state (the calls
    /// phase only ever holds `&Message`), and the fold — the only mutation —
    /// happens after every call has finished, in element order. So the result
    /// cannot depend on `max_concurrency` or on the order calls complete in.
    async fn run_for_each(
        &self,
        workflow: &Workflow,
        task: &Task,
        for_each: &ForEach,
        message: &mut Message,
        mut trace: Option<&mut ExecutionTrace>,
        pass: PassCtx,
    ) -> Result<TaskControlFlow> {
        use futures_util::stream::{FuturesUnordered, StreamExt};

        let tracing = trace.is_some();

        // ---- `over`, once ----
        // An `over` that is not an array, and an empty one, each record one
        // entry with no element index — so `metadata.progress` is written
        // exactly as it is for any task, and `continue_on_error` applies.
        let items = match self.evaluate_for_each_over(task, for_each, message) {
            Ok(items) if !items.is_empty() => items,
            other => {
                let failed = other.is_err();
                if other.is_ok() && !for_each.into_parts.is_empty() {
                    self.write_empty_results(for_each, message);
                }
                let clocks = self.open_task_clocks(tracing, message);
                let result = other.map(|_| {
                    let changes = self.empty_results_change(for_each, message);
                    (TaskOutcome::Success, changes)
                });
                self.emit_task_event(workflow, task, &result, clocks.obs_start);
                let flow = self.handle_task_result(
                    result,
                    &workflow.id_arc,
                    &task.id_arc,
                    fan_out_pass(task, clocks.errors_before),
                    message,
                    pass,
                )?;
                pass.note_executed(trace, &workflow.id, &task.id, message, clocks, None);
                let halt = matches!(flow, TaskControlFlow::HaltWorkflow)
                    || (failed && task.halt_on == HaltOn::Failure);
                return Ok(fan_out_flow(task, halt));
            }
        };
        let n = items.len();

        // ---- the calls ----
        let call = ElementCall {
            workflow,
            task,
            for_each,
            loop_counter: pass.loop_counter,
            timed: tracing || self.observer.is_some(),
        };
        let mut outcomes: Vec<Option<ElementOutcome>> = (0..n).map(|_| None).collect();
        {
            // Shared, on purpose: every call clones from this same state, and
            // nothing may change it until all of them are done.
            let base: &Message = message;
            let mut pending = items.into_iter().enumerate();
            let mut in_flight = FuturesUnordered::new();
            let mut stop = false;
            loop {
                while !stop && in_flight.len() < for_each.max_concurrency {
                    let Some((index, item)) = pending.next() else {
                        break;
                    };
                    in_flight.push(self.run_element(call, base, index, item));
                }
                let Some(outcome) = in_flight.next().await else {
                    break;
                };
                stop |= outcome.stops_fan_out(task.continue_on_error);
                let index = outcome.index;
                outcomes[index] = Some(outcome);
            }
        }

        // ---- fold back, in element order ----
        // Stops at the first element that fails the task (the `?` below) or
        // halts (the `break`); the outcomes after it are dropped unfolded.
        if !for_each.into_parts.is_empty() {
            // One slot per element, so an element that never ran — or failed —
            // is a `null` at its own index rather than a gap.
            set_nested_value_parts(
                &mut message.context,
                &for_each.into_parts,
                OwnedDataValue::Array(vec![OwnedDataValue::Null; n]),
            );
        }
        let mut halt = false;
        let mut any_failed = false;
        for outcome in outcomes.into_iter().flatten() {
            let failed = outcome.failed();
            let ElementOutcome {
                index,
                result,
                changes,
                new_errors,
                collected,
                timing,
            } = outcome;

            // Sampled before this element's own errors land, so the
            // error-context mirror in `handle_task_result` sees them as this
            // element's.
            let errors_before = message.errors.len();
            message
                .errors
                .extend(new_errors.into_iter().map(|mut error| {
                    error.element_index.get_or_insert(index);
                    error
                }));

            for change in &changes {
                if change.removed {
                    remove_nested_value(&mut message.context, &change.path);
                } else {
                    set_nested_value(&mut message.context, &change.path, change.new_value.clone());
                }
            }

            let mut changes = if message.capture_changes {
                changes
            } else {
                Vec::new()
            };
            if let Some(into) = for_each.into.as_deref() {
                let value = if failed {
                    OwnedDataValue::Null
                } else {
                    collected
                };
                if message.capture_changes {
                    changes.push(Change {
                        path: Arc::from(format!("{into}.{index}")),
                        old_value: OwnedDataValue::Null,
                        new_value: value.clone(),
                        removed: false,
                    });
                }
                let mut parts = for_each.into_parts.to_vec();
                parts.push(Arc::from(index.to_string()));
                set_nested_value_parts(&mut message.context, &parts, value);
            }

            any_failed |= failed;
            let result = result.map(|outcome| (outcome, changes));
            let element = PassCtx {
                element_index: Some(index),
                ..pass
            };
            let duration = timing
                .map(|t| Duration::from_micros(t.duration_us))
                .unwrap_or_default();
            self.emit_task_finished(workflow, task, &result, duration);
            let flow = self.handle_task_result(
                result,
                &workflow.id_arc,
                &task.id_arc,
                fan_out_pass(task, errors_before),
                message,
                element,
            )?;
            if let Some(timing) = timing {
                element.note_executed_timed(
                    trace.as_deref_mut(),
                    &workflow.id,
                    &task.id,
                    message,
                    timing,
                );
            }
            // A halting element ends the fold here, after its own record has
            // landed — later elements contribute nothing, finished or not,
            // exactly as they never started at `max_concurrency: 1`. Under
            // `fan_out_pass` only `TaskOutcome::Halt` reaches this branch.
            if matches!(flow, TaskControlFlow::HaltWorkflow) {
                halt = true;
                break;
            }
        }

        if halt {
            info!(
                "Task {} halted workflow {} during its fan-out",
                task.id, workflow.id
            );
        }
        Ok(fan_out_flow(
            task,
            halt || (task.halt_on == HaltOn::Failure && any_failed),
        ))
    }

    /// One fan-out call, against its own copy of `base`.
    ///
    /// Takes `&Message`, not `&mut`: that is the isolation guarantee. The copy
    /// carries the element at `temp_data.<as>` and its index at
    /// `temp_data.<as>_index`, records every write regardless of the real
    /// message's `capture_changes` (the fold replays them), and is reduced to
    /// an [`ElementOutcome`] as soon as the call ends.
    async fn run_element(
        &self,
        call: ElementCall<'_>,
        base: &Message,
        index: usize,
        item: OwnedDataValue,
    ) -> ElementOutcome {
        let fe = call.for_each;
        let mut copy = base.clone();
        copy.capture_changes = true;
        set_nested_value_parts(&mut copy.context, &fe.item_parts, item);
        set_nested_value_parts(
            &mut copy.context,
            &fe.index_parts,
            OwnedDataValue::from(index as u64),
        );
        let errors_before = copy.errors.len();

        let started_at = call.timed.then(Utc::now);
        let result = self
            .task_executor
            .execute_in_workflow(
                call.task,
                &mut copy,
                Some(TaskIdentity {
                    workflow_id: &call.workflow.id_arc,
                    task_id: &call.task.id_arc,
                }),
                CallStamp {
                    loop_counter: call.loop_counter,
                    element_index: Some(index),
                },
            )
            .await;
        let timing = started_at.map(|started_at| StepTiming {
            started_at,
            duration_us: duration_us_between(started_at, Utc::now()),
        });

        let (result, changes) = match result {
            Ok((outcome, changes)) => (Ok(outcome), changes),
            Err(e) => (Err(e), Vec::new()),
        };
        let collected = if fe.collect_parts.is_empty() {
            OwnedDataValue::Null
        } else {
            get_nested_value_parts(&copy.context, &fe.collect_parts)
                .cloned()
                .unwrap_or(OwnedDataValue::Null)
        };
        ElementOutcome {
            index,
            result,
            changes,
            new_errors: copy.errors.split_off(errors_before),
            collected,
            timing,
        }
    }

    /// Evaluate `for_each.over` once, to the element list.
    ///
    /// Anything but an array — `null` included — fails the task. The error is
    /// task-classified so the task's `continue_on_error` applies to it; a
    /// budget ceiling keeps its own `BUDGET_EXCEEDED` code.
    fn evaluate_for_each_over(
        &self,
        task: &Task,
        for_each: &ForEach,
        message: &Message,
    ) -> Result<Vec<OwnedDataValue>> {
        let Some(compiled) = for_each.compiled_over.as_ref() else {
            return Err(DataflowError::Task(format!(
                "Task {}: for_each.over is not compiled — build the engine through \
                 Engine::builder or Engine::new rather than running the workflow directly",
                task.id
            )));
        };
        match eval_to_owned(&self.engine, compiled, &message.context) {
            Ok(OwnedDataValue::Array(items)) => Ok(items),
            Ok(other) => Err(DataflowError::Task(format!(
                "Task {}: for_each.over must evaluate to an array, got {}",
                task.id,
                describe_kind(&other)
            ))),
            Err(e) => match from_datalogic_eval(&e) {
                budget @ DataflowError::BudgetExceeded(_) => Err(budget),
                other => Err(DataflowError::Task(format!(
                    "Task {}: for_each.over failed to evaluate: {other}",
                    task.id
                ))),
            },
        }
    }

    /// `into = []`, for a fan-out over an empty array.
    fn write_empty_results(&self, for_each: &ForEach, message: &mut Message) {
        set_nested_value_parts(
            &mut message.context,
            &for_each.into_parts,
            OwnedDataValue::Array(Vec::new()),
        );
    }

    /// The audit `Change` for [`Self::write_empty_results`], when changes are
    /// captured and `into` is set. Called after the write, so `new_value` is
    /// what landed.
    fn empty_results_change(&self, for_each: &ForEach, message: &Message) -> Vec<Change> {
        match for_each.into.as_deref() {
            Some(into) if message.capture_changes => vec![Change {
                path: Arc::from(into),
                old_value: OwnedDataValue::Null,
                new_value: OwnedDataValue::Array(Vec::new()),
                removed: false,
            }],
            _ => Vec::new(),
        }
    }

    /// Evaluate `loop.over` once, after setup, to the element list.
    ///
    /// Anything but an array is a workflow error naming the loop — `null`
    /// included. A missing batch is a bug the author wants to see, not a silent
    /// zero-iteration run; zero iterations is spelled `[]`.
    fn evaluate_over(
        &self,
        workflow: &Workflow,
        config: &LoopConfig,
        message: &Message,
    ) -> Result<Vec<OwnedDataValue>> {
        let Some(compiled) = config.compiled_over.as_ref() else {
            return Err(DataflowError::Workflow(format!(
                "Workflow {}: loop.over is not compiled — build the engine through \
                 Engine::builder or Engine::new rather than running the workflow directly",
                workflow.id
            )));
        };
        match eval_to_owned(&self.engine, compiled, &message.context) {
            Ok(OwnedDataValue::Array(items)) => Ok(items),
            Ok(other) => Err(DataflowError::Workflow(format!(
                "Workflow {}: loop.over must evaluate to an array, got {}",
                workflow.id,
                describe_kind(&other)
            ))),
            Err(e) => Err(DataflowError::Workflow(format!(
                "Workflow {}: loop.over failed to evaluate: {}",
                workflow.id,
                from_datalogic_eval(&e)
            ))),
        }
    }

    /// One pass over a task list: evaluate the workflow condition, then run
    /// the list once. Over the body this is the whole of a non-looping
    /// workflow, and one sweep of a looping one; over a loop's `setup` it is
    /// the once-only phase before the first sweep.
    ///
    /// The workflow condition is folded into the *first* sync stretch's arena
    /// scope: one `ArenaContext::from_owned` walk serves both the condition
    /// eval and the leading run of sync built-in tasks. The owned path
    /// (`eval_to_owned`) deep-borrowed the entire context — including the
    /// heavy `data.input` payload — for the condition, and `execute_tasks`
    /// then walked the same context again to build the first stretch's arena
    /// form. Mixed sync+async workflows now pay one walk where they paid two.
    /// No `.await` occurs inside the scope, preserving the `!Send` arena
    /// invariant.
    async fn execute_pass(
        &self,
        list: PassList<'_>,
        message: &mut Message,
        mut trace: Option<&mut ExecutionTrace>,
        pass: PassCtx,
        span: &mut WorkflowSpan,
    ) -> Result<PassOutcome> {
        /// Outcome of the folded condition-plus-first-stretch arena scope.
        enum FirstStretch {
            /// Workflow condition evaluated false — skip the workflow.
            Skipped,
            /// A filter task halted the workflow inside the first stretch.
            Halted,
            /// Continue with the remaining tasks, resuming at this index —
            /// the first async boundary, or further on when a skipped group's
            /// span reached past it.
            Continue(usize),
        }

        let PassList { workflow, tasks } = list;
        let first_boundary = next_async_boundary(tasks, 0);
        // One gate for the whole pass: a group can open in the folded first
        // stretch and close somewhere in the async tail.
        let mut gate = GroupGate::default();

        let first: Result<FirstStretch> =
            if workflow.compiled_condition.is_none() && first_boundary == 0 {
                // No condition and the workflow leads with an async task —
                // nothing to fold; don't build an arena context for nothing.
                // Unconditional, so the workflow runs and the span opens here.
                self.begin_workflow(span, workflow);
                Ok(FirstStretch::Continue(0))
            } else {
                with_arena(|arena| -> Result<FirstStretch> {
                    let mut arena_ctx = ArenaContext::from_owned(&message.context, arena);

                    let should_execute = match workflow.compiled_condition.as_ref() {
                        None => true,
                        Some(compiled) => evaluate_condition_in_arena(
                            &self.engine,
                            Some(compiled),
                            arena_ctx.as_data_value(),
                            arena,
                        )?,
                    };
                    if !should_execute {
                        return Ok(FirstStretch::Skipped);
                    }
                    // Admitted: this is the earliest point the workflow is
                    // known to run, so it is where the span opens. Emitting any
                    // earlier would report a workflow its condition rejected.
                    self.begin_workflow(span, workflow);
                    if first_boundary == 0 {
                        return Ok(FirstStretch::Continue(0));
                    }
                    let outcome = self.run_tasks_slice_in_arena(
                        TaskSlice {
                            tasks: &tasks[..first_boundary],
                            offset: 0,
                            gate: &mut gate,
                            list,
                        },
                        message,
                        &mut arena_ctx,
                        trace.as_deref_mut(),
                        pass,
                    )?;
                    Ok(match outcome {
                        SliceOutcome::Halted => FirstStretch::Halted,
                        SliceOutcome::JumpTo(target) => FirstStretch::Continue(target),
                        SliceOutcome::Completed => FirstStretch::Continue(first_boundary),
                    })
                })
            };

        // Drive the remaining (async-containing) tail. The workflow-level error
        // contract lives in the caller, which is the one place that knows
        // whether this pass was a whole workflow or one sweep of a loop.
        match first? {
            FirstStretch::Skipped => Ok(PassOutcome::ConditionFalse),
            FirstStretch::Halted => Ok(PassOutcome::Halted),
            FirstStretch::Continue(resume_at) => {
                let halted = self
                    .execute_tasks(list, message, trace, pass, resume_at, &mut gate)
                    .await?;
                Ok(if halted {
                    PassOutcome::Halted
                } else {
                    PassOutcome::Completed
                })
            }
        }
    }

    /// Record a `WORKFLOW_ERROR` to `message.errors` and log at the level
    /// `continue_on_error` implies. Returns `true` when the caller should stop
    /// processing further workflows (i.e. `continue_on_error` is `false`).
    ///
    /// Shared by `execute_inner` (returns from its own `Result<bool>`) and
    /// `execute_sync_workflow_run` (returns from its `with_arena` closure or
    /// continues the loop) — the recording and log-level decision are
    /// identical; only what happens next differs by call site.
    fn record_workflow_error(
        &self,
        workflow: &Workflow,
        message: &mut Message,
        e: &DataflowError,
    ) -> bool {
        message.errors.push(
            ErrorInfo::builder(
                "WORKFLOW_ERROR",
                format!("Workflow {} error: {}", workflow.id, e),
            )
            .workflow_id(&workflow.id)
            .build(),
        );

        if workflow.continue_on_error {
            warn!(
                "Workflow {} encountered error but continuing: {:?}",
                workflow.id, e
            );
            false
        } else {
            error!("Workflow {} failed: {:?}", workflow.id, e);
            true
        }
    }

    /// Execute the tasks of a workflow from index `start` onward.
    ///
    /// Groups consecutive synchronous built-in tasks into a single
    /// `with_arena` scope so the arena form of `message.context` is built
    /// once at the start of the stretch and reused across `parse_json`,
    /// `map`, `validation`, `log`, and `filter`. Async tasks (HTTP, Kafka,
    /// custom handlers) break the stretch — the arena flushes any pending
    /// state back to `OwnedDataValue` automatically (since each sync task
    /// already mutates `message.context` in place) and the next stretch
    /// rebuilds the arena form.
    ///
    /// `start` is non-zero when `execute_inner` already ran the leading sync
    /// stretch inside the folded condition scope.
    ///
    /// When `trace` is `Some`, the loop also records `ExecutionStep` entries
    /// after each task (skipped/executed) including per-mapping snapshots
    /// for `Map` tasks.
    ///
    /// Returns `Ok(true)` when a task halted the workflow.
    async fn execute_tasks(
        &self,
        list: PassList<'_>,
        message: &mut Message,
        mut trace: Option<&mut ExecutionTrace>,
        pass: PassCtx,
        start: usize,
        gate: &mut GroupGate,
    ) -> Result<bool> {
        let PassList { workflow, tasks } = list;
        let mut idx = start;
        while idx < tasks.len() {
            let stretch_end = next_async_boundary(tasks, idx);

            if stretch_end > idx {
                // Run [idx, stretch_end) as a sync stretch inside one arena.
                match self.run_sync_stretch(
                    TaskSlice {
                        tasks: &tasks[idx..stretch_end],
                        offset: idx,
                        gate,
                        list,
                    },
                    message,
                    trace.as_deref_mut(),
                    pass,
                )? {
                    SliceOutcome::Halted => return Ok(true),
                    // A group opening inside the stretch was skipped and its
                    // span reaches past the stretch — resume where it ends.
                    SliceOutcome::JumpTo(target) => {
                        idx = target;
                        continue;
                    }
                    SliceOutcome::Completed => idx = stretch_end,
                }
            }

            if idx < tasks.len() {
                // Single async task (or non-sync-builtin) at `idx`.
                let task = &tasks[idx];

                match admit_task(
                    list,
                    task,
                    idx,
                    gate,
                    trace.as_deref_mut(),
                    pass,
                    |compiled| evaluate_condition(&self.engine, compiled, &message.context),
                )? {
                    Admission::Halt => return Ok(true),
                    Admission::Jump(target) => {
                        idx = target;
                        continue;
                    }
                    Admission::Skip => {
                        idx += 1;
                        continue;
                    }
                    Admission::Run => {}
                }

                // A fan-out runs its own calls and records one entry per
                // element. `for_each` is only ever on a handler-backed
                // function, so it is always reached here, never in a sync
                // stretch.
                if let Some(for_each) = task.for_each.as_ref() {
                    let flow = self
                        .run_for_each(
                            workflow,
                            task,
                            for_each,
                            message,
                            trace.as_deref_mut(),
                            pass,
                        )
                        .await?;
                    if matches!(flow, TaskControlFlow::HaltWorkflow) {
                        return Ok(true);
                    }
                    idx += 1;
                    continue;
                }

                let clocks = self.open_task_clocks(trace.is_some(), message);

                let result = self
                    .task_executor
                    .execute_in_workflow(
                        task,
                        message,
                        Some(TaskIdentity {
                            workflow_id: &workflow.id_arc,
                            task_id: &task.id_arc,
                        }),
                        CallStamp {
                            loop_counter: pass.loop_counter,
                            element_index: None,
                        },
                    )
                    .await;

                // Before `handle_task_result`, whose `?` would drop failed tasks.
                self.emit_task_event(workflow, task, &result, clocks.obs_start);

                // No arena refresh here: no `ArenaContext` is live on this path,
                // and `run_sync_stretch` rebuilds one from `message.context` at
                // the start of the next stretch.
                let control_flow = self.handle_task_result(
                    result,
                    &workflow.id_arc,
                    &task.id_arc,
                    TaskPass {
                        continue_on_error: task.continue_on_error,
                        terminal: task.terminal,
                        halt_on: task.halt_on,
                        errors_before: clocks.errors_before,
                    },
                    message,
                    pass,
                )?;

                // Async tasks at the boundary have no per-mapping snapshots —
                // they're either HTTP/Kafka/Enrich or a custom handler.
                pass.note_executed(
                    trace.as_deref_mut(),
                    &workflow.id,
                    &task.id,
                    message,
                    clocks,
                    None,
                );

                if matches!(control_flow, TaskControlFlow::HaltWorkflow) {
                    return Ok(true);
                }
                idx += 1;
            }
        }

        // A terminal group closing on the last task still has to halt: for a
        // workflow carrying a `loop`, halting breaks the loop where completing
        // would start another sweep.
        Ok(gate.close_through(tasks.len()))
    }

    /// Execute a contiguous run of sync-builtin tasks inside one
    /// `with_arena` scope. The arena context is built once at the start and
    /// refreshed in place after each mutating task. Returns `Ok(true)` if a
    /// filter task halted the workflow.
    ///
    /// This is the single-workflow entry; the cross-workflow path
    /// (`execute_sync_workflow_run`) shares the same task loop via
    /// `run_tasks_slice_in_arena` but carries one `ArenaContext` across several
    /// workflows.
    fn run_sync_stretch(
        &self,
        slice: TaskSlice<'_, '_>,
        message: &mut Message,
        trace: Option<&mut ExecutionTrace>,
        pass: PassCtx,
    ) -> Result<SliceOutcome> {
        with_arena(|arena| -> Result<SliceOutcome> {
            let mut arena_ctx = ArenaContext::from_owned(&message.context, arena);
            self.run_tasks_slice_in_arena(slice, message, &mut arena_ctx, trace, pass)
        })
    }

    /// Run `tasks` against an already-built `ArenaContext`, evaluating each
    /// task's condition in-arena and refreshing the cache after each mutating
    /// task. Returns `Ok(true)` if a filter task halted the workflow.
    ///
    /// Factored out of `run_sync_stretch` so both the single-workflow stretch
    /// and the cross-workflow shared-arena run (`execute_sync_workflow_run`)
    /// share one implementation. The caller owns the `ArenaContext` lifetime,
    /// so the cross-workflow path can reuse the same arena form of
    /// `message.context` across consecutive workflows instead of rebuilding it.
    fn run_tasks_slice_in_arena<'arena>(
        &self,
        slice: TaskSlice<'_, 'arena>,
        message: &mut Message,
        arena_ctx: &mut ArenaContext<'arena>,
        mut trace: Option<&mut ExecutionTrace>,
        pass: PassCtx,
    ) -> Result<SliceOutcome> {
        let arena = arena_ctx.arena();
        let TaskSlice {
            tasks,
            offset,
            gate,
            list,
        } = slice;
        let workflow = list.workflow;
        let slice_end = offset + tasks.len();

        let mut i = 0;
        while i < tasks.len() {
            let task = &tasks[i];
            let abs = offset + i;

            // Conditions evaluate against the arena form so we don't re-borrow
            // the thread-local `RefCell`. A `None` compiled condition (the
            // compiler folds the default literal `true` to `None`) skips both
            // the eval and the per-task arena context slice build.
            match admit_task(
                list,
                task,
                abs,
                gate,
                trace.as_deref_mut(),
                pass,
                |compiled| {
                    evaluate_condition_in_arena(
                        &self.engine,
                        compiled,
                        arena_ctx.as_data_value(),
                        arena,
                    )
                },
            )? {
                Admission::Halt => return Ok(SliceOutcome::Halted),
                Admission::Jump(target) => {
                    // A span reaching past this slice is the caller's to
                    // resume — it owns the tasks beyond `slice_end`.
                    if target >= slice_end {
                        return Ok(SliceOutcome::JumpTo(target));
                    }
                    i = target - offset;
                    continue;
                }
                Admission::Skip => {
                    i += 1;
                    continue;
                }
                Admission::Run => {}
            }

            // Per-task snapshot buffer — only used for Map tasks in trace
            // mode, and only when the trace's policy wants them. Allocating an
            // empty Vec is cheap and the buffer stays empty for non-Map tasks.
            let mut mapping_snapshots: Vec<Value> = Vec::new();
            let want_mapping_contexts = trace
                .as_deref()
                .is_some_and(|t| t.options().mapping_contexts);
            let mapping_snapshots_buf = if want_mapping_contexts {
                Some(&mut mapping_snapshots)
            } else {
                None
            };

            let clocks = self.open_task_clocks(trace.is_some(), message);

            let result =
                self.execute_sync_task_in_arena(task, message, arena_ctx, mapping_snapshots_buf);

            // Before `handle_task_result`, whose `?` would drop failed tasks.
            self.emit_task_event(workflow, task, &result, clocks.obs_start);

            let flow = self.handle_task_result(
                result,
                &workflow.id_arc,
                &task.id_arc,
                TaskPass {
                    continue_on_error: task.continue_on_error,
                    terminal: task.terminal,
                    halt_on: task.halt_on,
                    errors_before: clocks.errors_before,
                },
                message,
                pass,
            );

            // Refresh the slots `handle_task_result` wrote so the next task —
            // and, in the cross-workflow path, the next workflow's condition —
            // sees them, without re-arenaing unrelated metadata children
            // (mapped `metadata.routing.*`, chained workflow state, …) after
            // every task.
            //
            // Deliberately *before* the `?`. An `Err` here does not necessarily
            // end this arena scope: `execute_sync_workflow_run` continues into
            // the next workflow carrying this same `ArenaContext` whenever the
            // failing task had `continue_on_error: false` but its workflow had
            // `continue_on_error: true`, and that workflow's condition would
            // otherwise be evaluated against a stale `metadata.progress`.
            arena_ctx.refresh_for_path(&message.context, "metadata.progress");
            // Gated on a failure actually being recorded: this walk deep-copies
            // the target subtree into the arena, so running it after every
            // successful task would be a permanent cost on the hot path.
            if let Some(cfg) = self.error_context_refresh(message, clocks.errors_before) {
                arena_ctx.refresh_for_path_parts(&message.context, &cfg.path_parts);
            }

            let control_flow = flow?;

            pass.note_executed(
                trace.as_deref_mut(),
                &workflow.id,
                &task.id,
                message,
                clocks,
                // A `map` task in trace mode collects one snapshot per mapping;
                // every other sync built-in leaves the buffer empty.
                Some(mapping_snapshots).filter(|s| !s.is_empty()),
            );

            if matches!(control_flow, TaskControlFlow::HaltWorkflow) {
                return Ok(SliceOutcome::Halted);
            }
            i += 1;
        }
        Ok(SliceOutcome::Completed)
    }

    /// Drive a message through `workflows` in order, grouping maximal runs of
    /// consecutive `fully_sync` workflows into a single shared-arena scope
    /// (`execute_sync_workflow_run`) and falling back to the per-workflow
    /// `.await` path (`execute_inner`) for any workflow containing an async
    /// task.
    ///
    /// A thin `&[&Workflow]` wrapper over `Self::run_all_borrowed`, which is
    /// the actual shared entry all four `Engine::process_message*` variants
    /// call directly (against `&[Workflow]` from the engine's own registry,
    /// with no per-message `Vec<&Workflow>` collect). This method exists for
    /// a caller that already holds borrowed references.
    pub async fn run_all(
        &self,
        workflows: &[&Workflow],
        message: &mut Message,
        trace: Option<&mut ExecutionTrace>,
        now: DateTime<Utc>,
    ) -> Result<()> {
        self.run_all_borrowed(workflows, message, trace, now).await
    }

    /// Generic driver behind [`Self::run_all`]: accepts any slice whose
    /// elements borrow as `Workflow` — `&[Workflow]` directly from the
    /// engine's registry (no per-message `Vec<&Workflow>` collect) or the
    /// `&[&Workflow]` shape the public entry keeps for compatibility.
    pub(crate) async fn run_all_borrowed<W: std::borrow::Borrow<Workflow>>(
        &self,
        workflows: &[W],
        message: &mut Message,
        trace: Option<&mut ExecutionTrace>,
        now: DateTime<Utc>,
    ) -> Result<()> {
        let started_at = self.observer_clock();
        if let Some(observer) = self.observer.as_ref() {
            observer.message_started(&MessageStarted {
                message_id: message.id(),
                workflows_considered: workflows.len(),
            });
        }
        let outcome = self.run_all_inner(workflows, message, trace, now).await;
        if let Some(observer) = self.observer.as_ref() {
            observer.message_finished(&MessageFinished {
                message_id: message.id(),
                duration: started_at
                    .map(|s| Duration::from_micros(duration_us_between(s, Utc::now())))
                    .unwrap_or_default(),
                errors: message.errors().len(),
                stopped_early: outcome.is_err(),
            });
        }
        outcome
    }

    /// The driver proper. Split out so `message_finished` fires on the early
    /// `Err` path too — an observer measuring a run must see the runs that
    /// stopped, which are the interesting ones.
    async fn run_all_inner<W: std::borrow::Borrow<Workflow>>(
        &self,
        workflows: &[W],
        message: &mut Message,
        mut trace: Option<&mut ExecutionTrace>,
        now: DateTime<Utc>,
    ) -> Result<()> {
        let mut i = 0;
        while i < workflows.len() {
            if joins_sync_run(workflows[i].borrow()) {
                // Extend over the maximal run of consecutive fully-sync
                // workflows and execute them in one shared arena scope.
                let mut j = i + 1;
                while j < workflows.len() && joins_sync_run(workflows[j].borrow()) {
                    j += 1;
                }
                self.execute_sync_workflow_run(
                    &workflows[i..j],
                    message,
                    trace.as_deref_mut(),
                    now,
                )?;
                i = j;
            } else {
                // Mixed sync+async (or fully-async) workflow: the existing
                // driver interleaves per-stretch arenas with `.await`.
                self.execute_inner(workflows[i].borrow(), message, trace.as_deref_mut(), now)
                    .await?;
                i += 1;
            }
        }
        Ok(())
    }

    /// Execute a maximal run of consecutive fully-sync workflows inside ONE
    /// shared `with_arena` scope. The message context is deep-walked into the
    /// arena once for the whole run, then carried — with the existing
    /// incremental `refresh_for_path` after each mutating task — across
    /// workflow boundaries, instead of being rebuilt per workflow.
    ///
    /// Per-workflow semantics are preserved exactly: each workflow's condition
    /// is evaluated (in-arena), a false condition skips only that workflow, a
    /// filter-halt stops only that workflow, and task errors are wrapped with
    /// the workflow id and honor `continue_on_error` (continue, or propagate
    /// `Err` out of the run to stop the whole message) — mirroring
    /// `execute_inner`.
    ///
    /// **Tokio safety:** this method is synchronous and the `fully_sync`
    /// precondition guarantees every task is a sync built-in, so no `.await`
    /// occurs while the `!Send` arena borrow is live. The borrow checker
    /// enforces this — the shared `ArenaContext` cannot escape the closure.
    fn execute_sync_workflow_run<W: std::borrow::Borrow<Workflow>>(
        &self,
        workflows: &[W],
        message: &mut Message,
        mut trace: Option<&mut ExecutionTrace>,
        now: DateTime<Utc>,
    ) -> Result<()> {
        // `joins_sync_run` keeps looping workflows out of this path, so every
        // workflow here runs exactly one pass and carries no loop counter.
        debug_assert!(
            workflows.iter().all(|w| joins_sync_run(w.borrow())),
            "only non-looping fully-sync workflows may join a shared-arena run"
        );
        let pass = PassCtx::once(now);

        with_arena(|arena| -> Result<()> {
            let mut arena_ctx = ArenaContext::from_owned(&message.context, arena);

            for workflow in workflows {
                let workflow: &Workflow = workflow.borrow();

                // Same gate as `execute_inner`. This is the site a fully-sync
                // workflow actually reaches: `fully_sync` routes every
                // map/log/validation/filter-only workflow here and never through
                // `execute_inner`, so gating only there would silently not apply
                // to most workflows.
                if !rollout_admits(workflow, message) {
                    note_workflow_skip(
                        trace.as_deref_mut(),
                        &workflow.id,
                        "outside rollout bucket",
                    );
                    continue;
                }

                // Workflow condition in-arena: a folded `None` skips the eval;
                // a real condition reuses the carried context instead of the
                // owned-path `eval_to_owned` deep-walk.
                let should_execute = match workflow.compiled_condition.as_ref() {
                    None => true,
                    Some(compiled) => evaluate_condition_in_arena(
                        &self.engine,
                        Some(compiled),
                        arena_ctx.as_data_value(),
                        arena,
                    )?,
                };

                if !should_execute {
                    note_workflow_skip(trace.as_deref_mut(), &workflow.id, "condition not met");
                    continue;
                }

                // Admitted, so the span opens here — the sync run's own gate
                // site, which `execute_inner` never sees for these workflows.
                let mut span = WorkflowSpan::default();
                self.begin_workflow(&mut span, workflow);

                // Group state is per-workflow: this run carries one arena
                // across several workflows, but never a group.
                let mut gate = GroupGate::default();
                match self.run_tasks_slice_in_arena(
                    TaskSlice {
                        tasks: &workflow.tasks,
                        offset: 0,
                        gate: &mut gate,
                        list: PassList::body(workflow),
                    },
                    message,
                    &mut arena_ctx,
                    trace.as_deref_mut(),
                    pass,
                ) {
                    // A halt stops only this workflow; carry on with the next
                    // one (and keep the shared arena context). The slice spans
                    // the whole task list, so a jump can only land at its end.
                    Ok(outcome) => {
                        self.end_workflow(&span, workflow, matches!(outcome, SliceOutcome::Halted));
                        info!("Successfully completed workflow: {}", workflow.id);
                    }
                    Err(e) => {
                        // Closed before the early return, for the same reason
                        // as the loop path: the failing workflow's span is the
                        // one an observer most wants.
                        self.end_workflow(&span, workflow, false);
                        // Single-channel contract — mirror `execute_inner`.
                        if self.record_workflow_error(workflow, message, &e) {
                            return Err(e);
                        }
                    }
                }
            }
            Ok(())
        })
    }

    /// Dispatch a single sync-builtin task via the consolidated
    /// `FunctionConfig::try_execute_in_arena`. `next_async_boundary` guarantees
    /// the stretch contents are sync built-ins, so the `None` arm is
    /// unreachable in practice.
    ///
    /// `mapping_snapshots` is only consulted by the `Map` variant; non-Map
    /// sync builtins ignore it. Pass `None` from the production path.
    fn execute_sync_task_in_arena<'arena>(
        &self,
        task: &'arena Task,
        message: &mut Message,
        arena_ctx: &mut ArenaContext<'arena>,
        mapping_snapshots: Option<&mut Vec<Value>>,
    ) -> Result<(TaskOutcome, Vec<Change>)> {
        debug!(
            "Executing sync task in arena: {} ({})",
            task.id,
            task.function.function_name()
        );
        debug_assert!(
            task.function.is_sync_builtin(),
            "execute_sync_task_in_arena called with non-sync-builtin task: {}",
            task.function.function_name()
        );
        // In debug builds the assert above catches mis-dispatch; in release
        // we still surface the invariant violation as a recoverable engine
        // error rather than panicking via `unreachable!`.
        task.function
            .try_execute_in_arena(message, arena_ctx, &self.engine, mapping_snapshots)
            .ok_or_else(|| {
                DataflowError::Task(format!(
                    "execute_sync_task_in_arena dispatched to non-sync-builtin task '{}' \
                     (engine bug — sync-stretch should only contain sync-builtin tasks)",
                    task.function.function_name()
                ))
            })?
    }

    /// Mirror every error this task contributed to `message.errors` into the
    /// configured context path.
    ///
    /// Taking the delta beyond `task.errors_before` rather than recording at each
    /// push site is what makes coverage match `errors()` exactly. Two of the four
    /// per-task producers never reach a failure arm at all: the `validation`
    /// built-in appends its per-rule failures and then returns `Status(400)`,
    /// which lands in the *success* arm, and `TaskContext::add_error` can fire on
    /// a task that succeeds outright. Neither is visible to a host that wraps
    /// handlers either, since the sync built-ins never reach the registry.
    ///
    /// Must run *after* the two pushes `handle_task_result` performs itself
    /// (`TASK_STATUS_ERROR` and the task error), or they fall outside the delta —
    /// which would silently drop every `map` failure, since `map` returns
    /// `Status(500)` without touching `errors` itself.
    ///
    /// Returns nothing — the sync stretch reads [`Self::error_context_refresh`]
    /// instead, so the arena refresh stays off the no-failure path.
    #[inline]
    fn mirror_task_errors(
        &self,
        message: &mut Message,
        workflow_id: &str,
        task_id: &str,
        status: u16,
        task: TaskPass,
    ) {
        let Some(cfg) = self.error_context.as_ref() else {
            return;
        };
        // Disjoint borrows of two fields of `Message`, so the split is needed to
        // read the error tail while mutating the context.
        let Message {
            context, errors, ..
        } = message;
        let new_errors = errors.get(task.errors_before..).unwrap_or(&[]);
        append_error_records(context, cfg, workflow_id, task_id, status, new_errors);
    }

    /// Whether the sync stretch must refresh the arena for the error-context
    /// path after this task — i.e. the option is on and the task contributed at
    /// least one error.
    #[inline]
    fn error_context_refresh<'a>(
        &'a self,
        message: &Message,
        errors_before: usize,
    ) -> Option<&'a Arc<ErrorContextConfig>> {
        let cfg = self.error_context.as_ref()?;
        if message.errors.len() > errors_before {
            Some(cfg)
        } else {
            None
        }
    }

    /// Handle the result of a task execution.
    ///
    /// `workflow_id_arc` and `task_id_arc` are the compile-time cached
    /// `Arc<str>` mirrors of `workflow.id` / `task.id`; we Arc-clone them into
    /// each `AuditTrail` rather than reallocating from the `&str` form.
    fn handle_task_result(
        &self,
        result: Result<(TaskOutcome, Vec<Change>)>,
        workflow_id_arc: &Arc<str>,
        task_id_arc: &Arc<str>,
        task: TaskPass,
        message: &mut Message,
        pass: PassCtx,
    ) -> Result<TaskControlFlow> {
        let workflow_id: &str = workflow_id_arc;
        let task_id: &str = task_id_arc;
        let continue_on_error = task.continue_on_error;
        match result {
            Ok((TaskOutcome::Skip, _)) => {
                // No audit trail, no progress write, and no error-context record
                // — the task has explicitly opted out of the per-task record
                // (filter gate set to `Skip`). `audit_status()` is `None` here,
                // so a record would need a fabricated `status`, breaking the
                // fixed four-key shape that makes the path predictable to branch
                // on. Reaching this with errors recorded takes a handler that
                // calls `add_error` and *then* skips; the entry is still on
                // `message.errors()`.
                debug!("Task {} signaled skip", task_id);
                Ok(TaskControlFlow::Continue)
            }
            Ok((outcome, changes)) => {
                // `Skip` already returned above; the remaining variants all
                // record an audit entry. `audit_status()` is `Some` for
                // Success/Status/Halt — expect is for documentation only.
                let status = outcome
                    .audit_status()
                    .expect("Skip handled above; remaining variants emit audit status");
                // `Task::terminal` and `Task::halt_on` reach the same halt as
                // `TaskOutcome::Halt`, but they are applied *after* the status
                // classification below — see the `flow` fold. Deciding here would
                // make halting the first branch of the chain, so a terminal task
                // returning 500 would stop without recording `TASK_STATUS_ERROR`
                // and without propagating when `continue_on_error` is false.
                // `halts_at` is the single definition of "failed" for `halt_on`.
                let halt_requested = outcome.halts_workflow() || task.halts_at(status);

                // Record audit trail. workflow_id_arc/task_id_arc are populated
                // by LogicCompiler at engine construction; cloning them is a
                // refcount bump, not a string copy. `now` is shared with all
                // other AuditTrails in this process_message call.
                message.audit_trail.push(AuditTrail {
                    timestamp: pass.now,
                    workflow_id: Arc::clone(workflow_id_arc),
                    task_id: Arc::clone(task_id_arc),
                    status: status as usize,
                    changes,
                    loop_counter: pass.loop_counter,
                    element_index: pass.element_index,
                });

                // Update progress metadata for workflow chaining. Always
                // emitted: when multiple workflows are registered in the same
                // engine, downstream workflows route on
                // `metadata.progress.{workflow_id,task_id,status_code}` to
                // advance through linear sequences. After the first task the
                // slot already holds the expected 3-key object, so the write
                // overwrites the three values in place — only the two id
                // `String` allocs remain. (This beat both three separate
                // `set_nested_value` calls and the batched slot replace on
                // the realistic workload.)
                write_progress_metadata(&mut message.context, workflow_id, task_id, status);

                // Decide the control flow first rather than returning from
                // inside each branch, so the error-context mirror below runs on
                // exactly one path. The halt and `!continue_on_error` exits would
                // otherwise each need their own call, and a future exit added
                // without one would silently stop recording.
                let flow = if (400..500).contains(&status) {
                    warn!("Task {} returned client error status: {}", task_id, status);
                    Ok(TaskControlFlow::Continue)
                } else if status >= 500 {
                    error!("Task {} returned server error status: {}", task_id, status);
                    // Single-channel contract: surface 5xx outcomes through
                    // `message.errors` as well as the audit trail, so callers
                    // that scan `errors()` see a 5xx-status task even when
                    // the workflow continues past it.
                    let mut info = ErrorInfo::builder(
                        "TASK_STATUS_ERROR",
                        format!("Task {} returned status {}", task_id, status),
                    )
                    .workflow_id(workflow_id)
                    .task_id(task_id);
                    if let Some(index) = pass.element_index {
                        info = info.element_index(index);
                    }
                    message.errors.push(info.build());
                    if continue_on_error {
                        Ok(TaskControlFlow::Continue)
                    } else {
                        Err(DataflowError::Task(format!(
                            "Task {} failed with status {}",
                            task_id, status
                        )))
                    }
                } else {
                    Ok(TaskControlFlow::Continue)
                };

                // Upgrade a `Continue` to a halt, leaving the 5xx `Err` and the
                // recording above untouched. `TaskOutcome::Halt`'s own status is
                // 299 — neither 4xx nor 5xx — so its behaviour is unchanged.
                let flow = match flow {
                    Ok(TaskControlFlow::Continue) if halt_requested => {
                        info!("Task {} halted workflow {}", task_id, workflow_id);
                        Ok(TaskControlFlow::HaltWorkflow)
                    }
                    other => other,
                };

                // After the `TASK_STATUS_ERROR` push above, so it lands inside
                // this task's delta.
                self.mirror_task_errors(message, workflow_id, task_id, status, task);
                flow
            }
            Err(e) => {
                error!("Task {} failed: {:?}", task_id, e);

                // Record error in audit trail (Arc clones are refcount bumps).
                message.audit_trail.push(AuditTrail {
                    timestamp: pass.now,
                    workflow_id: Arc::clone(workflow_id_arc),
                    task_id: Arc::clone(task_id_arc),
                    status: 500,
                    changes: vec![],
                    loop_counter: pass.loop_counter,
                    element_index: pass.element_index,
                });

                // Same invariant as the Ok arm: `metadata.progress` is written
                // after every task, unconditionally, so a downstream workflow
                // gating on it still sees this task ran even though it errored.
                write_progress_metadata(&mut message.context, workflow_id, task_id, 500);

                // Add error to message. A service-classified error contributes
                // its own `kind` as the code and carries its operator-only
                // `detail`; everything else takes its variant's code.
                // Deliberately lifted at the task site only: the two
                // `WORKFLOW_ERROR` wrappers wrap the same propagated error, so
                // lifting there too would put two entries with the same
                // `code` on the message — making "count errors by code"
                // double-count — and would stop `WORKFLOW_ERROR` reliably
                // meaning "a workflow stopped".
                //
                // `format!("{}", e)` stays caller-safe because `Service`'s
                // `Display` is `{message}` — the detail is never interpolated.
                let mut info = ErrorInfo::builder(
                    service_error_code(&e),
                    format!("Task {} error: {}", task_id, e),
                )
                .workflow_id(workflow_id)
                .task_id(task_id);
                if let Some(detail) = e.detail() {
                    info = info.detail(detail);
                }
                if let Some(index) = pass.element_index {
                    info = info.element_index(index);
                }
                message.errors.push(info.build());

                // `500` matches the audit entry and the progress write above: a
                // handler `Err` has no status of its own.
                self.mirror_task_errors(message, workflow_id, task_id, 500, task);

                if !continue_on_error {
                    Err(e)
                } else if task.halts_at(500) {
                    // Either flag halts here: `terminal` because the author said
                    // "nothing after this runs" whatever happened, `halt_on`
                    // because this is a failure. `500` is the status already
                    // stamped on the audit entry, the progress write and the
                    // error mirror above — a handler `Err` has none of its own.
                    // The error stays on `message.errors()` either way.
                    info!(
                        "Task {} halted workflow {} after failing",
                        task_id, workflow_id
                    );
                    Ok(TaskControlFlow::HaltWorkflow)
                } else {
                    Ok(TaskControlFlow::Continue)
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::engine::compiler::LogicCompiler;
    use serde_json::json;
    use std::collections::HashMap;

    /// Test-only helper: build an `OwnedDataValue` from a `json!` literal.
    fn dv(v: serde_json::Value) -> OwnedDataValue {
        OwnedDataValue::from(&v)
    }

    /// Compile `json` into a single runnable workflow plus its engine.
    fn compiled(json: &str) -> (Workflow, Arc<datalogic_rs::Engine>) {
        let compiler = LogicCompiler::new();
        let workflow = Workflow::from_json(json).expect("workflow should parse");
        let compiled = compiler
            .compile_workflows(vec![workflow])
            .expect("workflow should compile");
        (
            compiled.into_iter().next().expect("one workflow"),
            compiler.into_engine(),
        )
    }

    /// A `WorkflowExecutor` over an empty handler registry.
    fn executor(engine: Arc<datalogic_rs::Engine>) -> WorkflowExecutor {
        let task_executor = Arc::new(TaskExecutor::new(
            Arc::new(HashMap::new()),
            Arc::clone(&engine),
        ));
        WorkflowExecutor::new(task_executor, engine)
    }

    /// A `WorkflowExecutor` that mirrors failure codes to `metadata.errors`.
    fn executor_with_error_context(engine: Arc<datalogic_rs::Engine>) -> WorkflowExecutor {
        let cfg = ErrorContextConfig::new("metadata.errors".to_string(), 32)
            .expect("metadata.errors is a valid path");
        executor(engine).with_error_context(Arc::new(cfg))
    }

    #[tokio::test]
    async fn appending_records_mid_stretch_keeps_the_arena_cache_consistent() {
        // A failing `validation` followed by a `map`, both sync built-ins, so
        // they share one `ArenaContext`. Two things are under test:
        //
        // 1. the `map` reads the record appended by the `validation`, which only
        //    works if the append refreshed the arena; and
        // 2. `apply_mutation_parts_write_through`'s `#[cfg(test)]`
        //    `assert_matches_owned` runs on the `map`'s write, giving free
        //    differential verification that the refresh left the arena cache
        //    identical to a from-scratch rebuild of the owned context. That
        //    assertion is compiled out for the `tests/` binaries, so it can only
        //    be exercised from here.
        let (workflow, engine) = compiled(
            r#"{ "id": "w", "name": "w", "tasks": [
                { "id": "check", "name": "check", "continue_on_error": true,
                  "function": {"name": "validation", "input": {"rules": [
                      {"logic": false, "message": "nope"}]}}},
                { "id": "react", "name": "react",
                  "function": {"name": "map", "input": {"mappings": [
                      {"path": "data.seen", "logic": {"var": "metadata.errors.0.code"}}]}}}
            ]}"#,
        );
        let mut message = Message::from_value(&json!({}));

        executor_with_error_context(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("continue_on_error keeps the workflow running");

        assert_eq!(
            message.context["data"].get("seen"),
            Some(&dv(json!("VALIDATION_ERROR"))),
            "the map must read the record the validation appended in the same stretch"
        );
    }

    #[tokio::test]
    async fn the_error_context_path_is_untouched_when_every_task_succeeds() {
        let (workflow, engine) = compiled(&format!(
            r#"{{ "id": "w", "name": "w", "tasks": [{COUNTER_BODY}] }}"#
        ));
        let mut message = Message::from_value(&json!({}));

        executor_with_error_context(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("workflow should complete");

        assert_eq!(
            message.context["metadata"].get("errors"),
            None,
            "a clean run leaves the key absent, not an empty array"
        );
    }

    /// Every `loop_counter` recorded on the audit trail, in order.
    fn counters(message: &Message) -> Vec<Option<i64>> {
        message
            .audit_trail
            .iter()
            .map(|entry| entry.loop_counter)
            .collect()
    }

    /// A one-task `map` workflow body writing `data.n` from the counter.
    const COUNTER_BODY: &str = r#"{"id": "t", "name": "t", "function": {"name": "map",
        "input": {"mappings": [{"path": "data.n", "logic": {"var": "temp_data.i"}}]}}}"#;

    #[tokio::test]
    async fn loop_without_a_condition_runs_exactly_max_sweeps() {
        let (workflow, engine) = compiled(&format!(
            r#"{{ "id": "w", "name": "w", "loop": {{"counter": "i", "max": 3}},
                  "tasks": [{COUNTER_BODY}] }}"#
        ));
        let mut message = Message::from_value(&json!({}));

        let executed = executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("loop should complete");

        assert!(executed);
        // One audit entry per sweep, each stamped with its counter.
        assert_eq!(counters(&message), vec![Some(0), Some(1), Some(2)]);
        // The counter is left at the bound the loop stopped on.
        assert_eq!(message.context["temp_data"].get("i"), Some(&dv(json!(3))));
        // The body observed each value; the last one survives.
        assert_eq!(message.context["data"].get("n"), Some(&dv(json!(2))));
    }

    #[tokio::test]
    async fn loop_exits_early_when_the_condition_goes_false() {
        // Bounded at 10 but the condition stops it at 4.
        let (workflow, engine) = compiled(&format!(
            r#"{{ "id": "w", "name": "w",
                  "condition": {{"<": [{{"var": "temp_data.i"}}, 4]}},
                  "loop": {{"counter": "i", "max": 10}},
                  "tasks": [{COUNTER_BODY}] }}"#
        ));
        let mut message = Message::from_value(&json!({}));

        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("loop should complete");

        assert_eq!(counters(&message), vec![Some(0), Some(1), Some(2), Some(3)]);
    }

    #[tokio::test]
    async fn loop_whose_condition_is_false_on_the_first_sweep_is_a_plain_skip() {
        let (workflow, engine) = compiled(&format!(
            r#"{{ "id": "w", "name": "w", "condition": false,
                  "loop": {{"counter": "i", "max": 5}},
                  "tasks": [{COUNTER_BODY}] }}"#
        ));
        let mut message = Message::from_value(&json!({}));

        let executed = executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("a skip is not an error");

        assert!(!executed, "a never-entered loop reports as skipped");
        assert!(message.audit_trail.is_empty());
    }

    // -----------------------------------------------------------------
    // loop.setup / over / as / scratch (#60)
    // -----------------------------------------------------------------

    /// The body records the element id and the index it was visited at.
    const OVER_BODY: &str = r#"{"id": "t", "name": "t", "function": {"name": "map",
        "input": {"mappings": [
            {"path": "data.last", "logic": {"var": "temp_data.item.id"}},
            {"path": "data.last_i", "logic": {"var": "temp_data.i"}}]}}}"#;

    fn items_message() -> Message {
        Message::builder()
            .data(dv(
                json!({"items": [{"id": "a"}, {"id": "b"}, {"id": "c"}]}),
            ))
            .build()
    }

    #[tokio::test]
    async fn an_over_loop_visits_each_element_with_its_index() {
        let (workflow, engine) = compiled(&format!(
            r#"{{ "id": "w", "name": "w",
                  "loop": {{"counter": "i", "max": 100, "over": {{"var": "data.items"}}, "as": "item"}},
                  "tasks": [{OVER_BODY}] }}"#
        ));
        let mut message = items_message();

        let executed = executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("loop should complete");

        assert!(executed);
        assert_eq!(
            counters(&message),
            vec![Some(0), Some(1), Some(2)],
            "one sweep per element, not one per max"
        );
        assert_eq!(message.context["data"].get("last"), Some(&dv(json!("c"))));
        assert_eq!(message.context["data"].get("last_i"), Some(&dv(json!(2))));
        assert_eq!(
            message.context["temp_data"].get("item"),
            Some(&dv(json!({"id": "c"}))),
            "the element slot is left holding the last element, like the counter"
        );
        assert_eq!(message.context["temp_data"].get("i"), Some(&dv(json!(3))));
    }

    #[tokio::test]
    async fn max_still_bounds_an_over_loop() {
        let (workflow, engine) = compiled(&format!(
            r#"{{ "id": "w", "name": "w",
                  "loop": {{"counter": "i", "max": 2, "over": {{"var": "data.items"}}, "as": "item"}},
                  "tasks": [{OVER_BODY}] }}"#
        ));
        let mut message = items_message();
        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .unwrap();
        assert_eq!(counters(&message), vec![Some(0), Some(1)]);
    }

    #[tokio::test]
    async fn init_offsets_and_increment_strides_the_element_index() {
        let (workflow, engine) = compiled(&format!(
            r#"{{ "id": "w", "name": "w",
                  "loop": {{"counter": "i", "init": 1, "increment": 2, "max": 100,
                            "over": {{"var": "data.items"}}, "as": "item"}},
                  "tasks": [{OVER_BODY}] }}"#
        ));
        let mut message = Message::builder()
            .data(dv(json!({
                "items": [{"id": "a"}, {"id": "b"}, {"id": "c"}, {"id": "d"}, {"id": "e"}, {"id": "f"}]
            })))
            .build();
        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .unwrap();
        assert_eq!(counters(&message), vec![Some(1), Some(3), Some(5)]);
        assert_eq!(message.context["data"].get("last"), Some(&dv(json!("f"))));
    }

    #[tokio::test]
    async fn an_empty_over_runs_no_iteration_but_the_workflow_still_executed() {
        let (workflow, engine) = compiled(&format!(
            r#"{{ "id": "w", "name": "w",
                  "loop": {{"counter": "i", "max": 100, "over": [], "as": "item"}},
                  "tasks": [{OVER_BODY}] }}"#
        ));
        let mut message = Message::from_value(&json!({}));
        let executed = executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .unwrap();
        assert!(
            executed,
            "admitted by its condition, zero elements to visit"
        );
        assert!(message.audit_trail.is_empty());
        assert_eq!(
            message.context["temp_data"].get("item"),
            None,
            "never written"
        );
    }

    #[tokio::test]
    async fn a_non_array_over_is_a_workflow_error() {
        let make = |continue_on_error: bool| {
            compiled(&format!(
                r#"{{ "id": "w", "name": "w", "continue_on_error": {continue_on_error},
                      "loop": {{"counter": "i", "max": 100, "over": {{"var": "data.items"}}, "as": "item"}},
                      "tasks": [{OVER_BODY}] }}"#
            ))
        };
        for (items, kind) in [
            (json!("nope"), "a string"),
            (json!(null), "null"),
            (json!({"a": 1}), "an object"),
        ] {
            let (workflow, engine) = make(false);
            let mut message = Message::builder().data(dv(json!({"items": items}))).build();
            let err = executor(engine)
                .execute(&workflow, &mut message, Utc::now())
                .await
                .expect_err("not an array");
            assert!(
                err.to_string().contains("loop.over"),
                "names the loop: {err}"
            );
            assert!(err.to_string().contains(kind), "says what it got: {err}");
            assert_eq!(message.errors.len(), 1);
            assert_eq!(message.errors[0].code, "WORKFLOW_ERROR");
            assert!(message.audit_trail.is_empty(), "no sweep ran");
        }

        // With continue_on_error the error is recorded and the next workflow
        // still runs — but this one runs no iteration.
        let (workflow, engine) = make(true);
        let mut message = Message::builder().data(dv(json!({"items": 5}))).build();
        let executed = executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .unwrap();
        assert!(executed);
        assert_eq!(message.errors[0].code, "WORKFLOW_ERROR");
        assert!(message.audit_trail.is_empty());
    }

    /// A setup list that copies `data.items` into `temp_data.batch` and stamps
    /// `data.setup_ran` — the batch-read step of the issue.
    const SETUP: &str = r#"[{"id": "read", "name": "read", "function": {"name": "map",
        "input": {"mappings": [
            {"path": "temp_data.batch", "logic": {"var": "data.items"}},
            {"path": "data.setup_ran", "logic": true}]}}}]"#;

    #[tokio::test]
    async fn setup_runs_once_before_the_first_iteration_without_a_loop_counter() {
        let (workflow, engine) = compiled(&format!(
            r#"{{ "id": "w", "name": "w",
                  "loop": {{"counter": "i", "max": 100, "setup": {SETUP},
                            "over": {{"var": "temp_data.batch"}}, "as": "item"}},
                  "tasks": [{OVER_BODY}] }}"#
        ));
        let mut message = items_message();
        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .unwrap();

        assert_eq!(
            counters(&message),
            vec![None, Some(0), Some(1), Some(2)],
            "setup is stamped like a non-looping task, then one entry per element"
        );
        assert_eq!(message.audit_trail[0].task_id.as_ref(), "read");
        assert_eq!(message.context["data"].get("last"), Some(&dv(json!("c"))));
    }

    #[tokio::test]
    async fn a_false_condition_skips_setup_and_the_loop_alike() {
        let (workflow, engine) = compiled(&format!(
            r#"{{ "id": "w", "name": "w", "condition": false,
                  "loop": {{"counter": "i", "max": 100, "setup": {SETUP},
                            "over": {{"var": "temp_data.batch"}}, "as": "item"}},
                  "tasks": [{OVER_BODY}] }}"#
        ));
        let mut message = items_message();
        let executed = executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .unwrap();
        assert!(!executed);
        assert!(message.audit_trail.is_empty(), "setup did not run");
        assert_eq!(message.context["data"].get("setup_ran"), None);
    }

    #[tokio::test]
    async fn the_condition_is_re_evaluated_per_iteration_after_setup() {
        let (workflow, engine) = compiled(&format!(
            r#"{{ "id": "w", "name": "w",
                  "condition": {{"!=": [{{"var": "temp_data.i"}}, 2]}},
                  "loop": {{"counter": "i", "max": 100, "setup": {SETUP},
                            "over": {{"var": "temp_data.batch"}}, "as": "item"}},
                  "tasks": [{OVER_BODY}] }}"#
        ));
        let mut message = items_message();
        // Before setup `temp_data.i` is absent, so `!= 2` admits; iterations
        // 0 and 1 run; the condition ends the loop at 2.
        let executed = executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .unwrap();
        assert!(executed);
        assert_eq!(counters(&message), vec![None, Some(0), Some(1)]);
    }

    #[tokio::test]
    async fn a_condition_false_on_the_first_iteration_after_setup_still_reports_executed() {
        let (workflow, engine) = compiled(&format!(
            r#"{{ "id": "w", "name": "w",
                  "condition": {{"!": [{{"var": "temp_data.batch"}}]}},
                  "loop": {{"counter": "i", "max": 100, "setup": {SETUP},
                            "over": {{"var": "temp_data.batch"}}, "as": "item"}},
                  "tasks": [{OVER_BODY}] }}"#
        ));
        let mut message = items_message();
        // Admitted before setup (no batch yet); setup writes the batch, so the
        // condition is false on the first iteration. Setup ran, so this is an
        // executed workflow that visited nothing — not a skip.
        let executed = executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .unwrap();
        assert!(executed);
        assert_eq!(counters(&message), vec![None]);
    }

    #[tokio::test]
    async fn a_terminal_setup_step_ends_the_workflow_before_any_iteration() {
        let (workflow, engine) = compiled(&format!(
            r#"{{ "id": "w", "name": "w",
                  "loop": {{"counter": "i", "max": 100,
                            "setup": [{{"id": "stop", "name": "stop", "terminal": true,
                                        "function": {{"name": "map", "input": {{"mappings": []}}}}}}],
                            "over": {{"var": "data.items"}}, "as": "item"}},
                  "tasks": [{OVER_BODY}] }}"#
        ));
        let mut message = items_message();
        let executed = executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .unwrap();
        assert!(executed);
        assert_eq!(counters(&message), vec![None], "only the setup step ran");
        assert!(message.errors.is_empty());
    }

    #[tokio::test]
    async fn a_setup_error_stops_the_message_unless_the_workflow_continues_on_error() {
        let make = |continue_on_error: bool| {
            compiled(&format!(
                r#"{{ "id": "w", "name": "w", "continue_on_error": {continue_on_error},
                      "loop": {{"counter": "i", "max": 100,
                                "setup": [{{"id": "boom", "name": "boom",
                                            "function": {{"name": "map", "input": {{"mappings": [
                                              {{"path": {{"var": "data.root"}}, "unset": true}}]}}}}}}],
                                "over": {{"var": "data.items"}}, "as": "item"}},
                      "tasks": [{OVER_BODY}] }}"#
            ))
        };
        // A computed removal path resolving to a context root is refused at run
        // time with status 500, which without task-level continue_on_error
        // becomes an `Err` from the pass.
        let data = || dv(json!({"items": [1], "root": "temp_data"}));

        let (workflow, engine) = make(false);
        let mut message = Message::builder().data(data()).build();
        assert!(
            executor(engine)
                .execute(&workflow, &mut message, Utc::now())
                .await
                .is_err()
        );
        assert_eq!(
            counters(&message),
            vec![None],
            "setup recorded, no iteration"
        );

        let (workflow, engine) = make(true);
        let mut message = Message::builder().data(data()).build();
        let executed = executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .unwrap();
        assert!(executed);
        assert_eq!(
            counters(&message),
            vec![None],
            "recorded and ended — no first iteration without setup"
        );
        assert!(message.errors.iter().any(|e| e.code == "WORKFLOW_ERROR"));
    }

    /// Task `flag` writes `temp_data.it.flag` on iteration 0 only; task `read`
    /// records whether the flag is set, at a computed path per index.
    const SCRATCH_BODY: &str = r#"
        {"id": "flag", "name": "flag", "condition": {"==": [{"var": "temp_data.i"}, 0]},
         "function": {"name": "map", "input": {"mappings": [
            {"path": "temp_data.it.flag", "logic": true}]}}},
        {"id": "read", "name": "read",
         "function": {"name": "map", "input": {"mappings": [
            {"path": {"cat": ["data.f", {"var": "temp_data.i"}]},
             "logic": {"!!": [{"var": "temp_data.it.flag"}]}}]}}}"#;

    async fn run_scratch_body(scratch: &str) -> Message {
        let (workflow, engine) = compiled(&format!(
            r#"{{ "id": "w", "name": "w",
                  "loop": {{"counter": "i", "max": 3{scratch}}},
                  "tasks": [{SCRATCH_BODY}] }}"#
        ));
        let mut message = Message::builder()
            .temp_data(dv(json!({"keep": "me"})))
            .build();
        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .unwrap();
        message
    }

    #[tokio::test]
    async fn scratch_is_reset_at_the_start_of_every_iteration() {
        let leaky = run_scratch_body("").await;
        assert_eq!(
            leaky.context["data"].get("f1"),
            Some(&dv(json!(true))),
            "without scratch the slot leaks into the next iteration"
        );

        let clean = run_scratch_body(r#", "scratch": "it""#).await;
        assert_eq!(clean.context["data"].get("f0"), Some(&dv(json!(true))));
        assert_eq!(clean.context["data"].get("f1"), Some(&dv(json!(false))));
        assert_eq!(clean.context["data"].get("f2"), Some(&dv(json!(false))));
        assert_eq!(
            clean.context["temp_data"].get("keep"),
            Some(&dv(json!("me"))),
            "the reset is narrow"
        );
        assert_eq!(
            clean.context["temp_data"].get("it"),
            Some(&dv(json!({}))),
            "left holding the last iteration's (empty) scratch"
        );
    }

    #[tokio::test]
    async fn an_uncompiled_over_is_an_error_not_a_panic() {
        // A workflow that never went through `compile_workflows`.
        let workflow = Workflow::from_json(&format!(
            r#"{{ "id": "w", "name": "w",
                  "loop": {{"counter": "i", "max": 3, "over": {{"var": "data.items"}}}},
                  "tasks": [{COUNTER_BODY}] }}"#
        ))
        .unwrap();
        let engine = LogicCompiler::new().into_engine();
        let mut message = items_message();
        let err = executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect_err("not compiled");
        assert!(err.to_string().contains("loop.over"), "got: {err}");
    }

    #[tokio::test]
    async fn filter_halt_breaks_the_whole_loop_not_just_one_sweep() {
        let (workflow, engine) = compiled(
            r#"{ "id": "w", "name": "w", "loop": {"counter": "i", "max": 10},
                 "tasks": [
                   {"id": "gate", "name": "gate", "function": {"name": "filter",
                     "input": {"condition": {"<": [{"var": "temp_data.i"}, 2]},
                               "on_reject": "halt"}}},
                   {"id": "body", "name": "body", "function": {"name": "map",
                     "input": {"mappings": [
                        {"path": "data.n", "logic": {"var": "temp_data.i"}}]}}}] }"#,
        );
        let mut message = Message::from_value(&json!({}));

        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("a halt is not an error");

        // Sweeps 0 and 1 run both tasks; sweep 2's gate halts and ends the
        // loop rather than moving on to sweep 3.
        let ids: Vec<&str> = message
            .audit_trail
            .iter()
            .map(|entry| entry.task_id.as_ref())
            .collect();
        assert_eq!(ids, ["gate", "body", "gate", "body", "gate"]);
        assert_eq!(
            counters(&message),
            vec![Some(0), Some(0), Some(1), Some(1), Some(2)]
        );
    }

    #[tokio::test]
    async fn init_and_increment_drive_the_counter() {
        let (workflow, engine) = compiled(&format!(
            r#"{{ "id": "w", "name": "w",
                  "loop": {{"counter": "i", "init": 10, "increment": 5, "max": 25}},
                  "tasks": [{COUNTER_BODY}] }}"#
        ));
        let mut message = Message::from_value(&json!({}));

        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("loop should complete");

        assert_eq!(counters(&message), vec![Some(10), Some(15), Some(20)]);
    }

    #[tokio::test]
    async fn a_loop_without_a_named_counter_still_records_it_on_the_audit_trail() {
        let (workflow, engine) = compiled(
            r#"{ "id": "w", "name": "w", "loop": {"max": 2},
                 "tasks": [{"id": "t", "name": "t",
                            "function": {"name": "map", "input": {"mappings": []}}}] }"#,
        );
        let mut message = Message::from_value(&json!({}));

        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("loop should complete");

        assert_eq!(counters(&message), vec![Some(0), Some(1)]);
        // Nothing was written to temp_data — the counter was never named.
        assert_eq!(message.context["temp_data"], dv(json!({})));
    }

    #[tokio::test]
    async fn a_non_looping_workflow_records_no_loop_counter() {
        let (workflow, engine) = compiled(
            r#"{ "id": "w", "name": "w",
                 "tasks": [{"id": "t", "name": "t",
                            "function": {"name": "map", "input": {"mappings": []}}}] }"#,
        );
        let mut message = Message::from_value(&json!({}));

        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("should complete");

        assert_eq!(counters(&message), vec![None]);
    }

    #[tokio::test]
    async fn progress_metadata_is_written_on_every_sweep() {
        // `metadata.progress` is load-bearing for cross-workflow chaining; a
        // loop must not gate it.
        let (workflow, engine) = compiled(&format!(
            r#"{{ "id": "w", "name": "w", "loop": {{"counter": "i", "max": 3}},
                  "tasks": [{COUNTER_BODY}] }}"#
        ));
        let mut message = Message::from_value(&json!({}));

        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("loop should complete");

        let progress = message.context["metadata"]
            .get("progress")
            .expect("progress must be written");
        assert_eq!(progress.get("workflow_id"), Some(&dv(json!("w"))));
        assert_eq!(progress.get("task_id"), Some(&dv(json!("t"))));
        assert_eq!(progress.get("status_code"), Some(&dv(json!(200))));
    }

    #[tokio::test]
    async fn the_engine_owns_the_counter_even_if_a_body_task_writes_it() {
        // A body task writing the counter path is overwritten at the next
        // increment, so termination reasoning stays local to LoopConfig.
        let (workflow, engine) = compiled(
            r#"{ "id": "w", "name": "w", "loop": {"counter": "i", "max": 3},
                 "tasks": [{"id": "t", "name": "t", "function": {"name": "map",
                    "input": {"mappings": [{"path": "temp_data.i", "logic": 99}]}}}] }"#,
        );
        let mut message = Message::from_value(&json!({}));

        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("loop should complete");

        assert_eq!(
            counters(&message),
            vec![Some(0), Some(1), Some(2)],
            "the body's write must not stall or skew the loop"
        );
    }

    /// Run a bare counting loop with the given bounds and return the counter
    /// values the sweeps actually recorded.
    async fn counter_sequence(init: i64, increment: i64, max: i64) -> Vec<Option<i64>> {
        let (workflow, engine) = compiled(&format!(
            r#"{{ "id": "w", "name": "w",
                  "loop": {{"counter": "i", "init": {init},
                            "increment": {increment}, "max": {max}}},
                  "tasks": [{{"id": "t", "name": "t",
                              "function": {{"name": "map", "input": {{"mappings": []}}}}}}] }}"#
        ));
        let mut message = Message::from_value(&json!({}));
        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("loop should complete");
        counters(&message)
    }

    #[tokio::test]
    async fn counter_sequence_matrix_over_init_increment_and_max() {
        // The half-open `counter < max` bound, swept across signs and step
        // sizes. Each expected list is the exact sequence of sweeps.
        let cases: Vec<(i64, i64, i64, Vec<i64>)> = vec![
            // Defaults: 0-based, step 1 — the array-index case.
            (0, 1, 1, vec![0]),
            (0, 1, 2, vec![0, 1]),
            (0, 1, 5, vec![0, 1, 2, 3, 4]),
            // Non-unit steps, including a range the step does not divide.
            (0, 2, 6, vec![0, 2, 4]),
            (0, 3, 10, vec![0, 3, 6, 9]),
            (0, 5, 3, vec![0]),
            (0, 100, 1, vec![0]),
            // Non-zero starts.
            (10, 5, 25, vec![10, 15, 20]),
            (3, 1, 6, vec![3, 4, 5]),
            // Negative and mixed-sign ranges.
            (-3, 1, 2, vec![-3, -2, -1, 0, 1]),
            (-4, 2, 1, vec![-4, -2, 0]),
            (-10, 5, -5, vec![-10]),
        ];

        for (init, increment, max, expected) in cases {
            let got = counter_sequence(init, increment, max).await;
            let expected: Vec<Option<i64>> = expected.into_iter().map(Some).collect();
            assert_eq!(got, expected, "init={init} increment={increment} max={max}");
        }
    }

    #[tokio::test]
    async fn the_counter_advance_saturates_instead_of_overflowing() {
        // A huge increment must end the loop, not wrap into a negative counter
        // and spin. Both the giant-step and the near-i64::MAX start are
        // exercised, since either could overflow a plain `+`.
        assert_eq!(
            counter_sequence(0, i64::MAX, 5).await,
            vec![Some(0)],
            "one sweep, then the advance saturates past max"
        );
        assert_eq!(
            counter_sequence(i64::MAX - 1, 1, i64::MAX).await,
            vec![Some(i64::MAX - 1)],
            "the last representable sweep still terminates"
        );
        assert_eq!(
            counter_sequence(i64::MAX - 2, i64::MAX, i64::MAX).await,
            vec![Some(i64::MAX - 2)]
        );
    }

    #[tokio::test]
    async fn a_task_condition_is_re_evaluated_against_the_counter_every_sweep() {
        // Per-sweep task conditions are the mechanism for "do this only on
        // some iterations"; a stale condition cache would break it.
        let (workflow, engine) = compiled(
            r#"{ "id": "w", "name": "w", "loop": {"counter": "i", "max": 4},
                 "tasks": [
                   {"id": "evens", "name": "evens",
                    "condition": {"==": [{"%": [{"var": "temp_data.i"}, 2]}, 0]},
                    "function": {"name": "map", "input": {"mappings": []}}},
                   {"id": "always", "name": "always",
                    "function": {"name": "map", "input": {"mappings": []}}}] }"#,
        );
        let mut message = Message::from_value(&json!({}));

        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("loop should complete");

        let entries: Vec<(&str, Option<i64>)> = message
            .audit_trail
            .iter()
            .map(|e| (e.task_id.as_ref(), e.loop_counter))
            .collect();
        assert_eq!(
            entries,
            [
                ("evens", Some(0)),
                ("always", Some(0)),
                ("always", Some(1)),
                ("evens", Some(2)),
                ("always", Some(2)),
                ("always", Some(3)),
            ],
            "the gated task runs only on even counters"
        );
    }

    #[tokio::test]
    async fn a_filter_skip_does_not_keep_the_loop_alive_or_record_entries() {
        // `TaskOutcome::Skip` records no audit entry and no progress write.
        // The loop is driven by its bound, not by whether tasks recorded
        // anything, so it still runs exactly `max` sweeps.
        let (workflow, engine) = compiled(
            r#"{ "id": "w", "name": "w", "loop": {"counter": "i", "max": 3},
                 "tasks": [{"id": "gate", "name": "gate", "function": {"name": "filter",
                    "input": {"condition": false, "on_reject": "skip"}}}] }"#,
        );
        let mut message = Message::from_value(&json!({}));

        let executed = executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("skip is not an error");

        assert!(executed, "sweeps ran even though every task skipped");
        assert!(message.audit_trail.is_empty(), "Skip records no entry");
        assert_eq!(
            message.context["temp_data"].get("i"),
            Some(&dv(json!(3))),
            "the loop still ran to its bound"
        );
    }

    #[tokio::test]
    async fn a_4xx_task_status_is_recorded_per_sweep_without_stopping_the_loop() {
        // A failing `validation` yields 400: warned, recorded, loop continues.
        let (workflow, engine) = compiled(
            r#"{ "id": "w", "name": "w", "loop": {"counter": "i", "max": 3},
                 "tasks": [{"id": "check", "name": "check", "function": {"name": "validation",
                    "input": {"rules": [{"logic": {"==": [1, 2]}, "message": "nope"}]}}}] }"#,
        );
        let mut message = Message::from_value(&json!({}));

        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("a 4xx does not stop the workflow");

        assert_eq!(counters(&message), vec![Some(0), Some(1), Some(2)]);
        assert!(
            message.audit_trail.iter().all(|e| e.status == 400),
            "every sweep recorded the 4xx"
        );
    }

    #[tokio::test]
    async fn the_rollout_gate_excludes_a_looping_workflow_before_any_sweep() {
        // The gate runs ahead of the loop, so an excluded workflow writes no
        // counter at all — it must be indistinguishable from a plain skip.
        let (workflow, engine) = compiled(
            r#"{ "id": "w", "name": "w",
                 "rollout": {"bucket_start": 0, "bucket_end": 50},
                 "loop": {"counter": "i", "max": 5},
                 "tasks": [{"id": "t", "name": "t",
                            "function": {"name": "map", "input": {"mappings": []}}}] }"#,
        );
        let mut message = Message::builder().routing_bucket(75).build();

        let executed = executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("an excluded workflow is not an error");

        assert!(!executed);
        assert!(message.audit_trail.is_empty());
        assert_eq!(
            message.context["temp_data"].get("i"),
            None,
            "no counter is written for an excluded workflow"
        );
    }

    #[tokio::test]
    async fn a_nested_counter_path_is_created_and_advanced() {
        let (workflow, engine) = compiled(
            r#"{ "id": "w", "name": "w",
                 "loop": {"counter": "cursor.index", "max": 3},
                 "tasks": [{"id": "t", "name": "t", "function": {"name": "map",
                    "input": {"mappings": [
                       {"path": "data.seen", "logic": {"var": "temp_data.cursor.index"}}]}}}] }"#,
        );
        let mut message = Message::from_value(&json!({}));

        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("loop should complete");

        assert_eq!(
            message.context["temp_data"]["cursor"].get("index"),
            Some(&dv(json!(3)))
        );
        assert_eq!(
            message.context["data"].get("seen"),
            Some(&dv(json!(2))),
            "the body read the nested counter"
        );
    }

    #[tokio::test]
    async fn writing_the_counter_preserves_unrelated_temp_data() {
        let (workflow, engine) = compiled(
            r#"{ "id": "w", "name": "w", "loop": {"counter": "i", "max": 2},
                 "tasks": [{"id": "t", "name": "t",
                            "function": {"name": "map", "input": {"mappings": []}}}] }"#,
        );
        let mut message = Message::builder()
            .temp_data(dv(json!({"keep": "me", "nested": {"a": 1}})))
            .build();

        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("loop should complete");

        assert_eq!(
            message.context["temp_data"].get("keep"),
            Some(&dv(json!("me")))
        );
        assert_eq!(
            message.context["temp_data"]["nested"].get("a"),
            Some(&dv(json!(1)))
        );
        assert_eq!(message.context["temp_data"].get("i"), Some(&dv(json!(2))));
    }

    #[tokio::test]
    async fn the_counter_overwrites_a_pre_existing_value_at_that_path() {
        // The engine owns the path: whatever was there before the loop is
        // replaced by `init` on the first sweep.
        let (workflow, engine) = compiled(
            r#"{ "id": "w", "name": "w", "loop": {"counter": "i", "init": 5, "max": 7},
                 "tasks": [{"id": "t", "name": "t",
                            "function": {"name": "map", "input": {"mappings": []}}}] }"#,
        );
        let mut message = Message::builder()
            .temp_data(dv(json!({"i": "not a number"})))
            .build();

        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("loop should complete");

        assert_eq!(counters(&message), vec![Some(5), Some(6)]);
        assert_eq!(message.context["temp_data"].get("i"), Some(&dv(json!(7))));
    }

    #[tokio::test]
    async fn a_loop_records_audit_entries_with_capture_changes_off() {
        // `capture_changes(false)` suppresses the per-change diff, not the
        // audit entries themselves — so the loop counter is still recorded.
        let (workflow, engine) = compiled(
            r#"{ "id": "w", "name": "w", "loop": {"counter": "i", "max": 2},
                 "tasks": [{"id": "t", "name": "t", "function": {"name": "map",
                    "input": {"mappings": [
                       {"path": "data.n", "logic": {"var": "temp_data.i"}}]}}}] }"#,
        );
        let mut message = Message::builder().capture_changes(false).build();

        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("loop should complete");

        assert_eq!(counters(&message), vec![Some(0), Some(1)]);
        assert!(
            message.audit_trail.iter().all(|e| e.changes.is_empty()),
            "no diffs captured, but the entries are still there"
        );
    }

    #[tokio::test]
    async fn two_loops_sharing_a_counter_name_do_not_interfere() {
        // Each loop re-initialises the path it owns, so the second starts from
        // its own `init` rather than inheriting where the first stopped.
        let first = r#"{ "id": "a", "name": "a", "priority": 0,
             "loop": {"counter": "i", "max": 2},
             "tasks": [{"id": "t", "name": "t",
                        "function": {"name": "map", "input": {"mappings": []}}}] }"#;
        let second = r#"{ "id": "b", "name": "b", "priority": 1,
             "loop": {"counter": "i", "init": 10, "max": 12},
             "tasks": [{"id": "t", "name": "t",
                        "function": {"name": "map", "input": {"mappings": []}}}] }"#;

        let compiler = LogicCompiler::new();
        let workflows = compiler
            .compile_workflows(vec![
                Workflow::from_json(first).unwrap(),
                Workflow::from_json(second).unwrap(),
            ])
            .expect("should compile");
        let exec = executor(compiler.into_engine());
        let mut message = Message::from_value(&json!({}));

        exec.run_all_borrowed(&workflows, &mut message, None, Utc::now())
            .await
            .expect("both loops should complete");

        let per_workflow: Vec<(&str, Option<i64>)> = message
            .audit_trail
            .iter()
            .map(|e| (e.workflow_id.as_ref(), e.loop_counter))
            .collect();
        assert_eq!(
            per_workflow,
            [
                ("a", Some(0)),
                ("a", Some(1)),
                ("b", Some(10)),
                ("b", Some(11)),
            ]
        );
    }

    #[tokio::test]
    async fn a_looping_workflow_between_sync_workflows_does_not_break_the_sync_run() {
        // Regression guard for the `joins_sync_run` change: a loop workflow is
        // excluded from the shared-arena run, which must split the run around
        // it rather than dropping its neighbours.
        let sync_wf = |id: &str, priority: u32| {
            format!(
                r#"{{ "id": "{id}", "name": "{id}", "priority": {priority},
                      "tasks": [{{"id": "t", "name": "t", "function": {{"name": "map",
                        "input": {{"mappings": [
                          {{"path": "data.{id}", "logic": true}}]}}}}}}] }}"#
            )
        };
        let loop_wf = r#"{ "id": "mid", "name": "mid", "priority": 1,
             "loop": {"counter": "i", "max": 2},
             "tasks": [{"id": "t", "name": "t", "function": {"name": "map",
                "input": {"mappings": [{"path": "data.mid", "logic": true}]}}}] }"#;

        let compiler = LogicCompiler::new();
        let workflows = compiler
            .compile_workflows(vec![
                Workflow::from_json(&sync_wf("before", 0)).unwrap(),
                Workflow::from_json(loop_wf).unwrap(),
                Workflow::from_json(&sync_wf("after", 2)).unwrap(),
            ])
            .expect("should compile");
        // All three are sync-only, but the loop must not join a shared run.
        assert!(workflows.iter().all(|w| w.fully_sync));
        assert!(!joins_sync_run(&workflows[1]));

        let exec = executor(compiler.into_engine());
        let mut message = Message::from_value(&json!({}));

        exec.run_all_borrowed(&workflows, &mut message, None, Utc::now())
            .await
            .expect("all three should run");

        for id in ["before", "mid", "after"] {
            assert_eq!(
                message.context["data"].get(id),
                Some(&dv(json!(true))),
                "workflow {id} must have run"
            );
        }
        let order: Vec<(&str, Option<i64>)> = message
            .audit_trail
            .iter()
            .map(|e| (e.workflow_id.as_ref(), e.loop_counter))
            .collect();
        assert_eq!(
            order,
            [
                ("before", None),
                ("mid", Some(0)),
                ("mid", Some(1)),
                ("after", None),
            ],
            "priority order is preserved across the split"
        );
    }

    #[tokio::test]
    async fn consecutive_non_looping_sync_workflows_still_share_one_run() {
        // The other half of the same regression: without a loop in the way,
        // every fully-sync workflow still groups as it always did.
        let compiler = LogicCompiler::new();
        let workflows = compiler
            .compile_workflows(vec![
                Workflow::from_json(
                    r#"{ "id": "a", "name": "a", "priority": 0, "tasks": [{"id": "t", "name": "t",
                         "function": {"name": "map", "input": {"mappings": [
                           {"path": "data.a", "logic": 1}]}}}] }"#,
                )
                .unwrap(),
                Workflow::from_json(
                    r#"{ "id": "b", "name": "b", "priority": 1,
                         "condition": {"==": [{"var": "data.a"}, 1]},
                         "tasks": [{"id": "t", "name": "t",
                         "function": {"name": "map", "input": {"mappings": [
                           {"path": "data.b", "logic": 2}]}}}] }"#,
                )
                .unwrap(),
            ])
            .expect("should compile");
        assert!(workflows.iter().all(joins_sync_run));

        let exec = executor(compiler.into_engine());
        let mut message = Message::from_value(&json!({}));
        exec.run_all_borrowed(&workflows, &mut message, None, Utc::now())
            .await
            .expect("both should run");

        // `b`'s condition reads what `a` wrote, which only works if the shared
        // arena context was refreshed across the workflow boundary.
        assert_eq!(message.context["data"].get("b"), Some(&dv(json!(2))));
        assert_eq!(counters(&message), vec![None, None]);
    }

    #[tokio::test]
    async fn a_loop_body_can_index_an_array_by_its_counter() {
        // The per-item pattern, using only core operators.
        let (workflow, engine) = compiled(
            r#"{ "id": "w", "name": "w", "loop": {"counter": "i", "max": 3},
                 "tasks": [{"id": "pick", "name": "pick", "function": {"name": "map",
                    "input": {"mappings": [
                       {"path": "data.picked",
                        "logic": {"merge": [{"var": "data.picked"},
                                            [{"val": [["data", "items",
                                                       {"var": "temp_data.i"}]]}]]}}]}}}] }"#,
        );
        let mut message = Message::builder()
            .data(dv(json!({"items": ["a", "b", "c"], "picked": []})))
            .build();

        executor(engine)
            .execute(&workflow, &mut message, Utc::now())
            .await
            .expect("loop should complete");

        assert_eq!(
            serde_json::Value::from(&message.context["data"]["picked"]),
            json!(["a", "b", "c"]),
            "each sweep appended the item at its own index"
        );
    }

    #[tokio::test]
    async fn test_workflow_executor_skip_condition() {
        // Create a workflow with a false condition
        let workflow_json = r#"{
            "id": "test_workflow",
            "name": "Test Workflow",
            "condition": false,
            "tasks": [{
                "id": "dummy_task",
                "name": "Dummy Task",
                "function": {
                    "name": "map",
                    "input": {"mappings": []}
                }
            }]
        }"#;

        let compiler = LogicCompiler::new();
        let mut workflow = Workflow::from_json(workflow_json).unwrap();

        // Compile the workflow condition
        let workflows = compiler.compile_workflows(vec![workflow.clone()]).unwrap();
        if let Some(compiled_workflow) = workflows.iter().find(|w| w.id == "test_workflow") {
            workflow = compiled_workflow.clone();
        }

        let engine = compiler.into_engine();
        let task_executor = Arc::new(TaskExecutor::new(
            Arc::new(HashMap::new()),
            Arc::clone(&engine),
        ));
        let workflow_executor = WorkflowExecutor::new(task_executor, engine);

        let mut message = Message::from_value(&json!({}));

        // Execute workflow - should be skipped due to false condition
        let executed = workflow_executor
            .execute(&workflow, &mut message, Utc::now())
            .await
            .unwrap();
        assert!(!executed);
        assert_eq!(message.audit_trail.len(), 0);
    }

    #[tokio::test]
    async fn test_workflow_executor_execute_success() {
        // Create a workflow with a true condition
        let workflow_json = r#"{
            "id": "test_workflow",
            "name": "Test Workflow",
            "condition": true,
            "tasks": [{
                "id": "dummy_task",
                "name": "Dummy Task",
                "function": {
                    "name": "map",
                    "input": {"mappings": []}
                }
            }]
        }"#;

        let compiler = LogicCompiler::new();
        let mut workflow = Workflow::from_json(workflow_json).unwrap();

        // Compile the workflow
        let workflows = compiler.compile_workflows(vec![workflow.clone()]).unwrap();
        if let Some(compiled_workflow) = workflows.iter().find(|w| w.id == "test_workflow") {
            workflow = compiled_workflow.clone();
        }

        let engine = compiler.into_engine();
        let task_executor = Arc::new(TaskExecutor::new(
            Arc::new(HashMap::new()),
            Arc::clone(&engine),
        ));
        let workflow_executor = WorkflowExecutor::new(task_executor, engine);

        let mut message = Message::from_value(&json!({}));

        // Execute workflow - should succeed with empty task list
        let executed = workflow_executor
            .execute(&workflow, &mut message, Utc::now())
            .await
            .unwrap();
        assert!(executed);
    }
}