From aa185af04bef419a9b6744f9d51cc7eb1adb3e53 Mon Sep 17 00:00:00 2001 From: oxoxDev Date: Wed, 12 Aug 2026 16:13:55 +0530 Subject: [PATCH 1/2] chore: bump vendored tinyflows to 0.6.1 (#675) Advances the vendor/openhuman submodule to pick up tinyflows 0.6.1, which threads a parent run's cancellation token into sub_workflow children. The [patch] path dependency now resolves to 0.6.1 in Cargo.lock. --- Cargo.lock | 43 +++++++++++++++---------------------------- vendor/openhuman | 2 +- 2 files changed, 16 insertions(+), 29 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 01b4fc186..ee8a07a3e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -178,7 +178,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]] @@ -189,7 +189,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -2346,7 +2346,7 @@ dependencies = [ "libc", "option-ext", "redox_users 0.5.2", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -2736,7 +2736,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -4162,7 +4162,7 @@ dependencies = [ "js-sys", "log", "wasm-bindgen", - "windows-core 0.62.2", + "windows-core 0.57.0", ] [[package]] @@ -5544,7 +5544,7 @@ dependencies = [ "png 0.18.1", "serde", "thiserror 2.0.20", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -5747,7 +5747,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.59.0", ] [[package]] @@ -7361,7 +7361,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -7974,7 +7974,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -8814,7 +8814,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]] @@ -9573,7 +9573,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -9595,7 +9595,7 @@ dependencies = [ "parking_lot", "rustix", "signal-hook", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -9873,7 +9873,7 @@ dependencies = [ [[package]] name = "tinyflows" -version = "0.6.0" +version = "0.6.1" dependencies = [ "async-trait", "axum", @@ -10426,7 +10426,7 @@ dependencies = [ "png 0.18.1", "serde", "thiserror 2.0.20", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -11561,7 +11561,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]] @@ -11685,19 +11685,6 @@ dependencies = [ "windows-strings 0.4.2", ] -[[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 0.2.1", - "windows-result 0.4.1", - "windows-strings 0.5.1", -] - [[package]] name = "windows-future" version = "0.2.1" diff --git a/vendor/openhuman b/vendor/openhuman index d5ba6ab2c..bdd99f5b5 160000 --- a/vendor/openhuman +++ b/vendor/openhuman @@ -1 +1 @@ -Subproject commit d5ba6ab2c978ed24dd8ab74d38ae925f965a4682 +Subproject commit bdd99f5b5397fccc69d125f7579091995a6e9e71 From 1e0410d39c46da546c82fa06b5a93c61272f3568 Mon Sep 17 00:00:00 2001 From: oxoxDev Date: Wed, 12 Aug 2026 16:13:55 +0530 Subject: [PATCH 2/2] test(workflows): pin sub_workflow cancel propagation (#675) A full-stack regression: parent trigger -> sub_workflow(child) -> done, child trigger -> slow -> marker -> done. Cancelling while the child's slow node is mid-flight must reach the child run so its downstream marker node never executes, the run settles cancelled, and it winds down cleanly (bounded by the in-flight node's remainder, not the hard-abort grace). Verified to fail before the tinyflows fix (marker ran) and pass after. --- src/workflows/runner.rs | 189 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 189 insertions(+) diff --git a/src/workflows/runner.rs b/src/workflows/runner.rs index 50ec3ac27..f2f36284f 100644 --- a/src/workflows/runner.rs +++ b/src/workflows/runner.rs @@ -2178,6 +2178,195 @@ to = "done" ); } + /// A provider that records the last user message of every inference call and + /// **holds the `slow` node open until the operator cancels** (bounded), so a + /// cancel deterministically lands while a child `sub_workflow` node is + /// mid-flight. It distinguishes child nodes by a marker string authored into + /// each node's `prompt`: the node after `slow` must never be invoked once a + /// parent cancel has propagated into the child run. + struct RecordingSlowProvider { + seen: Arc>>, + 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]