oxdock_core/exec/
steps.rs

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

oxdock_core/exec/
steps.rs
