scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-06 08:18:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Thread EngineServices through Handler::execute() to eliminate circular dependency

Browse source 
ParallelHandler and SubPipelineHandler needed Arc<HandlerRegistry> at
construction time but also lived inside the registry, creating a circular
dependency. The previous fix special-cased ParallelHandler as a separate
field on PipelineEngine with a resolve_handler() override.

Instead, add an EngineServices struct (registry + emitter) passed through
Handler::execute(). ParallelHandler and SubPipelineHandler become unit
structs that get what they need at execution time. No special-casing,
both register normally in default_registry() like every other handler.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-02-23 16:31:55 -05:00
parent 74a6b219cc
commit 95a4ab24d5
13 changed files with 263 additions and 227 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

69
crates/attractor/src/engine.rs
View file

@ -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<HandlerRegistry>,
emitter: Arc<EventEmitter>,
parallel_handler: ParallelHandler,
services: EngineServices,
pub interviewer: Option<Arc<dyn Interviewer>>,
}
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(&registry), 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<dyn Interviewer>,
) -> Self {
let registry = Arc::new(registry);
let emitter = Arc::new(emitter);
let parallel_handler =
ParallelHandler::new(Arc::clone(&registry), 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<Outcome, AttractorError> {
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<Outcome, AttractorError> {
tokio::time::sleep(Duration::from_millis(self.sleep_ms)).await;
Ok(Outcome::success())

35
crates/attractor/src/handler/codergen.rs
View file

@ -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<Outcome, AttractorError> {
// 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);

15
crates/attractor/src/handler/conditional.rs
View file

@ -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<Outcome, AttractorError> {
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);

15
crates/attractor/src/handler/exit.rs
View file

@ -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<Outcome, AttractorError> {
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);

27
crates/attractor/src/handler/fan_in.rs
View file

@ -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<Outcome, AttractorError> {
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);

21
crates/attractor/src/handler/manager_loop.rs
View file

@ -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<Outcome, AttractorError> {
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);

12
crates/attractor/src/handler/mod.rs
View file

@ -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<HandlerRegistry>,
pub emitter: Arc<EventEmitter>,
}
/// 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<Outcome, AttractorError>;
/// 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<Outcome, AttractorError> {
Ok(Outcome::success())
}
@ -211,6 +222,7 @@ mod tests {
_context: &Context,
_graph: &Graph,
_logs_root: &Path,
_services: &EngineServices,
) -> Result<Outcome, AttractorError> {
Ok(Outcome::success())
}

72
crates/attractor/src/handler/parallel.rs
View file

@ -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<HandlerRegistry>,
emitter: Arc<EventEmitter>,
}
impl ParallelHandler {
#[must_use]
pub fn new(registry: Arc<HandlerRegistry>, emitter: Arc<EventEmitter>) -> 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<Outcome, AttractorError> {
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(&registry),
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<HandlerRegistry> {
fn make_services() -> EngineServices {
let registry = HandlerRegistry::new(Box::new(StartHandler));
Arc::new(registry)
}
fn make_emitter() -> Arc<EventEmitter> {
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();

14
crates/attractor/src/handler/start.rs
View file

@ -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<Outcome, AttractorError> {
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);

87
crates/attractor/src/handler/sub_pipeline.rs
View file

@ -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<HandlerRegistry>,
_emitter: Arc<EventEmitter>,
}
impl SubPipelineHandler {
#[must_use]
pub fn new(registry: Arc<HandlerRegistry>, emitter: Arc<EventEmitter>) -> 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<Outcome, AttractorError> {
// 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<HandlerRegistry> {
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<EventEmitter> {
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<Outcome, AttractorError> {
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<Outcome, AttractorError> {
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);

21
crates/attractor/src/handler/tool.rs
View file

@ -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<Outcome, AttractorError> {
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);

19
crates/attractor/src/handler/wait_human.rs
View file

@ -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<Outcome, AttractorError> {
// 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);

83
crates/attractor/tests/integration.rs
View file

@ -431,6 +431,7 @@ impl Handler for AlwaysFailHandler {
_context: &attractor::context::Context,
_graph: &Graph,
_logs_root: &Path,
_services: &attractor::handler::EngineServices,
) -> Result<Outcome, attractor::error::AttractorError> {
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<Outcome, attractor::error::AttractorError> {
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<Outcome, AttractorError> {
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<Outcome, AttractorError> {
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<Outcome, AttractorError> {
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(&registry), 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<Outcome, AttractorError> {
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<dyn Interviewer> = 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<Outcome, AttractorError> {
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<Outcome, AttractorError> {
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<Outcome, AttractorError> {
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 {