Append -attempt_{n} to node directories on revisits

When a pipeline revisits a node (goal gate retries, loops), each visit
now gets a distinct stage directory instead of silently overwriting the
previous one. First visit keeps the clean `nodes/{id}/` path; visit 2+
produces `nodes/{id}-attempt_{n}/`.

The engine always tracks visit counts and sets
`internal.node_visit_count` in context. Handlers read the count via
`visit_from_context()` and pass it to the updated `node_dir()`.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

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(&current_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.

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?;
 

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?;
 

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;

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!({

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();
 

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<Outcome, attractor::error::AttractorError> {
+            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");
 }