Skip to content

fix(engine): thread parent cancellation token into sub_workflow children #31

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
senamakel merged 6 commits into from
Aug 12, 2026
Merged

fix(engine): thread parent cancellation token into sub_workflow children #31

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
80f0071
fix(engine): thread parent cancellation token into sub_workflow children
oxoxDev Aug 12, 2026
1fc376d
test(engine): pin cross-boundary cancellation propagation (T1-T5)
oxoxDev Aug 12, 2026
76233d4
chore: release tinyflows 0.6.1
oxoxDev Aug 12, 2026
92adf45
Merge remote-tracking branch 'origin/main' into fix/675-subworkflow-c…
oxoxDev Aug 12, 2026
7077a3c
fix(engine): add token to the shell-node test NodeContext after mergi…
oxoxDev Aug 12, 2026
f579915
test(engine): scope the T5 defensive-arm guardrail to production source
oxoxDev Aug 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[package]
name = "tinyflows"
version = "0.6.0"
version = "0.6.1"
edition = "2024"
rust-version = "1.85"
license = "GPL-3.0-or-later"
Expand Down
17 changes: 16 additions & 1 deletion src/engine.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -808,6 +808,15 @@ pub const MAX_SUB_WORKFLOW_DEPTH: u64 = 8;
/// child can read it back from `ctx.run` and enforce [`MAX_SUB_WORKFLOW_DEPTH`].
/// Used only by the `sub_workflow` node's recursive execution.
///
/// `token` is the **parent run's** cancellation token, forwarded so cancelling
/// the parent winds the whole subtree down: the child observes the same flipped
/// flag at its next node boundary and returns a cancelled [`RunOutcome`] instead
/// of running to completion orphaned from the parent. The child in turn hands
/// this token to its own node contexts, so a deeper `sub_workflow` propagates it
/// on — the whole nesting chain shares one signal. Historically this seeded a
/// fresh [`CancellationToken`], which severed cancellation at every sub-workflow
/// boundary.
///
/// # Errors
/// Same as [`run`].
pub(crate) async fn run_sub_workflow(
Expand All @@ -816,6 +825,7 @@ pub(crate) async fn run_sub_workflow(
capabilities: &Capabilities,
depth: u64,
max_depth: u64,
token: CancellationToken,
) -> Result<RunOutcome> {
let checkpointer: Arc<dyn Checkpointer<Value>> =
Arc::new(InMemoryCheckpointer::<Value>::default());
Expand All @@ -833,7 +843,7 @@ pub(crate) async fn run_sub_workflow(
"sub_workflow_depth": depth,
"max_sub_workflow_depth": max_depth,
})),
CancellationToken::new(),
token,
)
.await?;
Ok(outcome)
Expand Down Expand Up @@ -1266,6 +1276,10 @@ fn build_graph(
run: &run_meta,
nodes: &nodes_state,
caps: &caps,
// Handed to the executor so a nested engine call (today the
// `sub_workflow` node) can thread this run's cancellation
// into its child; a plain executor never reads it.
token: token.clone(),
};
// BUG-8: bound THIS attempt (not the whole retry loop) to
// `node_timeout`. Race the attempt future against a
Expand Down Expand Up @@ -1386,6 +1400,7 @@ fn build_graph(
run: &run_meta,
nodes: &nodes_state,
caps: &caps,
token: token.clone(),
};
let scope = crate::nodes::expr_scope(&ctx);
crate::expr::resolve_traced(&node.config, &scope).1
Expand Down
1 change: 1 addition & 0 deletions src/nodes/control_flow/condition.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,7 @@ mod tests {
run: &run,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let out = ConditionNode.execute(ctx).await.expect("execute");
(
Expand Down
1 change: 1 addition & 0 deletions src/nodes/control_flow/dedup.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -325,6 +325,7 @@ mod tests {
run: &run,
nodes: &Value::Null,
caps,
token: crate::engine::CancellationToken::new(),
};
DedupNode.execute(ctx).await.expect("execute")
}
Expand Down
1 change: 1 addition & 0 deletions src/nodes/control_flow/loop_node.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,6 +180,7 @@ mod tests {
run: &Value::Null,
nodes: &nodes,
caps: &caps,
token: crate::engine::CancellationToken::new(),

@tinysweeper tinysweeper Bot Aug 12, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

priority medium security uncertain

Propagate the parent cancellation token instead of creating a new one in the l

The diff adds token: crate::engine::CancellationToken::new() to the execution context built inside the loop node's run method. CancellationToken::new() creates a token in the non-cancelled state; it has no link to any parent token that an engine or caller might use to cancel the workflow. Because this is a loop node, the body (or sub-nodes) executed with this context will be effectively uncancellable from the outside. If the loop condition or iteration source is influenced by workflow data or external input, a caller cannot interrupt a runaway loop, creating a denial-of-service path. The loop node should propagate the cancellation token it received from its own execution context rather than minting a new one.

[RULE] Keep the workflow model declarative; no arbitrary embedded scripting—code execution is a sandboxed capability. ·

})
.await
}
Expand Down
2 changes: 2 additions & 0 deletions src/nodes/control_flow/merge.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@ mod tests {
run: &Value::Null,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};

let output = MergeNode.execute(ctx).await.expect("execute");
Expand All @@ -71,6 +72,7 @@ mod tests {
run: &Value::Null,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
MergeNode.execute(ctx).await.expect("execute").items
}
Expand Down
4 changes: 4 additions & 0 deletions src/nodes/control_flow/split_out.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,7 @@ mod tests {
run: &Value::Null,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};

let output = SplitOutNode.execute(ctx).await.expect("execute");
Expand All @@ -110,6 +111,7 @@ mod tests {
run: &Value::Null,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};

let output = SplitOutNode.execute(ctx).await.expect("execute");
Expand All @@ -131,6 +133,7 @@ mod tests {
run: &Value::Null,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};

let output = SplitOutNode.execute(ctx).await.expect("execute");
Expand All @@ -149,6 +152,7 @@ mod tests {
run: &Value::Null,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
SplitOutNode.execute(ctx).await.expect("execute").items
}
Expand Down
2 changes: 2 additions & 0 deletions src/nodes/control_flow/switch.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,7 @@ mod tests {
run: &run,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let out = SwitchNode.execute(ctx).await.expect("execute");
(out.port.expect("switch always routes to a port"), out.items)
Expand All @@ -112,6 +113,7 @@ mod tests {
run: &run,
nodes: &nodes,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let out = SwitchNode.execute(ctx).await.expect("execute");
assert_eq!(out.port.as_deref(), Some("urgent"));
Expand Down
2 changes: 2 additions & 0 deletions src/nodes/control_flow/transform.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,7 @@ mod tests {
run: &run,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
TransformNode.execute(ctx).await.expect("execute").items
}
Expand All @@ -110,6 +111,7 @@ mod tests {
run: &run,
nodes: &nodes,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let out = TransformNode.execute(ctx).await.expect("execute").items;
assert_eq!(out[0].json["who"], json!("a@b.com"));
Expand Down
8 changes: 8 additions & 0 deletions src/nodes/integration/agent.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -291,6 +291,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
})
.await
.expect("execute");
Expand All @@ -306,6 +307,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
})
.await
.expect("execute");
Expand All @@ -327,6 +329,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let out = AgentNode.execute(ctx).await.expect("execute");
assert_eq!(out.items.len(), 1);
Expand All @@ -352,6 +355,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let out = AgentNode.execute(ctx).await.expect("execute");
assert_eq!(out.items[0].json["json"]["completion"]["prompt"], "X");
Expand All @@ -369,6 +373,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let out = AgentNode.execute(ctx).await.expect("execute");
assert_eq!(out.items[0].json["json"]["connection"], Value::Null);
Expand All @@ -392,6 +397,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let out = AgentNode.execute(ctx).await.expect("execute");
assert_eq!(out.items.len(), 1);
Expand Down Expand Up @@ -419,6 +425,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps,
token: crate::engine::CancellationToken::new(),
};
AgentNode
.execute(ctx)
Expand Down Expand Up @@ -579,6 +586,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let err = AgentNode
.execute(ctx)
Expand Down
1 change: 1 addition & 0 deletions src/nodes/integration/code.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
CodeNode.execute(ctx).await.expect("execute").items
}
Expand Down
3 changes: 3 additions & 0 deletions src/nodes/integration/http_request.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,6 +139,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let out = HttpRequestNode.execute(ctx).await.expect("execute");
assert_eq!(out.items.len(), 1);
Expand Down Expand Up @@ -175,6 +176,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let out = HttpRequestNode.execute(ctx).await.expect("execute");
assert_eq!(out.items[0].json["json"]["request"]["url"], "https://a");
Expand Down Expand Up @@ -202,6 +204,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let out = HttpRequestNode.execute(ctx).await.expect("execute");
assert_eq!(out.items[0].json["json"]["connection"], Value::Null);
Expand Down
8 changes: 8 additions & 0 deletions src/nodes/integration/memory.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -404,6 +404,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let out = MemoryNode.execute(ctx).await.expect("execute");
assert_eq!(out.items.len(), 2, "per_item default maps over input");
Expand Down Expand Up @@ -458,6 +459,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let out = MemoryNode.execute(ctx).await.expect("execute");
assert_eq!(out.items[0].json["json"]["opts"]["operation"], "search");
Expand All @@ -476,6 +478,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let err = MemoryNode
.execute(ctx)
Expand All @@ -501,6 +504,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let err = MemoryNode
.execute(ctx)
Expand Down Expand Up @@ -543,6 +547,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let err = MemoryNode
.execute(ctx)
Expand All @@ -567,6 +572,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let err = MemoryNode
.execute(ctx)
Expand All @@ -591,6 +597,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let err = MemoryNode
.execute(ctx)
Expand Down Expand Up @@ -619,6 +626,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let out = MemoryNode.execute(ctx).await.expect("execute");
assert_eq!(out.items.len(), 1, "once mode emits a single item");
Expand Down
3 changes: 3 additions & 0 deletions src/nodes/integration/output_parser.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,7 @@ mod tests {
run: &Value::Null,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
let out = OutputParserNode.execute(ctx).await.expect("execute");
assert_eq!(out.items, input);
Expand Down Expand Up @@ -109,6 +110,7 @@ mod tests {
run: &Value::Null,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
};
OutputParserNode.execute(ctx).await.expect("execute").items
}
Expand Down Expand Up @@ -171,6 +173,7 @@ mod tests {
run: &run_meta,
nodes: &Value::Null,
caps,
token: crate::engine::CancellationToken::new(),
};
OutputParserNode.execute(ctx).await.map(|o| o.items)
}
Expand Down
1 change: 1 addition & 0 deletions src/nodes/integration/shell_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ async fn execute_with(caps: Capabilities, config: Value) -> Result<NodeOutput> {
run: &Value::Null,
nodes: &Value::Null,
caps: &caps,
token: crate::engine::CancellationToken::new(),
})
.await
}
Expand Down
Loading