}
                    all_outputs.push(current_input.clone());
                }

                // -- Operator nodes (#4980) ----------------------------------
                //
                // Operator-node arms never call `agent_resolver`. They
                // record a `StepResult` with an empty `agent_id` / a
                // synthetic `agent_name` so the run history surfaces the
                // step uniformly, leave `current_input` untouched (so
                // downstream `{{input}}` substitutions still see the
                // previous step's output), and emit a structured log so
                // operators can see what happened in the daemon log
                // without diffing run records.
                //
                // `Wait` is the only one with real semantics in this PR;
                // the other four log a `warn!` and return success so the
                // wire format is usable from day one while the deferred
                // design questions (Gate.condition syntax,
                // Approval operator-identity, Transform.code shape,
                // Branch jump semantics) are still open. See #4980.
                StepMode::Wait { duration_secs } => {
                    let start = std::time::Instant::now();
                    // Reject the step before we sleep when the manifest
                    // requested longer than the documented cap. The
                    // validator already rejects this on workflow
                    // registration, but defensive double-checking here
                    // covers persisted-pre-cap workflows reloaded after
                    // an upgrade.
                    if *duration_secs > MAX_WAIT_SECS {
                        let err = format!(
                            "Wait step '{}' duration_secs={duration_secs} exceeds cap {MAX_WAIT_SECS}",
                            step.name
                        );
                        warn!(error = %err, "Wait step rejected by cap");
                        if let Some(mut r) = self.runs.get_mut(&run_id) {
                            if !matches!(r.state, WorkflowRunState::Cancelled) {
                                r.state = WorkflowRunState::Failed;
                                r.error = Some(err.clone());
                                r.completed_at = Some(Utc::now());
                            }
                        }
                        return Err(err);
                    }
                    let dur = std::time::Duration::from_secs(*duration_secs);
                    let notify = self.cancel_notify.get(&run_id).map(|n| Arc::clone(&*n));

                    // Race the sleep against a cancellation signal so a
                    // long Wait honours `cancel_run` at sub-step
                    // granularity. Without this, a `Wait { 86400 }`
                    // would ignore cancellation for a full day before
                    // the step boundary observed the Cancelled state.
                    let cancelled = if let Some(n) = notify {
                        tokio::select! {
                            _ = tokio::time::sleep(dur) => false,
                            _ = n.notified() => true,
                        }
                    } else {
                        tokio::time::sleep(dur).await;
                        false
                    };

                    if cancelled {
                        info!(
                            run_id = %run_id,
                            step = i + 1,
                            name = %step.name,
                            "Wait step cancelled mid-sleep"
                        );
                        return Err("workflow run cancelled".into());
                    }

                    let duration_ms = start.elapsed().as_millis() as u64;
                    let output = current_input.clone();
                    let step_result = StepResult {
                        step_name: step.name.clone(),
                        agent_id: String::new(),
                        agent_name: "_operator:wait".to_string(),
                        prompt: Self::operator_prompt_trace(
                            "wait",
                            serde_json::json!({ "duration_secs": duration_secs }),
                        ),
                        output: output.clone(),
                        input_tokens: 0,
                        output_tokens: 0,
                        duration_ms,
                    };
                    if let Some(mut r) = self.runs.get_mut(&run_id) {
                        r.step_results.push(step_result);
                    }
                    if let Some(ref var) = step.output_var {
                        variables.insert(var.clone(), output.clone());
                    }
                    all_outputs.push(output);
                    info!(
                        step = i + 1,
                        name = %step.name,
                        duration_secs,
                        duration_ms,
                        "Wait step completed"
                    );
                }

                StepMode::Gate { condition } => {
                    let start = std::time::Instant::now();
                    let eval = evaluate_gate_condition(condition, &current_input);
                    let duration_ms = start.elapsed().as_millis() as u64;
                    let condition_json =
                        serde_json::to_value(condition).unwrap_or(serde_json::Value::Null);
                    let input_trace = Self::truncate_operator_input_trace(&current_input);
                    match eval {
                        Ok(()) => {
                            let output = current_input.clone();
                            let step_result = StepResult {
                                step_name: step.name.clone(),
                                agent_id: String::new(),
                                agent_name: "_operator:gate".to_string(),
                                // Unified operator-prompt-trace JSON
                                // shape (#4980 follow-up nit #4): every
                                // operator records `{op,...}` so a future
                                // dashboard renderer dispatches on `op`
                                // alone. Carries the truncated decision
                                // input so a debugger can see *what* the
                                // comparator saw (nit #5).
                                prompt: Self::operator_prompt_trace(
                                    "gate",
                                    serde_json::json!({
                                        "condition": condition_json,
                                        "passed": true,
                                        "input": input_trace,
                                    }),
                                ),
                                output: output.clone(),
                                input_tokens: 0,
                                output_tokens: 0,
                                duration_ms,
                            };
                            if let Some(mut r) = self.runs.get_mut(&run_id) {
                                r.step_results.push(step_result);
                            }
                            if let Some(ref var) = step.output_var {
                                variables.insert(var.clone(), output.clone());
                            }
                            all_outputs.push(output);
                            info!(
                                step = i + 1,
                                name = %step.name,
                                field = ?condition.field,
                                op = %condition.op,
                                duration_ms,
                                "Gate step passed"
                            );
                        }
                        Err(reason) => {
                            // A failed gate halts the run with a recorded
                            // reason. We surface a synthetic StepResult so
                            // the operator can see *which* step blocked the
                            // workflow in the dashboard run history; the
                            // run itself transitions to Failed via the
                            // standard error path below.
                            let step_result = StepResult {
                                step_name: step.name.clone(),
                                agent_id: String::new(),
                                agent_name: "_operator:gate".to_string(),
                                prompt: Self::operator_prompt_trace(
                                    "gate",
                                    serde_json::json!({
                                        "condition": condition_json,
                                        "passed": false,
                                        "input": input_trace,
                                    }),
                                ),
                                output: reason.clone(),
                                input_tokens: 0,
                                output_tokens: 0,
                                duration_ms,
                            };
                            if let Some(mut r) = self.runs.get_mut(&run_id) {
                                r.step_results.push(step_result);
                            }
                            let err =
                                format!("Gate step '{}' blocked workflow: {reason}", step.name);
                            warn!(
                                step = i + 1,
                                name = %step.name,
                                field = ?condition.field,
                                op = %condition.op,
                                reason = %reason,
                                "Gate step blocked workflow"
                            );
                            if let Some(mut r) = self.runs.get_mut(&run_id) {
                                if !matches!(r.state, WorkflowRunState::Cancelled) {
                                    r.state = WorkflowRunState::Failed;
                                    r.error = Some(err.clone());
                                    r.completed_at = Some(Utc::now());
                                }
                            }
                            return Err(err);
                        }
                    }
                }

                StepMode::Approval {
                    recipients,
                    timeout_secs,
                } => {
                    // Cross-issue dependency marker, not a vanilla TODO:
                    // the Approval executor needs the async-task-tracker
                    // landing in #4983 to suspend the run on a channel
                    // and resume it when a human replies. Until #4983
                    // lands the stub stays a structured warn-and-noop so
                    // a workflow that includes Approval still completes
                    // visibly rather than failing closed.
                    // TODO(#4983): wire real Approval executor once the
                    // long-pending async-task tracker is available.
                    warn!(
                        step = i + 1,
                        name = %step.name,
                        recipients = ?recipients,
                        timeout_secs = ?timeout_secs,
                        "Approval executor not yet implemented — blocked on async-task-tracker landing in #4983 (refs #4980)"
                    );
                    Self::record_operator_noop_step_result(
                        &self.runs,
                        run_id,
                        step,
                        "_operator:approval",
                        &current_input,
                        &mut variables,
                        &mut all_outputs,
                    );
                }

                StepMode::Transform { code } => {
                    let start = std::time::Instant::now();
                    // Tera's context iterates its insertion order, so
                    // copy `variables` into a `BTreeMap` for
                    // determinism (#3298). The Tera renderer never
                    // reaches an LLM prompt directly today, but the
                    // rendered output flows into `current_input` and
                    // is consumed by downstream agent steps via
                    // `{{input}}` expansion — a non-deterministic iteration
                    // order through `vars` would silently change
                    // prompts across processes and invalidate the
                    // provider prompt cache.
                    let bt_vars: std::collections::BTreeMap<String, String> = variables
                        .iter()
                        .map(|(k, v)| (k.clone(), v.clone()))
                        .collect();
                    let result = render_transform_template(code, &current_input, &bt_vars);
                    let duration_ms = start.elapsed().as_millis() as u64;

                    match result {
                        Ok(rendered) => {
                            // Cap the rendered payload size: a template
                            // like `{% for i in range(end=1e7) %}x{% endfor %}`
                            // expands to tens of MiB, which pollutes
                            // `current_input` (read by every downstream
                            // `{{input}}` agent step) and the persisted
                            // `step_result.output`. Halt with a typed
                            // reason rather than silently propagating a
                            // huge blob.
                            if rendered.len() > MAX_TRANSFORM_OUTPUT_BYTES {
                                let err = format!(
                                    "Transform step '{}' rendered {} bytes (cap {MAX_TRANSFORM_OUTPUT_BYTES})",
                                    step.name,
                                    rendered.len()
                                );
                                warn!(
                                    step = i + 1,
                                    name = %step.name,
                                    rendered_bytes = rendered.len(),
                                    cap = MAX_TRANSFORM_OUTPUT_BYTES,
                                    "Transform step exceeded output cap"
                                );
                                if let Some(mut r) = self.runs.get_mut(&run_id) {
                                    if !matches!(r.state, WorkflowRunState::Cancelled) {
                                        r.state = WorkflowRunState::Failed;
                                        r.error = Some(err.clone());
                                        r.completed_at = Some(Utc::now());
                                    }
                                }
                                return Err(err);
                            }
                            let step_result = StepResult {
                                step_name: step.name.clone(),
                                agent_id: String::new(),
                                agent_name: "_operator:transform".to_string(),
                                prompt: Self::operator_prompt_trace(
                                    "transform",
                                    serde_json::json!({ "code": code }),
                                ),
                                output: rendered.clone(),
                                input_tokens: 0,
                                output_tokens: 0,
                                duration_ms,
                            };
                            if let Some(mut r) = self.runs.get_mut(&run_id) {
                                r.step_results.push(step_result);
                            }
                            if let Some(ref var) = step.output_var {
                                variables.insert(var.clone(), rendered.clone());
                            }
                            all_outputs.push(rendered.clone());
                            current_input = rendered;
                            info!(
                                step = i + 1,
                                name = %step.name,
                                duration_ms,
                                "Transform step rendered"
                            );
                        }
                        Err(reason) => {
                            let err = format!("Transform step '{}' failed: {reason}", step.name);
                            warn!(
                                step = i + 1,
                                name = %step.name,
                                reason = %reason,
                                "Transform step failed"
                            );
                            // Record a synthetic StepResult so the
                            // operator can see which transform step
                            // blew up in the run history; the
                            // `output` slot carries the Tera error
                            // (line + column included by Tera).
                            let step_result = StepResult {
                                step_name: step.name.clone(),
                                agent_id: String::new(),
                                agent_name: "_operator:transform".to_string(),
                                prompt: Self::operator_prompt_trace(
                                    "transform",
                                    serde_json::json!({ "code": code }),
                                ),
                                output: reason.clone(),
                                input_tokens: 0,
                                output_tokens: 0,
                                duration_ms,
                            };
                            if let Some(mut r) = self.runs.get_mut(&run_id) {
                                r.step_results.push(step_result);
                            }
                            if let Some(mut r) = self.runs.get_mut(&run_id) {
                                if !matches!(r.state, WorkflowRunState::Cancelled) {
                                    r.state = WorkflowRunState::Failed;
                                    r.error = Some(err.clone());
                                    r.completed_at = Some(Utc::now());
                                }
                            }
                            return Err(err);
                        }
                    }
                }

                StepMode::Branch { arms } => {
                    let start = std::time::Instant::now();
                    // Resolve the value to match against. Parse as JSON
                    // when possible — that lets numeric and structural
                    // match values (`0.8`, `{"status":"ok"}`) compare
                    // by JSON deep-equality rather than string form. A
                    // non-JSON predecessor compares its raw output
                    // against the string form of each arm's
                    // `match_value`.
                    let parsed_input: Option<serde_json::Value> =
                        serde_json::from_str(&current_input).ok();
                    let matched_arm_idx = arms.iter().position(|arm| match &parsed_input {
                        Some(parsed) => parsed == &arm.match_value,
                        None => {
                            let rhs = match &arm.match_value {
                                serde_json::Value::String(s) => s.clone(),
                                other => other.to_string(),
                            };
                            current_input == rhs
                        }
                    });
                    let duration_ms = start.elapsed().as_millis() as u64;

                    match matched_arm_idx {
                        Some(arm_idx) => {
                            let arm = &arms[arm_idx];
                            // Resolve the target step name to its
                            // index. The dispatcher only honours
                            // FORWARD jumps — a backward jump would
                            // let an unbounded loop hide inside a
                            // Branch when `Loop` already exists for
                            // that semantic.
                            //
                            // Defensive uniqueness check: duplicate
                            // step-name detection lives in
                            // `build_dependency_graph`, which is only
                            // reached via `topological_sort`. Sequential
                            // workflows that have no `depends_on` edges
                            // can skip that path entirely, so we refuse
                            // an ambiguous target here rather than let
                            // `iter().position` pick the silent first
                            // match.
                            let mut target_iter = workflow
                                .steps
                                .iter()
                                .enumerate()
                                .filter(|(_, s)| s.name == arm.then);
                            let first = target_iter.next();
                            let second = target_iter.next();
                            let target_idx = match (first, second) {
                                (Some((idx, _)), None) => Some(idx),
                                (None, _) => None,
                                (Some(_), Some(_)) => {
                                    let err = format!(
                                        "Branch step '{}' target name '{}' is ambiguous: \
                                         multiple steps share that name",
                                        step.name, arm.then
                                    );
                                    warn!(
                                        error = %err,
                                        "Branch step blocked workflow on ambiguous target"
                                    );
                                    if let Some(mut r) = self.runs.get_mut(&run_id) {
                                        if !matches!(r.state, WorkflowRunState::Cancelled) {
                                            r.state = WorkflowRunState::Failed;
                                            r.error = Some(err.clone());
                                            r.completed_at = Some(Utc::now());
                                        }
                                    }
                                    return Err(err);
                                }
                            };
                            match target_idx {
                                Some(t) if t > i => {
                                    let output = current_input.clone();
                                    // Carry the truncated decision input
                                    // into the trace so an operator
                                    // debugging a "wrong arm fired"
                                    // report can see the value the
                                    // comparator saw, not just the arm
                                    // index (#4980 review nit #5).
                                    let input_trace =
                                        Self::truncate_operator_input_trace(&current_input);
                                    let step_result = StepResult {
                                        step_name: step.name.clone(),
                                        agent_id: String::new(),
                                        agent_name: "_operator:branch".to_string(),
                                        prompt: Self::operator_prompt_trace(
                                            "branch",
                                            serde_json::json!({
                                                "target": arm.then,
                                                "arm_idx": arm_idx,
                                                "arms": arms.len(),
                                                "matched": true,
                                                "input": input_trace,
                                            }),
                                        ),
                                        output: output.clone(),
                                        input_tokens: 0,
                                        output_tokens: 0,
                                        duration_ms,
                                    };
                                    if let Some(mut r) = self.runs.get_mut(&run_id) {
                                        r.step_results.push(step_result);
                                    }
                                    if let Some(ref var) = step.output_var {
                                        variables.insert(var.clone(), output.clone());
                                    }
                                    all_outputs.push(output);
                                    info!(
                                        step = i + 1,
                                        name = %step.name,
                                        target = %arm.then,
                                        target_idx = t,
                                        duration_ms,
                                        "Branch jumped to target step"
                                    );
                                    i = t;
                                    continue;
                                }
                                Some(t) => {
                                    let err = format!(
                                        "Branch step '{}' target '{}' (index {}) is at or before current step (index {}) — backward jumps not allowed",
                                        step.name, arm.then, t, i
                                    );
                                    warn!(error = %err, "Branch step blocked workflow");
                                    if let Some(mut r) = self.runs.get_mut(&run_id) {
                                        if !matches!(r.state, WorkflowRunState::Cancelled) {
                                            r.state = WorkflowRunState::Failed;
                                            r.error = Some(err.clone());
                                            r.completed_at = Some(Utc::now());
                                        }
                                    }
                                    return Err(err);
                                }
                                None => {
                                    let err = format!(
                                        "Branch step '{}' target step '{}' not found in workflow",
                                        step.name, arm.then
                                    );
                                    warn!(error = %err, "Branch step blocked workflow");
                                    if let Some(mut r) = self.runs.get_mut(&run_id) {
                                        if !matches!(r.state, WorkflowRunState::Cancelled) {
                                            r.state = WorkflowRunState::Failed;
                                            r.error = Some(err.clone());
                                            r.completed_at = Some(Utc::now());
                                        }
                                    }
                                    return Err(err);
                                }
                            }
                        }
                        None => {
                            // No arm matched. We could fall through
                            // (treating Branch as a no-op when nothing
                            // matches) but that hides operator
                            // mistakes — they almost certainly meant
                            // for *some* arm to match. Halt with a
                            // typed reason; a future additive shape
                            // (`default_then: Option<String>`) can
                            // relax this when explicitly opted into.
                            let input_trace = Self::truncate_operator_input_trace(&current_input);
                            let reason = format!(
                                "Branch step '{}' had no matching arm for output: {}",
                                step.name, input_trace
                            );
                            warn!(reason = %reason, "Branch step blocked workflow");
                            let step_result = StepResult {
                                step_name: step.name.clone(),
                                agent_id: String::new(),
                                agent_name: "_operator:branch".to_string(),
                                prompt: Self::operator_prompt_trace(
                                    "branch",
                                    serde_json::json!({
                                        "arms": arms.len(),
                                        "matched": false,
                                        "input": input_trace,
                                    }),
                                ),
                                output: reason.clone(),
                                input_tokens: 0,
                                output_tokens: 0,
                                duration_ms,
                            };
                            if let Some(mut r) = self.runs.get_mut(&run_id) {
                                r.step_results.push(step_result);
                            }
                            if let Some(mut r) = self.runs.get_mut(&run_id) {
                                if !matches!(r.state, WorkflowRunState::Cancelled) {
                                    r.state = WorkflowRunState::Failed;
                                    r.error = Some(reason.clone());
                                    r.completed_at = Some(Utc::now());
                                }
                            }
                            return Err(reason);
                        }
                    }
                }

                StepMode::Operator {
                    notify,
                    actions,
                    timeout_secs,
                    timeout_action,
                } => {
                    // #4977 step 1/N — skeleton executor.
                    //
                    // What runs today: record a synthetic StepResult
                    // under the `_operator:operator` agent name
                    // (matching the `_operator:` namespace from
                    // #5035), log the configuration for observability,
                    // and request a pause on the run via the existing
                    // `pause_request` mechanism. The pause is honored
                    // by the pause-request gate at the *next* iteration
                    // of the while loop, so the run transitions to
                    // `Paused` with a fresh resume token before any
                    // downstream step executes. Operators / dashboards
                    // resume the run via the existing `resume_run`
                    // API once the human has acted.
                    //
                    // TODO(#4977 step 2): dispatch channel
                    // notifications via the channel bridge so the
                    // configured `notify` recipients actually receive
                    // the artifact + action buttons. Today the
                    // skeleton only logs which recipients *would*
                    // have been notified.
                    //
                    // TODO(#4977 step 2): spawn a timeout watchdog
                    // honouring `timeout_secs` + `timeout_action` so
                    // a `OperatorTimeoutAction::Approve` / `Reject` /
                    // `Fail` resolves the pause automatically when
                    // the budget elapses. Today `timeout_action` is
                    // recorded on the trace but not enforced.
                    //
                    // TODO(#4977 step 2): wire the operator HTTP
                    // actions endpoint so an operator's
                    // `OperatorAction::Approve` / `Reject` / `Edit`
                    // / `ProvideInput` / `FreeformInput` response
                    // translates into a `resume_run` call with the
                    // appropriate step output.
                    let input_trace = Self::truncate_operator_input_trace(&current_input);
                    let notify_count = notify.len();
                    let action_count = actions.len();
                    let timeout_action_label = match timeout_action {
                        OperatorTimeoutAction::Approve => "approve",
                        OperatorTimeoutAction::Reject => "reject",
                        OperatorTimeoutAction::Fail => "fail",
                        OperatorTimeoutAction::Continue => "continue",
                    };
                    let actions_json: Vec<serde_json::Value> = actions
                        .iter()
                        .map(|a| serde_json::to_value(a).unwrap_or(serde_json::Value::Null))
                        .collect();
                    let output = current_input.clone();
                    let step_result = StepResult {
                        step_name: step.name.clone(),
                        agent_id: String::new(),
                        agent_name: "_operator:operator".to_string(),
                        prompt: Self::operator_prompt_trace(
                            "operator",
                            serde_json::json!({
                                "notify": notify,
                                "actions": actions_json,
                                "timeout_secs": timeout_secs,
                                "timeout_action": timeout_action_label,
                                "input": input_trace,
                            }),
                        ),
                        output: output.clone(),
                        input_tokens: 0,
                        output_tokens: 0,
                        duration_ms: 0,
                    };
                    if let Some(mut r) = self.runs.get_mut(&run_id) {
                        r.step_results.push(step_result);
                    }
                    if let Some(ref var) = step.output_var {
                        variables.insert(var.clone(), output.clone());
                    }
                    all_outputs.push(output);

                    info!(
                        step = i + 1,
                        name = %step.name,
                        notify_count,
                        action_count,
                        timeout_secs = ?timeout_secs,
                        timeout_action = %timeout_action_label,
                        "Operator step entered — pausing run for human-in-the-loop (#4977 skeleton)"
                    );

                    // Lodge a pause request. The pause-request gate at
                    // the top of the next loop iteration atomically
                    // takes this request, snapshots state, and
                    // transitions the run to Paused with a resume
                    // token. We advance `i` past this step *before*
                    // pausing so the resume picks up at the NEXT
                    // step, treating the operator's response as the
                    // operator step's output (already recorded
                    // above).
                    let reason = format!(
                        "operator step '{}' awaiting human response ({} recipient(s), {} action(s))",
                        step.name, notify_count, action_count,
                    );
                    let token = Uuid::new_v4();
                    let hash = Self::hash_resume_token(&token);
                    if let Some(mut r) = self.runs.get_mut(&run_id) {
                        // Only lodge if no caller-driven pause is
                        // already pending — the operator step's
                        // pause is implicit and must not clobber a
                        // pre-existing one (idempotency parity with
                        // `pause_run`).
                        if r.pause_request.is_none() {
                            r.pause_request = Some(PauseRequest {
                                reason: reason.clone(),
                                resume_token_hash: hash,
                            });
                        }
                    }
                    // The plaintext token is not returned to the
                    // caller of `execute_run` today — it is recorded
                    // in the structured log + step result trace so
                    // operators can retrieve it via the dashboard /
                    // API. The follow-up issue will surface the token
                    // through a dedicated operator-step event.
                    info!(
                        run_id = %run_id,
                        step = i + 1,
                        resume_token = %token,
                        "Operator step pause token generated (capture from logs until step 2 lands)"
                    );

                    // Advance the cursor so the pause snapshot
                    // points at the *next* step. We then drive the
                    // pause snapshot inline — relying on the
                    // pause-request gate at the top of the next
                    // iteration is unsound when the operator step is
                    // the last step in the workflow: the `while i <
                    // workflow.steps.len()` condition would fail
                    // before the gate runs, and the function would
                    // fall through to the Completed transition with
                    // an orphan `pause_request` (caught by
                    // `execute_run_operator_step_pauses_with_resume_token`).
                    // Mirror the gate's atomic take-and-snapshot
                    // exactly so the two paths stay in lockstep.
                    i += 1;
                    let pending_pause = if let Some(mut run) = self.runs.get_mut(&run_id) {
                        if let Some(pause) = run.pause_request.take() {
                            run.paused_step_index = Some(i);
                            run.paused_variables = variables
                                .iter()
                                .map(|(k, v)| (k.clone(), v.clone()))
                                .collect();
                            run.paused_current_input = Some(current_input.clone());
                            run.state = WorkflowRunState::Paused {
                                resume_token_hash: pause.resume_token_hash.clone(),
                                reason: pause.reason.clone(),
                                paused_at: Utc::now(),
                            };
                            Some(pause)
                        } else {
                            None
                        }
                    } else {
                        None
                    };
                    if let Some(pause) = pending_pause {
                        // Persist immediately — same SIGKILL-safety
                        // reasoning as the loop-top gate.
                        if let Some(run) = self.runs.get(&run_id) {
                            self.upsert_run_to_store(&run);
                        }
                        info!(
                            run_id = %run_id,
                            resume_step = i,
                            reason = %pause.reason,
                            "Workflow run paused at operator step boundary"
                        );
                        return Ok(current_input);
                    }
                    // No pause was actually lodged (idempotency
                    // branch above declined because a caller-driven
                    // pause was already pending). Continue so the
                    // loop-top gate handles that pre-existing pause
                    // on the next iteration.
                    continue;
                }
            }

            i += 1;
        }

        // Mark workflow as completed. Clear the pause snapshot fields and
        // any orphan `pause_request` that may have been lodged after the
        // last step boundary check — a pause requested between the loop's
        // final iteration and this transition would otherwise survive on a
        // Completed run as dead data.
        let final_output = current_input.clone();
        if let Some(mut r) = self.runs.get_mut(&run_id) {
            r.state = WorkflowRunState::Completed;
            r.output = Some(final_output.clone());
            r.completed_at = Some(Utc::now());
            r.pause_request = None;
            r.paused_step_index = None;
            r.paused_variables.clear();
            r.paused_current_input = None;
        }

        info!(run_id = %run_id, "Workflow completed successfully");
        Ok(final_output)
    }

    /// DAG-based workflow execution.
    ///
    /// Steps are topologically sorted into layers based on `depends_on`.
    /// Steps within the same layer run concurrently; layers execute
    /// sequentially. Each step receives the workflow input plus any
    /// variables produced by its dependencies via `output_var`.
    async fn execute_run_dag<F, Fut>(
        &self,
        run_id: WorkflowRunId,
        workflow: &Workflow,
        input: &str,
        agent_resolver: &impl Fn(&StepAgent) -> Option<(AgentId, String, bool)>,
        send_message: &F,
    ) -> Result<String, String>
    where
        F: Fn(AgentId, String, Option<SessionMode>) -> Fut + Sync,
        Fut: std::future::Future<Output = Result<(String, u64, u64), String>> + Send,
    {
        // Pause/resume support is sequential-path only in #3335. If a pause
        // request was lodged before the engine routed into the DAG branch,
        // refuse cleanly rather than silently dropping the request — the
        // DAG executor would never observe `pause_request` and the run
        // would just complete normally, leaving the caller's `resume_run`
        // call hanging. The follow-up to add per-layer pause checkpoints
        // tracks against #3335.
        let dag_pause_requested = self
            .runs
            .get(&run_id)
            .and_then(|r| r.pause_request.as_ref().map(|p| p.reason.clone()));
        if let Some(reason) = dag_pause_requested {
            // Mark the run Failed and consume the lingering pause_request
            // before returning. Without this the run stays Running with
            // `pause_request: Some(_)` forever, looking like a live
            // workflow that just never executes — and `cleanup_terminal_pause_state`
            // (called by execute_run after we return) wouldn't run because
            // state isn't terminal. Set Failed so the cleanup pass picks it up.
            if let Some(mut run) = self.runs.get_mut(&run_id) {
                run.state = WorkflowRunState::Failed;
                run.error = Some(format!(
                    "DAG workflow refused to start: pause requested ({reason}) \
                     but pause/resume is supported on the sequential path only \
                     (#3335 follow-up)"
                ));
                run.completed_at = Some(Utc::now());
            }
            return Err(format!(
                "Pause requested ({reason}) but the workflow uses DAG dependencies; \
                 pause/resume is supported on the sequential path only (#3335 follow-up)"
            ));
        }

        let layers = Self::topological_sort(&workflow.steps)?;
        let mut variables: HashMap<String, String> = HashMap::new();
        // Seed per-key vars from object-shaped input JSON so that DAG step
        // prompts can substitute `{{cover}}` / `{{topic}}` etc. directly
        // from the caller's input (#4982 — gap 3 / rich invocation). The
        // sequential dispatch path performs the same seeding above.
        Self::seed_input_vars_from_json(input, &mut variables);
        // Track which step names have failed so we can skip dependents
        let mut failed_steps: std::collections::HashSet<String> = std::collections::HashSet::new();
        let mut last_output = input.to_string();

        info!(
            run_id = %run_id,
            layers = layers.len(),
            "Executing workflow in DAG mode"
        );

        for (layer_idx, layer) in layers.iter().enumerate() {
            debug!(
                layer = layer_idx + 1,
                steps = layer.len(),
                "Executing DAG layer"
            );

            if layer.len() == 1 {
                // Single step in layer — execute directly (no concurrency overhead)
                let step_idx = layer[0];
                let step = &workflow.steps[step_idx];

                // Check if any dependency failed
                let dep_failed = step.depends_on.iter().any(|dep| failed_steps.contains(dep));

                if dep_failed {
                    match step.error_mode {
                        ErrorMode::Fail => {
                            let error_msg =
                                format!("Step '{}' skipped: dependency failed", step.name);
                            if let Some(mut r) = self.runs.get_mut(&run_id) {
                                r.state = WorkflowRunState::Failed;
                                r.error = Some(error_msg.clone());
                                r.completed_at = Some(Utc::now());
                            }
                            return Err(error_msg);
                        }
                        _ => {
                            warn!(
                                step = %step.name,
                                "Skipping step due to failed dependency"
                            );
                            failed_steps.insert(step.name.clone());
                            continue;