diff --git a/crates/attractor/src/engine.rs b/crates/attractor/src/engine.rs index f57f991ce..e456396de 100644 --- a/crates/attractor/src/engine.rs +++ b/crates/attractor/src/engine.rs @@ -15,8 +15,7 @@ use crate::context::Context; use crate::error::{AttractorError, Result}; use crate::event::{EventEmitter, PipelineEvent}; use crate::graph::{Edge, Graph, Node}; -use crate::handler::parallel::ParallelHandler; -use crate::handler::HandlerRegistry; +use crate::handler::{EngineServices, HandlerRegistry}; use crate::interviewer::Interviewer; use crate::outcome::{Outcome, StageStatus}; use crate::preamble::build_preamble; @@ -452,23 +451,18 @@ pub struct RunConfig { /// The pipeline execution engine. pub struct PipelineEngine { - registry: Arc, - emitter: Arc, - parallel_handler: ParallelHandler, + services: EngineServices, pub interviewer: Option>, } impl PipelineEngine { #[must_use] pub fn new(registry: HandlerRegistry, emitter: EventEmitter) -> Self { - let registry = Arc::new(registry); - let emitter = Arc::new(emitter); - let parallel_handler = - ParallelHandler::new(Arc::clone(®istry), Arc::clone(&emitter)); Self { - registry, - emitter, - parallel_handler, + services: EngineServices { + registry: Arc::new(registry), + emitter: Arc::new(emitter), + }, interviewer: None, } } @@ -480,27 +474,15 @@ impl PipelineEngine { emitter: EventEmitter, interviewer: Arc, ) -> Self { - let registry = Arc::new(registry); - let emitter = Arc::new(emitter); - let parallel_handler = - ParallelHandler::new(Arc::clone(®istry), Arc::clone(&emitter)); Self { - registry, - emitter, - parallel_handler, + services: EngineServices { + registry: Arc::new(registry), + emitter: Arc::new(emitter), + }, interviewer: Some(interviewer), } } - /// Resolve the handler for a node, returning the parallel handler for - /// parallel nodes and delegating to the registry for everything else. - fn resolve_handler(&self, node: &Node) -> &dyn crate::handler::Handler { - if node.handler_type() == Some("parallel") { - return &self.parallel_handler; - } - self.registry.resolve(node) - } - /// Call inform on the interviewer, if one is configured. fn inform(&self, message: &str, stage: &str) { if let Some(ref interviewer) = self.interviewer { @@ -538,14 +520,14 @@ impl PipelineEngine { policy: &RetryPolicy, stage_index: usize, ) -> Result<(Outcome, u32)> { - let handler = self.resolve_handler(node); + let handler = self.services.registry.resolve(node); let node_timeout = node.timeout(); for attempt in 1..=policy.max_attempts { // Gap #11: Panic safety -- catch panics from handler execution let result = { - let future = handler.execute(node, context, graph, logs_root); + let future = handler.execute(node, context, graph, logs_root, &self.services); let panic_safe = AssertUnwindSafe(future).catch_unwind(); // Gap #2: Timeout enforcement -- wrap with tokio::time::timeout let timed_result = if let Some(duration) = node_timeout { @@ -582,13 +564,13 @@ impl PipelineEngine { // Gap #7: Check should_retry predicate before retrying if attempt < policy.max_attempts && handler.should_retry(&e) { let delay = policy.backoff.delay_for_attempt(attempt); - self.emitter.emit(&PipelineEvent::StageFailed { + self.services.emitter.emit(&PipelineEvent::StageFailed { name: node.label().to_string(), index: stage_index, error: e.to_string(), will_retry: true, }); - self.emitter.emit(&PipelineEvent::StageRetrying { + self.services.emitter.emit(&PipelineEvent::StageRetrying { name: node.label().to_string(), index: stage_index, attempt: usize::try_from(attempt).unwrap_or(usize::MAX), @@ -611,7 +593,7 @@ impl PipelineEngine { StageStatus::Retry => { if attempt < policy.max_attempts { let delay = policy.backoff.delay_for_attempt(attempt); - self.emitter.emit(&PipelineEvent::StageRetrying { + self.services.emitter.emit(&PipelineEvent::StageRetrying { name: node.label().to_string(), index: stage_index, attempt: usize::try_from(attempt).unwrap_or(usize::MAX), @@ -677,7 +659,7 @@ impl PipelineEngine { let run_start = Instant::now(); let run_id = uuid::Uuid::new_v4().to_string(); - self.emitter.emit(&PipelineEvent::PipelineStarted { + self.services.emitter.emit(&PipelineEvent::PipelineStarted { name: graph.name.clone(), id: run_id, }); @@ -774,7 +756,7 @@ impl PipelineEngine { let duration_ms = millis_u64(run_start.elapsed()); let error_msg = format!("goal gate unsatisfied for node {failed_node_id} and no retry target"); - self.emitter.emit(&PipelineEvent::PipelineFailed { + self.services.emitter.emit(&PipelineEvent::PipelineFailed { error: error_msg.clone(), duration_ms, }); @@ -828,7 +810,7 @@ impl PipelineEngine { context.set("current_node", serde_json::json!(&node.id)); let retry_policy = build_retry_policy(node, graph); - self.emitter.emit(&PipelineEvent::StageStarted { + self.services.emitter.emit(&PipelineEvent::StageStarted { name: node.label().to_string(), index: stage_index, }); @@ -861,7 +843,7 @@ impl PipelineEngine { let stage_duration_ms = millis_u64(stage_start.elapsed()); if outcome.status == StageStatus::Fail { - self.emitter.emit(&PipelineEvent::StageFailed { + self.services.emitter.emit(&PipelineEvent::StageFailed { name: node.label().to_string(), index: stage_index, error: outcome @@ -872,10 +854,13 @@ impl PipelineEngine { will_retry: false, }); } else { - self.emitter.emit(&PipelineEvent::StageCompleted { + self.services.emitter.emit(&PipelineEvent::StageCompleted { name: node.label().to_string(), index: stage_index, duration_ms: stage_duration_ms, + status: outcome.status.to_string(), + preferred_label: outcome.preferred_label.clone(), + suggested_next_ids: outcome.suggested_next_ids.clone(), }); self.inform( &format!("Stage completed: {}", node.label()), @@ -916,7 +901,7 @@ impl PipelineEngine { if let Err(e) = checkpoint.save(&checkpoint_path) { context.append_log(format!("checkpoint save failed: {e}")); } else { - self.emitter.emit(&PipelineEvent::CheckpointSaved { + self.services.emitter.emit(&PipelineEvent::CheckpointSaved { node_id: node.id.clone(), }); } @@ -936,7 +921,7 @@ impl PipelineEngine { "stage {} failed with no outgoing fail edge", node.id ); - self.emitter.emit(&PipelineEvent::PipelineFailed { + self.services.emitter.emit(&PipelineEvent::PipelineFailed { error: error_msg.clone(), duration_ms, }); @@ -962,7 +947,7 @@ impl PipelineEngine { } let duration_ms = millis_u64(run_start.elapsed()); - self.emitter.emit(&PipelineEvent::PipelineCompleted { + self.services.emitter.emit(&PipelineEvent::PipelineCompleted { duration_ms, artifact_count: 0, }); @@ -998,6 +983,7 @@ mod tests { _context: &Context, _graph: &Graph, _logs_root: &Path, + _services: &crate::handler::EngineServices, ) -> std::result::Result { Ok(Outcome::fail("always fails")) } @@ -1016,6 +1002,7 @@ mod tests { _context: &Context, _graph: &Graph, _logs_root: &Path, + _services: &crate::handler::EngineServices, ) -> std::result::Result { tokio::time::sleep(Duration::from_millis(self.sleep_ms)).await; Ok(Outcome::success()) diff --git a/crates/attractor/src/handler/codergen.rs b/crates/attractor/src/handler/codergen.rs index 146055a3e..b2440a212 100644 --- a/crates/attractor/src/handler/codergen.rs +++ b/crates/attractor/src/handler/codergen.rs @@ -7,7 +7,7 @@ use crate::error::AttractorError; use crate::graph::{Graph, Node}; use crate::outcome::Outcome; -use super::Handler; +use super::{EngineServices, Handler}; /// Result from a `CodergenBackend` invocation. pub enum CodergenResult { @@ -172,6 +172,7 @@ impl Handler for CodergenHandler { context: &Context, graph: &Graph, logs_root: &Path, + _services: &EngineServices, ) -> Result { // 1. Build prompt let raw_prompt = node @@ -257,9 +258,19 @@ impl Handler for CodergenHandler { #[cfg(test)] mod tests { use super::*; + use crate::event::EventEmitter; use crate::graph::AttrValue; + use crate::handler::start::StartHandler; + use crate::handler::HandlerRegistry; use tempfile::TempDir; + fn make_services() -> EngineServices { + EngineServices { + registry: std::sync::Arc::new(HandlerRegistry::new(Box::new(StartHandler))), + emitter: std::sync::Arc::new(EventEmitter::new()), + } + } + #[tokio::test] async fn codergen_handler_simulation_mode() { let handler = CodergenHandler::new(None); @@ -273,7 +284,7 @@ mod tests { let tmp = TempDir::new().unwrap(); let outcome = handler - .execute(&node, &context, &graph, tmp.path()) + .execute(&node, &context, &graph, tmp.path(), &make_services()) .await .unwrap(); assert_eq!(outcome.status, crate::outcome::StageStatus::Success); @@ -311,7 +322,7 @@ mod tests { let tmp = TempDir::new().unwrap(); handler - .execute(&node, &context, &graph, tmp.path()) + .execute(&node, &context, &graph, tmp.path(), &make_services()) .await .unwrap(); @@ -333,7 +344,7 @@ mod tests { let tmp = TempDir::new().unwrap(); handler - .execute(&node, &context, &graph, tmp.path()) + .execute(&node, &context, &graph, tmp.path(), &make_services()) .await .unwrap(); @@ -351,7 +362,7 @@ mod tests { let tmp = TempDir::new().unwrap(); let outcome = handler - .execute(&node, &context, &graph, tmp.path()) + .execute(&node, &context, &graph, tmp.path(), &make_services()) .await .unwrap(); @@ -397,7 +408,7 @@ mod tests { let tmp = TempDir::new().unwrap(); let outcome = handler - .execute(&node, &context, &graph, tmp.path()) + .execute(&node, &context, &graph, tmp.path(), &make_services()) .await .unwrap(); assert_eq!(outcome.status, crate::outcome::StageStatus::Skipped); @@ -421,7 +432,7 @@ mod tests { let tmp = TempDir::new().unwrap(); let outcome = handler - .execute(&node, &context, &graph, tmp.path()) + .execute(&node, &context, &graph, tmp.path(), &make_services()) .await .unwrap(); assert_eq!(outcome.status, crate::outcome::StageStatus::Success); @@ -440,7 +451,7 @@ mod tests { let tmp = TempDir::new().unwrap(); let outcome = handler - .execute(&node, &context, &graph, tmp.path()) + .execute(&node, &context, &graph, tmp.path(), &make_services()) .await .unwrap(); // Post-hook failure should not fail the node @@ -519,7 +530,7 @@ mod tests { let tmp = TempDir::new().unwrap(); handler - .execute(&node, &context, &graph, tmp.path()) + .execute(&node, &context, &graph, tmp.path(), &make_services()) .await .unwrap(); @@ -563,7 +574,7 @@ mod tests { let tmp = TempDir::new().unwrap(); handler - .execute(&node, &context, &graph, tmp.path()) + .execute(&node, &context, &graph, tmp.path(), &make_services()) .await .unwrap(); @@ -594,7 +605,7 @@ mod tests { let graph = Graph::new("test"); let tmp = TempDir::new().unwrap(); - let result = handler.execute(&node, &context, &graph, tmp.path()).await; + let result = handler.execute(&node, &context, &graph, tmp.path(), &make_services()).await; let err = result.unwrap_err(); assert!(err.is_retryable()); assert!(err.to_string().contains("Request timed out")); @@ -706,7 +717,7 @@ Some text in between. let tmp = TempDir::new().unwrap(); let outcome = handler - .execute(&node, &context, &graph, tmp.path()) + .execute(&node, &context, &graph, tmp.path(), &make_services()) .await .unwrap(); assert_eq!(outcome.status, crate::outcome::StageStatus::Fail); diff --git a/crates/attractor/src/handler/conditional.rs b/crates/attractor/src/handler/conditional.rs index 4e1251655..68b4e24dc 100644 --- a/crates/attractor/src/handler/conditional.rs +++ b/crates/attractor/src/handler/conditional.rs @@ -7,7 +7,7 @@ use crate::error::AttractorError; use crate::graph::{Graph, Node}; use crate::outcome::Outcome; -use super::Handler; +use super::{EngineServices, Handler}; /// Conditional routing handler. Returns SUCCESS with a note; actual routing /// is handled by the engine's edge selection algorithm. @@ -21,6 +21,7 @@ impl Handler for ConditionalHandler { _context: &Context, _graph: &Graph, _logs_root: &Path, + _services: &EngineServices, ) -> Result { let mut outcome = Outcome::success(); outcome.notes = Some(format!("Conditional node evaluated: {}", node.id)); @@ -31,6 +32,16 @@ impl Handler for ConditionalHandler { #[cfg(test)] mod tests { use super::*; + use crate::event::EventEmitter; + use crate::handler::start::StartHandler; + use crate::handler::HandlerRegistry; + + fn make_services() -> EngineServices { + EngineServices { + registry: std::sync::Arc::new(HandlerRegistry::new(Box::new(StartHandler))), + emitter: std::sync::Arc::new(EventEmitter::new()), + } + } #[tokio::test] async fn conditional_handler_returns_success_with_note() { @@ -40,7 +51,7 @@ mod tests { let graph = Graph::new("test"); let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(&node, &context, &graph, logs_root) + .execute(&node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, crate::outcome::StageStatus::Success); diff --git a/crates/attractor/src/handler/exit.rs b/crates/attractor/src/handler/exit.rs index 97ac5fd88..c00b28e77 100644 --- a/crates/attractor/src/handler/exit.rs +++ b/crates/attractor/src/handler/exit.rs @@ -7,7 +7,7 @@ use crate::error::AttractorError; use crate::graph::{Graph, Node}; use crate::outcome::Outcome; -use super::Handler; +use super::{EngineServices, Handler}; /// No-op handler for pipeline exit point. Returns SUCCESS immediately. pub struct ExitHandler; @@ -20,6 +20,7 @@ impl Handler for ExitHandler { _context: &Context, _graph: &Graph, _logs_root: &Path, + _services: &EngineServices, ) -> Result { Ok(Outcome::success()) } @@ -28,6 +29,16 @@ impl Handler for ExitHandler { #[cfg(test)] mod tests { use super::*; + use crate::event::EventEmitter; + use crate::handler::start::StartHandler; + use crate::handler::HandlerRegistry; + + fn make_services() -> EngineServices { + EngineServices { + registry: std::sync::Arc::new(HandlerRegistry::new(Box::new(StartHandler))), + emitter: std::sync::Arc::new(EventEmitter::new()), + } + } #[tokio::test] async fn exit_handler_returns_success() { @@ -37,7 +48,7 @@ mod tests { let graph = Graph::new("test"); let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(&node, &context, &graph, logs_root) + .execute(&node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, crate::outcome::StageStatus::Success); diff --git a/crates/attractor/src/handler/fan_in.rs b/crates/attractor/src/handler/fan_in.rs index 40cd41d64..5a6075bbb 100644 --- a/crates/attractor/src/handler/fan_in.rs +++ b/crates/attractor/src/handler/fan_in.rs @@ -8,7 +8,7 @@ use crate::graph::{Graph, Node}; use crate::outcome::Outcome; use super::codergen::{CodergenBackend, CodergenResult}; -use super::Handler; +use super::{EngineServices, Handler}; /// Consolidates results from a preceding parallel node and selects the best candidate. pub struct FanInHandler { @@ -30,6 +30,7 @@ impl Handler for FanInHandler { context: &Context, _graph: &Graph, logs_root: &Path, + _services: &EngineServices, ) -> Result { let results = context.get("parallel.results"); let Some(results) = results else { @@ -233,8 +234,18 @@ async fn llm_evaluate( #[cfg(test)] mod tests { use super::*; + use crate::event::EventEmitter; + use crate::handler::start::StartHandler; + use crate::handler::HandlerRegistry; use crate::outcome::StageStatus; + fn make_services() -> EngineServices { + EngineServices { + registry: std::sync::Arc::new(HandlerRegistry::new(Box::new(StartHandler))), + emitter: std::sync::Arc::new(EventEmitter::new()), + } + } + #[tokio::test] async fn fan_in_no_results() { let handler = FanInHandler::new(None); @@ -244,7 +255,7 @@ mod tests { let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(&node, &context, &graph, logs_root) + .execute(&node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Fail); @@ -266,7 +277,7 @@ mod tests { let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(&node, &context, &graph, logs_root) + .execute(&node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Success); @@ -293,7 +304,7 @@ mod tests { let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(&node, &context, &graph, logs_root) + .execute(&node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!( @@ -330,7 +341,7 @@ mod tests { let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(&node, &context, &graph, logs_root) + .execute(&node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Success); @@ -380,7 +391,7 @@ mod tests { let tmp = TempDir::new().unwrap(); let outcome = handler - .execute(&node, &context, &graph, tmp.path()) + .execute(&node, &context, &graph, tmp.path(), &make_services()) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Success); @@ -419,7 +430,7 @@ mod tests { let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(&node, &context, &graph, logs_root) + .execute(&node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Fail); @@ -447,7 +458,7 @@ mod tests { let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(&node, &context, &graph, logs_root) + .execute(&node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Success); diff --git a/crates/attractor/src/handler/manager_loop.rs b/crates/attractor/src/handler/manager_loop.rs index 9064a7967..393ff9700 100644 --- a/crates/attractor/src/handler/manager_loop.rs +++ b/crates/attractor/src/handler/manager_loop.rs @@ -9,7 +9,7 @@ use crate::error::AttractorError; use crate::graph::{Graph, Node}; use crate::outcome::{Outcome, StageStatus}; -use super::Handler; +use super::{EngineServices, Handler}; /// Trait for observing child pipeline state during the manager loop. #[async_trait] @@ -73,6 +73,7 @@ impl Handler for ManagerLoopHandler { context: &Context, _graph: &Graph, _logs_root: &Path, + _services: &EngineServices, ) -> Result { let poll_interval = node .attrs @@ -209,7 +210,17 @@ impl Handler for ManagerLoopHandler { #[cfg(test)] mod tests { use super::*; + use crate::event::EventEmitter; use crate::graph::AttrValue; + use crate::handler::start::StartHandler; + use crate::handler::HandlerRegistry; + + fn make_services() -> EngineServices { + EngineServices { + registry: std::sync::Arc::new(HandlerRegistry::new(Box::new(StartHandler))), + emitter: std::sync::Arc::new(EventEmitter::new()), + } + } #[tokio::test] async fn manager_loop_max_cycles_exceeded() { @@ -228,7 +239,7 @@ mod tests { let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(&node, &context, &graph, logs_root) + .execute(&node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Fail); @@ -265,7 +276,7 @@ mod tests { let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(&node, &context, &graph, logs_root) + .execute(&node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Success); @@ -293,7 +304,7 @@ mod tests { let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(&node, &context, &graph, logs_root) + .execute(&node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Fail); @@ -326,7 +337,7 @@ mod tests { let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(&node, &context, &graph, logs_root) + .execute(&node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Success); diff --git a/crates/attractor/src/handler/mod.rs b/crates/attractor/src/handler/mod.rs index b96615e49..45b617689 100644 --- a/crates/attractor/src/handler/mod.rs +++ b/crates/attractor/src/handler/mod.rs @@ -17,10 +17,17 @@ use async_trait::async_trait; use crate::context::Context; use crate::error::AttractorError; +use crate::event::EventEmitter; use crate::graph::{shape_to_handler_type, Graph, Node}; use crate::interviewer::Interviewer; use crate::outcome::Outcome; +/// Shared services available to all handlers during execution. +pub struct EngineServices { + pub registry: Arc, + pub emitter: Arc, +} + /// The handler interface for node execution. #[async_trait] pub trait Handler: Send + Sync { @@ -30,6 +37,7 @@ pub trait Handler: Send + Sync { context: &Context, graph: &Graph, logs_root: &Path, + services: &EngineServices, ) -> Result; /// Determines whether an error should be retried. @@ -105,10 +113,12 @@ pub fn default_registry( Box::new(wait_human::WaitHumanHandler::new(interviewer)), ); registry.register("tool", Box::new(tool::ToolHandler)); + registry.register("parallel", Box::new(parallel::ParallelHandler)); registry.register( "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::new(None)), @@ -133,6 +143,7 @@ mod tests { _context: &Context, _graph: &Graph, _logs_root: &Path, + _services: &EngineServices, ) -> Result { Ok(Outcome::success()) } @@ -211,6 +222,7 @@ mod tests { _context: &Context, _graph: &Graph, _logs_root: &Path, + _services: &EngineServices, ) -> Result { Ok(Outcome::success()) } diff --git a/crates/attractor/src/handler/parallel.rs b/crates/attractor/src/handler/parallel.rs index 78ec1dc2a..b6fba9985 100644 --- a/crates/attractor/src/handler/parallel.rs +++ b/crates/attractor/src/handler/parallel.rs @@ -7,11 +7,11 @@ use tokio::sync::Semaphore; use crate::context::Context; use crate::error::AttractorError; -use crate::event::{EventEmitter, PipelineEvent}; +use crate::event::PipelineEvent; use crate::graph::{Graph, Node}; use crate::outcome::{Outcome, StageStatus}; -use super::{Handler, HandlerRegistry}; +use super::{EngineServices, Handler}; /// Convert a Duration's milliseconds to u64, saturating on overflow. fn millis_u64(d: std::time::Duration) -> u64 { @@ -20,17 +20,7 @@ fn millis_u64(d: std::time::Duration) -> u64 { /// Fans out execution to multiple branches concurrently. /// Each branch gets an isolated context clone and runs independently. -pub struct ParallelHandler { - registry: Arc, - emitter: Arc, -} - -impl ParallelHandler { - #[must_use] - pub fn new(registry: Arc, emitter: Arc) -> Self { - Self { registry, emitter } - } -} +pub struct ParallelHandler; /// Parse join policy from node attributes. #[derive(Debug, Clone)] @@ -87,6 +77,7 @@ impl Handler for ParallelHandler { context: &Context, graph: &Graph, logs_root: &Path, + services: &EngineServices, ) -> Result { let parallel_start = Instant::now(); let branches = graph.outgoing_edges(&node.id); @@ -94,7 +85,7 @@ impl Handler for ParallelHandler { return Ok(Outcome::fail("No branches for parallel node")); } - self.emitter.emit(&PipelineEvent::ParallelStarted { + services.emitter.emit(&PipelineEvent::ParallelStarted { branch_count: branches.len(), }); @@ -124,8 +115,8 @@ impl Handler for ParallelHandler { for (branch_index, edge) in branches.iter().enumerate() { let target_id = edge.to.clone(); let branch_context = context.clone_context(); - let registry = Arc::clone(&self.registry); - let emitter = Arc::clone(&self.emitter); + let registry = Arc::clone(&services.registry); + let emitter = Arc::clone(&services.emitter); let graph = graph.clone(); let logs_root = logs_root.to_path_buf(); let sem = Arc::clone(&semaphore); @@ -155,9 +146,13 @@ impl Handler for ParallelHandler { }); }; + let branch_services = EngineServices { + registry: Arc::clone(®istry), + emitter: Arc::clone(&emitter), + }; let handler = registry.resolve(target_node); let outcome = handler - .execute(target_node, &branch_context, &graph, &logs_root) + .execute(target_node, &branch_context, &graph, &logs_root, &branch_services) .await?; let success = outcome.status == StageStatus::Success @@ -239,7 +234,7 @@ impl Handler for ParallelHandler { context.set("parallel.results", serde_json::json!(results_json)); context.set("parallel.branch_count", serde_json::json!(total)); - self.emitter.emit(&PipelineEvent::ParallelCompleted { + services.emitter.emit(&PipelineEvent::ParallelCompleted { duration_ms: millis_u64(parallel_start.elapsed()), success_count, failure_count: fail_count, @@ -315,29 +310,29 @@ impl Handler for ParallelHandler { #[cfg(test)] mod tests { use super::*; + use crate::event::EventEmitter; use crate::graph::{AttrValue, Edge}; use crate::handler::start::StartHandler; + use crate::handler::HandlerRegistry; - fn make_registry() -> Arc { + fn make_services() -> EngineServices { let registry = HandlerRegistry::new(Box::new(StartHandler)); - Arc::new(registry) - } - - fn make_emitter() -> Arc { - Arc::new(EventEmitter::new()) + EngineServices { + registry: Arc::new(registry), + emitter: Arc::new(EventEmitter::new()), + } } #[tokio::test] async fn parallel_handler_no_branches() { - let registry = make_registry(); - let handler = ParallelHandler::new(registry, make_emitter()); + let services = make_services(); let node = Node::new("par"); let context = Context::new(); let graph = Graph::new("test"); let logs_root = Path::new("/tmp/test"); - let outcome = handler - .execute(&node, &context, &graph, logs_root) + let outcome = ParallelHandler + .execute(&node, &context, &graph, logs_root, &services) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Fail); @@ -345,8 +340,7 @@ mod tests { #[tokio::test] async fn parallel_handler_with_branches() { - let registry = make_registry(); - let handler = ParallelHandler::new(registry, make_emitter()); + let services = make_services(); let mut node = Node::new("par"); node.attrs.insert( "shape".to_string(), @@ -365,8 +359,8 @@ mod tests { graph.edges.push(Edge::new("par", "branch_b")); let logs_root = Path::new("/tmp/test"); - let outcome = handler - .execute(&node, &context, &graph, logs_root) + let outcome = ParallelHandler + .execute(&node, &context, &graph, logs_root, &services) .await .unwrap(); @@ -380,8 +374,7 @@ mod tests { #[tokio::test] async fn parallel_handler_first_success_policy() { - let registry = make_registry(); - let handler = ParallelHandler::new(registry, make_emitter()); + let services = make_services(); let mut node = Node::new("par"); node.attrs.insert( "join_policy".to_string(), @@ -396,8 +389,8 @@ mod tests { graph.edges.push(Edge::new("par", "branch_a")); let logs_root = Path::new("/tmp/test"); - let outcome = handler - .execute(&node, &context, &graph, logs_root) + let outcome = ParallelHandler + .execute(&node, &context, &graph, logs_root, &services) .await .unwrap(); @@ -406,8 +399,7 @@ mod tests { #[tokio::test] async fn parallel_handler_k_of_n_policy() { - let registry = make_registry(); - let handler = ParallelHandler::new(registry, make_emitter()); + let services = make_services(); let mut node = Node::new("par"); node.attrs.insert( "join_policy".to_string(), @@ -430,8 +422,8 @@ mod tests { graph.edges.push(Edge::new("par", "branch_c")); let logs_root = Path::new("/tmp/test"); - let outcome = handler - .execute(&node, &context, &graph, logs_root) + let outcome = ParallelHandler + .execute(&node, &context, &graph, logs_root, &services) .await .unwrap(); diff --git a/crates/attractor/src/handler/start.rs b/crates/attractor/src/handler/start.rs index 46ca51bff..82d715401 100644 --- a/crates/attractor/src/handler/start.rs +++ b/crates/attractor/src/handler/start.rs @@ -7,7 +7,7 @@ use crate::error::AttractorError; use crate::graph::{Graph, Node}; use crate::outcome::Outcome; -use super::Handler; +use super::{EngineServices, Handler}; /// No-op handler for pipeline entry point. Returns SUCCESS immediately. pub struct StartHandler; @@ -20,6 +20,7 @@ impl Handler for StartHandler { _context: &Context, _graph: &Graph, _logs_root: &Path, + _services: &EngineServices, ) -> Result { Ok(Outcome::success()) } @@ -28,6 +29,15 @@ impl Handler for StartHandler { #[cfg(test)] mod tests { use super::*; + use crate::event::EventEmitter; + use crate::handler::HandlerRegistry; + + fn make_services() -> EngineServices { + EngineServices { + registry: std::sync::Arc::new(HandlerRegistry::new(Box::new(StartHandler))), + emitter: std::sync::Arc::new(EventEmitter::new()), + } + } #[tokio::test] async fn start_handler_returns_success() { @@ -37,7 +47,7 @@ mod tests { let graph = Graph::new("test"); let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(&node, &context, &graph, logs_root) + .execute(&node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, crate::outcome::StageStatus::Success); diff --git a/crates/attractor/src/handler/sub_pipeline.rs b/crates/attractor/src/handler/sub_pipeline.rs index 1cb8cc802..e533faec2 100644 --- a/crates/attractor/src/handler/sub_pipeline.rs +++ b/crates/attractor/src/handler/sub_pipeline.rs @@ -1,34 +1,19 @@ use std::path::Path; -use std::sync::Arc; use async_trait::async_trait; use crate::context::Context; use crate::engine::select_edge; use crate::error::AttractorError; -use crate::event::EventEmitter; use crate::graph::{Graph, Node}; use crate::outcome::Outcome; use crate::pipeline::prepare_pipeline; -use super::{Handler, HandlerRegistry}; +use super::{EngineServices, Handler}; /// 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 { - registry: Arc, - _emitter: Arc, -} - -impl SubPipelineHandler { - #[must_use] - pub fn new(registry: Arc, emitter: Arc) -> Self { - Self { - registry, - _emitter: emitter, - } - } -} +pub struct SubPipelineHandler; /// Check whether a node is a terminal (exit) node. fn is_terminal(node: &Node) -> bool { @@ -43,6 +28,7 @@ impl Handler for SubPipelineHandler { 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()) { @@ -92,9 +78,9 @@ impl Handler for SubPipelineHandler { } // Execute the node handler - let handler = self.registry.resolve(sub_node); + let handler = services.registry.resolve(sub_node); last_outcome = handler - .execute(sub_node, &sub_context, &sub_graph, &sub_logs_root) + .execute(sub_node, &sub_context, &sub_graph, &sub_logs_root, services) .await?; // Apply context updates from the outcome @@ -132,26 +118,34 @@ impl Handler for SubPipelineHandler { #[cfg(test)] mod tests { use super::*; + use std::sync::Arc; + 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; - fn make_registry() -> Arc { + fn make_services() -> EngineServices { let mut registry = HandlerRegistry::new(Box::new(StartHandler)); registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - Arc::new(registry) + EngineServices { + registry: Arc::new(registry), + emitter: Arc::new(EventEmitter::new()), + } } - fn make_emitter() -> Arc { - Arc::new(EventEmitter::new()) + fn make_services_with_registry(registry: HandlerRegistry) -> EngineServices { + EngineServices { + registry: Arc::new(registry), + emitter: Arc::new(EventEmitter::new()), + } } #[tokio::test] async fn executes_simple_sub_pipeline() { - let registry = make_registry(); - let handler = SubPipelineHandler::new(registry, make_emitter()); + let services = make_services(); let mut node = Node::new("sub"); node.attrs.insert( @@ -170,8 +164,8 @@ mod tests { let graph = Graph::new("parent"); let tmp = tempfile::tempdir().unwrap(); - let outcome = handler - .execute(&node, &context, &graph, tmp.path()) + let outcome = SubPipelineHandler + .execute(&node, &context, &graph, tmp.path(), &services) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Success); @@ -179,8 +173,7 @@ mod tests { #[tokio::test] async fn parent_context_available_in_sub_pipeline() { - let registry = make_registry(); - let handler = SubPipelineHandler::new(registry, make_emitter()); + let services = make_services(); let mut node = Node::new("sub"); node.attrs.insert( @@ -200,8 +193,8 @@ mod tests { let graph = Graph::new("parent"); let tmp = tempfile::tempdir().unwrap(); - let outcome = handler - .execute(&node, &context, &graph, tmp.path()) + let outcome = SubPipelineHandler + .execute(&node, &context, &graph, tmp.path(), &services) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Success); @@ -223,6 +216,7 @@ mod tests { context: &Context, _graph: &Graph, _logs_root: &Path, + _services: &EngineServices, ) -> Result { context.set("sub.result", serde_json::json!("from_sub")); Ok(Outcome::success()) @@ -232,9 +226,7 @@ mod tests { let mut registry = HandlerRegistry::new(Box::new(ContextSettingHandler)); registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let registry = Arc::new(registry); - - let handler = SubPipelineHandler::new(registry, make_emitter()); + let services = make_services_with_registry(registry); let mut node = Node::new("sub"); node.attrs.insert( @@ -254,8 +246,8 @@ mod tests { let graph = Graph::new("parent"); let tmp = tempfile::tempdir().unwrap(); - let outcome = handler - .execute(&node, &context, &graph, tmp.path()) + let outcome = SubPipelineHandler + .execute(&node, &context, &graph, tmp.path(), &services) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Success); @@ -283,6 +275,7 @@ mod tests { _context: &Context, _graph: &Graph, _logs_root: &Path, + _services: &EngineServices, ) -> Result { Ok(Outcome::fail("sub-pipeline failure")) } @@ -291,9 +284,7 @@ mod tests { let mut registry = HandlerRegistry::new(Box::new(AlwaysFailHandler)); registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let registry = Arc::new(registry); - - let handler = SubPipelineHandler::new(registry, make_emitter()); + 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 @@ -315,8 +306,8 @@ mod tests { let graph = Graph::new("parent"); let tmp = tempfile::tempdir().unwrap(); - let outcome = handler - .execute(&node, &context, &graph, tmp.path()) + let outcome = SubPipelineHandler + .execute(&node, &context, &graph, tmp.path(), &services) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Fail); @@ -324,16 +315,15 @@ mod tests { #[tokio::test] async fn missing_dot_source_returns_fail() { - let registry = make_registry(); - let handler = SubPipelineHandler::new(registry, make_emitter()); + 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 = handler - .execute(&node, &context, &graph, tmp.path()) + let outcome = SubPipelineHandler + .execute(&node, &context, &graph, tmp.path(), &services) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Fail); @@ -349,8 +339,7 @@ mod tests { #[tokio::test] async fn invalid_dot_source_returns_fail() { - let registry = make_registry(); - let handler = SubPipelineHandler::new(registry, make_emitter()); + let services = make_services(); let mut node = Node::new("sub"); node.attrs.insert( @@ -362,8 +351,8 @@ mod tests { let graph = Graph::new("parent"); let tmp = tempfile::tempdir().unwrap(); - let outcome = handler - .execute(&node, &context, &graph, tmp.path()) + let outcome = SubPipelineHandler + .execute(&node, &context, &graph, tmp.path(), &services) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Fail); diff --git a/crates/attractor/src/handler/tool.rs b/crates/attractor/src/handler/tool.rs index 0864cb57e..b785432e4 100644 --- a/crates/attractor/src/handler/tool.rs +++ b/crates/attractor/src/handler/tool.rs @@ -7,7 +7,7 @@ use crate::error::AttractorError; use crate::graph::{Graph, Node}; use crate::outcome::Outcome; -use super::Handler; +use super::{EngineServices, Handler}; /// Executes an external tool (shell command) configured via node attributes. pub struct ToolHandler; @@ -47,6 +47,7 @@ impl Handler for ToolHandler { _context: &Context, _graph: &Graph, _logs_root: &Path, + _services: &EngineServices, ) -> Result { let command = node .attrs @@ -87,10 +88,20 @@ impl Handler for ToolHandler { #[cfg(test)] mod tests { use super::*; + use crate::event::EventEmitter; use crate::graph::AttrValue; + use crate::handler::start::StartHandler; + use crate::handler::HandlerRegistry; use crate::outcome::StageStatus; use std::time::Duration; + fn make_services() -> EngineServices { + EngineServices { + registry: std::sync::Arc::new(HandlerRegistry::new(Box::new(StartHandler))), + emitter: std::sync::Arc::new(EventEmitter::new()), + } + } + #[tokio::test] async fn tool_handler_no_command() { let handler = ToolHandler; @@ -100,7 +111,7 @@ mod tests { let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(&node, &context, &graph, logs_root) + .execute(&node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Fail); @@ -123,7 +134,7 @@ mod tests { let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(&node, &context, &graph, logs_root) + .execute(&node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Success); @@ -145,7 +156,7 @@ mod tests { let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(&node, &context, &graph, logs_root) + .execute(&node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Fail); @@ -168,7 +179,7 @@ mod tests { let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(&node, &context, &graph, logs_root) + .execute(&node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, StageStatus::Fail); diff --git a/crates/attractor/src/handler/wait_human.rs b/crates/attractor/src/handler/wait_human.rs index 2750db2fa..c9e9048ea 100644 --- a/crates/attractor/src/handler/wait_human.rs +++ b/crates/attractor/src/handler/wait_human.rs @@ -13,7 +13,7 @@ use crate::interviewer::{ }; use crate::outcome::Outcome; -use super::Handler; +use super::{EngineServices, Handler}; /// Convert a Duration's milliseconds to u64, saturating on overflow. fn millis_u64(d: std::time::Duration) -> u64 { @@ -105,6 +105,7 @@ impl Handler for WaitHumanHandler { _context: &Context, graph: &Graph, _logs_root: &Path, + _services: &EngineServices, ) -> Result { // 1. Derive choices from outgoing edges let edges = graph.outgoing_edges(&node.id); @@ -276,9 +277,19 @@ fn answer_text(answer: &Answer) -> String { #[cfg(test)] mod tests { use super::*; + use crate::event::EventEmitter; use crate::graph::{AttrValue, Edge}; + use crate::handler::start::StartHandler; + use crate::handler::HandlerRegistry; use crate::interviewer::auto_approve::AutoApproveInterviewer; + fn make_services() -> EngineServices { + EngineServices { + registry: std::sync::Arc::new(HandlerRegistry::new(Box::new(StartHandler))), + emitter: std::sync::Arc::new(EventEmitter::new()), + } + } + fn build_graph_with_human_gate() -> Graph { let mut graph = Graph::new("test"); let mut gate = Node::new("gate"); @@ -345,7 +356,7 @@ mod tests { let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(node, &context, &graph, logs_root) + .execute(node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, crate::outcome::StageStatus::Success); @@ -369,7 +380,7 @@ mod tests { let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(node, &context, &graph, logs_root) + .execute(node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, crate::outcome::StageStatus::Fail); @@ -401,7 +412,7 @@ mod tests { let logs_root = Path::new("/tmp/test"); let outcome = handler - .execute(node, &context, &graph, logs_root) + .execute(node, &context, &graph, logs_root, &make_services()) .await .unwrap(); assert_eq!(outcome.status, crate::outcome::StageStatus::Success); diff --git a/crates/attractor/tests/integration.rs b/crates/attractor/tests/integration.rs index 5361e8758..6c288ee8f 100644 --- a/crates/attractor/tests/integration.rs +++ b/crates/attractor/tests/integration.rs @@ -431,6 +431,7 @@ impl Handler for AlwaysFailHandler { _context: &attractor::context::Context, _graph: &Graph, _logs_root: &Path, + _services: &attractor::handler::EngineServices, ) -> Result { Ok(Outcome::fail(format!("forced failure for {}", node.id))) } @@ -533,6 +534,7 @@ async fn goal_gate_routes_to_retry_target_when_present() { _context: &attractor::context::Context, _graph: &Graph, _logs_root: &Path, + _services: &attractor::handler::EngineServices, ) -> Result { let count = self .call_count @@ -849,6 +851,7 @@ async fn retry_on_failure_then_succeed() { _context: &Context, _graph: &Graph, _logs_root: &Path, + _services: &attractor::handler::EngineServices, ) -> Result { let count = self .call_count @@ -1071,6 +1074,7 @@ impl Handler for CounterHandler { _context: &Context, _graph: &Graph, _logs_root: &Path, + _services: &attractor::handler::EngineServices, ) -> Result { let count = self .call_count @@ -1094,6 +1098,7 @@ impl Handler for ContextSetterHandler { _context: &Context, _graph: &Graph, _logs_root: &Path, + _services: &attractor::handler::EngineServices, ) -> Result { let mut outcome = Outcome::success(); outcome @@ -1295,7 +1300,6 @@ async fn smoke_test_with_mock_codergen_backend() { async fn end_to_end_parallel_fan_out_fan_in() { use attractor::handler::fan_in::FanInHandler; use attractor::handler::parallel::ParallelHandler; - use std::sync::Arc; let input = r#"digraph parallel_test { start [shape=Mdiamond] @@ -1327,27 +1331,13 @@ async fn end_to_end_parallel_fan_out_fan_in() { "codergen", Box::new(CodergenHandler::new(Some(Box::new(MockCodergenBackend)))), ); - - let registry = Arc::new(registry); - - let emitter = Arc::new(EventEmitter::new()); - let parallel_handler = ParallelHandler::new(Arc::clone(®istry), Arc::clone(&emitter)); - let fan_in_handler = FanInHandler::new(Some(Box::new(MockCodergenBackend))); - - // Build a new registry with parallel and fan_in registered - let mut full_registry = HandlerRegistry::new( - Box::new(CodergenHandler::new(Some(Box::new(MockCodergenBackend)))), + registry.register("parallel", Box::new(ParallelHandler)); + registry.register( + "parallel.fan_in", + Box::new(FanInHandler::new(Some(Box::new(MockCodergenBackend)))), ); - full_registry.register("start", Box::new(StartHandler)); - full_registry.register("exit", Box::new(ExitHandler)); - full_registry.register( - "codergen", - Box::new(CodergenHandler::new(Some(Box::new(MockCodergenBackend)))), - ); - full_registry.register("parallel", Box::new(parallel_handler)); - full_registry.register("parallel.fan_in", Box::new(fan_in_handler)); - let engine = PipelineEngine::new(full_registry, EventEmitter::new()); + let engine = PipelineEngine::new(registry, EventEmitter::new()); let config = RunConfig { logs_root: dir.path().to_path_buf(), cancel_token: None, @@ -1818,6 +1808,7 @@ async fn branching_loop_back_on_failure() { _context: &Context, _graph: &Graph, _logs_root: &Path, + _services: &attractor::handler::EngineServices, ) -> Result { let count = self .call_count @@ -2059,39 +2050,27 @@ async fn scenario_parallel_expert_review() { let recorder = Arc::new(RecordingInterviewer::new(Box::new(AutoApproveInterviewer))); let dir = tempfile::tempdir().unwrap(); - let mut base_registry = HandlerRegistry::new(Box::new(CodergenHandler::new(Some( - Box::new(MockCodergenBackend), - )))); - base_registry.register("start", Box::new(StartHandler)); - base_registry.register("exit", Box::new(ExitHandler)); - base_registry.register( - "codergen", - Box::new(CodergenHandler::new(Some(Box::new(MockCodergenBackend)))), - ); - let base_registry = Arc::new(base_registry); - let emitter = Arc::new(EventEmitter::new()); - let parallel_handler = - ParallelHandler::new(Arc::clone(&base_registry), Arc::clone(&emitter)); - let fan_in_handler = FanInHandler::new(Some(Box::new(MockCodergenBackend))); - let interviewer: Arc = recorder.clone(); - let mut full_registry = HandlerRegistry::new(Box::new(CodergenHandler::new(Some( + let mut registry = HandlerRegistry::new(Box::new(CodergenHandler::new(Some( Box::new(MockCodergenBackend), )))); - full_registry.register("start", Box::new(StartHandler)); - full_registry.register("exit", Box::new(ExitHandler)); - full_registry.register( + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register( "codergen", Box::new(CodergenHandler::new(Some(Box::new(MockCodergenBackend)))), ); - full_registry.register("parallel", Box::new(parallel_handler)); - full_registry.register("parallel.fan_in", Box::new(fan_in_handler)); - full_registry.register( + registry.register("parallel", Box::new(ParallelHandler)); + registry.register( + "parallel.fan_in", + Box::new(FanInHandler::new(Some(Box::new(MockCodergenBackend)))), + ); + registry.register( "wait.human", Box::new(WaitHumanHandler::new(interviewer)), ); - let engine = PipelineEngine::new(full_registry, EventEmitter::new()); + let engine = PipelineEngine::new(registry, EventEmitter::new()); let config = RunConfig { logs_root: dir.path().to_path_buf(), cancel_token: None, @@ -2125,6 +2104,7 @@ async fn scenario_node_retries_on_retry_status() { _context: &Context, _graph: &Graph, _logs_root: &Path, + _services: &attractor::handler::EngineServices, ) -> Result { let count = self .call_count @@ -2349,6 +2329,7 @@ async fn manager_loop_stop_condition_satisfied_e2e() { _context: &Context, _graph: &Graph, _logs_root: &Path, + _services: &attractor::handler::EngineServices, ) -> Result { let mut outcome = Outcome::success(); outcome @@ -2780,6 +2761,7 @@ async fn custom_handler_registration_and_execution() { _context: &Context, _graph: &Graph, _logs_root: &Path, + _services: &attractor::handler::EngineServices, ) -> Result { let mut outcome = Outcome::success(); outcome @@ -3344,24 +3326,11 @@ async fn sub_pipeline_e2e_through_engine() { let dir = tempfile::tempdir().unwrap(); - // Build a registry that includes SubPipelineHandler - let child_registry = { - let mut r = HandlerRegistry::new(Box::new(CodergenHandler::new(None))); - r.register("start", Box::new(StartHandler)); - r.register("exit", Box::new(ExitHandler)); - r.register("codergen", Box::new(CodergenHandler::new(None))); - Arc::new(r) - }; - let emitter = Arc::new(EventEmitter::new()); - 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::new(child_registry, emitter)), - ); + registry.register("sub_pipeline", Box::new(SubPipelineHandler)); let engine = PipelineEngine::new(registry, EventEmitter::new()); let config = RunConfig {