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, ErrorInfo, Result, service_error_code};
use crate::engine::executor::{
    ArenaContext, evaluate_condition, evaluate_condition_in_arena, with_arena,
};
use crate::engine::functions::BoxedFunctionHandler;
use crate::engine::message::{AuditTrail, Change, Message};
use crate::engine::observer::{ExecutionObserver, TaskEvent};
use crate::engine::task::Task;
use crate::engine::task_executor::TaskExecutor;
use crate::engine::task_outcome::TaskOutcome;
use crate::engine::trace::{ExecutionStep, ExecutionTrace, duration_us_between};
use crate::engine::utils::set_nested_value;
use crate::engine::workflow::Workflow;
use chrono::{DateTime, Utc};
use core::time::Duration;
use datalogic_rs::Engine;
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,
}

/// 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));
    }
}

/// 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.
///
/// Nested `match` rather than a let-chain: MSRV is 1.85. See
/// `write_progress_metadata` below for the same reason.
fn rollout_admits(workflow: &Workflow, message: &Message) -> bool {
    match workflow.rollout {
        None => true,
        Some(r) => match message.routing_bucket() {
            None => true,
            Some(b) => r.accepts(b),
        },
    }
}

/// 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 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" => {
                *v = OwnedDataValue::String(workflow_id.to_string());
                matched += 1;
            }
            "task_id" => {
                *v = OwnedDataValue::String(task_id.to_string());
                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 — no Vec/Object/key-`String` allocations, just the
/// two unavoidable id `String`s. 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,
) {
    // Nested `if let` rather than a let-chain: let-chains are stable only from
    // Rust 1.88 and this crate's MSRV is 1.85. Keep it that way.
    if let OwnedDataValue::Object(top) = context {
        if 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 {
                        if 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),
    );
}

/// 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>>,
}

impl WorkflowExecutor {
    /// Create a new WorkflowExecutor
    pub fn new(task_executor: Arc<TaskExecutor>, engine: Arc<Engine>) -> Self {
        Self {
            task_executor,
            engine,
            observer: 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()
    }

    /// 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 let Some(observer) = self.observer.as_ref() {
            let status = match result {
                Ok((outcome, _)) => outcome.audit_status(),
                Err(_) => Some(500),
            };
            let duration = started_at
                .map(|s| Duration::from_micros(duration_us_between(s, Utc::now())))
                .unwrap_or_default();
            observer.task_finished(&TaskEvent {
                workflow_id: &workflow.id,
                task_id: &task.id,
                function: task.function.function_name(),
                status,
                duration,
            });
        }
    }

    /// 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())
    }

    /// 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()
    }

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

    /// Unified workflow-condition + task-loop driver. `trace` is `None` for
    /// the production path and `Some(&mut trace)` for the debug path —
    /// stepping is the only behavioural difference between them.
    ///
    /// 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_inner(
        &self,
        workflow: &Workflow,
        message: &mut Message,
        mut trace: Option<&mut ExecutionTrace>,
        now: DateTime<Utc>,
    ) -> Result<bool> {
        /// 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 (from the first async
            /// boundary onward).
            Continue,
        }

        // Traffic-split gate, ahead of the arena scope below 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);
        }

        let tasks = &workflow.tasks;
        let first_boundary = next_async_boundary(tasks, 0);

        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.
                Ok(FirstStretch::Continue)
            } 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);
                    }
                    if first_boundary == 0 {
                        return Ok(FirstStretch::Continue);
                    }
                    let halted = self.run_tasks_slice_in_arena(
                        &tasks[..first_boundary],
                        workflow,
                        message,
                        &mut arena_ctx,
                        trace.as_deref_mut(),
                        now,
                    )?;
                    Ok(if halted {
                        FirstStretch::Halted
                    } else {
                        FirstStretch::Continue
                    })
                })
            };

        // Drive the remaining (async-containing) tail, then apply the single
        // workflow-level error contract to whichever half failed.
        let run_result: Result<()> = match first {
            Ok(FirstStretch::Skipped) => {
                note_workflow_skip(trace.as_deref_mut(), &workflow.id, "condition not met");
                return Ok(false);
            }
            Ok(FirstStretch::Halted) => Ok(()),
            Ok(FirstStretch::Continue) => {
                self.execute_tasks(workflow, message, trace, now, first_boundary)
                    .await
            }
            Err(e) => Err(e),
        };

        match run_result {
            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)
                }
            }
        }
    }

    /// 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.
    async fn execute_tasks(
        &self,
        workflow: &Workflow,
        message: &mut Message,
        mut trace: Option<&mut ExecutionTrace>,
        now: DateTime<Utc>,
        start: usize,
    ) -> Result<()> {
        let tasks = &workflow.tasks;
        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.
                let halt = self.run_sync_stretch(
                    &tasks[idx..stretch_end],
                    workflow,
                    message,
                    trace.as_deref_mut(),
                    now,
                )?;
                if halt {
                    return Ok(());
                }
                idx = stretch_end;
            }

            if idx < tasks.len() {
                // Single async task (or non-sync-builtin) at `idx`.
                let task = &tasks[idx];
                let should_execute = evaluate_condition(
                    &self.engine,
                    task.compiled_condition.as_ref(),
                    &message.context,
                )?;

                if !should_execute {
                    debug!("Skipping task {} - condition not met", task.id);
                    if let Some(t) = trace.as_deref_mut() {
                        t.add_step(ExecutionStep::task_skipped(&workflow.id, &task.id));
                    }
                    idx += 1;
                    continue;
                }

                // 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 trace.is_some() {
                    Some(Utc::now())
                } else {
                    None
                };
                let obs_start = trace_start.or_else(|| self.observer_clock());

                let result = self.task_executor.execute(task, message).await;

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

                let control_flow = self.handle_task_result(
                    result,
                    &workflow.id_arc,
                    &task.id_arc,
                    task.continue_on_error,
                    message,
                    now,
                )?;

                // Async tasks at the boundary have no per-mapping snapshots —
                // they're either HTTP/Kafka/Enrich or a custom handler.
                if let Some(t) = trace.as_deref_mut() {
                    let started_at = trace_start.unwrap_or(now);
                    t.add_executed_step(
                        &workflow.id,
                        &task.id,
                        message,
                        started_at,
                        duration_us_between(started_at, Utc::now()),
                        None,
                    );
                }

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

        Ok(())
    }

    /// 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,
        tasks: &[Task],
        workflow: &Workflow,
        message: &mut Message,
        trace: Option<&mut ExecutionTrace>,
        now: DateTime<Utc>,
    ) -> Result<bool> {
        with_arena(|arena| -> Result<bool> {
            let mut arena_ctx = ArenaContext::from_owned(&message.context, arena);
            self.run_tasks_slice_in_arena(tasks, workflow, message, &mut arena_ctx, trace, now)
        })
    }

    /// 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,
        tasks: &'arena [Task],
        workflow: &Workflow,
        message: &mut Message,
        arena_ctx: &mut ArenaContext<'arena>,
        mut trace: Option<&mut ExecutionTrace>,
        now: DateTime<Utc>,
    ) -> Result<bool> {
        let arena = arena_ctx.arena();

        for task in tasks {
            // Task condition — evaluate against the arena form so we don't
            // re-borrow the thread-local `RefCell`. A `None` compiled
            // condition (compiler folds the default literal `true` to
            // `None`) skips both the eval and the per-task arena context
            // slice build.
            let should_execute = match task.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 {
                debug!("Skipping task {} - condition not met", task.id);
                if let Some(t) = trace.as_deref_mut() {
                    t.add_step(ExecutionStep::task_skipped(&workflow.id, &task.id));
                }
                continue;
            }

            // 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
            };

            // 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 trace.is_some() {
                Some(Utc::now())
            } else {
                None
            };
            let obs_start = trace_start.or_else(|| self.observer_clock());

            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, obs_start);

            let control_flow = self.handle_task_result(
                result,
                &workflow.id_arc,
                &task.id_arc,
                task.continue_on_error,
                message,
                now,
            )?;

            // The only context write `handle_task_result` performs is
            // `metadata.progress`. Refresh exactly that depth-2 slot so the
            // next task — and, in the cross-workflow path, the next
            // workflow's condition — sees it, without re-arenaing unrelated
            // metadata children (mapped `metadata.routing.*`, chained
            // workflow state, …) after every task.
            arena_ctx.refresh_for_path(&message.context, "metadata.progress");

            if let Some(t) = trace.as_deref_mut() {
                let started_at = trace_start.unwrap_or(now);
                let mapping_contexts = if mapping_snapshots.is_empty() {
                    None
                } else {
                    Some(mapping_snapshots)
                };
                t.add_executed_step(
                    &workflow.id,
                    &task.id,
                    message,
                    started_at,
                    duration_us_between(started_at, Utc::now()),
                    mapping_contexts,
                );
            }

            if matches!(control_flow, TaskControlFlow::HaltWorkflow) {
                return Ok(true);
            }
        }
        Ok(false)
    }

    /// 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,
        mut trace: Option<&mut ExecutionTrace>,
        now: DateTime<Utc>,
    ) -> Result<()> {
        let mut i = 0;
        while i < workflows.len() {
            if workflows[i].borrow().fully_sync {
                // 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() && workflows[j].borrow().fully_sync {
                    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<()> {
        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;
                }

                match self.run_tasks_slice_in_arena(
                    &workflow.tasks,
                    workflow,
                    message,
                    &mut arena_ctx,
                    trace.as_deref_mut(),
                    now,
                ) {
                    // Filter-halt stops only this workflow; carry on with the
                    // next one (and keep the shared arena context).
                    Ok(_halted) => {
                        info!("Successfully completed workflow: {}", workflow.id);
                    }
                    Err(e) => {
                        // 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()
                ))
            })?
    }

    /// 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>,
        continue_on_error: bool,
        message: &mut Message,
        now: DateTime<Utc>,
    ) -> Result<TaskControlFlow> {
        let workflow_id: &str = workflow_id_arc;
        let task_id: &str = task_id_arc;
        match result {
            Ok((TaskOutcome::Skip, _)) => {
                // No audit trail, no progress write — task has explicitly opted
                // out (filter gate set to `Skip`).
                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");
                let halt = outcome.halts_workflow();

                // 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: now,
                    workflow_id: Arc::clone(workflow_id_arc),
                    task_id: Arc::clone(task_id_arc),
                    status: status as usize,
                    changes,
                });

                // 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);

                if halt {
                    info!("Task {} halted workflow {}", task_id, workflow_id);
                    return Ok(TaskControlFlow::HaltWorkflow);
                }

                // Check status code
                if (400..500).contains(&status) {
                    warn!("Task {} returned client error status: {}", task_id, status);
                } 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.
                    message.errors.push(
                        ErrorInfo::builder(
                            "TASK_STATUS_ERROR",
                            format!("Task {} returned status {}", task_id, status),
                        )
                        .workflow_id(workflow_id)
                        .task_id(task_id)
                        .build(),
                    );
                    if !continue_on_error {
                        return Err(DataflowError::Task(format!(
                            "Task {} failed with status {}",
                            task_id, status
                        )));
                    }
                }
                Ok(TaskControlFlow::Continue)
            }
            Err(e) => {
                error!("Task {} failed: {:?}", task_id, e);

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

                // 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 keeps the historical `TASK_ERROR`.
                // 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);
                // Nested `if let`, not a let-chain: MSRV is 1.85.
                if let Some(detail) = e.detail() {
                    info = info.detail(detail);
                }
                message.errors.push(info.build());

                if !continue_on_error {
                    Err(e)
                } else {
                    Ok(TaskControlFlow::Continue)
                }
            }
        }
    }
}

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

    #[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);
    }
}