diff --git a/crates/attractor/src/engine.rs b/crates/attractor/src/engine.rs index 2058ff86d..56449863a 100644 --- a/crates/attractor/src/engine.rs +++ b/crates/attractor/src/engine.rs @@ -259,14 +259,29 @@ fn write_manifest(logs_root: &Path, graph: &Graph) { } } -/// Return the directory for a node's logs: `{logs_root}/nodes/{node_id}`. -pub fn node_dir(logs_root: &Path, node_id: &str) -> PathBuf { - logs_root.join("nodes").join(node_id) +/// Return the directory for a node's logs. +/// +/// First visit (`visit <= 1`): `{logs_root}/nodes/{node_id}` +/// Subsequent visits: `{logs_root}/nodes/{node_id}-attempt_{visit}` +pub fn node_dir(logs_root: &Path, node_id: &str, visit: usize) -> PathBuf { + if visit <= 1 { + logs_root.join("nodes").join(node_id) + } else { + logs_root.join("nodes").join(format!("{node_id}-attempt_{visit}")) + } +} + +/// Read the visit count from context, defaulting to 1 if not set. +pub fn visit_from_context(context: &Context) -> usize { + context + .get("internal.node_visit_count") + .and_then(|v| v.as_u64()) + .unwrap_or(1) as usize } /// Write status.json for a completed node into {`logs_root}/nodes/{node_id}/status.json`. -fn write_node_status(logs_root: &Path, node_id: &str, outcome: &Outcome) { - let node_dir = node_dir(logs_root, node_id); +fn write_node_status(logs_root: &Path, node_id: &str, visit: usize, outcome: &Outcome) { + let node_dir = node_dir(logs_root, node_id, visit); let _ = std::fs::create_dir_all(&node_dir); let status = serde_json::json!({ "status": outcome.status.to_string(), @@ -523,6 +538,7 @@ impl PipelineEngine { /// Execute a node handler with retry policy. /// Returns `(outcome, attempts_used)` where `attempts_used` is the 1-indexed count. + #[allow(clippy::too_many_arguments)] async fn execute_with_retry( &self, node: &Node, @@ -531,6 +547,7 @@ impl PipelineEngine { logs_root: &Path, policy: &RetryPolicy, stage_index: usize, + visit: usize, ) -> Result<(Outcome, u32)> { let handler = self.services.registry.resolve(node); @@ -565,7 +582,7 @@ impl PipelineEngine { } else { "handler panicked".to_string() }; - let panic_dir = node_dir(&logs_root, &node.id); + let panic_dir = node_dir(logs_root, &node.id, visit); let _ = std::fs::create_dir_all(&panic_dir); let _ = std::fs::write(panic_dir.join("panic.txt"), &msg); Err(AttractorError::Handler(msg)) @@ -723,6 +740,10 @@ impl PipelineEngine { context.append_log(log_entry.clone()); } completed_nodes = cp.completed_nodes.clone(); + // Rebuild visit counts from completed_nodes (which records every visit) + for id in &completed_nodes { + *node_visits.entry(id.clone()).or_insert(0) += 1; + } // Gap #5: Restore retry counters from checkpoint node_retries = cp.node_retries.clone(); // P1: Restore node outcomes for goal gate checks @@ -775,16 +796,14 @@ impl PipelineEngine { AttractorError::Engine(format!("node not found: {current_node_id}")) })?; - // Check per-node visit limit - if max_node_visits > 0 { - let count = node_visits.entry(current_node_id.clone()).or_insert(0); - *count += 1; - if *count > max_node_visits { - return Err(AttractorError::Engine(format!( - "node \"{}\" exceeded max visit limit of {max_node_visits}", - current_node_id - ))); - } + // Always track visit count (used for stage directory naming) + let count = node_visits.entry(current_node_id.clone()).or_insert(0); + *count += 1; + if max_node_visits > 0 && *count > max_node_visits { + return Err(AttractorError::Engine(format!( + "node \"{}\" exceeded max visit limit of {max_node_visits}", + current_node_id + ))); } // Step 1: Check for terminal node @@ -852,6 +871,8 @@ impl PipelineEngine { } // Step 2: Execute node handler with retry policy + let visit = *node_visits.get(¤t_node_id).unwrap_or(&1); + context.set("internal.node_visit_count", serde_json::json!(visit)); context.set("current_node", serde_json::json!(&node.id)); let retry_policy = build_retry_policy(node, graph); @@ -871,7 +892,7 @@ impl PipelineEngine { let stage_start = Instant::now(); let (mut outcome, attempts_used) = self - .execute_with_retry(node, &context, graph, &config.logs_root, &retry_policy, stage_index) + .execute_with_retry(node, &context, graph, &config.logs_root, &retry_policy, stage_index, visit) .await?; // Gap #5: Track retry count per node node_retries.insert(node.id.clone(), attempts_used); @@ -928,7 +949,7 @@ impl PipelineEngine { } // Write per-node status.json (spec 5.6) - write_node_status(&config.logs_root, &node.id, &outcome); + write_node_status(&config.logs_root, &node.id, visit, &outcome); // Offload large context values to artifact store before recording if let Err(e) = offload_large_values(&mut outcome.context_updates, &artifact_store) { @@ -2611,6 +2632,32 @@ mod tests { ); } + // --- node_dir visit-count tests --- + + #[test] + fn node_dir_first_visit() { + let root = Path::new("/tmp/logs"); + assert_eq!(node_dir(root, "work", 1), root.join("nodes").join("work")); + } + + #[test] + fn node_dir_second_visit() { + let root = Path::new("/tmp/logs"); + assert_eq!( + node_dir(root, "work", 2), + root.join("nodes").join("work-attempt_2") + ); + } + + #[test] + fn node_dir_fifth_visit() { + let root = Path::new("/tmp/logs"); + assert_eq!( + node_dir(root, "work", 5), + root.join("nodes").join("work-attempt_5") + ); + } + // --- panic.txt tests --- /// Handler that always panics. diff --git a/crates/attractor/src/handler/codergen.rs b/crates/attractor/src/handler/codergen.rs index 4bdc8cabb..1a4abcc01 100644 --- a/crates/attractor/src/handler/codergen.rs +++ b/crates/attractor/src/handler/codergen.rs @@ -212,7 +212,8 @@ impl Handler for CodergenHandler { }; // 2. Write prompt to logs - let stage_dir = crate::engine::node_dir(logs_root, &node.id); + let visit = crate::engine::visit_from_context(context); + let stage_dir = crate::engine::node_dir(logs_root, &node.id, visit); tokio::fs::create_dir_all(&stage_dir).await?; tokio::fs::write(stage_dir.join("prompt.md"), &prompt).await?; diff --git a/crates/attractor/src/handler/fan_in.rs b/crates/attractor/src/handler/fan_in.rs index fc57bf6b8..18a41c571 100644 --- a/crates/attractor/src/handler/fan_in.rs +++ b/crates/attractor/src/handler/fan_in.rs @@ -169,7 +169,8 @@ async fn llm_evaluate( ); // Write prompt to logs - let stage_dir = crate::engine::node_dir(logs_root, node_id); + let visit = crate::engine::visit_from_context(context); + let stage_dir = crate::engine::node_dir(logs_root, node_id, visit); tokio::fs::create_dir_all(&stage_dir).await?; tokio::fs::write(stage_dir.join("prompt.md"), &full_prompt).await?; diff --git a/crates/attractor/src/handler/parallel.rs b/crates/attractor/src/handler/parallel.rs index 6cb3f1e80..7bc130cf9 100644 --- a/crates/attractor/src/handler/parallel.rs +++ b/crates/attractor/src/handler/parallel.rs @@ -273,7 +273,8 @@ impl Handler for ParallelHandler { context.set("parallel.results", serde_json::json!(results_json)); context.set("parallel.branch_count", serde_json::json!(total)); - let node_dir = crate::engine::node_dir(logs_root, &node.id); + let visit = crate::engine::visit_from_context(context); + let node_dir = crate::engine::node_dir(logs_root, &node.id, visit); let _ = tokio::fs::create_dir_all(&node_dir).await; if let Ok(json) = serde_json::to_string_pretty(&results_json) { let _ = tokio::fs::write(node_dir.join("parallel_results.json"), json).await; diff --git a/crates/attractor/src/handler/script.rs b/crates/attractor/src/handler/script.rs index 8247bb270..3e42d58f2 100644 --- a/crates/attractor/src/handler/script.rs +++ b/crates/attractor/src/handler/script.rs @@ -21,7 +21,7 @@ impl Handler for ScriptHandler { async fn execute( &self, node: &Node, - _context: &Context, + context: &Context, _graph: &Graph, logs_root: &Path, _services: &EngineServices, @@ -49,7 +49,8 @@ impl Handler for ScriptHandler { ))); } - let stage_dir = crate::engine::node_dir(logs_root, &node.id); + let visit = crate::engine::visit_from_context(context); + let stage_dir = crate::engine::node_dir(logs_root, &node.id, visit); tokio::fs::create_dir_all(&stage_dir).await?; let invocation = serde_json::json!({ diff --git a/crates/attractor/src/handler/sub_pipeline.rs b/crates/attractor/src/handler/sub_pipeline.rs index fc3c0a6b6..651f38652 100644 --- a/crates/attractor/src/handler/sub_pipeline.rs +++ b/crates/attractor/src/handler/sub_pipeline.rs @@ -60,7 +60,8 @@ impl Handler for SubPipelineHandler { let before_snapshot = context.snapshot(); // 5. Walk the sub-graph - let sub_logs_root = crate::engine::node_dir(logs_root, &node.id); + let visit = crate::engine::visit_from_context(context); + let sub_logs_root = crate::engine::node_dir(logs_root, &node.id, visit); let mut current_node_id = start_node.clone(); let mut last_outcome = Outcome::success(); diff --git a/crates/attractor/tests/integration.rs b/crates/attractor/tests/integration.rs index 814e8f4b6..22313cecd 100644 --- a/crates/attractor/tests/integration.rs +++ b/crates/attractor/tests/integration.rs @@ -7082,4 +7082,107 @@ async fn artifact_pointers_rewritten_for_remote_execution_env() { "written content should be >100KB, got {} bytes", written[0].1.len() ); +} + +// --------------------------------------------------------------------------- +// Node directory visit-count naming +// --------------------------------------------------------------------------- + +/// Verify that revisited nodes get distinct stage directories: +/// visit 1 → `nodes/{id}/` +/// visit 2 → `nodes/{id}-attempt_2/` +#[tokio::test] +async fn node_dir_uses_visit_count_on_revisit() { + // Handler that fails on first call, succeeds on second. + struct FailOnceHandler { + call_count: std::sync::atomic::AtomicU32, + } + + #[async_trait::async_trait] + impl Handler for FailOnceHandler { + async fn execute( + &self, + _node: &Node, + _context: &attractor::context::Context, + _graph: &Graph, + _logs_root: &Path, + _services: &attractor::handler::EngineServices, + ) -> Result { + let n = self.call_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if n == 0 { + Ok(Outcome::fail("first attempt fails")) + } else { + Ok(Outcome::success()) + } + } + } + + // Graph: start -> gated_work -> exit + // gated_work has goal_gate=true, retry_target=start + // First visit fails → goal gate unsatisfied → retries from start + // Second visit succeeds → pipeline completes + let mut graph = Graph::new("VisitCountTest"); + + let mut start = Node::new("start"); + start.attrs.insert("shape".to_string(), AttrValue::String("Mdiamond".to_string())); + graph.nodes.insert("start".to_string(), start); + + let mut exit = Node::new("exit"); + exit.attrs.insert("shape".to_string(), AttrValue::String("Msquare".to_string())); + graph.nodes.insert("exit".to_string(), exit); + + let mut gated_work = Node::new("gated_work"); + gated_work.attrs.insert("goal_gate".to_string(), AttrValue::Boolean(true)); + gated_work.attrs.insert("max_retries".to_string(), AttrValue::Integer(0)); + gated_work.attrs.insert( + "retry_target".to_string(), + AttrValue::String("start".to_string()), + ); + gated_work.attrs.insert( + "type".to_string(), + AttrValue::String("fail_once".to_string()), + ); + graph.nodes.insert("gated_work".to_string(), gated_work); + + graph.edges.push(Edge::new("start", "gated_work")); + graph.edges.push(Edge::new("gated_work", "exit")); + + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register( + "fail_once", + Box::new(FailOnceHandler { + call_count: std::sync::atomic::AtomicU32::new(0), + }), + ); + + 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, + }; + + let outcome = engine.run(&graph, &config).await.expect("pipeline should succeed"); + assert_eq!(outcome.status, StageStatus::Success); + + // First visit: nodes/gated_work/status.json + let first = dir.path().join("nodes").join("gated_work").join("status.json"); + assert!(first.exists(), "first visit directory should exist at {}", first.display()); + + // Second visit: nodes/gated_work-attempt_2/status.json + let second = dir.path().join("nodes").join("gated_work-attempt_2").join("status.json"); + assert!(second.exists(), "second visit directory should exist at {}", second.display()); + + // Verify distinct content (first = fail, second = success) + let first_json: serde_json::Value = serde_json::from_str( + &std::fs::read_to_string(&first).unwrap() + ).unwrap(); + let second_json: serde_json::Value = serde_json::from_str( + &std::fs::read_to_string(&second).unwrap() + ).unwrap(); + assert_eq!(first_json["status"], "fail"); + assert_eq!(second_json["status"], "success"); } \ No newline at end of file