oxdock_core/exec/
steps.rs

1use std::collections::HashMap;
2use std::process::ExitStatus;
3use std::sync::Arc;
4use std::sync::Mutex;
5use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
6
7use anyhow::{Result, bail};
8use oxdock_fs::GuardedPath;
9use oxdock_parser::{Arg, AssertTarget, Step, StepKind, Value, guard_option_allows};
10use oxdock_process::{BackgroundHandle, CommandStdin, ProcessManager, SharedInput, SharedOutput};
11
12/// Create an ExitStatus from a raw exit code. Cross-platform.
13fn exit_status_from_code(code: i32) -> ExitStatus {
14    #[cfg(unix)]
15    {
16        use std::os::unix::process::ExitStatusExt;
17        ExitStatus::from_raw(code << 8)
18    }
19    #[cfg(windows)]
20    {
21        use std::os::windows::process::ExitStatusExt;
22        ExitStatus::from_raw(code as u32)
23    }
24}
25
26use super::handlers;
27use super::io::{ExactCapture, SlidingWindow, StreamHandle};
28use super::state::{ExecState, TaskEntry, TaskPhase};
29use oxdock_pipe::PipeInner;
30
31/// A background handle wrapping a `std::thread::JoinHandle` for ASYNC blocks
32/// that execute commands in a background thread.
33pub(super) struct ThreadJoinHandle {
34    join: Option<std::thread::JoinHandle<Result<()>>>,
35    cancel_token: Arc<AtomicBool>,
36    active_process: Arc<Mutex<Option<Box<dyn BackgroundHandle>>>>,
37    /// Identity of the worker thread, published by the child on entry.
38    /// Forked worker state shares the parent task registry via `Arc`, so a
39    /// parent that errors out can drop its registry reference while the
40    /// worker is still alive. The worker then becomes the last registry
41    /// owner and would drop (and join) its own handle on thread exit, which
42    /// is undefined behavior (`pthread_join` on self). Detect that case and
43    /// detach instead of joining.
44    worker: Arc<Mutex<Option<std::thread::ThreadId>>>,
45    /// Preserved error from the child thread, if any.
46    thread_error: Option<anyhow::Error>,
47}
48
49impl ThreadJoinHandle {
50    pub(super) fn new(
51        join: std::thread::JoinHandle<Result<()>>,
52        cancel_token: Arc<AtomicBool>,
53        active_process: Arc<Mutex<Option<Box<dyn BackgroundHandle>>>>,
54        worker: Arc<Mutex<Option<std::thread::ThreadId>>>,
55    ) -> Self {
56        Self {
57            join: Some(join),
58            cancel_token,
59            active_process,
60            worker,
61            thread_error: None,
62        }
63    }
64
65    /// Whether the caller is the worker thread owned by this handle.
66    fn is_self(&self) -> bool {
67        let guard = self.worker.lock().unwrap_or_else(|e| e.into_inner());
68        guard.is_some_and(|id| id == std::thread::current().id())
69    }
70
71    /// Reap the thread if finished, preserving any error.
72    fn reap(&mut self) {
73        if self.join.is_none() {
74            return;
75        }
76        if self.is_self() {
77            // The worker is dropping the last registry reference on its own
78            // exit path (parent already tore down or errored out). Detach
79            // instead of joining self, which is undefined behavior.
80            let _ = self.join.take();
81            return;
82        }
83        let handle = self.join.take().unwrap();
84        match handle.join() {
85            Ok(Ok(())) => {}
86            Ok(Err(e)) => {
87                self.thread_error = Some(e);
88            }
89            Err(panic) => {
90                let msg = if let Some(s) = panic.downcast_ref::<&str>() {
91                    s.to_string()
92                } else if let Some(s) = panic.downcast_ref::<String>() {
93                    s.clone()
94                } else {
95                    "thread panicked".to_string()
96                };
97                self.thread_error = Some(anyhow::anyhow!("{msg}"));
98            }
99        }
100    }
101}
102
103impl BackgroundHandle for ThreadJoinHandle {
104    fn try_wait(&mut self) -> Result<Option<ExitStatus>> {
105        if let Some(join) = &self.join {
106            if join.is_finished() {
107                self.reap();
108            } else {
109                return Ok(None);
110            }
111        }
112        // anyhow::Error is not Clone and this method may run repeatedly,
113        // so the preserved error is re-emitted rather than moved. The
114        // alternate display (`{err:#}`) flattens the full causal chain
115        // into the new message: `{err}` alone would drop every
116        // `Caused by` layer at the ASYNC task boundary.
117        if let Some(ref err) = self.thread_error {
118            Err(anyhow::anyhow!("{err:#}"))
119        } else {
120            Ok(Some(exit_status_from_code(0)))
121        }
122    }
123
124    fn kill(&mut self) -> Result<()> {
125        // Signal cancellation
126        self.cancel_token.store(true, Ordering::SeqCst);
127        // Kill any active OS process to interrupt blocking wait
128        if let Ok(mut guard) = self.active_process.lock()
129            && let Some(ref mut proc) = *guard
130        {
131            let _ = proc.kill();
132        }
133        // Join the thread to ensure it completes before returning
134        self.reap();
135        Ok(())
136    }
137
138    fn wait(&mut self) -> Result<ExitStatus> {
139        self.reap();
140        // Same chain-preserving re-emit as `try_wait` above.
141        if let Some(ref err) = self.thread_error {
142            Err(anyhow::anyhow!("{err:#}"))
143        } else {
144            Ok(exit_status_from_code(0))
145        }
146    }
147}
148
149impl Drop for ThreadJoinHandle {
150    fn drop(&mut self) {
151        let _ = self.kill();
152    }
153}
154
155/// Monotonically increasing generation counter for assert_windows key scoping.
156/// Each execute_steps invocation gets a unique generation, preventing key
157/// collisions between nested scopes (for_loop bodies, WithIo blocks).
158static ASSERT_GENERATION: AtomicUsize = AtomicUsize::new(0);
159
160/// Intra-thread control-flow signal (`BREAK`/`CONTINUE`/`RETURN`).
161/// Produced by steps, consumed by the nearest loop (`Break`/`Continue`) or
162/// `call_func` (`Return`). Anything reaching a thread boundary (`ASYNC`
163/// spawn, `await` reaping) or the pipeline top becomes a step-numbered
164/// error. `idx` is the 0-based index of the originating step in its own
165/// body, so boundary errors can name it.
166#[derive(Debug)]
167pub(super) enum Flow {
168    Done,
169    Break { idx: usize },
170    Continue { idx: usize },
171    Return { idx: usize, value: Value },
172}
173
174pub(super) fn allocate_assert_generation() -> usize {
175    ASSERT_GENERATION.fetch_add(1, Ordering::Relaxed)
176}
177
178/// Which stream a stream-targeted assertion observes.
179#[derive(Clone, Copy, PartialEq, Eq)]
180pub(super) enum AssertStream {
181    Stdout,
182    Stderr,
183}
184
185/// Extract the substring needle from a step asserting over a stream:
186/// `ASSERT_CONTAINS stdout|stderr`, handling both top-level and
187/// WITH_IO-wrapped variants. Returns the observed stream and the needle.
188fn extract_stream_needle(kind: &StepKind) -> Option<(AssertStream, &Arg)> {
189    let step = match kind {
190        StepKind::WithIo { cmd, .. } => cmd.as_ref(),
191        other => other,
192    };
193    match step {
194        StepKind::AssertContains { haystack, needle } => match haystack {
195            AssertTarget::Stdout => Some((AssertStream::Stdout, needle)),
196            AssertTarget::Stderr => Some((AssertStream::Stderr, needle)),
197            _ => None,
198        },
199        _ => None,
200    }
201}
202
203/// Whether the step needs the exact-match stdout accumulator:
204/// `ASSERT_EQ stdout`, top-level or WITH_IO-wrapped.
205fn needs_exact_stdout(kind: &StepKind) -> bool {
206    let step = match kind {
207        StepKind::WithIo { cmd, .. } => cmd.as_ref(),
208        other => other,
209    };
210    matches!(
211        step,
212        StepKind::AssertEq {
213            actual: AssertTarget::Stdout,
214            ..
215        }
216    )
217}
218
219/// An assertion first argument evaluated far enough to check:
220/// values stay typed, streams stay references to live buffers.
221pub(super) enum ResolvedAssertTarget {
222    Value(Value),
223    Stdout,
224    Stderr,
225    Pipe(Vec<u8>),
226}
227
228/// Evaluate an assertion target. `Arg::Expr` evaluates typed;
229/// strings, templates, and parts render to `String`; stream markers
230/// resolve to live buffers (peeked, never consumed). A `$var` holding a
231/// `PIPE` likewise peeks its backend bytes: lowering cannot know variable
232/// types, so the pipe dispatch lives here where the value exists.
233pub(super) fn resolve_assert_target<P: ProcessManager>(
234    target: &AssertTarget,
235    cx: &mut StepCtx<'_, P>,
236) -> Result<ResolvedAssertTarget> {
237    match target {
238        AssertTarget::Value(arg) => {
239            let value = super::args::evaluate_assert_operand(arg, cx)?;
240            if let Some(handle) = value.as_pipe_handle() {
241                let bytes =
242                    cx.state.io.peek_pipe_content(&handle).map_err(|e| {
243                        anyhow::anyhow!("step pipe assertion cannot read pipe: {e}")
244                    })?;
245                return Ok(ResolvedAssertTarget::Pipe(bytes));
246            }
247            Ok(ResolvedAssertTarget::Value(value))
248        }
249        AssertTarget::Stdout => Ok(ResolvedAssertTarget::Stdout),
250        AssertTarget::Stderr => Ok(ResolvedAssertTarget::Stderr),
251    }
252}
253
254/// Pre-register stream assertion observers so tees feed them data before
255/// the steps execute. Substring needles (`ASSERT_CONTAINS stdout|stderr`)
256/// get per-step `SlidingWindow`s; `ASSERT_EQ stdout` allocates the
257/// generation's exact accumulator. Uses `args::resolve_arg_state` for
258/// actual template expansion. Handles both top-level and WITH_IO-wrapped
259/// assertions via `extract_stream_needle` / `needs_exact_stdout`.
260pub(super) fn pre_register_assertions<P: ProcessManager>(
261    state: &mut ExecState<P>,
262    steps: &[Step],
263    generation: usize,
264) -> Result<()> {
265    let mut windows = match state.assert_windows.lock() {
266        Ok(guard) => guard,
267        Err(_) => bail!("assert_windows poisoned"),
268    };
269    let mut stderr_windows = match state.assert_windows_stderr.lock() {
270        Ok(guard) => guard,
271        Err(_) => bail!("assert_windows_stderr poisoned"),
272    };
273    let mut exact = match state.exact_stdout.lock() {
274        Ok(guard) => guard,
275        Err(_) => bail!("exact_stdout poisoned"),
276    };
277    for (idx, step) in steps.iter().enumerate() {
278        if let Some((stream, arg)) = extract_stream_needle(&step.kind) {
279            let resolved = super::args::resolve_arg_state(arg, state)?;
280            let map = match stream {
281                AssertStream::Stdout => &mut windows,
282                AssertStream::Stderr => &mut stderr_windows,
283            };
284            map.insert((generation, idx), SlidingWindow::new(resolved.into_bytes()));
285        }
286        if needs_exact_stdout(&step.kind) {
287            exact.entry(generation).or_insert_with(ExactCapture::new);
288        }
289    }
290    Ok(())
291}
292
293/// After an environment mutation (ENV or INHERIT_ENV), re-expand all
294/// substring assertion needles for the current generation to reflect new
295/// env values. Preserves ring buffer history via `update_needle`. Handles
296/// both top-level and WITH_IO-wrapped assertions. Exact accumulators hold
297/// no needle and need no sync.
298#[allow(clippy::collapsible_if)]
299pub(super) fn sync_iteration_assert_needles<P: ProcessManager>(
300    state: &ExecState<P>,
301    steps: &[Step],
302    generation: usize,
303) -> Result<()> {
304    let mut windows = match state.assert_windows.lock() {
305        Ok(guard) => guard,
306        Err(_) => bail!("assert_windows poisoned"),
307    };
308    let mut stderr_windows = match state.assert_windows_stderr.lock() {
309        Ok(guard) => guard,
310        Err(_) => bail!("assert_windows_stderr poisoned"),
311    };
312    for (idx, step) in steps.iter().enumerate() {
313        if let Some((stream, arg)) = extract_stream_needle(&step.kind) {
314            let map = match stream {
315                AssertStream::Stdout => &mut windows,
316                AssertStream::Stderr => &mut stderr_windows,
317            };
318            if let Some(w) = map.get_mut(&(generation, idx)) {
319                let resolved = super::args::resolve_arg_state(arg, state)?;
320                w.update_needle(resolved.into_bytes());
321            }
322        }
323    }
324    Ok(())
325}
326
327/// Per-step execution context handed to every command handler.
328///
329/// Host-registered functions receive this context: read script state through
330/// the public accessors (`get_var`, `get_env`, `cwd`) and return a `Value`.
331/// The fields stay crate-private so execution invariants hold for hosts.
332///
333/// Output contract (load-bearing for `LET`-capture, pipes, and stream assertions):
334/// handlers must emit stdout/stderr ONLY through `out`/`err` — via
335/// `write_stdout` or `StreamHandle::to_stdout`/`to_stderr` — and never write
336/// to host stdout directly. The step runner swaps these handles per context:
337/// `LET $x: STRING = <command>` installs a spillable capture sink, `WITH_IO`
338/// installs named-pipe endpoints, and the root installs the assertion tee. A handler that bypasses its context handles silently breaks all three.
339pub struct StepCtx<'a, P: ProcessManager> {
340    pub(super) state: &'a mut ExecState<P>,
341    pub(super) process: &'a mut P,
342    pub(super) stdin: CommandStdin,
343    pub(super) expose_stdin: bool,
344    pub(super) out: Option<StreamHandle>,
345    pub(super) err: Option<StreamHandle>,
346    /// Pipe backend backing `out`, when a `WITH_IO` stdout binding resolved
347    /// to a script pipe. Uniform context enrichment (populated for every
348    /// command, read only by consumers that need the backend, like the
349    /// network bridge). `None` for inherited, captured, and tee outputs,
350    /// and for OS pairs (kernel bytes are invisible).
351    pub(super) out_pipe: Option<Arc<PipeInner>>,
352    /// Pipe backend backing `stdin`, when a `WITH_IO` stdin binding
353    /// resolved to a script pipe. Lets bridge workers run timeout-bounded
354    /// reads without touching shared pipe semantics (`None` for OS pairs,
355    /// which fall back to blocking reads).
356    pub(super) stdin_pipe: Option<Arc<PipeInner>>,
357}
358
359impl<'a, P: ProcessManager> StepCtx<'a, P> {
360    /// Look up a script variable by name (innermost scope first).
361    pub fn get_var(&self, key: &str) -> Option<Value> {
362        self.state.get_var(key)
363    }
364
365    /// Look up an environment variable visible to the script.
366    pub fn get_env(&self, key: &str) -> Option<String> {
367        self.state.envs.get(key).cloned()
368    }
369
370    /// Snapshot of the script-visible environment: `ENV` assignments
371    /// layered over inherited entries, as currently scoped. Hosts staging
372    /// child processes layer this over the host environment (the same
373    /// contract `RUN` honors through `CommandContext`), so block-scoped
374    /// `ENV` reaches the child and reverts at scope exit with no extra
375    /// machinery.
376    pub fn env_snapshot(&self) -> HashMap<String, String> {
377        self.state.envs.as_ref().clone()
378    }
379
380    /// Current working directory (guarded; stays inside the workspace).
381    pub fn cwd(&self) -> &GuardedPath {
382        &self.state.cwd
383    }
384
385    /// Mint a fresh unbound pipe handle, like bare `LET $p: PIPE`. The
386    /// backend materializes lazily on first binding; return it from a
387    /// host function to hand the DSL a pipe it can bind. Tagged with the
388    /// current task so promotion checks see the declaration origin.
389    pub fn new_pipe(&self) -> Value {
390        Value::pipe_fresh_in_task(self.state.task_id)
391    }
392
393    /// Borrow the read half of a `PIPE` value for byte streaming (see
394    /// [`PipeStream`](super::PipeStream)). Unbound handles materialize as script pipes —
395    /// hosts cannot spawn `RUN`, so script is the only sensible kind,
396    /// and a later `RUN` binding adapts through the shared path. DSL,
397    /// bridge, and host bindings on an OS-materialized handle resolve
398    /// through the single-take bridge: the first call takes, repeats bail
399    /// loudly (same contract as DSL consumers; use script-backed pipes
400    /// for repeat or multi access).
401    pub fn pipe_reader(&self, value: &Value) -> Result<SharedInput> {
402        use oxdock_pipe::{Materialized, materialize};
403        let Some(handle) = value.as_pipe_handle() else {
404            anyhow::bail!(
405                "host pipe_reader needs a PIPE value, got {}",
406                value.type_name()
407            );
408        };
409        match materialize(&handle, false)? {
410            Materialized::Script(backend) => Ok(backend.reader_handle()),
411            #[cfg(not(miri))]
412            Materialized::Os(entry) => {
413                let owned = entry.reader.take().map_err(|_| {
414                    anyhow::anyhow!(
415                        "OS pipe handle has already been consumed by another binding; declare a fresh LET $x: PIPE for a new session"
416                    )
417                })?;
418                Ok(Arc::new(Mutex::new(owned)))
419            }
420        }
421    }
422
423    /// Borrow the write half of a `PIPE` value for byte streaming (see
424    /// [`PipeStream`](super::PipeStream)). Same materialization and take-once contract as
425    /// [`StepCtx::pipe_reader`].
426    pub fn pipe_writer(&self, value: &Value) -> Result<SharedOutput> {
427        use oxdock_pipe::{Materialized, materialize};
428        let Some(handle) = value.as_pipe_handle() else {
429            anyhow::bail!(
430                "host pipe_writer needs a PIPE value, got {}",
431                value.type_name()
432            );
433        };
434        match materialize(&handle, false)? {
435            Materialized::Script(backend) => Ok(backend.writer_handle()),
436            #[cfg(not(miri))]
437            Materialized::Os(entry) => {
438                let owned = entry.writer.take().map_err(|_| {
439                    anyhow::anyhow!(
440                        "OS pipe handle has already been consumed by another binding; declare a fresh LET $x: PIPE for a new session"
441                    )
442                })?;
443                Ok(Arc::new(Mutex::new(owned)))
444            }
445        }
446    }
447
448    /// Explicitly close a script pipe: readers drain buffered bytes, then
449    /// observe EOF regardless of live writers or keeper pins. Unbound
450    /// handles bail (closing a never-bound pipe is a caller bug), and
451    /// OS-materialized handles bail (kernel pairs close by dropping their
452    /// taken halves — drop the value instead).
453    pub fn close_pipe(&self, value: &Value) -> Result<()> {
454        let Some(handle) = value.as_pipe_handle() else {
455            anyhow::bail!(
456                "host close_pipe needs a PIPE value, got {}",
457                value.type_name()
458            );
459        };
460        let Some(backend) = oxdock_pipe::script_backend(&handle) else {
461            anyhow::bail!(
462                "host close_pipe needs a script-materialized pipe (unbound and OS handles cannot be force-closed)"
463            );
464        };
465        backend.force_close();
466        Ok(())
467    }
468
469    /// Whether the current task was cancelled (`CANCEL`/`TIMEOUT`). For
470    /// external host modules running blocking pumps: poll each tick so
471    /// silent-but-open pipes cannot strand the task thread.
472    pub fn is_cancelled(&self) -> bool {
473        self.state
474            .cancel_token
475            .load(std::sync::atomic::Ordering::SeqCst)
476    }
477
478    /// Whether this step runs on an `ASYNC` task thread. Blocking pumps
479    /// must refuse the main sequential flow.
480    pub fn is_async_task(&self) -> bool {
481        self.state.inside_async
482    }
483
484    /// Resolve an explicitly passed `PIPE` value to its script backend for
485    /// timeout-bounded reads (`read_into_timeout`). This is value-based on
486    /// purpose: the ambient `out_pipe`/`stdin_pipe` fields only populate via
487    /// engine-level `WITH_IO` resolution, which never runs for host function
488    /// calls. Returns `None` for unbound and OS-materialized handles, which
489    /// fall back to blocking reads.
490    pub fn pipe_backend(&self, value: &Value) -> Option<Arc<PipeInner>> {
491        let handle = value.as_pipe_handle()?;
492        oxdock_pipe::script_backend(&handle)
493    }
494}
495
496#[allow(clippy::too_many_arguments)]
497pub(super) fn execute_steps<P: ProcessManager>(
498    state: &mut ExecState<P>,
499    process: &mut P,
500    steps: &[Step],
501    stdin: CommandStdin,
502    expose_stdin: bool,
503    out: Option<StreamHandle>,
504    err: Option<StreamHandle>,
505    wait_at_end: bool,
506) -> Result<Flow> {
507    let generation = allocate_assert_generation();
508    let flow = match execute_steps_inner(
509        state,
510        process,
511        generation,
512        steps,
513        stdin,
514        expose_stdin,
515        out,
516        err,
517        wait_at_end,
518    ) {
519        Ok(flow) => flow,
520        Err(e) => {
521            // A step failed before end-of-pipeline reaping ran. Join
522            // background work now so the parent owns teardown: otherwise the
523            // parent drops its task-registry reference while a worker still
524            // lives, leaving the worker as the last registry owner to drop
525            // (and join) its own handle on thread exit.
526            teardown_tasks_on_error(state);
527            cleanup_assertion_generation(state, generation)?;
528            return Err(e);
529        }
530    };
531    // Cleanup: remove all assertion state for this generation
532    cleanup_assertion_generation(state, generation)?;
533    Ok(flow)
534}
535
536/// Remove per-generation assertion observers. Runs on success and on step
537/// failure so a failed pipeline never leaks windows into later runs.
538fn cleanup_assertion_generation<P: ProcessManager>(
539    state: &mut ExecState<P>,
540    generation: usize,
541) -> Result<()> {
542    let mut windows = match state.assert_windows.lock() {
543        Ok(guard) => guard,
544        Err(_) => bail!("assert_windows poisoned"),
545    };
546    windows.retain(|(g, _), _| *g != generation);
547    let mut stderr_windows = match state.assert_windows_stderr.lock() {
548        Ok(guard) => guard,
549        Err(_) => bail!("assert_windows_stderr poisoned"),
550    };
551    stderr_windows.retain(|(g, _), _| *g != generation);
552    let mut exact = match state.exact_stdout.lock() {
553        Ok(guard) => guard,
554        Err(_) => bail!("exact_stdout poisoned"),
555    };
556    exact.retain(|g, _| *g != generation);
557    Ok(())
558}
559
560/// Join background work after a step failure, mirroring the end-of-pipeline
561/// fail-fast teardown. Anonymous handles always belong to the current
562/// thread. Named entries are root-owned: worker threads must never block on
563/// sibling tasks, which may depend on the worker via AWAIT.
564fn teardown_tasks_on_error<P: ProcessManager>(state: &mut ExecState<P>) {
565    for survivor in state.bg_children.iter_mut() {
566        let _ = survivor.kill();
567    }
568    state.bg_children.clear();
569    if state.inside_async {
570        return;
571    }
572    let entries: Vec<Arc<TaskEntry>> = {
573        let named = state.named_tasks.lock().unwrap_or_else(|e| e.into_inner());
574        named.values().cloned().collect()
575    };
576    let mut to_kill: Vec<(Arc<TaskEntry>, Box<dyn BackgroundHandle>)> = Vec::new();
577    for entry in &entries {
578        let mut guard = entry.state.lock().unwrap_or_else(|e| e.into_inner());
579        match guard.phase {
580            TaskPhase::Running | TaskPhase::Awaiting => {
581                guard.phase = TaskPhase::Cancelled;
582                if let Some(handle) = guard.handle.take() {
583                    to_kill.push((Arc::clone(entry), handle));
584                }
585            }
586            TaskPhase::Cancelled | TaskPhase::Completed => {}
587        }
588    }
589    for (entry, mut handle) in to_kill {
590        let _ = handle.kill();
591        entry.finish_teardown();
592    }
593}
594
595/// Execute a single step with an explicit generation and index.
596/// Used by `with_io` to preserve the parent step's index for assertion window keys.
597#[allow(clippy::too_many_arguments)]
598pub(super) fn execute_single_step_with_generation<P: ProcessManager>(
599    state: &mut ExecState<P>,
600    process: &mut P,
601    cmd: &StepKind,
602    generation: usize,
603    idx: usize,
604    stdin: CommandStdin,
605    expose_stdin: bool,
606    out: Option<StreamHandle>,
607    err: Option<StreamHandle>,
608    out_pipe: Option<Arc<PipeInner>>,
609    stdin_pipe: Option<Arc<PipeInner>>,
610) -> Result<Flow> {
611    let mut cx = StepCtx {
612        state,
613        process,
614        stdin,
615        expose_stdin,
616        out,
617        err,
618        out_pipe,
619        stdin_pipe,
620    };
621    // Compound steps (loops, functions, scoped wrappers) participate in
622    // Flow and dispatch through the Flow path; every other variant runs
623    // the leaf pipeline below and yields Done.
624    match cmd {
625        StepKind::FuncDef { .. }
626        | StepKind::Call { .. }
627        | StepKind::Return { .. }
628        | StepKind::While { .. }
629        | StepKind::Break
630        | StepKind::Continue
631        | StepKind::For { .. }
632        | StepKind::If { .. }
633        | StepKind::Timeout { .. }
634        | StepKind::WithIo { .. }
635        | StepKind::AssignCapture { .. } => {
636            return dispatch_flow_step(cmd, &mut cx, generation, idx);
637        }
638        _ => {}
639    }
640    match cmd {
641        StepKind::Run(arg) => {
642            let cmd = super::args::resolve_arg(arg, &mut cx)?;
643            let cmd = super::args::expand_dsl_vars(&cmd, cx.state);
644            handlers::run(&mut cx, idx, &cmd)
645        }
646        StepKind::RunExec { argv } => {
647            let resolved = handlers::resolve_run_exec_argv(argv, &mut cx)?;
648            handlers::run_argv(&mut cx, idx, &resolved)
649        }
650        StepKind::Echo(arg) => {
651            let msg = super::args::resolve_arg(arg, &mut cx)?;
652            handlers::echo(&mut cx, &msg)
653        }
654        StepKind::AsyncBlock { .. } => handlers::dispatch_async_block(cmd, &mut cx),
655        StepKind::Workdir(arg) => {
656            let path = super::args::resolve_arg(arg, &mut cx)?;
657            handlers::workdir(&mut cx, idx, &path)
658        }
659        StepKind::Workspace(target) => handlers::workspace(&mut cx, target),
660        StepKind::Env { key, value } => {
661            let resolved = super::args::resolve_arg(value, &mut cx)?;
662            handlers::env(&mut cx, key, &resolved)
663        }
664        StepKind::InheritEnv { keys } => {
665            handlers::inherit_env(&mut cx, keys)?;
666            sync_iteration_assert_needles(
667                cx.state,
668                &[Step {
669                    guard: None,
670                    kind: cmd.clone(),
671                    scope_enter: 0,
672                    scope_exit: 0,
673                }],
674                generation,
675            )?;
676            Ok(())
677        }
678        StepKind::Copy {
679            from_workspace,
680            from,
681            to,
682        } => {
683            let from_resolved = super::args::resolve_arg(from, &mut cx)?;
684            let to_resolved = super::args::resolve_arg(to, &mut cx)?;
685            handlers::copy(
686                &mut cx,
687                idx,
688                from_workspace.clone(),
689                &from_resolved,
690                &to_resolved,
691            )
692        }
693        StepKind::CopyGit {
694            rev,
695            from,
696            to,
697            include_dirty,
698        } => {
699            let rev_resolved = super::args::resolve_arg(rev, &mut cx)?;
700            let from_resolved = super::args::resolve_arg(from, &mut cx)?;
701            let to_resolved = super::args::resolve_arg(to, &mut cx)?;
702            handlers::copy_git(
703                &mut cx,
704                idx,
705                &rev_resolved,
706                &from_resolved,
707                &to_resolved,
708                *include_dirty,
709            )
710        }
711        StepKind::HashSha256 { path } => {
712            let path_resolved = super::args::resolve_arg(path, &mut cx)?;
713            handlers::hash_sha256(&mut cx, idx, &path_resolved)
714        }
715        StepKind::Symlink {
716            from_workspace,
717            from,
718            to,
719        } => {
720            let from_resolved = super::args::resolve_arg(from, &mut cx)?;
721            let to_resolved = super::args::resolve_arg(to, &mut cx)?;
722            handlers::symlink(
723                &mut cx,
724                idx,
725                from_workspace.clone(),
726                &from_resolved,
727                &to_resolved,
728            )
729        }
730        StepKind::Mkdir(arg) => {
731            let path = super::args::resolve_arg(arg, &mut cx)?;
732            handlers::mkdir(&mut cx, idx, &path)
733        }
734        StepKind::Ls(arg) => {
735            let resolved = super::args::resolve_arg_opt(arg, &mut cx)?;
736            handlers::ls(&mut cx, idx, &resolved)
737        }
738        StepKind::Cwd => handlers::cwd(&mut cx, idx),
739        StepKind::Read(arg) => {
740            let resolved = super::args::resolve_arg_opt(arg, &mut cx)?;
741            handlers::read(&mut cx, idx, &resolved)
742        }
743        StepKind::ReadLine { var } => handlers::read_line(&mut cx, idx, var),
744        StepKind::ListAppend { list, item } => {
745            let value = super::args::evaluate_assert_operand(item, &mut cx)?;
746            handlers::push_into(&mut cx, idx, list, value)
747        }
748        StepKind::Write { path, contents } => {
749            let path_resolved = super::args::resolve_arg(path, &mut cx)?;
750            let contents_resolved = super::args::resolve_arg_opt(contents, &mut cx)?;
751            handlers::write(&mut cx, idx, &path_resolved, contents_resolved.as_deref())
752        }
753        StepKind::Append { path, contents } => {
754            let path_resolved = super::args::resolve_arg(path, &mut cx)?;
755            let contents_resolved = super::args::resolve_arg_opt(contents, &mut cx)?;
756            handlers::append(&mut cx, idx, &path_resolved, contents_resolved.as_deref())
757        }
758        StepKind::Expand { path, overrides } => {
759            let path_resolved = super::args::resolve_arg_opt(path, &mut cx)?;
760            let overrides_resolved = super::args::resolve_overrides(overrides, &mut cx)?;
761            handlers::replace(&mut cx, idx, &path_resolved, &overrides_resolved)
762        }
763        StepKind::AssertEq {
764            hash,
765            actual,
766            expected,
767        } => {
768            let target = resolve_assert_target(actual, &mut cx)?;
769            let expected_resolved = match expected {
770                Some(e) => Some(super::args::evaluate_assert_operand(e, &mut cx)?),
771                None => None,
772            };
773            handlers::assert_eq(
774                &mut cx,
775                idx,
776                generation,
777                idx,
778                hash,
779                &target,
780                expected_resolved.as_ref(),
781            )
782        }
783        StepKind::AssertContains { haystack, needle } => {
784            let target = resolve_assert_target(haystack, &mut cx)?;
785            handlers::assert_contains(&mut cx, idx, generation, idx, &target, needle)
786        }
787        StepKind::WithIoBlock { .. } => {
788            bail!("WITH_IO block should have been expanded during parsing")
789        }
790        StepKind::Exit(code) => {
791            let code = super::args::resolve_arg_as_int(code, &mut cx)?;
792            handlers::exit(&mut cx, code)
793        }
794        StepKind::Assign {
795            var,
796            decl_type,
797            expr,
798        } => handlers::assign(&mut cx, var, decl_type.clone(), expr),
799        StepKind::Set { var, expr } => handlers::set_var_value(&mut cx, var, expr),
800        StepKind::AssignAsync {
801            var,
802            decl_type,
803            body,
804        } => handlers::dispatch_assign_async(var, decl_type.clone(), body, &mut cx),
805        StepKind::Await { var } => handlers::dispatch_await(var, &mut cx),
806        StepKind::AwaitCapture {
807            out_var,
808            out_type,
809            task_var,
810        } => handlers::dispatch_await_capture(out_var, out_type.clone(), task_var, &mut cx),
811        StepKind::Cancel { var } => handlers::dispatch_cancel(var, &mut cx),
812        StepKind::Sleep { duration } => {
813            let duration = super::args::resolve_arg_as_duration(duration, &mut cx)?;
814            handlers::sleep(&mut cx, idx, &duration)
815        }
816        StepKind::FuncDef { .. }
817        | StepKind::Call { .. }
818        | StepKind::Return { .. }
819        | StepKind::While { .. }
820        | StepKind::Break
821        | StepKind::Continue
822        | StepKind::For { .. }
823        | StepKind::If { .. }
824        | StepKind::Timeout { .. }
825        | StepKind::WithIo { .. }
826        | StepKind::AssignCapture { .. } => {
827            unreachable!("compound steps dispatch before this match")
828        }
829    }?;
830    Ok(Flow::Done)
831}
832
833#[allow(clippy::too_many_arguments)]
834fn execute_steps_inner<P: ProcessManager>(
835    state: &mut ExecState<P>,
836    process: &mut P,
837    generation: usize,
838    steps: &[Step],
839    stdin: CommandStdin,
840    expose_stdin: bool,
841    out: Option<StreamHandle>,
842    err: Option<StreamHandle>,
843    wait_at_end: bool,
844) -> Result<Flow> {
845    // Pre-register assertion windows for this generation
846    pre_register_assertions(state, steps, generation)?;
847
848    for (idx, step) in steps.iter().enumerate() {
849        // Check for cancellation before each step
850        if state.cancel_token.load(Ordering::SeqCst) {
851            bail!("ASYNC task cancelled");
852        }
853        if step.scope_enter > 0 {
854            for _ in 0..step.scope_enter {
855                state.push_scope();
856            }
857        }
858
859        let should_run = guard_option_allows(step.guard.as_ref(), &state.envs);
860        let flow_result: Result<Flow> = if !should_run {
861            Ok(Flow::Done)
862        } else {
863            let mut cx = StepCtx {
864                state,
865                process,
866                stdin: stdin.clone(),
867                expose_stdin,
868                out: out.clone(),
869                err: err.clone(),
870                out_pipe: None,
871                stdin_pipe: None,
872            };
873            // Function/loop control steps dispatch through the Flow path;
874            // every other variant runs the leaf pipeline and yields Done.
875            let flow_result: Result<Flow> = match &step.kind {
876                StepKind::FuncDef { .. }
877                | StepKind::Call { .. }
878                | StepKind::Return { .. }
879                | StepKind::While { .. }
880                | StepKind::Break
881                | StepKind::Continue
882                | StepKind::For { .. }
883                | StepKind::If { .. }
884                | StepKind::Timeout { .. }
885                | StepKind::WithIo { .. }
886                | StepKind::AssignCapture { .. } => {
887                    dispatch_flow_step(&step.kind, &mut cx, generation, idx)
888                }
889                _ => {
890                    match &step.kind {
891                        StepKind::InheritEnv { keys } => {
892                            handlers::inherit_env(&mut cx, keys)?;
893                            sync_iteration_assert_needles(cx.state, steps, generation)?;
894                            Ok(())
895                        }
896                        StepKind::Workdir(arg) => {
897                            let path = super::args::resolve_arg(arg, &mut cx)?;
898                            handlers::workdir(&mut cx, idx, &path)
899                        }
900                        StepKind::Workspace(target) => handlers::workspace(&mut cx, target),
901                        StepKind::Env { key, value } => {
902                            let resolved = super::args::resolve_arg(value, &mut cx)?;
903                            handlers::env(&mut cx, key, &resolved)?;
904                            sync_iteration_assert_needles(cx.state, steps, generation)?;
905                            Ok(())
906                        }
907                        StepKind::Run(arg) => {
908                            let cmd = super::args::resolve_arg(arg, &mut cx)?;
909                            let cmd = super::args::expand_dsl_vars(&cmd, cx.state);
910                            handlers::run(&mut cx, idx, &cmd)
911                        }
912                        StepKind::RunExec { argv } => {
913                            let resolved = handlers::resolve_run_exec_argv(argv, &mut cx)?;
914                            handlers::run_argv(&mut cx, idx, &resolved)
915                        }
916                        StepKind::Echo(arg) => {
917                            let msg = super::args::resolve_arg(arg, &mut cx)?;
918                            handlers::echo(&mut cx, &msg)
919                        }
920                        StepKind::AsyncBlock { .. } => {
921                            handlers::dispatch_async_block(&step.kind, &mut cx)
922                        }
923                        StepKind::Copy {
924                            from_workspace,
925                            from,
926                            to,
927                        } => {
928                            let from_resolved = super::args::resolve_arg(from, &mut cx)?;
929                            let to_resolved = super::args::resolve_arg(to, &mut cx)?;
930                            handlers::copy(
931                                &mut cx,
932                                idx,
933                                from_workspace.clone(),
934                                &from_resolved,
935                                &to_resolved,
936                            )
937                        }
938                        StepKind::CopyGit {
939                            rev,
940                            from,
941                            to,
942                            include_dirty,
943                        } => {
944                            let rev_resolved = super::args::resolve_arg(rev, &mut cx)?;
945                            let from_resolved = super::args::resolve_arg(from, &mut cx)?;
946                            let to_resolved = super::args::resolve_arg(to, &mut cx)?;
947                            handlers::copy_git(
948                                &mut cx,
949                                idx,
950                                &rev_resolved,
951                                &from_resolved,
952                                &to_resolved,
953                                *include_dirty,
954                            )
955                        }
956                        StepKind::HashSha256 { path } => {
957                            let path_resolved = super::args::resolve_arg(path, &mut cx)?;
958                            handlers::hash_sha256(&mut cx, idx, &path_resolved)
959                        }
960                        StepKind::Symlink {
961                            from_workspace,
962                            from,
963                            to,
964                        } => {
965                            let from_resolved = super::args::resolve_arg(from, &mut cx)?;
966                            let to_resolved = super::args::resolve_arg(to, &mut cx)?;
967                            handlers::symlink(
968                                &mut cx,
969                                idx,
970                                from_workspace.clone(),
971                                &from_resolved,
972                                &to_resolved,
973                            )
974                        }
975                        StepKind::Mkdir(arg) => {
976                            let path = super::args::resolve_arg(arg, &mut cx)?;
977                            handlers::mkdir(&mut cx, idx, &path)
978                        }
979                        StepKind::Ls(arg) => {
980                            let resolved = super::args::resolve_arg_opt(arg, &mut cx)?;
981                            handlers::ls(&mut cx, idx, &resolved)
982                        }
983                        StepKind::Cwd => handlers::cwd(&mut cx, idx),
984                        StepKind::Read(arg) => {
985                            let resolved = super::args::resolve_arg_opt(arg, &mut cx)?;
986                            handlers::read(&mut cx, idx, &resolved)
987                        }
988                        StepKind::ReadLine { var } => handlers::read_line(&mut cx, idx, var),
989                        StepKind::ListAppend { list, item } => {
990                            let value = super::args::evaluate_assert_operand(item, &mut cx)?;
991                            handlers::push_into(&mut cx, idx, list, value)
992                        }
993                        StepKind::Write { path, contents } => {
994                            let path_resolved = super::args::resolve_arg(path, &mut cx)?;
995                            let contents_resolved =
996                                super::args::resolve_arg_opt(contents, &mut cx)?;
997                            handlers::write(
998                                &mut cx,
999                                idx,
1000                                &path_resolved,
1001                                contents_resolved.as_deref(),
1002                            )
1003                        }
1004                        StepKind::Append { path, contents } => {
1005                            let path_resolved = super::args::resolve_arg(path, &mut cx)?;
1006                            let contents_resolved =
1007                                super::args::resolve_arg_opt(contents, &mut cx)?;
1008                            handlers::append(
1009                                &mut cx,
1010                                idx,
1011                                &path_resolved,
1012                                contents_resolved.as_deref(),
1013                            )
1014                        }
1015                        StepKind::Expand { path, overrides } => {
1016                            let path_resolved = super::args::resolve_arg_opt(path, &mut cx)?;
1017                            let overrides_resolved =
1018                                super::args::resolve_overrides(overrides, &mut cx)?;
1019                            handlers::replace(&mut cx, idx, &path_resolved, &overrides_resolved)
1020                        }
1021                        StepKind::AssertEq {
1022                            hash,
1023                            actual,
1024                            expected,
1025                        } => {
1026                            let target = resolve_assert_target(actual, &mut cx)?;
1027                            let expected_resolved = match expected {
1028                                Some(e) => Some(super::args::evaluate_assert_operand(e, &mut cx)?),
1029                                None => None,
1030                            };
1031                            handlers::assert_eq(
1032                                &mut cx,
1033                                idx,
1034                                generation,
1035                                idx,
1036                                hash,
1037                                &target,
1038                                expected_resolved.as_ref(),
1039                            )
1040                        }
1041                        StepKind::AssertContains { haystack, needle } => {
1042                            let target = resolve_assert_target(haystack, &mut cx)?;
1043                            handlers::assert_contains(
1044                                &mut cx, idx, generation, idx, &target, needle,
1045                            )
1046                        }
1047                        StepKind::WithIoBlock { .. } => {
1048                            bail!("WITH_IO block should have been expanded during parsing")
1049                        }
1050                        StepKind::Exit(code) => {
1051                            let code = super::args::resolve_arg_as_int(code, &mut cx)?;
1052                            handlers::exit(&mut cx, code)
1053                        }
1054                        StepKind::Assign {
1055                            var,
1056                            decl_type,
1057                            expr,
1058                        } => handlers::assign(&mut cx, var, decl_type.clone(), expr),
1059                        StepKind::Set { var, expr } => handlers::set_var_value(&mut cx, var, expr),
1060                        StepKind::AssignAsync {
1061                            var,
1062                            decl_type,
1063                            body,
1064                        } => handlers::dispatch_assign_async(var, decl_type.clone(), body, &mut cx),
1065                        StepKind::Await { var } => handlers::dispatch_await(var, &mut cx),
1066                        StepKind::AwaitCapture {
1067                            out_var,
1068                            out_type,
1069                            task_var,
1070                        } => handlers::dispatch_await_capture(
1071                            out_var,
1072                            out_type.clone(),
1073                            task_var,
1074                            &mut cx,
1075                        ),
1076                        StepKind::Cancel { var } => handlers::dispatch_cancel(var, &mut cx),
1077                        StepKind::Sleep { duration } => {
1078                            let duration = super::args::resolve_arg_as_duration(duration, &mut cx)?;
1079                            handlers::sleep(&mut cx, idx, &duration)
1080                        }
1081                        StepKind::FuncDef { .. }
1082                        | StepKind::Call { .. }
1083                        | StepKind::Return { .. }
1084                        | StepKind::While { .. }
1085                        | StepKind::Break
1086                        | StepKind::Continue
1087                        | StepKind::For { .. }
1088                        | StepKind::If { .. }
1089                        | StepKind::Timeout { .. }
1090                        | StepKind::WithIo { .. }
1091                        | StepKind::AssignCapture { .. } => {
1092                            unreachable!("compound steps dispatch in the outer match")
1093                        }
1094                    }?;
1095                    Ok(Flow::Done)
1096                }
1097            };
1098            flow_result
1099        };
1100
1101        let restore_result = restore_scopes(state, step.scope_exit);
1102        // Keeper expiry: drop spawn-time pins whose final producer step
1103        // just completed, so later consumer steps in the same task observe
1104        // EOF. Gated on slice identity, so nested bodies executing through
1105        // this same loop never discharge the worker's top-level map.
1106        let expiry_drained = if let Some(expiry) = state.keeper_expiry.as_mut() {
1107            expiry.expire_step(steps, idx)
1108        } else {
1109            false
1110        };
1111        if expiry_drained {
1112            state.keeper_expiry = None;
1113        }
1114        let flow = flow_result?;
1115        restore_result?;
1116        match flow {
1117            Flow::Done => {}
1118            Flow::Break { .. } | Flow::Continue { .. } | Flow::Return { .. } => {
1119                return Ok(flow);
1120            }
1121        }
1122    }
1123
1124    // Poll anonymous background handles at end-of-pipeline. The shared
1125    // named_tasks entries are reaped only by the root context: task threads
1126    // must never block on sibling tasks, which may depend on this thread
1127    // via AWAIT (three-way deadlock). Un-awaited named tasks are still
1128    // reaped by the root end-poll, and explicitly awaited tasks join via
1129    // AWAIT. Entries are retained as Cancelled/Completed tombstones so
1130    // later AWAIT/CANCEL report precise errors.
1131    let reap_named = !state.inside_async;
1132    let has_bg = !state.bg_children.is_empty();
1133    let named_pending = |state: &ExecState<P>| {
1134        reap_named
1135            && state
1136                .named_tasks
1137                .lock()
1138                .unwrap_or_else(|e| e.into_inner())
1139                .values()
1140                .any(|entry| !entry.state.lock().unwrap_or_else(|e| e.into_inner()).reaped)
1141    };
1142    let has_named = named_pending(state);
1143    if wait_at_end && (has_bg || has_named) {
1144        loop {
1145            let mut failed_status: Option<anyhow::Error> = None;
1146
1147            // Cancellation (deadline watcher or parent teardown) must break
1148            // the poll loop: without this, a stuck background handle would
1149            // hang the reaper forever. Flows into the shared fail-fast
1150            // teardown below.
1151            if failed_status.is_none() && state.cancel_token.load(Ordering::SeqCst) {
1152                failed_status = Some(anyhow::anyhow!("ASYNC task cancelled"));
1153            }
1154
1155            // 1. Poll anonymous background handles
1156            let mut i = 0;
1157            while i < state.bg_children.len() {
1158                match state.bg_children[i].try_wait() {
1159                    Ok(Some(status)) => {
1160                        if !status.success() && failed_status.is_none() {
1161                            failed_status =
1162                                Some(anyhow::anyhow!("ASYNC process exited with status {status}"));
1163                            break;
1164                        }
1165                        state.bg_children.swap_remove(i);
1166                    }
1167                    Ok(None) => {
1168                        i += 1;
1169                    }
1170                    Err(e) => {
1171                        if failed_status.is_none() {
1172                            failed_status = Some(e);
1173                        }
1174                        break;
1175                    }
1176                }
1177            }
1178
1179            // 2. Poll un-awaited named tasks (root context only). Each entry
1180            // is probed under a short entry lock; terminal entries are
1181            // retained as tombstones, never removed.
1182            if failed_status.is_none() && reap_named {
1183                let entries: Vec<(u64, Arc<TaskEntry>)> = {
1184                    let named = state.named_tasks.lock().unwrap_or_else(|e| e.into_inner());
1185                    named
1186                        .iter()
1187                        .map(|(id, entry)| (*id, Arc::clone(entry)))
1188                        .collect()
1189                };
1190                for (id, entry) in &entries {
1191                    enum Poll {
1192                        Pending,
1193                        CompletedOk,
1194                        CompletedErr(anyhow::Error),
1195                    }
1196                    let poll = {
1197                        let mut guard = entry.state.lock().unwrap_or_else(|e| e.into_inner());
1198                        match guard.phase {
1199                            TaskPhase::Running | TaskPhase::Awaiting => {
1200                                match guard.handle.as_mut() {
1201                                    Some(handle) => match handle.try_wait() {
1202                                        Ok(Some(status)) => {
1203                                            let _ = guard.handle.take();
1204                                            guard.phase = TaskPhase::Completed;
1205                                            if status.success() {
1206                                                Poll::CompletedOk
1207                                            } else {
1208                                                Poll::CompletedErr(anyhow::anyhow!(
1209                                                    "named ASYNC task {id} exited with status {status}"
1210                                                ))
1211                                            }
1212                                        }
1213                                        Ok(None) => Poll::Pending,
1214                                        Err(e) => {
1215                                            let _ = guard.handle.take();
1216                                            guard.phase = TaskPhase::Completed;
1217                                            Poll::CompletedErr(e)
1218                                        }
1219                                    },
1220                                    // Handle taken by a concurrent CANCEL/AWAIT
1221                                    // teardown; the barrier below rendezvouses.
1222                                    None => Poll::Pending,
1223                                }
1224                            }
1225                            TaskPhase::Cancelled | TaskPhase::Completed => Poll::Pending,
1226                        }
1227                    };
1228                    match poll {
1229                        Poll::Pending => {}
1230                        Poll::CompletedOk => {
1231                            entry.finish_teardown();
1232                        }
1233                        Poll::CompletedErr(e) => {
1234                            entry.finish_teardown();
1235                            if failed_status.is_none() {
1236                                failed_status = Some(e);
1237                            }
1238                            break;
1239                        }
1240                    }
1241                }
1242            }
1243
1244            // 3. Fail-fast teardown (named entries are root-owned; task
1245            // threads only tear down their own anonymous children).
1246            // Handles are taken under short locks and killed outside every
1247            // lock; tombstones are retained.
1248            if let Some(err) = failed_status {
1249                for survivor in state.bg_children.iter_mut() {
1250                    let _ = survivor.kill();
1251                }
1252                if reap_named {
1253                    let entries: Vec<Arc<TaskEntry>> = {
1254                        let named = state.named_tasks.lock().unwrap_or_else(|e| e.into_inner());
1255                        named.values().cloned().collect()
1256                    };
1257                    let mut to_kill: Vec<(Arc<TaskEntry>, Box<dyn BackgroundHandle>)> = Vec::new();
1258                    for entry in &entries {
1259                        let mut guard = entry.state.lock().unwrap_or_else(|e| e.into_inner());
1260                        match guard.phase {
1261                            TaskPhase::Running | TaskPhase::Awaiting => {
1262                                guard.phase = TaskPhase::Cancelled;
1263                                if let Some(handle) = guard.handle.take() {
1264                                    to_kill.push((Arc::clone(entry), handle));
1265                                }
1266                            }
1267                            TaskPhase::Cancelled | TaskPhase::Completed => {}
1268                        }
1269                    }
1270                    for (entry, mut handle) in to_kill {
1271                        let _ = handle.kill();
1272                        entry.finish_teardown();
1273                    }
1274                }
1275                state.bg_children.clear();
1276                return Err(err);
1277            }
1278
1279            let bg_empty = state.bg_children.is_empty();
1280            if bg_empty && !named_pending(state) {
1281                return Ok(Flow::Done);
1282            }
1283            // Rendezvous: a concurrent CANCEL/AWAIT on another thread may
1284            // own teardown of a Cancelled-but-unreaped entry. Wait for it
1285            // instead of spinning, so this thread never outruns the join.
1286            if reap_named {
1287                let unreaped: Vec<Arc<TaskEntry>> = {
1288                    let named = state.named_tasks.lock().unwrap_or_else(|e| e.into_inner());
1289                    named
1290                        .values()
1291                        .filter(|entry| {
1292                            let guard = entry.state.lock().unwrap_or_else(|e| e.into_inner());
1293                            matches!(guard.phase, TaskPhase::Cancelled) && !guard.reaped
1294                        })
1295                        .cloned()
1296                        .collect()
1297                };
1298                for entry in &unreaped {
1299                    entry.wait_reaped();
1300                }
1301            }
1302            std::thread::sleep(std::time::Duration::from_millis(10));
1303        }
1304    }
1305
1306    Ok(Flow::Done)
1307}
1308
1309fn restore_scopes<P: ProcessManager>(state: &mut ExecState<P>, count: usize) -> Result<()> {
1310    for _ in 0..count {
1311        state.pop_scope()?;
1312    }
1313    Ok(())
1314}
1315
1316/// Execute steps inside a fresh lexical scope (IF branches, TIMEOUT bodies).
1317/// Blocks scope everything (LET/ENV/WORKDIR/WORKSPACE); only pipes and
1318/// filesystem effects cross. Restores even when the body fails. Propagates
1319/// Flow signals (BREAK/CONTINUE/RETURN) to the caller after restoring.
1320#[allow(clippy::too_many_arguments)]
1321pub(super) fn execute_scoped_steps<P: ProcessManager>(
1322    state: &mut ExecState<P>,
1323    process: &mut P,
1324    steps: &[Step],
1325    stdin: CommandStdin,
1326    expose_stdin: bool,
1327    out: Option<StreamHandle>,
1328    err: Option<StreamHandle>,
1329    wait_at_end: bool,
1330) -> Result<Flow> {
1331    state.push_scope();
1332    let res = execute_steps(
1333        state,
1334        process,
1335        steps,
1336        stdin,
1337        expose_stdin,
1338        out,
1339        err,
1340        wait_at_end,
1341    );
1342    // Restore the scope even when the body failed, but never let an
1343    // unwinding failure mask the body's own error.
1344    let pop_res = state.pop_scope();
1345    match (res, pop_res) {
1346        (Ok(flow), Ok(())) => Ok(flow),
1347        (Err(e), _) => Err(e),
1348        (Ok(_), Err(e)) => Err(e),
1349    }
1350}
1351
1352/// Dispatch one compound step (loops, functions, scoped wrappers) through
1353/// the Flow path. Called with the caller's generation/idx so assertion
1354/// windows and error attribution match the leaf pipeline.
1355fn dispatch_flow_step<P: ProcessManager>(
1356    cmd: &StepKind,
1357    cx: &mut StepCtx<'_, P>,
1358    generation: usize,
1359    idx: usize,
1360) -> Result<Flow> {
1361    match cmd {
1362        StepKind::FuncDef { name, params, body } => {
1363            handlers::define_func(cx, name, params, body)?;
1364            Ok(Flow::Done)
1365        }
1366        StepKind::Call { name, args } => {
1367            let _ = handlers::call_func_value(cx, idx, name, args)?;
1368            Ok(Flow::Done)
1369        }
1370        StepKind::Return { expr } => handlers::handle_return(cx, idx, expr),
1371        StepKind::While { cond, body } => handlers::while_loop(cx, idx, cond, body),
1372        StepKind::Break => Ok(Flow::Break { idx }),
1373        StepKind::Continue => Ok(Flow::Continue { idx }),
1374        StepKind::For {
1375            key_var,
1376            key_type,
1377            var,
1378            var_type,
1379            in_expr,
1380            body,
1381        } => handlers::for_loop(
1382            cx,
1383            key_var.as_deref(),
1384            key_type.clone(),
1385            var,
1386            var_type.clone(),
1387            in_expr,
1388            body,
1389        ),
1390        StepKind::If {
1391            cond,
1392            then_body,
1393            else_ifs,
1394            else_body,
1395        } => handlers::if_then(cx, cond, then_body, else_ifs, else_body),
1396        StepKind::Timeout { duration, body } => {
1397            let duration = super::args::resolve_arg_as_duration(duration, cx)?;
1398            handlers::timeout(cx, idx, &duration, body)
1399        }
1400        StepKind::WithIo { bindings, cmd } => handlers::with_io(cx, generation, idx, bindings, cmd),
1401        StepKind::AssignCapture {
1402            var,
1403            decl_type,
1404            cmd,
1405        } => handlers::assign_capture(cx, generation, idx, var, decl_type.clone(), cmd),
1406        _ => {
1407            unreachable!("dispatch_flow_step handles only compound steps")
1408        }
1409    }
1410}
oxdock_core/exec/steps.rs

oxdock_core/exec/
steps.rs
