diff --git a/crates/arc-workflows/src/context/keys.rs b/crates/arc-workflows/src/context/keys.rs index 8abf20014..41363ccda 100644 --- a/crates/arc-workflows/src/context/keys.rs +++ b/crates/arc-workflows/src/context/keys.rs @@ -24,6 +24,7 @@ pub const INTERNAL_WORK_DIR: &str = "internal.work_dir"; pub const INTERNAL_FIDELITY: &str = "internal.fidelity"; pub const INTERNAL_THREAD_ID: &str = "internal.thread_id"; pub const INTERNAL_NODE_VISIT_COUNT: &str = "internal.node_visit_count"; +pub const INTERNAL_PARENT_PREAMBLE: &str = "internal.parent_preamble"; // --- current.* keys --- pub const CURRENT_PREAMBLE: &str = "current.preamble"; @@ -74,6 +75,16 @@ pub fn retry_count_key(node_id: &str) -> String { format!("{INTERNAL_RETRY_COUNT_PREFIX}{node_id}") } +/// Returns `true` for engine-internal keys that should not propagate from child +/// to parent workflow contexts. +#[must_use] +pub fn is_engine_internal_key(key: &str) -> bool { + key.starts_with(INTERNAL_PREFIX) + || key.starts_with(GRAPH_PREFIX) + || key.starts_with(THREAD_PREFIX) + || key.starts_with(CURRENT_PREFIX) +} + /// Fidelity mode controlling how much prior context is provided to LLM sessions. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Fidelity { @@ -202,4 +213,24 @@ mod tests { fn fidelity_unknown_mode_errors() { assert!("bogus".parse::().is_err()); } + + #[test] + fn is_engine_internal_key_classifies_correctly() { + // Keys that ARE engine-internal (should not propagate) + assert!(is_engine_internal_key("internal.run_id")); + assert!(is_engine_internal_key("internal.fidelity")); + assert!(is_engine_internal_key("internal.parent_preamble")); + assert!(is_engine_internal_key("graph.goal")); + assert!(is_engine_internal_key("thread.main.current_node")); + assert!(is_engine_internal_key("current.preamble")); + assert!(is_engine_internal_key("current_node")); + + // Keys that are NOT engine-internal (should propagate) + assert!(!is_engine_internal_key("response.plan")); + assert!(!is_engine_internal_key("command.output")); + assert!(!is_engine_internal_key("outcome")); + assert!(!is_engine_internal_key("last_stage")); + assert!(!is_engine_internal_key("review.result")); + assert!(!is_engine_internal_key("user.name")); + } } diff --git a/crates/arc-workflows/src/handler/manager_loop.rs b/crates/arc-workflows/src/handler/manager_loop.rs index df3f799d7..6878029ce 100644 --- a/crates/arc-workflows/src/handler/manager_loop.rs +++ b/crates/arc-workflows/src/handler/manager_loop.rs @@ -7,6 +7,7 @@ use std::time::Duration; use async_trait::async_trait; use crate::condition::evaluate_condition; +use crate::context::keys; use crate::context::Context; use crate::engine::{RunConfig, WorkflowRunEngine}; use crate::error::ArcError; @@ -148,8 +149,15 @@ impl Handler for SubWorkflowHandler { labels: HashMap::new(), }; - // Clone parent context for child; snapshot before for diffing + // Clone parent context for child; inject parent preamble let child_context = context.clone_context(); + let parent_preamble = context.preamble(); + if !parent_preamble.is_empty() { + child_context.set( + keys::INTERNAL_PARENT_PREAMBLE, + serde_json::json!(parent_preamble), + ); + } let before_snapshot = context.snapshot(); // Spawn child engine @@ -171,9 +179,19 @@ impl Handler for SubWorkflowHandler { Err(e) => return Ok(Outcome::fail_classify(format!("Child task panicked: {e}"))), }; - // Compute context diff + // Compute context diff, filtering engine-internal keys let after_snapshot = child_final_context.snapshot(); - let diff = context_diff(&before_snapshot, &after_snapshot); + let raw_diff = context_diff(&before_snapshot, &after_snapshot); + let diff: HashMap = raw_diff + .into_iter() + .filter(|(key, _)| !keys::is_engine_internal_key(key)) + .collect(); + + tracing::debug!( + node = %node.id, + propagated_keys = ?diff.keys().collect::>(), + "Sub-workflow context diff filtered" + ); let mut outcome = Outcome { status: child_outcome.status.clone(), @@ -642,4 +660,204 @@ mod tests { let diff = context_diff(&before, &after); assert!(diff.is_empty()); } + + #[test] + fn context_diff_excludes_engine_internal_keys() { + let before = HashMap::new(); + let mut after = HashMap::new(); + after.insert("graph.goal".to_string(), serde_json::json!("child goal")); + after.insert( + "internal.run_id".to_string(), + serde_json::json!("child-run"), + ); + after.insert( + "thread.main.current_node".to_string(), + serde_json::json!("exit"), + ); + after.insert("current_node".to_string(), serde_json::json!("exit")); + after.insert( + "response.plan".to_string(), + serde_json::json!("the plan"), + ); + after.insert( + "review.result".to_string(), + serde_json::json!("approved"), + ); + + let raw_diff = context_diff(&before, &after); + let filtered: HashMap = raw_diff + .into_iter() + .filter(|(key, _)| !keys::is_engine_internal_key(key)) + .collect(); + + assert_eq!(filtered.len(), 2); + assert!(filtered.contains_key("response.plan")); + assert!(filtered.contains_key("review.result")); + } + + #[tokio::test] + async fn context_flows_parent_to_child_and_back_excludes_internals() { + struct ContextEchoHandler; + + #[async_trait] + impl Handler for ContextEchoHandler { + async fn execute( + &self, + _node: &Node, + context: &Context, + _graph: &Graph, + _logs_root: &Path, + _services: &EngineServices, + ) -> Result { + let target = context.get_string("review.target", ""); + let mut outcome = Outcome::success(); + outcome + .context_updates + .insert("review.result".to_string(), serde_json::json!("approved")); + outcome + .context_updates + .insert("review.echo".to_string(), serde_json::json!(target)); + Ok(outcome) + } + } + + let mut registry = HandlerRegistry::new(Box::new(ContextEchoHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + let services = EngineServices { + registry: std::sync::Arc::new(registry), + emitter: std::sync::Arc::new(EventEmitter::new()), + sandbox: std::sync::Arc::new(arc_agent::LocalSandbox::new( + std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")), + )), + git_state: std::sync::RwLock::new(None), + hook_runner: None, + }; + + let handler = SubWorkflowHandler; + let mut node = Node::new("manager"); + node.attrs.insert( + "stack.child_dot_source".to_string(), + AttrValue::String( + "digraph Child { start [shape=Mdiamond]; work [shape=box]; exit [shape=Msquare]; start -> work -> exit }" + .to_string(), + ), + ); + node.attrs + .insert("manager.max_cycles".to_string(), AttrValue::Integer(100)); + node.attrs.insert( + "manager.poll_interval".to_string(), + AttrValue::Duration(Duration::from_millis(10)), + ); + + let context = Context::new(); + context.set("review.target", serde_json::json!("src/main.rs")); + + let graph = Graph::new("test"); + let dir = tempfile::tempdir().unwrap(); + + let outcome = handler + .execute(&node, &context, &graph, dir.path(), &services) + .await + .unwrap(); + + assert_eq!(outcome.status, StageStatus::Success); + // User-defined keys propagate + assert_eq!( + outcome.context_updates.get("review.result"), + Some(&serde_json::json!("approved")) + ); + // Engine-internal keys do NOT propagate + assert!(!outcome.context_updates.contains_key("internal.run_id")); + assert!(!outcome.context_updates.contains_key("graph.goal")); + assert!(!outcome + .context_updates + .keys() + .any(|k| k.starts_with("thread."))); + assert!(!outcome + .context_updates + .keys() + .any(|k| k.starts_with("current"))); + } + + #[tokio::test] + async fn child_receives_parent_preamble() { + struct PreambleEchoHandler; + + #[async_trait] + impl Handler for PreambleEchoHandler { + async fn execute( + &self, + _node: &Node, + context: &Context, + _graph: &Graph, + _logs_root: &Path, + _services: &EngineServices, + ) -> Result { + let parent_preamble = + context.get_string(keys::INTERNAL_PARENT_PREAMBLE, ""); + let mut outcome = Outcome::success(); + outcome.context_updates.insert( + "echo.parent_preamble".to_string(), + serde_json::json!(parent_preamble), + ); + Ok(outcome) + } + } + + let mut registry = HandlerRegistry::new(Box::new(PreambleEchoHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + let services = EngineServices { + registry: std::sync::Arc::new(registry), + emitter: std::sync::Arc::new(EventEmitter::new()), + sandbox: std::sync::Arc::new(arc_agent::LocalSandbox::new( + std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")), + )), + git_state: std::sync::RwLock::new(None), + hook_runner: None, + }; + + let handler = SubWorkflowHandler; + let mut node = Node::new("manager"); + node.attrs.insert( + "stack.child_dot_source".to_string(), + AttrValue::String( + "digraph Child { start [shape=Mdiamond]; work [shape=box]; exit [shape=Msquare]; start -> work -> exit }" + .to_string(), + ), + ); + node.attrs + .insert("manager.max_cycles".to_string(), AttrValue::Integer(100)); + node.attrs.insert( + "manager.poll_interval".to_string(), + AttrValue::Duration(Duration::from_millis(10)), + ); + + // Set a preamble on the parent context + let context = Context::new(); + context.set( + keys::CURRENT_PREAMBLE, + serde_json::json!("Parent did step A and step B"), + ); + + let graph = Graph::new("test"); + let dir = tempfile::tempdir().unwrap(); + + let outcome = handler + .execute(&node, &context, &graph, dir.path(), &services) + .await + .unwrap(); + + assert_eq!(outcome.status, StageStatus::Success); + let echoed = outcome + .context_updates + .get("echo.parent_preamble") + .and_then(|v| v.as_str()) + .unwrap_or(""); + assert!( + echoed.contains("Parent did step A and step B"), + "Child should receive the parent preamble, got: {echoed}" + ); + } } diff --git a/crates/arc-workflows/src/preamble.rs b/crates/arc-workflows/src/preamble.rs index eef9ba08c..25f80e10d 100644 --- a/crates/arc-workflows/src/preamble.rs +++ b/crates/arc-workflows/src/preamble.rs @@ -29,7 +29,7 @@ pub fn build_preamble( let goal = graph.goal(); let run_id = context.run_id(); - match fidelity { + let preamble = match fidelity { Fidelity::Full => String::new(), Fidelity::Truncate => { format!("Goal: {goal}\nRun ID: {run_id}\n") @@ -64,6 +64,13 @@ pub fn build_preamble( graph, SummaryDetail::High, ), + }; + + let parent_preamble = context.get_string(keys::INTERNAL_PARENT_PREAMBLE, ""); + if !parent_preamble.is_empty() && !preamble.is_empty() { + format!("## Parent workflow context\n{parent_preamble}\n\n## Current sub-workflow\n{preamble}") + } else { + preamble } } @@ -2031,4 +2038,58 @@ mod tests { "should not show stages header when empty" ); } + + #[test] + fn build_preamble_prepends_parent_preamble_when_present() { + let graph = Graph::new("test"); + let context = Context::new(); + context.set( + keys::INTERNAL_PARENT_PREAMBLE, + serde_json::json!("Parent completed plan and review"), + ); + let completed_nodes: Vec = Vec::new(); + let node_outcomes: HashMap = HashMap::new(); + + let preamble = build_preamble( + keys::Fidelity::Compact, + &context, + &graph, + &completed_nodes, + &node_outcomes, + ); + + assert!( + preamble.contains("## Parent workflow context"), + "should contain parent section header" + ); + assert!( + preamble.contains("Parent completed plan and review"), + "should contain parent preamble text" + ); + assert!( + preamble.contains("## Current sub-workflow"), + "should contain current sub-workflow section header" + ); + } + + #[test] + fn build_preamble_no_parent_preamble_when_absent() { + let graph = Graph::new("test"); + let context = Context::new(); + let completed_nodes: Vec = Vec::new(); + let node_outcomes: HashMap = HashMap::new(); + + let preamble = build_preamble( + keys::Fidelity::Compact, + &context, + &graph, + &completed_nodes, + &node_outcomes, + ); + + assert!( + !preamble.contains("Parent workflow context"), + "should not contain parent section when no parent preamble" + ); + } }