Fix sub-workflow context diff leaking child internals into parent

Filter engine-internal keys (internal.*, graph.*, thread.*, current*)
from the context diff returned by SubWorkflowHandler, preventing child
run state from overwriting parent values. Pass the parent's preamble
into the child context so child workflows have awareness of what the
parent already accomplished.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

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::<Fidelity>().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"));
+    }
 }

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<String, serde_json::Value> = raw_diff
+                        .into_iter()
+                        .filter(|(key, _)| !keys::is_engine_internal_key(key))
+                        .collect();
+
+                    tracing::debug!(
+                        node = %node.id,
+                        propagated_keys = ?diff.keys().collect::<Vec<_>>(),
+                        "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<String, serde_json::Value> = 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<Outcome, ArcError> {
+                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<Outcome, ArcError> {
+                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}"
+        );
+    }
 }

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<String> = Vec::new();
+        let node_outcomes: HashMap<String, Outcome> = 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<String> = Vec::new();
+        let node_outcomes: HashMap<String, Outcome> = 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"
+        );
+    }
 }