mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-10-06 02:48:25 +00:00
Replace ChildObserver mock with real child engine spawning in ManagerLoopHandler
The manager loop (house node) now parses a child DOT pipeline, spawns a real PipelineEngine, and monitors it via a tokio::select poll loop. Context is cloned into the child and diffed on completion to propagate updates back to the parent. - Add PipelineEngine::from_services() to share parent's Arc services - Add PipelineEngine::run_with_context() returning (Outcome, Context) - Rewrite ManagerLoopHandler as unit struct with real child engine spawning - Support both inline DOT (stack.child_dot_source) and file (stack.child_dotfile) - Delete ChildObserver trait entirely - Update all existing tests and add new e2e tests for context flow and dotfile Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp
parent
1aca8231bc
commit
47e16d4c0d
4 changed files with 721 additions and 268 deletions
|
|
@ -767,6 +767,20 @@ impl PipelineEngine {
|
|||
}
|
||||
}
|
||||
|
||||
/// Create a child engine that shares a parent's `Arc` services (registry, emitter, env).
|
||||
#[must_use]
|
||||
pub fn from_services(services: &EngineServices) -> Self {
|
||||
Self {
|
||||
services: EngineServices {
|
||||
registry: Arc::clone(&services.registry),
|
||||
emitter: Arc::clone(&services.emitter),
|
||||
execution_env: Arc::clone(&services.execution_env),
|
||||
git_state: std::sync::RwLock::new(None),
|
||||
},
|
||||
interviewer: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Create a new engine with an interviewer for `inform()` callbacks.
|
||||
#[must_use]
|
||||
pub fn with_interviewer(
|
||||
|
|
@ -939,8 +953,29 @@ impl PipelineEngine {
|
|||
/// Returns an error if no start node is found, a node is missing, or a goal gate fails
|
||||
/// without a retry target.
|
||||
pub async fn run(&self, graph: &Graph, config: &RunConfig) -> Result<Outcome> {
|
||||
self.run_internal(graph, config, None, None, LoopState::default())
|
||||
.await
|
||||
let (outcome, _context) =
|
||||
self.run_internal(graph, config, None, None, None, LoopState::default())
|
||||
.await?;
|
||||
Ok(outcome)
|
||||
}
|
||||
|
||||
/// Run a pipeline seeded with an existing context. Returns both the outcome
|
||||
/// and the final context so the caller can diff changes.
|
||||
pub async fn run_with_context(
|
||||
&self,
|
||||
graph: &Graph,
|
||||
config: &RunConfig,
|
||||
seed_context: Context,
|
||||
) -> Result<(Outcome, Context)> {
|
||||
self.run_internal(
|
||||
graph,
|
||||
config,
|
||||
None,
|
||||
None,
|
||||
Some(seed_context),
|
||||
LoopState::default(),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Resume from a checkpoint. Restores context, completed nodes, and continues
|
||||
|
|
@ -960,8 +995,10 @@ impl PipelineEngine {
|
|||
loop_failure_signatures: checkpoint.loop_failure_signatures.clone(),
|
||||
restart_failure_signatures: checkpoint.restart_failure_signatures.clone(),
|
||||
};
|
||||
self.run_internal(graph, config, Some(checkpoint), None, loop_state)
|
||||
.await
|
||||
let (outcome, _context) = self
|
||||
.run_internal(graph, config, Some(checkpoint), None, None, loop_state)
|
||||
.await?;
|
||||
Ok(outcome)
|
||||
}
|
||||
|
||||
/// Internal run implementation supporting optional checkpoint resume and `start_at` override.
|
||||
|
|
@ -971,8 +1008,9 @@ impl PipelineEngine {
|
|||
config: &RunConfig,
|
||||
resume_checkpoint: Option<&Checkpoint>,
|
||||
start_at: Option<&str>,
|
||||
seed_context: Option<Context>,
|
||||
mut loop_state: LoopState,
|
||||
) -> Result<Outcome> {
|
||||
) -> Result<(Outcome, Context)> {
|
||||
let run_start = Instant::now();
|
||||
let run_id = config.run_id.clone();
|
||||
let artifact_store = ArtifactStore::new(Some(config.logs_root.clone()));
|
||||
|
|
@ -1089,7 +1127,7 @@ impl PipelineEngine {
|
|||
stage_index = 0;
|
||||
current_node_id = start.to_string();
|
||||
} else {
|
||||
context = Context::new();
|
||||
context = seed_context.unwrap_or_default();
|
||||
Self::mirror_graph_attributes(graph, &context);
|
||||
completed_nodes = Vec::new();
|
||||
stage_index = 0;
|
||||
|
|
@ -1192,7 +1230,7 @@ impl PipelineEngine {
|
|||
duration_ms,
|
||||
git_commit_sha: last_git_sha.clone(),
|
||||
});
|
||||
return Ok(Outcome::fail(error_msg));
|
||||
return Ok((Outcome::fail(error_msg), context));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1596,6 +1634,7 @@ impl PipelineEngine {
|
|||
config,
|
||||
None,
|
||||
Some(&edge.to),
|
||||
None,
|
||||
loop_state,
|
||||
))
|
||||
.await;
|
||||
|
|
@ -1651,7 +1690,7 @@ impl PipelineEngine {
|
|||
.get(completed_nodes.last().unwrap_or(&String::new()))
|
||||
.cloned()
|
||||
.unwrap_or_else(Outcome::success);
|
||||
Ok(last_outcome)
|
||||
Ok((last_outcome, context))
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,47 +1,23 @@
|
|||
use std::collections::HashMap;
|
||||
use std::path::Path;
|
||||
use std::time::{Duration, Instant};
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use async_trait::async_trait;
|
||||
|
||||
use crate::condition::evaluate_condition;
|
||||
use crate::context::Context;
|
||||
use crate::engine::{PipelineEngine, RunConfig};
|
||||
use crate::error::ArcError;
|
||||
use crate::graph::{Graph, Node};
|
||||
use crate::outcome::{Outcome, StageStatus};
|
||||
use crate::pipeline::prepare_pipeline;
|
||||
|
||||
use super::{EngineServices, Handler};
|
||||
|
||||
/// Trait for observing child pipeline state during the manager loop.
|
||||
#[async_trait]
|
||||
pub trait ChildObserver: Send + Sync {
|
||||
/// Launch the child pipeline. Called before the observation loop when `child_autostart` is true.
|
||||
async fn launch_child(
|
||||
&self,
|
||||
_dotfile: &str,
|
||||
_workdir: &str,
|
||||
_context: &Context,
|
||||
) -> Result<(), ArcError> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Ingest child telemetry into the context.
|
||||
async fn observe(&self, context: &Context) -> Result<(), ArcError>;
|
||||
|
||||
/// Optionally steer the child pipeline (e.g., write intervention instructions).
|
||||
async fn steer(&self, context: &Context, node: &Node) -> Result<(), ArcError>;
|
||||
}
|
||||
|
||||
/// Orchestrates observe/steer/wait cycles over a child pipeline.
|
||||
pub struct ManagerLoopHandler {
|
||||
observer: Option<Box<dyn ChildObserver>>,
|
||||
}
|
||||
|
||||
impl ManagerLoopHandler {
|
||||
#[must_use]
|
||||
pub fn new(observer: Option<Box<dyn ChildObserver>>) -> Self {
|
||||
Self { observer }
|
||||
}
|
||||
}
|
||||
/// Orchestrates a child pipeline engine, polling for completion or stop conditions.
|
||||
pub struct ManagerLoopHandler;
|
||||
|
||||
/// Parse a duration string like "45s", "200ms", "5m" into a Duration.
|
||||
/// Falls back to 45 seconds on parse failure.
|
||||
|
|
@ -65,6 +41,33 @@ fn parse_duration_str(s: &str) -> Duration {
|
|||
Duration::from_secs(45)
|
||||
}
|
||||
|
||||
/// Read DOT source from node attributes: inline `stack.child_dot_source` or
|
||||
/// file path `stack.child_dotfile`.
|
||||
fn read_child_dot(node: &Node) -> Result<String, ArcError> {
|
||||
if let Some(dot) = node.attrs.get("stack.child_dot_source").and_then(|v| v.as_str()) {
|
||||
return Ok(dot.to_string());
|
||||
}
|
||||
if let Some(path) = node.attrs.get("stack.child_dotfile").and_then(|v| v.as_str()) {
|
||||
return std::fs::read_to_string(path)
|
||||
.map_err(|e| ArcError::Handler(format!("Failed to read child dotfile {path}: {e}")));
|
||||
}
|
||||
Err(ArcError::Handler("No child DOT source".to_string()))
|
||||
}
|
||||
|
||||
/// Compute the context diff: keys that changed or were added relative to `before`.
|
||||
fn context_diff(
|
||||
before: &HashMap<String, serde_json::Value>,
|
||||
after: &HashMap<String, serde_json::Value>,
|
||||
) -> HashMap<String, serde_json::Value> {
|
||||
let mut diff = HashMap::new();
|
||||
for (key, value) in after {
|
||||
if before.get(key) != Some(value) {
|
||||
diff.insert(key.clone(), value.clone());
|
||||
}
|
||||
}
|
||||
diff
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Handler for ManagerLoopHandler {
|
||||
async fn execute(
|
||||
|
|
@ -72,8 +75,8 @@ impl Handler for ManagerLoopHandler {
|
|||
node: &Node,
|
||||
context: &Context,
|
||||
_graph: &Graph,
|
||||
_logs_root: &Path,
|
||||
_services: &EngineServices,
|
||||
logs_root: &Path,
|
||||
services: &EngineServices,
|
||||
) -> Result<Outcome, ArcError> {
|
||||
let poll_interval = node
|
||||
.attrs
|
||||
|
|
@ -101,104 +104,113 @@ impl Handler for ManagerLoopHandler {
|
|||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("");
|
||||
|
||||
let actions_str = node
|
||||
.attrs
|
||||
.get("manager.actions")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("observe,wait");
|
||||
let do_observe = actions_str.contains("observe");
|
||||
let do_steer = actions_str.contains("steer");
|
||||
let do_wait = actions_str.contains("wait");
|
||||
// Read and parse child DOT source
|
||||
let dot_source = match read_child_dot(node) {
|
||||
Ok(s) => s,
|
||||
Err(e) => return Ok(Outcome::fail(e.to_string())),
|
||||
};
|
||||
|
||||
// Child autostart: launch child pipeline before the observation loop
|
||||
let child_autostart = node
|
||||
.attrs
|
||||
.get("stack.child_autostart")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("true");
|
||||
if child_autostart != "false" {
|
||||
if let Some(ref observer) = self.observer {
|
||||
let child_dotfile = node
|
||||
.attrs
|
||||
.get("stack.child_dotfile")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("");
|
||||
let child_workdir = node
|
||||
.attrs
|
||||
.get("stack.child_workdir")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("");
|
||||
observer
|
||||
.launch_child(child_dotfile, child_workdir, context)
|
||||
.await?;
|
||||
let child_graph = match prepare_pipeline(&dot_source) {
|
||||
Ok(g) => g,
|
||||
Err(e) => {
|
||||
return Ok(Outcome::fail(format!(
|
||||
"Failed to parse child pipeline: {e}"
|
||||
)))
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// Steer cooldown tracking
|
||||
let steer_cooldown = node
|
||||
.attrs
|
||||
.get("manager.steer_cooldown")
|
||||
.and_then(|v| v.as_str())
|
||||
.map_or(Duration::ZERO, parse_duration_str);
|
||||
let mut last_steer_time: Option<Instant> = None;
|
||||
// Build child RunConfig
|
||||
let visit = context
|
||||
.get("internal.node_visit")
|
||||
.and_then(|v| v.as_u64())
|
||||
.unwrap_or(1);
|
||||
let child_logs = logs_root.join(format!("nodes/{}_{visit}/child", node.id));
|
||||
let _ = std::fs::create_dir_all(&child_logs);
|
||||
|
||||
// Observation loop
|
||||
let parent_run_id = context
|
||||
.get("internal.run_id")
|
||||
.and_then(|v| v.as_str().map(String::from))
|
||||
.unwrap_or_default();
|
||||
let cancel_token = Arc::new(AtomicBool::new(false));
|
||||
let child_cancel = Arc::clone(&cancel_token);
|
||||
|
||||
let child_config = RunConfig {
|
||||
logs_root: child_logs,
|
||||
cancel_token: Some(cancel_token),
|
||||
dry_run: false,
|
||||
run_id: format!("{parent_run_id}_child_{}", node.id),
|
||||
git_checkpoint: None,
|
||||
base_sha: None,
|
||||
run_branch: None,
|
||||
meta_branch: None,
|
||||
labels: HashMap::new(),
|
||||
};
|
||||
|
||||
// Clone parent context for child; snapshot before for diffing
|
||||
let child_context = context.clone_context();
|
||||
let before_snapshot = context.snapshot();
|
||||
|
||||
// Spawn child engine
|
||||
let engine = PipelineEngine::from_services(services);
|
||||
let mut child_handle =
|
||||
tokio::spawn(
|
||||
async move { engine.run_with_context(&child_graph, &child_config, child_context).await },
|
||||
);
|
||||
|
||||
// Poll loop
|
||||
for cycle in 1..=max_cycles {
|
||||
// Observe
|
||||
if do_observe {
|
||||
if let Some(ref observer) = self.observer {
|
||||
observer.observe(context).await?;
|
||||
}
|
||||
}
|
||||
tokio::select! {
|
||||
result = &mut child_handle => {
|
||||
// Child finished
|
||||
let (child_outcome, child_final_context) = match result {
|
||||
Ok(Ok(pair)) => pair,
|
||||
Ok(Err(e)) => return Ok(Outcome::fail(format!("Child engine error: {e}"))),
|
||||
Err(e) => return Ok(Outcome::fail(format!("Child task panicked: {e}"))),
|
||||
};
|
||||
|
||||
// Steer (with cooldown)
|
||||
if do_steer {
|
||||
let cooldown_elapsed = match last_steer_time {
|
||||
Some(t) => t.elapsed() >= steer_cooldown,
|
||||
None => true,
|
||||
};
|
||||
if cooldown_elapsed {
|
||||
if let Some(ref observer) = self.observer {
|
||||
observer.steer(context, node).await?;
|
||||
last_steer_time = Some(Instant::now());
|
||||
// Compute context diff
|
||||
let after_snapshot = child_final_context.snapshot();
|
||||
let diff = context_diff(&before_snapshot, &after_snapshot);
|
||||
|
||||
let mut outcome = Outcome {
|
||||
status: child_outcome.status.clone(),
|
||||
notes: Some(format!("Child completed at cycle {cycle}")),
|
||||
context_updates: diff,
|
||||
..Outcome::success()
|
||||
};
|
||||
|
||||
if child_outcome.status == StageStatus::Fail {
|
||||
outcome.failure_reason = child_outcome.failure_reason;
|
||||
}
|
||||
|
||||
return Ok(outcome);
|
||||
}
|
||||
_ = tokio::time::sleep(poll_interval) => {
|
||||
// Check stop condition
|
||||
if !stop_condition.is_empty() {
|
||||
let dummy_outcome = Outcome::success();
|
||||
if evaluate_condition(stop_condition, &dummy_outcome, context) {
|
||||
child_cancel.store(true, Ordering::Relaxed);
|
||||
// Give child a moment to wind down
|
||||
let _ = tokio::time::timeout(
|
||||
Duration::from_millis(100),
|
||||
&mut child_handle,
|
||||
).await;
|
||||
return Ok(Outcome {
|
||||
status: StageStatus::Success,
|
||||
notes: Some(format!("Stop condition satisfied at cycle {cycle}")),
|
||||
..Outcome::success()
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Check child status from context
|
||||
let child_status = context.get_string("context.stack.child.status", "");
|
||||
if child_status == "completed" || child_status == "failed" {
|
||||
let child_outcome = context.get_string("context.stack.child.outcome", "");
|
||||
if child_outcome == "success" {
|
||||
return Ok(Outcome {
|
||||
status: StageStatus::Success,
|
||||
notes: Some(format!("Child completed at cycle {cycle}")),
|
||||
..Outcome::success()
|
||||
});
|
||||
}
|
||||
if child_status == "failed" {
|
||||
return Ok(Outcome::fail(format!("Child failed at cycle {cycle}")));
|
||||
}
|
||||
}
|
||||
|
||||
// Evaluate stop condition
|
||||
if !stop_condition.is_empty() {
|
||||
let dummy_outcome = Outcome::success();
|
||||
if evaluate_condition(stop_condition, &dummy_outcome, context) {
|
||||
return Ok(Outcome {
|
||||
status: StageStatus::Success,
|
||||
notes: Some(format!("Stop condition satisfied at cycle {cycle}")),
|
||||
..Outcome::success()
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Wait
|
||||
if do_wait {
|
||||
tokio::time::sleep(poll_interval).await;
|
||||
}
|
||||
}
|
||||
|
||||
// Max cycles exceeded — cancel child
|
||||
child_cancel.store(true, Ordering::Relaxed);
|
||||
let _ = tokio::time::timeout(Duration::from_millis(100), &mut child_handle).await;
|
||||
|
||||
Ok(Outcome::fail(format!(
|
||||
"Max cycles ({max_cycles}) exceeded for manager loop node: {}",
|
||||
node.id
|
||||
|
|
@ -211,12 +223,16 @@ mod tests {
|
|||
use super::*;
|
||||
use crate::event::EventEmitter;
|
||||
use crate::graph::AttrValue;
|
||||
use crate::handler::exit::ExitHandler;
|
||||
use crate::handler::start::StartHandler;
|
||||
use crate::handler::HandlerRegistry;
|
||||
|
||||
fn make_services() -> EngineServices {
|
||||
let mut registry = HandlerRegistry::new(Box::new(StartHandler));
|
||||
registry.register("start", Box::new(StartHandler));
|
||||
registry.register("exit", Box::new(ExitHandler));
|
||||
EngineServices {
|
||||
registry: std::sync::Arc::new(HandlerRegistry::new(Box::new(StartHandler))),
|
||||
registry: std::sync::Arc::new(registry),
|
||||
emitter: std::sync::Arc::new(EventEmitter::new()),
|
||||
execution_env: std::sync::Arc::new(arc_agent::LocalExecutionEnvironment::new(
|
||||
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
|
||||
|
|
@ -225,22 +241,264 @@ mod tests {
|
|||
}
|
||||
}
|
||||
|
||||
fn child_dot_succeeds() -> &'static str {
|
||||
"digraph Child { start [shape=Mdiamond]; exit [shape=Msquare]; start -> exit }"
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn manager_loop_max_cycles_exceeded() {
|
||||
let handler = ManagerLoopHandler::new(None);
|
||||
async fn child_pipeline_succeeds() {
|
||||
let handler = ManagerLoopHandler;
|
||||
let mut node = Node::new("manager");
|
||||
node.attrs.insert(
|
||||
"stack.child_dot_source".to_string(),
|
||||
AttrValue::String(child_dot_succeeds().to_string()),
|
||||
);
|
||||
node.attrs
|
||||
.insert("manager.max_cycles".to_string(), AttrValue::Integer(100));
|
||||
node.attrs.insert(
|
||||
"manager.poll_interval".to_string(),
|
||||
AttrValue::Duration(Duration::from_millis(10)),
|
||||
);
|
||||
|
||||
let context = Context::new();
|
||||
let graph = Graph::new("test");
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
|
||||
let outcome = handler
|
||||
.execute(&node, &context, &graph, dir.path(), &make_services())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(outcome.status, StageStatus::Success);
|
||||
assert!(outcome.notes.as_deref().unwrap().contains("Child completed"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn no_dot_source_fails() {
|
||||
let handler = ManagerLoopHandler;
|
||||
let mut node = Node::new("manager");
|
||||
node.attrs
|
||||
.insert("manager.max_cycles".to_string(), AttrValue::Integer(10));
|
||||
node.attrs.insert(
|
||||
"manager.poll_interval".to_string(),
|
||||
AttrValue::Duration(Duration::from_millis(1)),
|
||||
);
|
||||
|
||||
let context = Context::new();
|
||||
let graph = Graph::new("test");
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
|
||||
let outcome = handler
|
||||
.execute(&node, &context, &graph, dir.path(), &make_services())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(outcome.status, StageStatus::Fail);
|
||||
assert!(outcome
|
||||
.failure_reason
|
||||
.as_deref()
|
||||
.unwrap()
|
||||
.contains("No child DOT source"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn invalid_dot_source_fails() {
|
||||
let handler = ManagerLoopHandler;
|
||||
let mut node = Node::new("manager");
|
||||
node.attrs.insert(
|
||||
"stack.child_dot_source".to_string(),
|
||||
AttrValue::String("not valid dot!!!".to_string()),
|
||||
);
|
||||
node.attrs
|
||||
.insert("manager.max_cycles".to_string(), AttrValue::Integer(10));
|
||||
node.attrs.insert(
|
||||
"manager.poll_interval".to_string(),
|
||||
AttrValue::Duration(Duration::from_millis(1)),
|
||||
);
|
||||
|
||||
let context = Context::new();
|
||||
let graph = Graph::new("test");
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
|
||||
let outcome = handler
|
||||
.execute(&node, &context, &graph, dir.path(), &make_services())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(outcome.status, StageStatus::Fail);
|
||||
assert!(outcome
|
||||
.failure_reason
|
||||
.as_deref()
|
||||
.unwrap()
|
||||
.contains("Failed to parse child pipeline"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn context_flows_parent_to_child_and_back() {
|
||||
// Register a handler that reads parent context and sets a result
|
||||
struct ContextEchoHandler;
|
||||
|
||||
#[async_trait]
|
||||
impl Handler for ContextEchoHandler {
|
||||
async fn execute(
|
||||
&self,
|
||||
_node: &Node,
|
||||
context: &Context,
|
||||
_graph: &Graph,
|
||||
_logs_root: &Path,
|
||||
_services: &EngineServices,
|
||||
) -> Result<Outcome, ArcError> {
|
||||
let target = context.get_string("review.target", "");
|
||||
let mut outcome = Outcome::success();
|
||||
outcome
|
||||
.context_updates
|
||||
.insert("review.result".to_string(), serde_json::json!("approved"));
|
||||
outcome.context_updates.insert(
|
||||
"review.echo".to_string(),
|
||||
serde_json::json!(target),
|
||||
);
|
||||
Ok(outcome)
|
||||
}
|
||||
}
|
||||
|
||||
let mut registry = HandlerRegistry::new(Box::new(ContextEchoHandler));
|
||||
registry.register("start", Box::new(StartHandler));
|
||||
registry.register("exit", Box::new(ExitHandler));
|
||||
let services = EngineServices {
|
||||
registry: std::sync::Arc::new(registry),
|
||||
emitter: std::sync::Arc::new(EventEmitter::new()),
|
||||
execution_env: std::sync::Arc::new(arc_agent::LocalExecutionEnvironment::new(
|
||||
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
|
||||
)),
|
||||
git_state: std::sync::RwLock::new(None),
|
||||
};
|
||||
|
||||
let handler = ManagerLoopHandler;
|
||||
let mut node = Node::new("manager");
|
||||
// Child pipeline with a "work" node (default handler = ContextEchoHandler)
|
||||
node.attrs.insert(
|
||||
"stack.child_dot_source".to_string(),
|
||||
AttrValue::String(
|
||||
"digraph Child { start [shape=Mdiamond]; work [shape=box]; exit [shape=Msquare]; start -> work -> exit }"
|
||||
.to_string(),
|
||||
),
|
||||
);
|
||||
node.attrs
|
||||
.insert("manager.max_cycles".to_string(), AttrValue::Integer(100));
|
||||
node.attrs.insert(
|
||||
"manager.poll_interval".to_string(),
|
||||
AttrValue::Duration(Duration::from_millis(10)),
|
||||
);
|
||||
|
||||
// Parent sets a context value the child should be able to read
|
||||
let context = Context::new();
|
||||
context.set("review.target", serde_json::json!("src/main.rs"));
|
||||
|
||||
let graph = Graph::new("test");
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
|
||||
let outcome = handler
|
||||
.execute(&node, &context, &graph, dir.path(), &services)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(outcome.status, StageStatus::Success);
|
||||
assert_eq!(
|
||||
outcome.context_updates.get("review.result"),
|
||||
Some(&serde_json::json!("approved"))
|
||||
);
|
||||
assert_eq!(
|
||||
outcome.context_updates.get("review.echo"),
|
||||
Some(&serde_json::json!("src/main.rs"))
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn child_dotfile_reads_from_file() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let dot_path = dir.path().join("child.dot");
|
||||
std::fs::write(&dot_path, child_dot_succeeds()).unwrap();
|
||||
|
||||
let handler = ManagerLoopHandler;
|
||||
let mut node = Node::new("manager");
|
||||
node.attrs.insert(
|
||||
"stack.child_dotfile".to_string(),
|
||||
AttrValue::String(dot_path.to_string_lossy().to_string()),
|
||||
);
|
||||
node.attrs
|
||||
.insert("manager.max_cycles".to_string(), AttrValue::Integer(100));
|
||||
node.attrs.insert(
|
||||
"manager.poll_interval".to_string(),
|
||||
AttrValue::Duration(Duration::from_millis(10)),
|
||||
);
|
||||
|
||||
let context = Context::new();
|
||||
let graph = Graph::new("test");
|
||||
|
||||
let outcome = handler
|
||||
.execute(&node, &context, &graph, dir.path(), &make_services())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(outcome.status, StageStatus::Success);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn max_cycles_exceeded_cancels_child() {
|
||||
// Use a child that takes a long time (many nodes with sleep won't work, so use a
|
||||
// child that succeeds quickly but set max_cycles=1 and very short poll)
|
||||
// Actually, to test max cycles exceeded we need a child that runs longer than
|
||||
// max_cycles * poll_interval. Use a child dot that's valid but we set max_cycles=1
|
||||
// with poll_interval=1ms so the child likely won't finish in time.
|
||||
//
|
||||
// But a simple start->exit child is almost instant. So we need a handler that
|
||||
// sleeps to make the child slow.
|
||||
struct SlowHandler;
|
||||
|
||||
#[async_trait]
|
||||
impl Handler for SlowHandler {
|
||||
async fn execute(
|
||||
&self,
|
||||
_node: &Node,
|
||||
_context: &Context,
|
||||
_graph: &Graph,
|
||||
_logs_root: &Path,
|
||||
_services: &EngineServices,
|
||||
) -> Result<Outcome, ArcError> {
|
||||
tokio::time::sleep(Duration::from_secs(10)).await;
|
||||
Ok(Outcome::success())
|
||||
}
|
||||
}
|
||||
|
||||
let mut registry = HandlerRegistry::new(Box::new(SlowHandler));
|
||||
registry.register("start", Box::new(StartHandler));
|
||||
registry.register("exit", Box::new(ExitHandler));
|
||||
let services = EngineServices {
|
||||
registry: std::sync::Arc::new(registry),
|
||||
emitter: std::sync::Arc::new(EventEmitter::new()),
|
||||
execution_env: std::sync::Arc::new(arc_agent::LocalExecutionEnvironment::new(
|
||||
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
|
||||
)),
|
||||
git_state: std::sync::RwLock::new(None),
|
||||
};
|
||||
|
||||
let handler = ManagerLoopHandler;
|
||||
let mut node = Node::new("manager");
|
||||
node.attrs.insert(
|
||||
"stack.child_dot_source".to_string(),
|
||||
AttrValue::String(
|
||||
"digraph Child { start [shape=Mdiamond]; slow [shape=box]; exit [shape=Msquare]; start -> slow -> exit }"
|
||||
.to_string(),
|
||||
),
|
||||
);
|
||||
node.attrs
|
||||
.insert("manager.max_cycles".to_string(), AttrValue::Integer(2));
|
||||
node.attrs.insert(
|
||||
"manager.poll_interval".to_string(),
|
||||
AttrValue::Duration(Duration::from_millis(1)),
|
||||
);
|
||||
|
||||
let context = Context::new();
|
||||
let graph = Graph::new("test");
|
||||
let logs_root = Path::new("/tmp/test");
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
|
||||
let outcome = handler
|
||||
.execute(&node, &context, &graph, logs_root, &make_services())
|
||||
.execute(&node, &context, &graph, dir.path(), &services)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(outcome.status, StageStatus::Fail);
|
||||
|
|
@ -252,67 +510,47 @@ mod tests {
|
|||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn manager_loop_child_completed_success() {
|
||||
let handler = ManagerLoopHandler::new(None);
|
||||
async fn stop_condition_cancels_child() {
|
||||
struct SlowHandler;
|
||||
|
||||
#[async_trait]
|
||||
impl Handler for SlowHandler {
|
||||
async fn execute(
|
||||
&self,
|
||||
_node: &Node,
|
||||
_context: &Context,
|
||||
_graph: &Graph,
|
||||
_logs_root: &Path,
|
||||
_services: &EngineServices,
|
||||
) -> Result<Outcome, ArcError> {
|
||||
tokio::time::sleep(Duration::from_secs(10)).await;
|
||||
Ok(Outcome::success())
|
||||
}
|
||||
}
|
||||
|
||||
let mut registry = HandlerRegistry::new(Box::new(SlowHandler));
|
||||
registry.register("start", Box::new(StartHandler));
|
||||
registry.register("exit", Box::new(ExitHandler));
|
||||
let services = EngineServices {
|
||||
registry: std::sync::Arc::new(registry),
|
||||
emitter: std::sync::Arc::new(EventEmitter::new()),
|
||||
execution_env: std::sync::Arc::new(arc_agent::LocalExecutionEnvironment::new(
|
||||
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
|
||||
)),
|
||||
git_state: std::sync::RwLock::new(None),
|
||||
};
|
||||
|
||||
let handler = ManagerLoopHandler;
|
||||
let mut node = Node::new("manager");
|
||||
node.attrs
|
||||
.insert("manager.max_cycles".to_string(), AttrValue::Integer(10));
|
||||
node.attrs.insert(
|
||||
"manager.poll_interval".to_string(),
|
||||
AttrValue::Duration(Duration::from_millis(1)),
|
||||
"stack.child_dot_source".to_string(),
|
||||
AttrValue::String(
|
||||
"digraph Child { start [shape=Mdiamond]; slow [shape=box]; exit [shape=Msquare]; start -> slow -> exit }"
|
||||
.to_string(),
|
||||
),
|
||||
);
|
||||
// Pre-set child status to "completed" and outcome to "success"
|
||||
let context = Context::new();
|
||||
context.set("context.stack.child.status", serde_json::json!("completed"));
|
||||
context.set("context.stack.child.outcome", serde_json::json!("success"));
|
||||
let graph = Graph::new("test");
|
||||
let logs_root = Path::new("/tmp/test");
|
||||
|
||||
let outcome = handler
|
||||
.execute(&node, &context, &graph, logs_root, &make_services())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(outcome.status, StageStatus::Success);
|
||||
assert!(outcome
|
||||
.notes
|
||||
.as_deref()
|
||||
.unwrap()
|
||||
.contains("Child completed"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn manager_loop_child_failed() {
|
||||
let handler = ManagerLoopHandler::new(None);
|
||||
let mut node = Node::new("manager");
|
||||
node.attrs
|
||||
.insert("manager.max_cycles".to_string(), AttrValue::Integer(10));
|
||||
node.attrs.insert(
|
||||
"manager.poll_interval".to_string(),
|
||||
AttrValue::Duration(Duration::from_millis(1)),
|
||||
);
|
||||
let context = Context::new();
|
||||
context.set("context.stack.child.status", serde_json::json!("failed"));
|
||||
let graph = Graph::new("test");
|
||||
let logs_root = Path::new("/tmp/test");
|
||||
|
||||
let outcome = handler
|
||||
.execute(&node, &context, &graph, logs_root, &make_services())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(outcome.status, StageStatus::Fail);
|
||||
assert!(outcome
|
||||
.failure_reason
|
||||
.as_deref()
|
||||
.unwrap()
|
||||
.contains("Child failed"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn manager_loop_stop_condition_satisfied() {
|
||||
let handler = ManagerLoopHandler::new(None);
|
||||
let mut node = Node::new("manager");
|
||||
node.attrs
|
||||
.insert("manager.max_cycles".to_string(), AttrValue::Integer(10));
|
||||
.insert("manager.max_cycles".to_string(), AttrValue::Integer(100));
|
||||
node.attrs.insert(
|
||||
"manager.poll_interval".to_string(),
|
||||
AttrValue::Duration(Duration::from_millis(1)),
|
||||
|
|
@ -321,13 +559,16 @@ mod tests {
|
|||
"manager.stop_condition".to_string(),
|
||||
AttrValue::String("context.done=true".to_string()),
|
||||
);
|
||||
|
||||
// Pre-set the stop condition so it fires on first poll
|
||||
let context = Context::new();
|
||||
context.set("done", serde_json::json!("true"));
|
||||
|
||||
let graph = Graph::new("test");
|
||||
let logs_root = Path::new("/tmp/test");
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
|
||||
let outcome = handler
|
||||
.execute(&node, &context, &graph, logs_root, &make_services())
|
||||
.execute(&node, &context, &graph, dir.path(), &services)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(outcome.status, StageStatus::Success);
|
||||
|
|
|
|||
|
|
@ -140,7 +140,7 @@ pub fn default_registry(
|
|||
registry.register("sub_pipeline", Box::new(sub_pipeline::SubPipelineHandler));
|
||||
registry.register(
|
||||
"stack.manager_loop",
|
||||
Box::new(manager_loop::ManagerLoopHandler::new(None)),
|
||||
Box::new(manager_loop::ManagerLoopHandler),
|
||||
);
|
||||
registry
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1255,7 +1255,7 @@ fn make_full_registry(interviewer: Arc<dyn Interviewer>) -> HandlerRegistry {
|
|||
registry.register("wait.human", Box::new(WaitHumanHandler::new(interviewer)));
|
||||
registry.register(
|
||||
"stack.manager_loop",
|
||||
Box::new(ManagerLoopHandler::new(None)),
|
||||
Box::new(ManagerLoopHandler),
|
||||
);
|
||||
registry
|
||||
}
|
||||
|
|
@ -2611,6 +2611,23 @@ async fn manager_loop_stop_condition_satisfied_e2e() {
|
|||
}
|
||||
}
|
||||
|
||||
// A slow handler so the child doesn't finish before the stop condition is checked
|
||||
struct SlowHandler;
|
||||
#[async_trait::async_trait]
|
||||
impl Handler for SlowHandler {
|
||||
async fn execute(
|
||||
&self,
|
||||
_node: &Node,
|
||||
_context: &Context,
|
||||
_graph: &Graph,
|
||||
_logs_root: &Path,
|
||||
_services: &arc_workflows::handler::EngineServices,
|
||||
) -> Result<Outcome, ArcError> {
|
||||
tokio::time::sleep(std::time::Duration::from_secs(10)).await;
|
||||
Ok(Outcome::success())
|
||||
}
|
||||
}
|
||||
|
||||
let mut graph = make_graph_with_start_exit("ManagerStopTest");
|
||||
let mut setter = Node::new("setter");
|
||||
setter.attrs.insert(
|
||||
|
|
@ -2623,6 +2640,13 @@ async fn manager_loop_stop_condition_satisfied_e2e() {
|
|||
"type".to_string(),
|
||||
AttrValue::String("stack.manager_loop".to_string()),
|
||||
);
|
||||
manager.attrs.insert(
|
||||
"stack.child_dot_source".to_string(),
|
||||
AttrValue::String(
|
||||
"digraph Child { start [shape=Mdiamond]; slow [shape=box]; exit [shape=Msquare]; start -> slow -> exit }"
|
||||
.to_string(),
|
||||
),
|
||||
);
|
||||
manager.attrs.insert(
|
||||
"manager.stop_condition".to_string(),
|
||||
AttrValue::String("context.done=true".to_string()),
|
||||
|
|
@ -2640,14 +2664,11 @@ async fn manager_loop_stop_condition_satisfied_e2e() {
|
|||
graph.edges.push(Edge::new("manager", "exit"));
|
||||
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let mut registry = HandlerRegistry::new(Box::new(StartHandler));
|
||||
let mut registry = HandlerRegistry::new(Box::new(SlowHandler));
|
||||
registry.register("start", Box::new(StartHandler));
|
||||
registry.register("exit", Box::new(ExitHandler));
|
||||
registry.register("done_setter", Box::new(DoneSetterHandler));
|
||||
registry.register(
|
||||
"stack.manager_loop",
|
||||
Box::new(ManagerLoopHandler::new(None)),
|
||||
);
|
||||
registry.register("stack.manager_loop", Box::new(ManagerLoopHandler));
|
||||
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
|
||||
let config = RunConfig {
|
||||
logs_root: dir.path().to_path_buf(),
|
||||
|
|
@ -2676,12 +2697,36 @@ async fn manager_loop_stop_condition_satisfied_e2e() {
|
|||
|
||||
#[tokio::test]
|
||||
async fn manager_loop_max_cycles_exceeded_e2e() {
|
||||
// A slow handler so the child doesn't finish before max cycles
|
||||
struct SlowHandler;
|
||||
#[async_trait::async_trait]
|
||||
impl Handler for SlowHandler {
|
||||
async fn execute(
|
||||
&self,
|
||||
_node: &Node,
|
||||
_context: &Context,
|
||||
_graph: &Graph,
|
||||
_logs_root: &Path,
|
||||
_services: &arc_workflows::handler::EngineServices,
|
||||
) -> Result<Outcome, ArcError> {
|
||||
tokio::time::sleep(std::time::Duration::from_secs(10)).await;
|
||||
Ok(Outcome::success())
|
||||
}
|
||||
}
|
||||
|
||||
let mut graph = make_graph_with_start_exit("ManagerMaxCyclesTest");
|
||||
let mut manager = Node::new("manager");
|
||||
manager.attrs.insert(
|
||||
"type".to_string(),
|
||||
AttrValue::String("stack.manager_loop".to_string()),
|
||||
);
|
||||
manager.attrs.insert(
|
||||
"stack.child_dot_source".to_string(),
|
||||
AttrValue::String(
|
||||
"digraph Child { start [shape=Mdiamond]; slow [shape=box]; exit [shape=Msquare]; start -> slow -> exit }"
|
||||
.to_string(),
|
||||
),
|
||||
);
|
||||
manager
|
||||
.attrs
|
||||
.insert("manager.max_cycles".to_string(), AttrValue::Integer(2));
|
||||
|
|
@ -2694,13 +2739,10 @@ async fn manager_loop_max_cycles_exceeded_e2e() {
|
|||
graph.edges.push(Edge::new("manager", "exit"));
|
||||
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let mut registry = HandlerRegistry::new(Box::new(StartHandler));
|
||||
let mut registry = HandlerRegistry::new(Box::new(SlowHandler));
|
||||
registry.register("start", Box::new(StartHandler));
|
||||
registry.register("exit", Box::new(ExitHandler));
|
||||
registry.register(
|
||||
"stack.manager_loop",
|
||||
Box::new(ManagerLoopHandler::new(None)),
|
||||
);
|
||||
registry.register("stack.manager_loop", Box::new(ManagerLoopHandler));
|
||||
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
|
||||
let config = RunConfig {
|
||||
logs_root: dir.path().to_path_buf(),
|
||||
|
|
@ -3290,52 +3332,11 @@ async fn sub_pipeline_e2e_through_engine() {
|
|||
}
|
||||
|
||||
// ===========================================================================
|
||||
// 19b. Manager loop with ChildObserver E2E (TS Scenario 10)
|
||||
// 19b. Manager loop runs child engine E2E
|
||||
// ===========================================================================
|
||||
|
||||
#[tokio::test]
|
||||
async fn manager_loop_with_child_observer_e2e() {
|
||||
use arc_workflows::handler::manager_loop::{ChildObserver, ManagerLoopHandler};
|
||||
use std::sync::atomic::{AtomicU32, Ordering};
|
||||
|
||||
struct SimulatingChildObserver {
|
||||
launch_count: AtomicU32,
|
||||
observe_count: AtomicU32,
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl ChildObserver for SimulatingChildObserver {
|
||||
async fn launch_child(
|
||||
&self,
|
||||
_dotfile: &str,
|
||||
_workdir: &str,
|
||||
_context: &arc_workflows::context::Context,
|
||||
) -> Result<(), arc_workflows::error::ArcError> {
|
||||
self.launch_count.fetch_add(1, Ordering::SeqCst);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn observe(
|
||||
&self,
|
||||
context: &arc_workflows::context::Context,
|
||||
) -> Result<(), arc_workflows::error::ArcError> {
|
||||
let cycle = self.observe_count.fetch_add(1, Ordering::SeqCst);
|
||||
if cycle >= 2 {
|
||||
context.set("context.stack.child.status", serde_json::json!("completed"));
|
||||
context.set("context.stack.child.outcome", serde_json::json!("success"));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn steer(
|
||||
&self,
|
||||
_context: &arc_workflows::context::Context,
|
||||
_node: &arc_workflows::graph::Node,
|
||||
) -> Result<(), arc_workflows::error::ArcError> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
async fn manager_loop_runs_child_engine_e2e() {
|
||||
let mut graph = Graph::new("ManagerLoopE2E");
|
||||
graph.attrs.insert(
|
||||
"goal".to_string(),
|
||||
|
|
@ -3361,39 +3362,30 @@ async fn manager_loop_with_child_observer_e2e() {
|
|||
"type".to_string(),
|
||||
AttrValue::String("stack.manager_loop".to_string()),
|
||||
);
|
||||
supervisor.attrs.insert(
|
||||
"stack.child_dot_source".to_string(),
|
||||
AttrValue::String(
|
||||
"digraph Child { start [shape=Mdiamond]; exit [shape=Msquare]; start -> exit }"
|
||||
.to_string(),
|
||||
),
|
||||
);
|
||||
supervisor.attrs.insert(
|
||||
"manager.poll_interval".to_string(),
|
||||
AttrValue::Duration(std::time::Duration::from_millis(1)),
|
||||
AttrValue::Duration(std::time::Duration::from_millis(10)),
|
||||
);
|
||||
supervisor
|
||||
.attrs
|
||||
.insert("manager.max_cycles".to_string(), AttrValue::Integer(50));
|
||||
supervisor.attrs.insert(
|
||||
"manager.actions".to_string(),
|
||||
AttrValue::String("observe,wait".to_string()),
|
||||
);
|
||||
supervisor.attrs.insert(
|
||||
"manager.stop_condition".to_string(),
|
||||
AttrValue::String(String::new()),
|
||||
);
|
||||
.insert("manager.max_cycles".to_string(), AttrValue::Integer(100));
|
||||
graph.nodes.insert("supervisor".to_string(), supervisor);
|
||||
|
||||
graph.edges.push(Edge::new("start", "supervisor"));
|
||||
graph.edges.push(Edge::new("supervisor", "exit"));
|
||||
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let observer = SimulatingChildObserver {
|
||||
launch_count: AtomicU32::new(0),
|
||||
observe_count: AtomicU32::new(0),
|
||||
};
|
||||
|
||||
let mut registry = HandlerRegistry::new(Box::new(StartHandler));
|
||||
registry.register("start", Box::new(StartHandler));
|
||||
registry.register("exit", Box::new(ExitHandler));
|
||||
registry.register(
|
||||
"stack.manager_loop",
|
||||
Box::new(ManagerLoopHandler::new(Some(Box::new(observer)))),
|
||||
);
|
||||
registry.register("stack.manager_loop", Box::new(ManagerLoopHandler));
|
||||
|
||||
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
|
||||
let config = RunConfig {
|
||||
|
|
@ -3422,7 +3414,6 @@ async fn manager_loop_with_child_observer_e2e() {
|
|||
"supervisor should be in completed_nodes"
|
||||
);
|
||||
|
||||
// The manager loop handler stores notes about child completion
|
||||
let supervisor_outcome = checkpoint.node_outcomes.get("supervisor");
|
||||
assert!(
|
||||
supervisor_outcome.is_some(),
|
||||
|
|
@ -3435,6 +3426,188 @@ async fn manager_loop_with_child_observer_e2e() {
|
|||
);
|
||||
}
|
||||
|
||||
// ===========================================================================
|
||||
// 19b-2. Manager loop: context flows parent → child → parent
|
||||
// ===========================================================================
|
||||
|
||||
#[tokio::test]
|
||||
async fn manager_loop_context_flows_e2e() {
|
||||
// Handler that reads parent's context value and sets a result
|
||||
struct ContextEchoHandler;
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl Handler for ContextEchoHandler {
|
||||
async fn execute(
|
||||
&self,
|
||||
_node: &Node,
|
||||
context: &Context,
|
||||
_graph: &Graph,
|
||||
_logs_root: &Path,
|
||||
_services: &arc_workflows::handler::EngineServices,
|
||||
) -> Result<Outcome, ArcError> {
|
||||
let target = context.get_string("review.target", "");
|
||||
let mut outcome = Outcome::success();
|
||||
outcome
|
||||
.context_updates
|
||||
.insert("review.result".to_string(), serde_json::json!("approved"));
|
||||
outcome.context_updates.insert(
|
||||
"review.echo".to_string(),
|
||||
serde_json::json!(target),
|
||||
);
|
||||
Ok(outcome)
|
||||
}
|
||||
}
|
||||
|
||||
let mut graph = make_graph_with_start_exit("ManagerContextFlowE2E");
|
||||
|
||||
// A setter node that puts review.target into context before the manager
|
||||
struct SetterHandler;
|
||||
#[async_trait::async_trait]
|
||||
impl Handler for SetterHandler {
|
||||
async fn execute(
|
||||
&self,
|
||||
_node: &Node,
|
||||
_context: &Context,
|
||||
_graph: &Graph,
|
||||
_logs_root: &Path,
|
||||
_services: &arc_workflows::handler::EngineServices,
|
||||
) -> Result<Outcome, ArcError> {
|
||||
let mut outcome = Outcome::success();
|
||||
outcome
|
||||
.context_updates
|
||||
.insert("review.target".to_string(), serde_json::json!("src/main.rs"));
|
||||
Ok(outcome)
|
||||
}
|
||||
}
|
||||
|
||||
let mut setter = Node::new("setter");
|
||||
setter.attrs.insert(
|
||||
"type".to_string(),
|
||||
AttrValue::String("setter".to_string()),
|
||||
);
|
||||
graph.nodes.insert("setter".to_string(), setter);
|
||||
|
||||
let mut supervisor = Node::new("supervisor");
|
||||
supervisor.attrs.insert(
|
||||
"type".to_string(),
|
||||
AttrValue::String("stack.manager_loop".to_string()),
|
||||
);
|
||||
supervisor.attrs.insert(
|
||||
"stack.child_dot_source".to_string(),
|
||||
AttrValue::String(
|
||||
"digraph Child { start [shape=Mdiamond]; work [shape=box]; exit [shape=Msquare]; start -> work -> exit }"
|
||||
.to_string(),
|
||||
),
|
||||
);
|
||||
supervisor.attrs.insert(
|
||||
"manager.poll_interval".to_string(),
|
||||
AttrValue::Duration(std::time::Duration::from_millis(10)),
|
||||
);
|
||||
supervisor
|
||||
.attrs
|
||||
.insert("manager.max_cycles".to_string(), AttrValue::Integer(100));
|
||||
graph.nodes.insert("supervisor".to_string(), supervisor);
|
||||
|
||||
graph.edges.push(Edge::new("start", "setter"));
|
||||
graph.edges.push(Edge::new("setter", "supervisor"));
|
||||
graph.edges.push(Edge::new("supervisor", "exit"));
|
||||
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
// Default handler = ContextEchoHandler (handles the child's "work" node)
|
||||
let mut registry = HandlerRegistry::new(Box::new(ContextEchoHandler));
|
||||
registry.register("start", Box::new(StartHandler));
|
||||
registry.register("exit", Box::new(ExitHandler));
|
||||
registry.register("setter", Box::new(SetterHandler));
|
||||
registry.register("stack.manager_loop", Box::new(ManagerLoopHandler));
|
||||
|
||||
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
|
||||
let config = RunConfig {
|
||||
logs_root: dir.path().to_path_buf(),
|
||||
cancel_token: None,
|
||||
dry_run: false,
|
||||
run_id: "test-run".into(),
|
||||
git_checkpoint: None,
|
||||
base_sha: None,
|
||||
run_branch: None,
|
||||
meta_branch: None,
|
||||
labels: std::collections::HashMap::new(),
|
||||
};
|
||||
|
||||
let outcome = engine.run(&graph, &config).await.expect("run");
|
||||
assert_eq!(outcome.status, StageStatus::Success);
|
||||
|
||||
// Check that child's context updates were propagated through the manager
|
||||
let checkpoint = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap();
|
||||
let sup_outcome = checkpoint.node_outcomes.get("supervisor").unwrap();
|
||||
assert_eq!(
|
||||
sup_outcome.context_updates.get("review.result"),
|
||||
Some(&serde_json::json!("approved")),
|
||||
"child's review.result should propagate to parent"
|
||||
);
|
||||
assert_eq!(
|
||||
sup_outcome.context_updates.get("review.echo"),
|
||||
Some(&serde_json::json!("src/main.rs")),
|
||||
"child should have read parent's review.target"
|
||||
);
|
||||
}
|
||||
|
||||
// ===========================================================================
|
||||
// 19b-3. Manager loop with child_dotfile E2E
|
||||
// ===========================================================================
|
||||
|
||||
#[tokio::test]
|
||||
async fn manager_loop_child_dotfile_e2e() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let dot_path = dir.path().join("child.dot");
|
||||
std::fs::write(
|
||||
&dot_path,
|
||||
"digraph Child { start [shape=Mdiamond]; exit [shape=Msquare]; start -> exit }",
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let mut graph = make_graph_with_start_exit("ManagerDotfileE2E");
|
||||
let mut supervisor = Node::new("supervisor");
|
||||
supervisor.attrs.insert(
|
||||
"type".to_string(),
|
||||
AttrValue::String("stack.manager_loop".to_string()),
|
||||
);
|
||||
supervisor.attrs.insert(
|
||||
"stack.child_dotfile".to_string(),
|
||||
AttrValue::String(dot_path.to_string_lossy().to_string()),
|
||||
);
|
||||
supervisor.attrs.insert(
|
||||
"manager.poll_interval".to_string(),
|
||||
AttrValue::Duration(std::time::Duration::from_millis(10)),
|
||||
);
|
||||
supervisor
|
||||
.attrs
|
||||
.insert("manager.max_cycles".to_string(), AttrValue::Integer(100));
|
||||
graph.nodes.insert("supervisor".to_string(), supervisor);
|
||||
graph.edges.push(Edge::new("start", "supervisor"));
|
||||
graph.edges.push(Edge::new("supervisor", "exit"));
|
||||
|
||||
let mut registry = HandlerRegistry::new(Box::new(StartHandler));
|
||||
registry.register("start", Box::new(StartHandler));
|
||||
registry.register("exit", Box::new(ExitHandler));
|
||||
registry.register("stack.manager_loop", Box::new(ManagerLoopHandler));
|
||||
|
||||
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
|
||||
let config = RunConfig {
|
||||
logs_root: dir.path().to_path_buf(),
|
||||
cancel_token: None,
|
||||
dry_run: false,
|
||||
run_id: "test-run".into(),
|
||||
git_checkpoint: None,
|
||||
base_sha: None,
|
||||
run_branch: None,
|
||||
meta_branch: None,
|
||||
labels: std::collections::HashMap::new(),
|
||||
};
|
||||
|
||||
let outcome = engine.run(&graph, &config).await.expect("run");
|
||||
assert_eq!(outcome.status, StageStatus::Success);
|
||||
}
|
||||
|
||||
// ===========================================================================
|
||||
// 19c. GraphMerge E2E (TS Scenario 11)
|
||||
// ===========================================================================
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue