From f45c1b1e2d187450701e1a5d26f9b18bfcdf65cd Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sun, 1 Mar 2026 23:38:53 -0500 Subject: [PATCH] Remove unused SubPipelineHandler No DOT pipelines reference the sub_pipeline handler type. The manager loop handler now covers the child pipeline spawning use case. Co-Authored-By: Claude Opus 4.6 --- crates/arc-workflows/src/handler/mod.rs | 2 - .../arc-workflows/src/handler/sub_pipeline.rs | 402 ------------------ crates/arc-workflows/tests/integration.rs | 62 --- 3 files changed, 466 deletions(-) delete mode 100644 crates/arc-workflows/src/handler/sub_pipeline.rs diff --git a/crates/arc-workflows/src/handler/mod.rs b/crates/arc-workflows/src/handler/mod.rs index 1bdd85002..9df890c75 100644 --- a/crates/arc-workflows/src/handler/mod.rs +++ b/crates/arc-workflows/src/handler/mod.rs @@ -6,7 +6,6 @@ pub mod manager_loop; pub mod parallel; pub mod script; pub mod start; -pub mod sub_pipeline; pub mod wait_human; use std::collections::HashMap; @@ -137,7 +136,6 @@ pub fn default_registry( "parallel.fan_in", Box::new(fan_in::FanInHandler::new(make_backend())), ); - registry.register("sub_pipeline", Box::new(sub_pipeline::SubPipelineHandler)); registry.register( "stack.manager_loop", Box::new(manager_loop::ManagerLoopHandler), diff --git a/crates/arc-workflows/src/handler/sub_pipeline.rs b/crates/arc-workflows/src/handler/sub_pipeline.rs deleted file mode 100644 index e5cd8c231..000000000 --- a/crates/arc-workflows/src/handler/sub_pipeline.rs +++ /dev/null @@ -1,402 +0,0 @@ -use std::path::Path; -use std::time::Instant; - -use async_trait::async_trait; - -use crate::context::Context; -use crate::engine::select_edge; -use crate::error::ArcError; -use crate::event::PipelineEvent; -use crate::graph::{Graph, Node}; -use crate::outcome::Outcome; -use crate::pipeline::prepare_pipeline; - -use super::{EngineServices, Handler}; - -/// Convert a Duration's milliseconds to u64, saturating on overflow. -fn millis_u64(d: std::time::Duration) -> u64 { - u64::try_from(d.as_millis()).unwrap_or(u64::MAX) -} - -/// Executes a sub-pipeline defined by inline DOT source in a node attribute. -/// The sub-pipeline runs with a cloned context; context updates propagate back. -pub struct SubPipelineHandler; - -/// Check whether a node is a terminal (exit) node. -fn is_terminal(node: &Node) -> bool { - node.shape() == "Msquare" || node.handler_type() == Some("exit") -} - -#[async_trait] -impl Handler for SubPipelineHandler { - async fn execute( - &self, - node: &Node, - context: &Context, - _graph: &Graph, - logs_root: &Path, - services: &EngineServices, - ) -> Result { - // 1. Get DOT source from node attribute - let dot_source = match node - .attrs - .get("sub_pipeline.dot_source") - .and_then(|v| v.as_str()) - { - Some(s) if !s.is_empty() => s, - _ => { - return Ok(Outcome::fail( - "No sub_pipeline.dot_source attribute specified", - )) - } - }; - - // 2. Parse the sub-pipeline DOT - let sub_graph = match prepare_pipeline(dot_source) { - Ok(g) => g, - Err(e) => return Ok(Outcome::fail(format!("Failed to parse sub-pipeline: {e}"))), - }; - - // 3. Find start node - let start_node = match sub_graph.find_start_node() { - Some(n) => n.id.clone(), - None => return Ok(Outcome::fail("Sub-pipeline has no start node")), - }; - - // 4. Clone parent context for isolation - let sub_context = context.clone_context(); - let before_snapshot = context.snapshot(); - - // 5. Walk the sub-graph - let visit = crate::engine::visit_from_context(context); - let sub_logs_root = crate::engine::node_dir(logs_root, &node.id, visit); - let mut current_node_id = start_node.clone(); - let mut last_outcome = Outcome::success(); - - services.emitter.emit(&PipelineEvent::SubgraphStarted { - node_id: node.id.clone(), - start_node, - }); - let subgraph_start = Instant::now(); - - let max_steps: usize = 1000; - let mut steps: usize = 0; - - while steps < max_steps { - steps += 1; - - let sub_node = match sub_graph.nodes.get(¤t_node_id) { - Some(n) => n, - None => { - return Ok(Outcome::fail(format!( - "Sub-pipeline node not found: {current_node_id}" - ))); - } - }; - - // Check for terminal node - if is_terminal(sub_node) { - break; - } - - // Execute the node handler - let handler = services.registry.resolve(sub_node); - last_outcome = handler - .execute(sub_node, &sub_context, &sub_graph, &sub_logs_root, services) - .await?; - - // Apply context updates from the outcome - sub_context.apply_updates(&last_outcome.context_updates); - sub_context.set( - "outcome", - serde_json::json!(last_outcome.status.to_string()), - ); - - // Select next edge - match select_edge(¤t_node_id, &last_outcome, &sub_context, &sub_graph) { - Some(edge) => { - current_node_id.clone_from(&edge.to); - } - None => break, - } - } - - services.emitter.emit(&PipelineEvent::SubgraphCompleted { - node_id: node.id.clone(), - steps_executed: steps, - status: last_outcome.status.to_string(), - duration_ms: millis_u64(subgraph_start.elapsed()), - }); - - // 6. Compute context diff (sub_context changes vs parent's original snapshot) - let after_snapshot = sub_context.snapshot(); - let mut context_updates = std::collections::HashMap::new(); - for (key, value) in &after_snapshot { - match before_snapshot.get(key) { - Some(old_value) if old_value == value => {} - _ => { - context_updates.insert(key.clone(), value.clone()); - } - } - } - - // 7. Return the last outcome with context updates propagated - let mut result = last_outcome; - result.context_updates.extend(context_updates); - Ok(result) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::event::EventEmitter; - use crate::graph::AttrValue; - use crate::handler::exit::ExitHandler; - use crate::handler::start::StartHandler; - use crate::handler::HandlerRegistry; - use crate::outcome::StageStatus; - use std::sync::Arc; - - fn local_env() -> Arc { - Arc::new(arc_agent::LocalExecutionEnvironment::new( - std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")), - )) - } - - fn make_services() -> EngineServices { - let mut registry = HandlerRegistry::new(Box::new(StartHandler)); - registry.register("start", Box::new(StartHandler)); - registry.register("exit", Box::new(ExitHandler)); - EngineServices { - registry: Arc::new(registry), - emitter: Arc::new(EventEmitter::new()), - execution_env: local_env(), - git_state: std::sync::RwLock::new(None), - } - } - - fn make_services_with_registry(registry: HandlerRegistry) -> EngineServices { - EngineServices { - registry: Arc::new(registry), - emitter: Arc::new(EventEmitter::new()), - execution_env: local_env(), - git_state: std::sync::RwLock::new(None), - } - } - - #[tokio::test] - async fn executes_simple_sub_pipeline() { - let services = make_services(); - - let mut node = Node::new("sub"); - node.attrs.insert( - "sub_pipeline.dot_source".to_string(), - AttrValue::String( - r"digraph Sub { - start [shape=Mdiamond] - exit [shape=Msquare] - start -> exit - }" - .to_string(), - ), - ); - - let context = Context::new(); - let graph = Graph::new("parent"); - let tmp = tempfile::tempdir().unwrap(); - - let outcome = SubPipelineHandler - .execute(&node, &context, &graph, tmp.path(), &services) - .await - .unwrap(); - assert_eq!(outcome.status, StageStatus::Success); - } - - #[tokio::test] - async fn parent_context_available_in_sub_pipeline() { - let services = make_services(); - - let mut node = Node::new("sub"); - node.attrs.insert( - "sub_pipeline.dot_source".to_string(), - AttrValue::String( - r"digraph Sub { - start [shape=Mdiamond] - exit [shape=Msquare] - start -> exit - }" - .to_string(), - ), - ); - - let context = Context::new(); - context.set("parent.value", serde_json::json!("hello")); - let graph = Graph::new("parent"); - let tmp = tempfile::tempdir().unwrap(); - - let outcome = SubPipelineHandler - .execute(&node, &context, &graph, tmp.path(), &services) - .await - .unwrap(); - assert_eq!(outcome.status, StageStatus::Success); - // The sub-pipeline clones the context, so the parent value should be - // available during sub-execution. After execution, any sub-pipeline - // context updates should be in the outcome's context_updates. - } - - #[tokio::test] - async fn context_updates_propagate_back() { - // Use a handler that sets a context value, register it in the sub-pipeline registry - struct ContextSettingHandler; - - #[async_trait] - impl Handler for ContextSettingHandler { - async fn execute( - &self, - _node: &Node, - context: &Context, - _graph: &Graph, - _logs_root: &Path, - _services: &EngineServices, - ) -> Result { - context.set("sub.result", serde_json::json!("from_sub")); - Ok(Outcome::success()) - } - } - - let mut registry = HandlerRegistry::new(Box::new(ContextSettingHandler)); - registry.register("start", Box::new(StartHandler)); - registry.register("exit", Box::new(ExitHandler)); - let services = make_services_with_registry(registry); - - let mut node = Node::new("sub"); - node.attrs.insert( - "sub_pipeline.dot_source".to_string(), - AttrValue::String( - r"digraph Sub { - start [shape=Mdiamond] - work [shape=box] - exit [shape=Msquare] - start -> work -> exit - }" - .to_string(), - ), - ); - - let context = Context::new(); - let graph = Graph::new("parent"); - let tmp = tempfile::tempdir().unwrap(); - - let outcome = SubPipelineHandler - .execute(&node, &context, &graph, tmp.path(), &services) - .await - .unwrap(); - assert_eq!(outcome.status, StageStatus::Success); - - // Context updates from the sub-pipeline should be in the outcome - assert!( - outcome.context_updates.contains_key("sub.result"), - "sub-pipeline context updates should propagate back" - ); - assert_eq!( - outcome.context_updates.get("sub.result"), - Some(&serde_json::json!("from_sub")) - ); - } - - #[tokio::test] - async fn failing_sub_pipeline_returns_fail() { - struct AlwaysFailHandler; - - #[async_trait] - impl Handler for AlwaysFailHandler { - async fn execute( - &self, - _node: &Node, - _context: &Context, - _graph: &Graph, - _logs_root: &Path, - _services: &EngineServices, - ) -> Result { - Ok(Outcome::fail("sub-pipeline failure")) - } - } - - let mut registry = HandlerRegistry::new(Box::new(AlwaysFailHandler)); - registry.register("start", Box::new(StartHandler)); - registry.register("exit", Box::new(ExitHandler)); - let services = make_services_with_registry(registry); - - let mut node = Node::new("sub"); - // Sub-pipeline where the work node fails and there's a fail edge to exit - node.attrs.insert( - "sub_pipeline.dot_source".to_string(), - AttrValue::String( - r#"digraph Sub { - start [shape=Mdiamond] - work [shape=box, max_retries="0"] - exit [shape=Msquare] - start -> work - work -> exit [condition="outcome=fail"] - }"# - .to_string(), - ), - ); - - let context = Context::new(); - let graph = Graph::new("parent"); - let tmp = tempfile::tempdir().unwrap(); - - let outcome = SubPipelineHandler - .execute(&node, &context, &graph, tmp.path(), &services) - .await - .unwrap(); - assert_eq!(outcome.status, StageStatus::Fail); - } - - #[tokio::test] - async fn missing_dot_source_returns_fail() { - let services = make_services(); - - let node = Node::new("sub"); - let context = Context::new(); - let graph = Graph::new("parent"); - let tmp = tempfile::tempdir().unwrap(); - - let outcome = SubPipelineHandler - .execute(&node, &context, &graph, tmp.path(), &services) - .await - .unwrap(); - assert_eq!(outcome.status, StageStatus::Fail); - assert!( - outcome - .failure_reason - .as_deref() - .unwrap() - .contains("sub_pipeline.dot_source"), - "should mention the missing attribute" - ); - } - - #[tokio::test] - async fn invalid_dot_source_returns_fail() { - let services = make_services(); - - let mut node = Node::new("sub"); - node.attrs.insert( - "sub_pipeline.dot_source".to_string(), - AttrValue::String("not valid dot".to_string()), - ); - - let context = Context::new(); - let graph = Graph::new("parent"); - let tmp = tempfile::tempdir().unwrap(); - - let outcome = SubPipelineHandler - .execute(&node, &context, &graph, tmp.path(), &services) - .await - .unwrap(); - assert_eq!(outcome.status, StageStatus::Fail); - } -} diff --git a/crates/arc-workflows/tests/integration.rs b/crates/arc-workflows/tests/integration.rs index 5dc3dd8dc..74f2f5001 100644 --- a/crates/arc-workflows/tests/integration.rs +++ b/crates/arc-workflows/tests/integration.rs @@ -3269,68 +3269,6 @@ async fn integration_smoke_plan_implement_review_done() { .any(|e| matches!(e, PipelineEvent::PipelineCompleted { .. }))); } -// =========================================================================== -// 19a. Sub-pipeline E2E (TS Scenario 9) -// =========================================================================== - -#[tokio::test] -async fn sub_pipeline_e2e_through_engine() { - use arc_workflows::handler::sub_pipeline::SubPipelineHandler; - - let input = r#"digraph SubPipelineE2E { - graph [goal="Test sub-pipeline"] - start [shape=Mdiamond] - exit [shape=Msquare] - generate [shape=box, prompt="Generate code"] - validate [type="sub_pipeline", sub_pipeline.dot_source="digraph Child { start [shape=Mdiamond]; lint [shape=box, prompt=\"Lint\"]; test [shape=box, prompt=\"Test\"]; exit [shape=Msquare]; start -> lint -> test -> exit }"] - - start -> generate -> validate -> exit - }"#; - - let graph = parse(input).expect("parse should succeed"); - - let dir = tempfile::tempdir().unwrap(); - - let mut registry = HandlerRegistry::new(Box::new(CodergenHandler::new(None))); - registry.register("start", Box::new(StartHandler)); - registry.register("exit", Box::new(ExitHandler)); - registry.register("codergen", Box::new(CodergenHandler::new(None))); - registry.register("sub_pipeline", Box::new(SubPipelineHandler)); - - let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); - let config = RunConfig { - logs_root: dir.path().to_path_buf(), - cancel_token: None, - dry_run: false, - run_id: "test-run".into(), - git_checkpoint: None, - base_sha: None, - run_branch: None, - meta_branch: None, - labels: std::collections::HashMap::new(), - }; - - let outcome = engine - .run(&graph, &config) - .await - .expect("sub-pipeline E2E should succeed"); - assert_eq!(outcome.status, StageStatus::Success); - - let checkpoint = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap(); - assert!( - checkpoint.completed_nodes.contains(&"generate".to_string()), - "generate should be in completed_nodes" - ); - assert!( - checkpoint.completed_nodes.contains(&"validate".to_string()), - "validate should be in completed_nodes" - ); - - // Context should have last_stage set by the validate node's sub-pipeline - let last_stage = checkpoint.context_values.get("last_stage"); - assert!(last_stage.is_some(), "last_stage should be set in context"); -} - // =========================================================================== // 19b. Manager loop runs child engine E2E // ===========================================================================