diff --git a/Cargo.lock b/Cargo.lock index 6175e9a23..2d9a2fb61 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -132,7 +132,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -143,7 +143,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -1364,7 +1364,7 @@ dependencies = [ "libc", "option-ext", "redox_users 0.5.2", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -1477,7 +1477,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -2208,7 +2208,7 @@ dependencies = [ "js-sys", "log", "wasm-bindgen", - "windows-core 0.62.2", + "windows-core", ] [[package]] @@ -3058,7 +3058,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -3672,7 +3672,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -4035,7 +4035,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -4485,7 +4485,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -4711,7 +4711,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -4929,13 +4929,13 @@ dependencies = [ [[package]] name = "tinyflows" -version = "0.6.0" +version = "0.6.1" dependencies = [ "async-trait", "axum", "futures-timer", "futures-util", - "getrandom 0.3.4", + "getrandom 0.4.3", "jaq-core", "jaq-json", "jaq-std", @@ -5807,7 +5807,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.48.0", ] [[package]] @@ -5822,7 +5822,7 @@ version = "0.57.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "12342cb4d8e3b046f3d80effd474a7a02447231330ef77d71daa6fbc40681143" dependencies = [ - "windows-core 0.57.0", + "windows-core", "windows-targets 0.52.6", ] @@ -5832,25 +5832,12 @@ version = "0.57.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d2ed2439a290666cd67ecce2b0ffaad89c2a56b976b736e6ece670297897832d" dependencies = [ - "windows-implement 0.57.0", - "windows-interface 0.57.0", + "windows-implement", + "windows-interface", "windows-result 0.1.2", "windows-targets 0.52.6", ] -[[package]] -name = "windows-core" -version = "0.62.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb" -dependencies = [ - "windows-implement 0.60.2", - "windows-interface 0.59.3", - "windows-link", - "windows-result 0.4.1", - "windows-strings", -] - [[package]] name = "windows-implement" version = "0.57.0" @@ -5862,17 +5849,6 @@ dependencies = [ "syn 2.0.119", ] -[[package]] -name = "windows-implement" -version = "0.60.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.119", -] - [[package]] name = "windows-interface" version = "0.57.0" @@ -5884,17 +5860,6 @@ dependencies = [ "syn 2.0.119", ] -[[package]] -name = "windows-interface" -version = "0.59.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.119", -] - [[package]] name = "windows-link" version = "0.2.1" diff --git a/src/workflows/caps/mod.rs b/src/workflows/caps/mod.rs index 2e1e3d8bb..0def401da 100644 --- a/src/workflows/caps/mod.rs +++ b/src/workflows/caps/mod.rs @@ -293,6 +293,15 @@ pub async fn build_capabilities( state, resolver, agent, + // New in tinyflows 0.6.1, pinned here via the #675 cancel-token chain + // (openhuman #5520 → tinyflows #31). A `shell` node runs an inline + // POSIX script through a host-configured runner; OC wires none, and + // whether a workflow may spawn shell processes on the company host is a + // policy question this repo has not answered. `None` fails such a node + // at run time with a capability error — the honest answer until that + // decision is made, and no company manifest can currently author a + // `shell` node (mirrors `memory` below). + shell: None, // New in tinyflows 0.6, which arrived with the #499 pin bump. Left // unwired deliberately rather than pointed at the company's context // store: a `memory` node would then read and WRITE agent memory on diff --git a/src/workflows/gate.rs b/src/workflows/gate.rs index aad6cd1d0..15c6ff398 100644 --- a/src/workflows/gate.rs +++ b/src/workflows/gate.rs @@ -357,13 +357,15 @@ fn call_of(node: &tinyflows::model::Node) -> Option<(String, Value, Option>>, + entered_slow: Arc, + cancel: crate::ports::workflow_runner::RunCancel, + } + + #[async_trait] + impl tinyagents::harness::model::ChatModel<()> for RecordingSlowProvider { + async fn invoke( + &self, + _state: &(), + request: tinyagents::harness::model::ModelRequest, + ) -> tinyagents::Result { + // Scan the whole conversation, not just the last user turn: the + // openhuman harness reshapes an agent node's authored instruction + // into a multi-message prompt, so the node's marker can land in any + // role. Matching the joined text keeps the probe robust to that. + let all_text = request + .messages + .iter() + .map(|m| m.text()) + .collect::>() + .join("\n"); + self.seen.lock().expect("seen mutex").push(all_text.clone()); + if all_text.contains("SLOW-NODE") { + // Announce arrival, then hold the child at this node until the run + // is cancelled (bounded, so a broken build cannot hang CI). This + // pins the cancel to land while `slow` is in flight and makes the + // wind-down a clean node-boundary stop, not a hard abort. + self.entered_slow.notify_waiters(); + tokio::select! { + () = self.cancel.cancelled() => {} + () = tokio::time::sleep(std::time::Duration::from_secs(5)) => {} + } + } + Ok(tinyagents::harness::model::ModelResponse::assistant( + "acknowledged".to_string(), + )) + } + } + + impl crate::harness::provider::HarnessModel for RecordingSlowProvider { + fn telemetry_provider_id(&self) -> String { + "recording-slow".to_string() + } + } + + /// **Full-stack cancel propagation into a `sub_workflow` child (issue #675).** + /// Parent `trigger → sub_workflow(child) → done`, child + /// `trigger → slow → marker → done` where `slow`/`marker` are agent nodes. + /// The operator cancels while the child's `slow` node is mid-flight; the + /// parent's `CancellationToken` must reach the child run so its `marker` node + /// never executes, the run settles `cancelled`, and it comes back promptly + /// (the clean node-boundary wind-down bounded by `slow`'s remainder — not the + /// hard-abort grace). Before the fix the child ran behind a fresh token, so + /// the cancel never crossed the boundary and `marker` executed. + #[tokio::test] + async fn a_parent_cancel_propagates_into_a_sub_workflow_child() { + let source = tempfile::tempdir().unwrap(); + write_wf( + source.path(), + "child", + r#" +id = "child" +name = "Child" +[[node]] +id = "start" +kind = "trigger" +name = "Start" +[[node]] +id = "slow" +kind = "agent" +name = "Slow" +agent = "ceo" +summary = "SLOW-NODE hold here until cancelled" +prompt = "SLOW-NODE hold here until cancelled" +[[node]] +id = "marker" +kind = "agent" +name = "Marker" +agent = "ceo" +summary = "MARKER-NODE must never run once cancellation propagates" +prompt = "MARKER-NODE must never run once cancellation propagates" +[[node]] +id = "done" +kind = "output" +name = "Done" +[[edge]] +from = "start" +to = "slow" +[[edge]] +from = "slow" +to = "marker" +[[edge]] +from = "marker" +to = "done" +"#, + ); + let parent = r#" +id = "parent" +name = "Parent" +[[node]] +id = "start" +kind = "trigger" +name = "Start" +[[node]] +id = "sub" +kind = "sub_workflow" +name = "Sub" +[node.config] +workflow_id = "child" +[[node]] +id = "done" +kind = "output" +name = "Done" +[[edge]] +from = "start" +to = "sub" +[[edge]] +from = "sub" +to = "done" +"#; + + let home = tempfile::tempdir().unwrap(); + let pool = Arc::new(HarnessPool::new()); + let rec = record(); + let ctx = WorkflowRunContext::new(false); + let seen = Arc::new(std::sync::Mutex::new(Vec::new())); + let entered = Arc::new(tokio::sync::Notify::new()); + + let mut deps = deps_with_source(home.path(), source.path()); + deps.provider = Arc::new(RecordingSlowProvider { + seen: seen.clone(), + entered_slow: entered.clone(), + cancel: ctx.cancel.clone(), + }); + deps.provider_slug = "recording-slow".to_string(); + pool.ensure(&rec, &deps).await.expect("roster builds"); + + let file = parse_workflow(parent).expect("parent parses"); + // Registered before the run starts so `slow` cannot slip past it. + let reached_slow = entered.notified(); + + let mut run = Box::pin(run_workflow(pool, deps, &rec, &file, Value::Null, &ctx)); + tokio::select! { + _ = &mut run => panic!( + "the run finished before `slow` was reached; provider saw: {:?}", + seen.lock().expect("seen mutex") + ), + () = reached_slow => {} + } + + // The operator presses Cancel while the child's `slow` node is in flight. + let pressed = std::time::Instant::now(); + ctx.cancel.cancel(); + let run = tokio::time::timeout(std::time::Duration::from_secs(30), run) + .await + .expect("the cancelled run never returned") + .expect("a cancelled run is Ok, not Err"); + let elapsed = pressed.elapsed(); + + assert!(run.cancelled, "the run must report that it was stopped"); + + let seen = seen.lock().expect("seen mutex").clone(); + assert!( + seen.iter().any(|m| m.contains("SLOW-NODE")), + "the in-flight child `slow` node ran: {seen:?}" + ); + assert!( + !seen.iter().any(|m| m.contains("MARKER-NODE")), + "cancellation must propagate into the child: its `marker` node should never run, \ + got {seen:?}" + ); + // A clean node-boundary wind-down bounded by `slow`'s remainder — nowhere + // near the hard-abort grace, which only a wedged (never-returning) node + // would reach. + assert!( + elapsed < CANCEL_HARD_ABORT_GRACE, + "the child wound down cleanly, so settle time should be well under the hard-abort \ + grace; took {elapsed:?}" + ); + } + /// T-cycle — two on-disk workflows referencing each other by id hard-reject /// with the static cycle message, not the depth backstop. #[tokio::test] diff --git a/vendor/openhuman b/vendor/openhuman index 2826259d0..acab11b93 160000 --- a/vendor/openhuman +++ b/vendor/openhuman @@ -1 +1 @@ -Subproject commit 2826259d0f4ae553a8837f942a13c9ca24882bde +Subproject commit acab11b93969e1c8304723c525e086ad50fe71e4