1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
// SPDX-License-Identifier: AGPL-3.0-only
//! Nested-body **cancellation and deadlines** end to end: when a `foreach`
//! element fails under `on_error: fail`, and when a `race` runs out of time,
//! the siblings still in flight must actually STOP — their steps cancelled,
//! their timers disarmed, their turn workers killed.
//!
//! The two failures these tests guard against:
//!
//! 1. `cancel_scoped_children` matching children with `starts_with("{prefix}.")`
//! cancels nothing nested: an element/branch instance is keyed
//! `parent[ix].step` / `parent{branch}.step` — neither starts with
//! `parent.`. The race WINNER path passes an already-scoped id, but the
//! three failure paths (`foreach` on_error fail, `parallel` on_error fail,
//! race timeout) pass the parent's bare id. A sibling's `sleep` then stays
//! armed, so the instance outlives the run that abandoned it and the
//! daemon — which counts a live timer as busy — cannot idle-exit.
//! 2. `timeout` is a COMMON_FIELD: the parser lifts it into `step.timeout_ms`
//! and never copies it into `spec`, so a `race` that reads its deadline back
//! out of `spec` gets null and waits forever.
//!
//! Both tests therefore assert that the process EXITS, promptly, under a hard
//! timeout: a failure fails fast instead of hanging CI for the full 90 s of
//! the sibling sleep it was supposed to cancel.
mod common;
use std::process::{Command, Stdio};
use std::time::{Duration, Instant};
/// The sleep a cancelled sibling is parked on. Long enough that waiting it out
/// is unmistakably a failure rather than a slow machine.
const SIBLING_SLEEP: &str = "90s";
/// How long a bounded life may take before the test calls it wedged. A build
/// that cancels its siblings exits in ~1.5 s (the failure plus `idle_grace`);
/// one that leaves a sibling timer armed needs 90 s, so anything in between
/// separates them cleanly.
const HARD_TIMEOUT: Duration = Duration::from_secs(20);
fn write_file(tag: &str, ext: &str, body: &str) -> String {
let path = common::unique_path(tag, ext);
std::fs::write(&path, body).expect("write test file");
path
}
/// A job-shaped v2 document with one inline workflow (JSON steps for brevity),
/// with no model and no MCP server — everything below is pure engine.
fn job(steps: &str) -> String {
write_file(
"agentd-nested-cancel",
"yaml",
&format!(
"config_version: \"1\"\nagent:\n name: nested-cancel\nworkflows:\n - name: pipe\n steps: {steps}\nlifecycle:\n run_until: idle\n idle_grace: 1s\nobservability:\n log_level: info\n log_content: true\n"
),
)
}
/// One life of the daemon, killed (and failed) if it does not exit in time.
/// Unlike `Command::output()` this never blocks forever: an uncancelled
/// sibling must surface as a failing assertion, not a 90-second hang.
/// Returns `(exit code, stderr, how long it took)`.
fn run_agentd_bounded(config: &str) -> (Option<i32>, String, Duration) {
let err_path = common::unique_path("agentd-nested-cancel", "err");
let err = std::fs::File::create(&err_path).expect("create stderr file");
let started = Instant::now();
let mut child = Command::new(env!("CARGO_BIN_EXE_agentd"))
.args(["--config", config])
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::from(err))
.spawn()
.expect("spawn agentd");
let deadline = started + HARD_TIMEOUT;
loop {
match child.try_wait().expect("wait for agentd") {
Some(status) => {
let log = std::fs::read_to_string(&err_path).unwrap_or_default();
let _ = std::fs::remove_file(&err_path);
return (status.code(), log, started.elapsed());
}
None if Instant::now() >= deadline => {
let _ = child.kill();
let _ = child.wait();
let log = std::fs::read_to_string(&err_path).unwrap_or_default();
let _ = std::fs::remove_file(&err_path);
panic!(
"agentd never exited (waited {HARD_TIMEOUT:?}): a nested body's siblings were never cancelled, so their {SIBLING_SLEEP} sleep timers keep the instance busy.\nstderr:\n{log}"
);
}
None => std::thread::sleep(Duration::from_millis(25)),
}
}
}
/// The telemetry lines named `name`, in emission order.
fn events(stderr: &str, name: &str) -> Vec<serde_json::Value> {
stderr
.lines()
.filter_map(|l| serde_json::from_str::<serde_json::Value>(l).ok())
.filter(|v| v["event"] == name)
.collect()
}
/// Whether a scoped step reached `name` in the telemetry (`step.start` proves
/// it really ran; `step.done` proves it reached a terminal status of its own —
/// which a CANCELLED step never does, because cancellation rewrites the status
/// in place rather than finishing the step).
fn saw_step(stderr: &str, name: &str, step: &str) -> bool {
events(stderr, name).iter().any(|e| e["step"] == step)
}
#[test]
fn a_failed_foreach_element_cancels_the_siblings_still_sleeping() {
// Two elements at `parallel: 2`, so both are in flight at once. The body
// routes on the item (a `switch`, not a `when` — this build has no CEL):
// element 0 parks on a 90 s sleep, element 1 fails outright. `on_error`
// defaults to `fail`, so the parent step fails and MUST take element 0's
// sleeper down with it.
let steps = format!(
r#"{{
"start": {{"kind": "once"}},
"each": {{"kind": "foreach", "depends_on": ["start"], "over": ["hold", "boom"], "batch": {{"parallel": 2}},
"body": {{"steps": {{
"route": {{"kind": "switch", "on": "{{{{item}}}}", "cases": {{"hold": "sleeper", "boom": "explode"}}}},
"sleeper": {{"kind": "sleep", "depends_on": ["route"], "duration": "{SIBLING_SLEEP}"}},
"explode": {{"kind": "fail", "depends_on": ["route"], "message": "element {{{{index}}}} exploded"}}}}}}}},
"done": {{"kind": "finish", "depends_on": ["each"], "status": "completed"}}
}}"#
);
let (code, stderr, took) = run_agentd_bounded(&job(&steps));
assert!(code.is_some(), "the daemon exited on its own:\n{stderr}");
// The premise: element 0 really did start its long sleep, and element 1
// really did fail — otherwise everything below holds vacuously.
assert!(
saw_step(&stderr, "step.start", "each[0].sleeper"),
"the sibling element started sleeping:\n{stderr}"
);
assert!(
events(&stderr, "step.done")
.iter()
.any(|e| e["step"] == "each[1].explode" && e["status"] == "failed"),
"the second element failed:\n{stderr}"
);
// The invariant: element 0's sleep is disarmed, so it never wakes and the
// instance goes idle at once instead of outliving the run by 90 seconds.
assert!(
!saw_step(&stderr, "step.done", "each[0].sleeper"),
"the cancelled sibling must never finish its sleep:\n{stderr}"
);
assert!(
took < Duration::from_secs(15),
"the instance idled out promptly (took {took:?}) — a still-armed sibling timer would hold it for {SIBLING_SLEEP}:\n{stderr}"
);
let done = events(&stderr, "run.done");
assert_eq!(
done.len(),
1,
"the run reached a terminal status:\n{stderr}"
);
assert_eq!(done[0]["status"], "failed", "{stderr}");
}
#[test]
fn a_race_timeout_fires_and_cancels_every_branch() {
// Neither branch can ever win: the race's own `timeout` is the only thing
// that can end this step, and when it fires both branches must be cancelled.
let steps = format!(
r#"{{
"start": {{"kind": "once"}},
"pick": {{"kind": "race", "depends_on": ["start"], "timeout": "300ms",
"branches": {{"slow": {{"steps": {{"s": {{"kind": "sleep", "duration": "{SIBLING_SLEEP}"}}}}}},
"slower": {{"steps": {{"s": {{"kind": "sleep", "duration": "{SIBLING_SLEEP}"}}}}}}}}}},
"done": {{"kind": "finish", "depends_on": ["pick"], "status": "completed"}}
}}"#
);
let (code, stderr, took) = run_agentd_bounded(&job(&steps));
assert!(code.is_some(), "the daemon exited on its own:\n{stderr}");
// The premise: both branches really were in flight when the deadline hit.
for branch in ["pick{slow}.s", "pick{slower}.s"] {
assert!(
saw_step(&stderr, "step.start", branch),
"branch step {branch} started:\n{stderr}"
);
assert!(
!saw_step(&stderr, "step.done", branch),
"branch step {branch} was cancelled, not slept out:\n{stderr}"
);
}
// The deadline is honoured at all — read out of a spec that never carries
// it, it is null, and the race waits on a 90 s branch instead.
assert!(
events(&stderr, "step.done")
.iter()
.any(|e| e["step"] == "pick" && e["status"] == "timeout"),
"the race timed out:\n{stderr}"
);
assert!(
took < Duration::from_secs(15),
"the instance idled out promptly (took {took:?}):\n{stderr}"
);
let done = events(&stderr, "run.done");
assert_eq!(
done.len(),
1,
"the run reached a terminal status:\n{stderr}"
);
assert_eq!(done[0]["status"], "failed", "{stderr}");
}