Add /graph G1: parallel fan-out with worktree isolation and merge-back

With KIGI_GRAPH_CONCURRENCY > 1 (default 3, clamp [1,8]) and >=2 Ready nodes,
drive_graph — the single dispatch loop shared by setup/advance/resume — runs
parallel batches: each node executes as a bounded worker<->verifier subagent
loop (KIGI_GRAPH_NODE_ROUNDS, default 3; general-purpose children with the
full implementer toolset; worktree isolation on round 1, resume keeps context
and worktree on later rounds; NODE_RESULT/NODE_VERDICT terminal contracts
parsed fail-closed with fence-stripping and line anchoring). Achieved nodes
merge back SEQUENTIALLY via kigi_workspace apply_worktree in Merge mode; a
conflict fails the node, block_dependents fires, and surviving chains keep
going. gn-final always runs serially on the full goal engine. Concurrency=1
is byte-identical to the serial G0 path; non-git projects degrade to serial.

Merge primitive hardened for real use (kigi-workspace): the 3-way apply is
now byte-safe (binary files no longer read as UTF-8 and silently deleted)
and gains the identical-content rule (ours==theirs => already present, not a
conflict) — without it, any dirty file inherited via PreserveWorkingTree
false-conflicted every node merge and multi-wave graphs self-poisoned.

Adversarial review pass (16 confirmed findings, all fixed): in-session
resume demotes orphaned Running nodes instead of wedge-pausing forever after
a mid-batch Esc; cancel re-sweeps subagents AFTER the turn abort so workers
spawned in the cancel window die too; empty child ids are never adopted as
resume targets (no unisolated escape to the shared tree); a successful
isolated round returning no worktree fails the node (soft-fallback can no
longer put N writers in one tree); main-HEAD movement during a batch aborts
merges instead of reverse-applying external commits; failed nodes still
charge the token budget; budget trips terminally fail the in-flight node;
runaway (>600s) rounds are cancelled by spawn id and retried via resume;
worker summaries are marker-sanitized before verifier embedding. Merged
worktrees are removed immediately (storage discipline); failed nodes keep
theirs for postmortem.

Tests: 4922 kigi-shell lib tests green (58 graph-specific), including
fan-out proven by a held-reply gate, real-git batch merge with cleanup
assertions, budget charging across verdicts, cap trimming, resume-after-
cancelled-batch, and backgrounded-round cancel semantics.

AGENTS.md

@@ -49,6 +49,17 @@ import) or any `KIMI_*` env var.
 - `third_party/` — vendored Mermaid rendering stack (untouched policy).
 - `bin/protoc` — dotslash launcher used by proto codegen.
 
+## Storage discipline
+
+- Tests that touch the filesystem MUST use `tempfile::TempDir` (drop
+  cleans up) — never bare `std::env::temp_dir()` + `create_dir_all`,
+  which leaks directories into the OS temp root forever.
+- `target/` grows past 150GB across repeated full-workspace builds
+  (incremental is already off); run `cargo clean` when it exceeds
+  ~50GB and at milestone boundaries.
+- Graph node worktrees are removed right after a successful merge-back;
+  only FAILED nodes keep theirs for postmortem.
+
 ## Test seams
 
 Cross-crate test hooks are behind the `test-support` cargo feature
@@ -86,6 +97,19 @@ edges stay deterministic Rust. The harness appends a terminal
   the verifier gates completion).
 - `/goal` and `/graph` are mutually exclusive while the graph owns the
   engine; e2e suite: `acp_session_tests/graph/graph_e2e_tests.rs`.
+- Parallel fan-out (G1): with `KIGI_GRAPH_CONCURRENCY > 1` (default 3,
+  clamp [1,8]) and ≥2 `Ready` nodes, `drive_graph` runs batches via
+  `acp_session_impl/graph_workers.rs` — per node a bounded
+  worker↔verifier subagent loop (`KIGI_GRAPH_NODE_ROUNDS`, default 3;
+  `general-purpose` children; worktree isolation on round 1, resume
+  keeps context+worktree on later rounds; `NODE_RESULT:` /
+  `NODE_VERDICT:` terminal contracts parsed fail-closed), then
+  SEQUENTIAL merge-back via `kigi_workspace::worktree::apply_worktree`
+  (`ApplyMode::Merge`); a conflict fails the node and blocks its
+  dependents while other chains continue. `gn-final` always runs
+  serially on the full goal engine. Concurrency=1 is byte-identical to
+  the serial G0 path. Ceiling: a worker round exceeding the foreground
+  subagent await budget (600s) is cancelled and retried via resume.
 
 ## Milestones (PRD §8.3)
 

crates/codegen/kigi-shell/src/agent/config.rs

@@ -1919,6 +1919,26 @@ impl Config {
     pub(crate) fn resolve_graph(&self) -> Resolved<bool> {
         BoolFlag::env("KIGI_GRAPH").default(false).resolve()
     }
+    /// Max graph nodes running concurrently (`KIGI_GRAPH_CONCURRENCY`).
+    /// 1 = serial (G0-identical); clamped to [1, 8] — the coordinator has
+    /// no cap of its own, so this is the only brake on worker fan-out.
+    pub(crate) fn resolve_graph_concurrency(&self) -> u32 {
+        std::env::var("KIGI_GRAPH_CONCURRENCY")
+            .ok()
+            .and_then(|v| v.parse::<u32>().ok())
+            .unwrap_or(3)
+            .clamp(1, 8)
+    }
+    /// Max worker↔verifier rounds per parallel graph node
+    /// (`KIGI_GRAPH_NODE_ROUNDS`); exhausting them fails the node.
+    /// Clamped to [1, 8].
+    pub(crate) fn resolve_graph_node_rounds(&self) -> u32 {
+        std::env::var("KIGI_GRAPH_NODE_ROUNDS")
+            .ok()
+            .and_then(|v| v.parse::<u32>().ok())
+            .unwrap_or(3)
+            .clamp(1, 8)
+    }
     /// Classifier, planner, and summary all default to goal mode itself: when
     /// `/goal` is on they are on unless config/env/remote says otherwise.
     /// `goal_enabled` is the session's already-resolved master switch (the same

crates/codegen/kigi-shell/src/session/acp_session.rs

@@ -95,6 +95,8 @@ pub use types::{TodoGateDecision, TodoGateReason};
 mod goal;
 #[path = "acp_session_impl/graph.rs"]
 mod graph;
+#[path = "acp_session_impl/graph_workers.rs"]
+mod graph_workers;
 #[path = "acp_session_impl/interjection.rs"]
 mod interjection;
 #[path = "acp_session_impl/tool_calls.rs"]
@@ -599,6 +601,12 @@ pub(crate) struct SessionActor {
     /// layered over the goal engine. Modeled after `goal_tracker` above;
     /// all graph state logic lives in `graph_tracker.rs`.
     pub(crate) graph_tracker: Arc<parking_lot::Mutex<crate::session::graph_tracker::GraphTracker>>,
+    /// Max graph nodes running concurrently (1 = serial G0 behavior).
+    /// Cached at actor construction from `resolve_graph_concurrency`.
+    pub(crate) graph_concurrency: u32,
+    /// Max worker↔verifier rounds per parallel graph node before the
+    /// node fails. Cached at actor construction.
+    pub(crate) graph_node_rounds: u32,
     /// `task_id`s of background tasks (and monitors) that originated during
     /// the goal turn — either spawned by the goal model itself or reparented
     /// from a harness verifier/planner subagent on its exit. Their late

crates/codegen/kigi-shell/src/session/acp_session_impl/graph.rs

@@ -38,7 +38,7 @@ pub(super) enum GraphSetupOutcome {
 /// Compose the goal objective for one graph node. The node spec is the
 /// contract; the graph context line keeps the node model from wandering
 /// into other nodes' scope.
-fn node_goal_objective(
+pub(super) fn node_goal_objective(
     graph_objective: &str,
     node: &GraphNode,
     position: usize,
@@ -157,18 +157,17 @@ impl SessionActor {
         self.persist_graph_state();
         tracing::info!(total, "graph: DAG installed, launching first node");
 
-        match self.launch_next_graph_node().await {
+        match self.drive_graph().await {
             Some(reminder) => GraphSetupOutcome::Inference {
                 reminder,
                 user_msg: format!(
-                    "Graph created: {total} nodes (incl. final verification). Starting node 1."
+                    "Graph created: {total} nodes (incl. final verification). Starting work."
                 ),
             },
-            // Validation guarantees at least one root, so no Ready node
-            // here means the budget gate tripped inside the launcher.
+            // The dispatch loop already settled (parallel completion,
+            // pause, budget) and messaged; point at the status view.
             None => GraphSetupOutcome::Message(
-                "Graph created but no node could start (budget exhausted?). See /graph status."
-                    .to_owned(),
+                "Graph did not enter a serial node. See /graph status.".to_owned(),
             ),
         }
     }
@@ -434,32 +433,64 @@ impl SessionActor {
             .lock()
             .mark_node_achieved(&node_id, rounds, node_tokens);
         self.persist_graph_state();
+        self.drive_graph().await
+    }
 
-        if self.graph_tracker.lock().all_achieved() {
-            let (total, objective) = {
-                let mut tracker = self.graph_tracker.lock();
-                tracker.complete();
-                let snapshot = tracker.snapshot();
-                (
-                    snapshot.map(|s| s.nodes.len()).unwrap_or(0),
-                    snapshot.map(|s| s.objective.clone()).unwrap_or_default(),
+    /// THE single dispatch loop, shared by setup/advance/resume: run
+    /// parallel batches while ≥2 nodes are `Ready` (and the cap allows),
+    /// then launch the next serial node on the goal engine and return
+    /// its reminder — or settle the graph (complete / wedged / paused /
+    /// budget) and return `None`. With `graph_concurrency == 1` this
+    /// degenerates to exactly the G0 serial behavior.
+    pub(super) async fn drive_graph(&self) -> Option<String> {
+        loop {
+            if !self.graph_tracker.lock().is_active() {
+                // Paused/limited during a batch (cancel cascade) or by a
+                // serial-launch failure — the pauser already messaged.
+                return None;
+            }
+            if self.graph_tracker.lock().remaining_budget() == Some(0) {
+                tracing::warn!("graph: budget exhausted at dispatch");
+                self.graph_tracker.lock().budget_limit();
+                self.persist_graph_state();
+                self.send_slash_command_output(
+                    "Graph token budget exhausted. Use /graph clear, then /graph <objective>.",
                 )
-            };
-            self.persist_graph_state();
-            self.reset_goal_engine_state().await;
-            tracing::info!(total, "graph: complete");
-            self.send_slash_command_output(&format!(
-                "Graph complete: all {total} nodes achieved (final verification included).\n\
-                 Objective: {objective}"
-            ))
-            .await;
-            return None;
-        }
-
-        match self.launch_next_graph_node().await {
-            Some(reminder) => Some(reminder),
-            None => {
-                if self.graph_tracker.lock().is_wedged() {
+                .await;
+                return None;
+            }
+            let ready: Vec<String> = self
+                .graph_tracker
+                .lock()
+                .snapshot()
+                .map(|s| {
+                    s.nodes
+                        .iter()
+                        .filter(|n| n.status == NodeStatus::Ready)
+                        .map(|n| n.id.clone())
+                        .collect()
+                })
+                .unwrap_or_default();
+            if ready.is_empty() {
+                if self.graph_tracker.lock().all_achieved() {
+                    let (total, objective) = {
+                        let mut tracker = self.graph_tracker.lock();
+                        tracker.complete();
+                        let snapshot = tracker.snapshot();
+                        (
+                            snapshot.map(|s| s.nodes.len()).unwrap_or(0),
+                            snapshot.map(|s| s.objective.clone()).unwrap_or_default(),
+                        )
+                    };
+                    self.persist_graph_state();
+                    self.reset_goal_engine_state().await;
+                    tracing::info!(total, "graph: complete");
+                    self.send_slash_command_output(&format!(
+                        "Graph complete: all {total} nodes achieved (final verification \
+                         included).\nObjective: {objective}"
+                    ))
+                    .await;
+                } else if self.graph_tracker.lock().is_wedged() {
                     tracing::warn!("graph: wedged (no runnable node, work remaining)");
                     self.graph_tracker.lock().pause_with_message(
                         GoalPauseReason::Verification,
@@ -467,13 +498,53 @@ impl SessionActor {
                     );
                     self.persist_graph_state();
                     self.send_slash_command_output(
-                        "Graph blocked: a dependency chain failed and no runnable node is left. \
-                         See /graph status.",
+                        "Graph blocked: a dependency chain failed and no runnable node is \
+                         left. See /graph status.",
+                    )
+                    .await;
+                } else {
+                    tracing::error!("graph: active with no ready node and work remaining");
+                    self.graph_tracker.lock().pause_with_message(
+                        GoalPauseReason::Infra,
+                        "Scheduler found no runnable node while work remains".to_owned(),
+                    );
+                    self.persist_graph_state();
+                    // Never pause invisibly: the user must know the
+                    // scheduler stopped and why.
+                    self.send_slash_command_output(
+                        "Graph paused: scheduler found no runnable node while work \
+                         remains. See /graph status.",
                     )
                     .await;
                 }
-                None
+                return None;
             }
+            // Parallel batches need per-node worktrees, which need a git
+            // repo. Outside one, degrade to serial (the goal engine works
+            // anywhere) instead of letting N workers share one tree.
+            let parallel_possible = kigi_workspace::session::git::find_git_root_from_path(
+                self.tool_context.cwd.as_path(),
+            )
+            .is_ok();
+            let cap = if parallel_possible {
+                self.graph_concurrency as usize
+            } else {
+                1
+            };
+            if cap <= 1 || ready.len() == 1 {
+                return self.launch_next_graph_node().await;
+            }
+            let batch: Vec<String> = ready.into_iter().take(cap).collect();
+            // No goal is Active during a batch, so arm the interrupt
+            // gate explicitly: a queued user prompt must stack FIFO
+            // behind the batch instead of cancelling the graph turn
+            // (the in-turn loop re-syncs the gate from goal status at
+            // every round start).
+            self.set_goal_loop_active_resource(true).await;
+            self.run_graph_parallel_batch(batch).await;
+            // Loop: the batch resolved nodes (achieved/failed); recompute
+            // and keep driving — another batch, a serial tail (gn-final),
+            // or settle.
         }
     }
 
@@ -560,7 +631,24 @@ impl SessionActor {
                 "Graph is budget-limited. Use /graph clear, then /graph <objective>.".to_owned(),
             ),
             Some(s) if s.is_paused() => {
-                self.graph_tracker.lock().resume();
+                {
+                    let mut tracker = self.graph_tracker.lock();
+                    tracker.resume();
+                    // A cancel during a parallel batch left its nodes
+                    // marked Running with no executor. Demote every
+                    // orphaned in-flight node back to Ready, keeping only
+                    // the node whose goal actually lives in the engine —
+                    // otherwise resume wedges forever (nothing Ready,
+                    // nothing achieved, is_wedged false).
+                    let keep = self
+                        .goal_tracker
+                        .lock()
+                        .snapshot()
+                        .is_some()
+                        .then(|| tracker.current_node_id().map(str::to_owned))
+                        .flatten();
+                    tracker.demote_orphaned_in_flight(keep.as_deref());
+                }
                 self.persist_graph_state();
                 tracing::info!("graph: resumed");
                 // Planning never finished? Re-plan before touching nodes.
@@ -607,13 +695,14 @@ impl SessionActor {
                         }
                     }
                 } else {
-                    match self.launch_next_graph_node().await {
+                    match self.drive_graph().await {
                         Some(reminder) => GraphSetupOutcome::Inference {
                             reminder,
                             user_msg: "Graph resumed.".to_owned(),
                         },
                         None => GraphSetupOutcome::Message(
-                            "Graph resumed but no node is runnable. See /graph status.".to_owned(),
+                            "Graph resumed but did not enter a serial node. See /graph status."
+                                .to_owned(),
                         ),
                     }
                 }
@@ -639,7 +728,7 @@ impl SessionActor {
                 }
                 self.graph_tracker.lock().install_nodes(nodes);
                 self.persist_graph_state();
-                Ok(self.launch_next_graph_node().await)
+                Ok(self.drive_graph().await)
             }
             Err(reason) => {
                 {

crates/codegen/kigi-shell/src/session/acp_session_impl/graph_workers.rs

@@ -0,0 +1,927 @@
+//! Parallel graph-node execution: worker/verifier subagent pairs.
+//!
+//! In parallel mode (`KIGI_GRAPH_CONCURRENCY > 1` with ≥2 `Ready`
+//! nodes) a node does NOT run on the session goal engine — it runs as a
+//! harness-internal `general-purpose` subagent (the implementer toolset)
+//! in its OWN git worktree, adversarially checked by a read-only
+//! verifier subagent, with a bounded worker↔verifier round loop
+//! (`graph_node_rounds`). Achieved nodes merge back into the main tree
+//! SEQUENTIALLY via `kigi_workspace`'s 3-way `apply_worktree`; a merge
+//! conflict fails the node (its dependents block; other chains
+//! continue). The terminal `gn-final` node always runs serially on the
+//! full goal engine because it depends on every other node.
+//!
+//! Known ceiling: a worker round that outlives the foreground subagent
+//! await budget (default 600s) is cancelled and counted as a failed
+//! round with an explicit gap; the next round resumes the same child
+//! session. Fetching results from auto-backgrounded children would need
+//! completed-store plumbing — deferred until real usage demands it.
+
+use std::sync::Arc;
+
+use kigi_tools::implementations::kigi::task::types::{
+    SubagentEvent, SubagentRequest, SubagentRuntimeOverrides,
+};
+
+use super::SessionActor;
+
+const WORKER_PROMPT_TEMPLATE: &str = include_str!("../templates/graph_node_worker_prompt.md");
+const VERIFIER_PROMPT_TEMPLATE: &str = include_str!("../templates/graph_node_verifier_prompt.md");
+
+// Terminal-contract parsing
+
+/// The worker's parsed claim, from the trailing `NODE_RESULT:` line.
+#[derive(Debug, PartialEq, Eq)]
+pub(crate) enum WorkerClaim {
+    Done { summary: String },
+    Blocked { reason: String },
+    Unparseable,
+}
+
+/// Drop ``` fenced code blocks so a QUOTED marker (the templates
+/// themselves contain fenced `NODE_RESULT:`/`NODE_VERDICT:` examples a
+/// child may echo) can never be parsed as the real terminal line.
+fn strip_fenced_blocks(output: &str) -> String {
+    let mut kept = String::with_capacity(output.len());
+    let mut in_fence = false;
+    for line in output.lines() {
+        if line.trim_start().starts_with("```") {
+            in_fence = !in_fence;
+            continue;
+        }
+        if !in_fence {
+            kept.push_str(line);
+            kept.push('\n');
+        }
+    }
+    kept
+}
+
+/// The LAST line that STARTS with `marker` (line-anchored — a
+/// mid-sentence mention never matches), plus everything after it.
+fn last_marker_line(output: &str, marker: &str) -> Option<(String, String)> {
+    let lines: Vec<&str> = output.lines().collect();
+    let idx = lines
+        .iter()
+        .rposition(|l| l.trim_start().trim_start_matches('`').starts_with(marker))?;
+    let value = lines[idx]
+        .trim_start()
+        .trim_start_matches('`')
+        .trim_start_matches(marker)
+        .trim()
+        .trim_matches('`')
+        .to_owned();
+    let tail = lines[idx + 1..].join("\n").trim().to_owned();
+    Some((value, tail))
+}
+
+/// Parse the last line-anchored `NODE_RESULT:` marker outside fenced
+/// blocks; text after the marker line is the summary/reason.
+/// Fail-closed: no marker ⇒ `Unparseable`.
+pub(crate) fn parse_worker_claim(output: &str) -> WorkerClaim {
+    let stripped = strip_fenced_blocks(output);
+    let Some((value, tail)) = last_marker_line(&stripped, "NODE_RESULT:") else {
+        return WorkerClaim::Unparseable;
+    };
+    match value.as_str() {
+        "done" => WorkerClaim::Done { summary: tail },
+        "blocked" => WorkerClaim::Blocked {
+            reason: if tail.is_empty() {
+                "no reason given".to_owned()
+            } else {
+                tail
+            },
+        },
+        _ => WorkerClaim::Unparseable,
+    }
+}
+
+/// The verifier's parsed verdict, from the trailing `NODE_VERDICT:` line.
+/// Fail-closed: anything unparseable is `NotAchieved` with that fact as
+/// the gap.
+#[derive(Debug, PartialEq, Eq)]
+pub(crate) enum NodeVerdict {
+    Achieved,
+    NotAchieved { gaps: Vec<String> },
+}
+
+pub(crate) fn parse_node_verdict(output: &str) -> NodeVerdict {
+    let stripped = strip_fenced_blocks(output);
+    let Some((value, tail)) = last_marker_line(&stripped, "NODE_VERDICT:") else {
+        return NodeVerdict::NotAchieved {
+            gaps: vec!["verifier response lacked a NODE_VERDICT line".to_owned()],
+        };
+    };
+    match value.as_str() {
+        "achieved" => NodeVerdict::Achieved,
+        "not_achieved" => {
+            let gaps: Vec<String> = tail
+                .lines()
+                .map(str::trim)
+                .filter(|l| !l.is_empty() && *l != "GAPS:")
+                .map(|l| l.trim_start_matches('-').trim().to_owned())
+                .filter(|l| !l.is_empty())
+                .collect();
+            NodeVerdict::NotAchieved {
+                gaps: if gaps.is_empty() {
+                    vec!["verifier rejected without naming gaps".to_owned()]
+                } else {
+                    gaps
+                },
+            }
+        }
+        other => NodeVerdict::NotAchieved {
+            gaps: vec![format!("unrecognized verdict token {other:?}")],
+        },
+    }
+}
+
+// Spawner seam (mockable in tests)
+
+pub(crate) struct WorkerSpawnSpec {
+    pub prompt: String,
+    pub description: String,
+    /// Explicit child cwd (verifiers run in the worker's worktree).
+    pub cwd: Option<String>,
+    /// Mint an isolated worktree for the child (first worker round).
+    pub isolation_worktree: bool,
+    /// Resume a prior child session (later worker rounds keep context
+    /// AND the worktree).
+    pub resume_from: Option<String>,
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct WorkerSpawnOutcome {
+    pub success: bool,
+    pub cancelled: bool,
+    pub backgrounded: bool,
+    pub output: String,
+    pub error: Option<String>,
+    pub child_session_id: String,
+    pub tokens_used: u64,
+    pub worktree_path: Option<String>,
+}
+
+#[async_trait::async_trait]
+pub(crate) trait GraphWorkerSpawner: Send + Sync {
+    /// Spawn one child and await its terminal result. `Err` = transport
+    /// failure (coordinator gone).
+    async fn spawn(&self, id: &str, spec: WorkerSpawnSpec) -> Result<WorkerSpawnOutcome, String>;
+    /// Best-effort cancel of a still-running child (budget overrun).
+    async fn cancel(&self, subagent_id: &str);
+}
+
+/// Production spawner: raw harness-internal `SubagentEvent::Spawn`,
+/// exactly the goal-classifier wire (`surface_completion: false`, no
+/// fork), plus worktree isolation / cwd override for node work.
+pub(crate) struct GraphWorkerChannelSpawner {
+    pub event_tx: tokio::sync::mpsc::UnboundedSender<SubagentEvent>,
+    pub parent_session_id: String,
+    pub parent_prompt_id: Option<String>,
+}
+
+#[async_trait::async_trait]
+impl GraphWorkerSpawner for GraphWorkerChannelSpawner {
+    async fn spawn(&self, id: &str, spec: WorkerSpawnSpec) -> Result<WorkerSpawnOutcome, String> {
+        let (result_tx, result_rx) = tokio::sync::oneshot::channel();
+        let request = SubagentRequest {
+            id: id.to_string(),
+            prompt: spec.prompt,
+            description: spec.description,
+            // The implementer toolset (full read/edit/bash inventory);
+            // verifier read-only-ness is prompt-enforced, same as the
+            // goal skeptic panel.
+            subagent_type: "general-purpose".to_string(),
+            parent_session_id: self.parent_session_id.clone(),
+            parent_prompt_id: self.parent_prompt_id.clone(),
+            resume_from: spec.resume_from,
+            cwd: spec.cwd,
+            runtime_overrides: SubagentRuntimeOverrides {
+                isolation: spec
+                    .isolation_worktree
+                    .then_some(kigi_tool_types::SubagentIsolationMode::Worktree),
+                ..Default::default()
+            },
+            run_in_background: false,
+            // Harness-internal: never surfaces to the model's idle reminder.
+            surface_completion: false,
+            fork_context: false,
+            result_tx,
+        };
+        if self
+            .event_tx
+            .send(SubagentEvent::Spawn(Box::new(request)))
+            .is_err()
+        {
+            return Err("subagent coordinator channel closed".to_owned());
+        }
+        let result = result_rx
+            .await
+            .map_err(|_| "subagent result channel dropped".to_owned())?;
+        Ok(WorkerSpawnOutcome {
+            success: result.success,
+            cancelled: result.cancelled,
+            backgrounded: result.backgrounded,
+            output: result.output.to_string(),
+            error: result.error.clone(),
+            child_session_id: result.child_session_id.clone(),
+            tokens_used: result.tokens_used,
+            worktree_path: result.worktree_path.clone(),
+        })
+    }
+
+    async fn cancel(&self, subagent_id: &str) {
+        use kigi_tools::implementations::kigi::task::types::{
+            SubagentCancelRequest, SubagentCancelTarget,
+        };
+        let (respond_to, ack) = tokio::sync::oneshot::channel();
+        let _ = self
+            .event_tx
+            .send(SubagentEvent::Cancel(SubagentCancelRequest {
+                target: SubagentCancelTarget::SubagentId(subagent_id.to_string()),
+                respond_to,
+            }));
+        let _ = ack.await;
+    }
+}
+
+// Per-node bounded closed loop
+
+#[derive(Debug)]
+pub(crate) struct NodeRunReport {
+    pub node_id: String,
+    pub achieved: bool,
+    /// Worker summary on success; failure reason otherwise.
+    pub detail: String,
+    pub rounds: u32,
+    pub tokens_used: i64,
+    pub worktree_path: Option<String>,
+    /// Last worker child session id (audit link, stored on the node).
+    pub worker_session_id: Option<String>,
+}
+
+fn worker_prompt(node_objective: &str, gaps: &[String]) -> String {
+    let mut p = String::with_capacity(WORKER_PROMPT_TEMPLATE.len() + node_objective.len() + 256);
+    p.push_str(WORKER_PROMPT_TEMPLATE);
+    p.push_str("\n\nNODE OBJECTIVE:\n");
+    p.push_str(node_objective);
+    if !gaps.is_empty() {
+        p.push_str("\n\nGAPS (from the previous verification round — close exactly these):\n");
+        for gap in gaps {
+            p.push_str("- ");
+            p.push_str(gap);
+            p.push('\n');
+        }
+    }
+    p
+}
+
+fn verifier_prompt(node_objective: &str, worker_summary: &str) -> String {
+    // Neutralize terminal-contract tokens in the worker-controlled
+    // summary so a lazy/adversarial claim cannot smuggle marker lines
+    // into the verifier's context.
+    let safe_summary = worker_summary
+        .replace("NODE_VERDICT", "NODE-VERDICT")
+        .replace("NODE_RESULT", "NODE-RESULT");
+    format!(
+        "{VERIFIER_PROMPT_TEMPLATE}\n\nNODE OBJECTIVE (the contract to judge):\n{node_objective}\n\n\
+         IMPLEMENTER'S CLAIM (audit it, do not trust it):\n{safe_summary}\n"
+    )
+}
+
+/// Drive one node through bounded worker↔verifier rounds. Never panics;
+/// every failure path returns a `NodeRunReport` with a precise reason.
+pub(crate) async fn run_node_to_verdict(
+    spawner: &Arc<dyn GraphWorkerSpawner>,
+    node_id: &str,
+    node_objective: &str,
+    rounds_cap: u32,
+) -> NodeRunReport {
+    let mut tokens: i64 = 0;
+    let mut gaps: Vec<String> = Vec::new();
+    let mut resume_from: Option<String> = None;
+    let mut worktree_path: Option<String> = None;
+    let mut last_gaps_summary = String::new();
+
+    for round in 1..=rounds_cap {
+        let spawn_id = format!("graph-{node_id}-w{round}-{}", uuid::Uuid::now_v7());
+        let spec_was_isolated = resume_from.is_none();
+        let spec = WorkerSpawnSpec {
+            prompt: worker_prompt(node_objective, &gaps),
+            description: format!("graph node worker ({node_id})"),
+            cwd: None,
+            // Fresh worktree only on the first round; resumes reuse it.
+            isolation_worktree: spec_was_isolated,
+            resume_from: resume_from.clone(),
+        };
+        tracing::info!(%node_id, round, resumed = resume_from.is_some(), "graph worker: round start");
+        let outcome = match spawner.spawn(&spawn_id, spec).await {
+            Ok(o) => o,
+            Err(err) => {
+                return NodeRunReport {
+                    node_id: node_id.to_owned(),
+                    achieved: false,
+                    detail: format!("worker transport failure: {err}"),
+                    rounds: round,
+                    tokens_used: tokens,
+                    worktree_path,
+                    worker_session_id: resume_from,
+                };
+            }
+        };
+        let round_requested_isolation = spec_was_isolated;
+        tokens = tokens.saturating_add(outcome.tokens_used as i64);
+        if outcome.worktree_path.is_some() {
+            worktree_path = outcome.worktree_path.clone();
+        }
+        // Adopt-guard: an in-band spawn failure carries an EMPTY child id;
+        // adopting it would make the next round a fresh UNISOLATED spawn
+        // in the shared tree while verify/merge still target the stale
+        // worktree. Keep the last valid id (or None ⇒ re-mint isolation).
+        if !outcome.child_session_id.is_empty() {
+            resume_from = Some(outcome.child_session_id.clone());
+        }
+
+        if outcome.cancelled {
+            return NodeRunReport {
+                node_id: node_id.to_owned(),
+                achieved: false,
+                detail: "worker cancelled".to_owned(),
+                rounds: round,
+                tokens_used: tokens,
+                worktree_path,
+                worker_session_id: resume_from,
+            };
+        }
+        if outcome.backgrounded {
+            // Ceiling (see module doc): cancel the runaway child and
+            // burn the round; the resume keeps its context.
+            tracing::warn!(%node_id, round, "graph worker: exceeded foreground await budget; cancelling round");
+            // Cancel by the SPAWN REQUEST id — the coordinator's cancel
+            // maps are keyed by it, not by the child session id.
+            spawner.cancel(&spawn_id).await;
+            gaps = vec![
+                "the previous round exceeded the foreground time budget and was cancelled; \
+                 split the remaining work into smaller, faster steps"
+                    .to_owned(),
+            ];
+            last_gaps_summary = gaps.join("; ");
+            continue;
+        }
+        if !outcome.success {
+            let err = outcome.error.unwrap_or_else(|| "unknown error".to_owned());
+            tracing::warn!(%node_id, round, %err, "graph worker: round failed");
+            gaps = vec![format!("the previous round failed with an error: {err}")];
+            last_gaps_summary = gaps.join("; ");
+            continue;
+        }
+
+        // Isolation guard: a SUCCESSFUL first round that came back with
+        // no worktree means isolation silently degraded (non-git dir,
+        // worktree creation failure, or the snapshot-disposal flag
+        // deleted it before we saw it). Running parallel writers in the
+        // shared tree — or merging a disposed tree — is never OK.
+        if round_requested_isolation && outcome.worktree_path.is_none() {
+            return NodeRunReport {
+                node_id: node_id.to_owned(),
+                achieved: false,
+                detail: "worktree isolation unavailable for this node (non-git directory, \
+                         worktree creation failure, or KIGI_SUBAGENT_WORKTREE_SNAPSHOT \
+                         disposal); parallel execution requires isolation"
+                    .to_owned(),
+                rounds: round,
+                tokens_used: tokens,
+                worktree_path: None,
+                worker_session_id: resume_from,
+            };
+        }
+
+        match parse_worker_claim(&outcome.output) {
+            WorkerClaim::Blocked { reason } => {
+                return NodeRunReport {
+                    node_id: node_id.to_owned(),
+                    achieved: false,
+                    detail: format!("worker reported blocked: {reason}"),
+                    rounds: round,
+                    tokens_used: tokens,
+                    worktree_path,
+                    worker_session_id: resume_from,
+                };
+            }
+            WorkerClaim::Unparseable => {
+                gaps = vec![
+                    "the previous round's final message lacked the required NODE_RESULT line"
+                        .to_owned(),
+                ];
+                last_gaps_summary = gaps.join("; ");
+                continue;
+            }
+            WorkerClaim::Done { summary } => {
+                let verify_id = format!("graph-{node_id}-v{round}-{}", uuid::Uuid::now_v7());
+                let verify_spec = WorkerSpawnSpec {
+                    prompt: verifier_prompt(node_objective, &summary),
+                    description: format!("graph node verifier ({node_id})"),
+                    // The verifier inspects the worker's worktree.
+                    cwd: worktree_path.clone(),
+                    isolation_worktree: false,
+                    resume_from: None,
+                };
+                let verdict = match spawner.spawn(&verify_id, verify_spec).await {
+                    Ok(v) => {
+                        tokens = tokens.saturating_add(v.tokens_used as i64);
+                        if v.success {
+                            parse_node_verdict(&v.output)
+                        } else {
+                            // Fail CLOSED: an unverified claim never passes.
+                            NodeVerdict::NotAchieved {
+                                gaps: vec![format!(
+                                    "verifier run failed ({}); the claim is unverified",
+                                    v.error.unwrap_or_else(|| "unknown error".to_owned())
+                                )],
+                            }
+                        }
+                    }
+                    Err(err) => NodeVerdict::NotAchieved {
+                        gaps: vec![format!("verifier transport failure: {err}")],
+                    },
+                };
+                match verdict {
+                    NodeVerdict::Achieved => {
+                        tracing::info!(%node_id, round, tokens, "graph worker: node verified achieved");
+                        return NodeRunReport {
+                            node_id: node_id.to_owned(),
+                            achieved: true,
+                            detail: summary,
+                            rounds: round,
+                            tokens_used: tokens,
+                            worktree_path,
+                            worker_session_id: resume_from,
+                        };
+                    }
+                    NodeVerdict::NotAchieved { gaps: new_gaps } => {
+                        tracing::info!(%node_id, round, gap_count = new_gaps.len(), "graph worker: verifier rejected round");
+                        last_gaps_summary = new_gaps.join("; ");
+                        gaps = new_gaps;
+                    }
+                }
+            }
+        }
+    }
+    NodeRunReport {
+        node_id: node_id.to_owned(),
+        achieved: false,
+        detail: format!(
+            "verification rejected after {rounds_cap} rounds; last gaps: {last_gaps_summary}"
+        ),
+        rounds: rounds_cap,
+        tokens_used: tokens,
+        worktree_path,
+        worker_session_id: resume_from,
+    }
+}
+
+/// `git rev-parse HEAD` of `dir`, `None` outside a git repo.
+async fn git_head(dir: &std::path::Path) -> Option<String> {
+    let output = tokio::process::Command::new("git")
+        .arg("-C")
+        .arg(dir)
+        .args(["rev-parse", "HEAD"])
+        .output()
+        .await
+        .ok()?;
+    output
+        .status
+        .success()
+        .then(|| String::from_utf8_lossy(&output.stdout).trim().to_owned())
+}
+
+// SessionActor integration
+
+impl SessionActor {
+    /// Production worker spawner wired to this session's coordinator.
+    fn graph_worker_spawner(&self) -> Option<Arc<dyn GraphWorkerSpawner>> {
+        let event_tx = self.tool_context.subagent_event_tx.clone()?;
+        let parent_prompt_id = self
+            .current_prompt_id
+            .lock()
+            .expect("current_prompt_id mutex poisoned")
+            .clone();
+        Some(Arc::new(GraphWorkerChannelSpawner {
+            event_tx,
+            parent_session_id: self.session_id_string(),
+            parent_prompt_id,
+        }))
+    }
+
+    /// Run one parallel batch of `Ready` nodes to their verdicts, then
+    /// merge achieved worktrees back SEQUENTIALLY in batch order. All
+    /// tracker mutations + persistence happen here; the caller re-reads
+    /// the tracker afterwards.
+    pub(super) async fn run_graph_parallel_batch(&self, node_ids: Vec<String>) {
+        let Some(spawner) = self.graph_worker_spawner() else {
+            tracing::error!("graph batch: no subagent coordinator; pausing graph");
+            self.graph_tracker.lock().pause_with_message(
+                crate::session::goal_tracker::GoalPauseReason::Infra,
+                "No subagent coordinator available for parallel execution".to_owned(),
+            );
+            self.persist_graph_state();
+            return;
+        };
+        // Compose objectives + mark Running under one lock pass.
+        let mut jobs: Vec<(String, String)> = Vec::with_capacity(node_ids.len());
+        {
+            let mut tracker = self.graph_tracker.lock();
+            let Some(snapshot) = tracker.snapshot() else {
+                return;
+            };
+            let total = snapshot.nodes.len();
+            let objective = snapshot.objective.clone();
+            for id in &node_ids {
+                if let Some(pos) = snapshot.nodes.iter().position(|n| n.id == *id) {
+                    let node = &snapshot.nodes[pos];
+                    jobs.push((
+                        id.clone(),
+                        super::graph::node_goal_objective(&objective, node, pos + 1, total),
+                    ));
+                }
+            }
+            for (id, _) in &jobs {
+                tracker.mark_node_running(id, String::new());
+            }
+            // `current_node` means "the node on the serial goal engine";
+            // batch nodes are tracked by their own Running status.
+            if let Some(s) = tracker.snapshot_mut() {
+                s.current_node = None;
+            }
+        }
+        self.persist_graph_state();
+        // Merge-base integrity: if the main repo HEAD moves during the
+        // batch (external commit), apply_worktree would diff against the
+        // wrong base and silently reverse-apply those commits. Capture
+        // HEAD now; every merge re-checks it.
+        let head_at_fanout = git_head(self.tool_context.cwd.as_path()).await;
+        let rounds_cap = self.graph_node_rounds;
+        tracing::info!(batch = jobs.len(), rounds_cap, "graph batch: fan-out");
+
+        let reports = futures::future::join_all(jobs.iter().map(|(id, objective)| {
+            let spawner = spawner.clone();
+            async move { run_node_to_verdict(&spawner, id, objective, rounds_cap).await }
+        }))
+        .await;
+
+        // Sequential merge + tracker resolution in batch order.
+        let mut achieved = 0usize;
+        let mut failed = 0usize;
+        for report in reports {
+            // Stamp the worker session id for audit (goal_id slot).
+            if let Some(worker_id) = &report.worker_session_id
+                && let Some(node) = self
+                    .graph_tracker
+                    .lock()
+                    .snapshot_mut()
+                    .and_then(|s| s.nodes.iter_mut().find(|n| n.id == report.node_id))
+            {
+                node.goal_id = Some(worker_id.clone());
+            }
+            if !report.achieved {
+                failed += 1;
+                {
+                    let mut tracker = self.graph_tracker.lock();
+                    // Budget integrity: a failed node's tokens were still
+                    // spent — charge them before failing the node.
+                    tracker.charge_node_tokens(&report.node_id, report.tokens_used);
+                    tracker.mark_node_failed(&report.node_id, report.detail.clone());
+                }
+                self.persist_graph_state();
+                continue;
+            }
+            match self
+                .merge_node_worktree(
+                    &report.node_id,
+                    report.worktree_path.as_deref(),
+                    head_at_fanout.as_deref(),
+                )
+                .await
+            {
+                Ok(()) => {
+                    achieved += 1;
+                    self.graph_tracker.lock().mark_node_achieved(
+                        &report.node_id,
+                        report.rounds,
+                        report.tokens_used,
+                    );
+                }
+                Err(detail) => {
+                    failed += 1;
+                    self.graph_tracker
+                        .lock()
+                        .mark_node_failed(&report.node_id, detail);
+                }
+            }
+            self.persist_graph_state();
+        }
+        tracing::info!(achieved, failed, "graph batch: settled");
+        self.send_slash_command_output(&format!(
+            "Graph batch settled: {achieved} node(s) achieved, {failed} failed."
+        ))
+        .await;
+    }
+
+    /// Merge one achieved node's worktree back into the main tree with
+    /// the 3-way apply. `None` worktree (isolation soft-fallback) means
+    /// the worker already wrote in the shared tree — nothing to merge.
+    pub(super) async fn merge_node_worktree(
+        &self,
+        node_id: &str,
+        worktree_path: Option<&str>,
+        expected_main_head: Option<&str>,
+    ) -> Result<(), String> {
+        let Some(worktree_path) = worktree_path else {
+            tracing::warn!(
+                %node_id,
+                "graph merge: worker ran without worktree isolation (soft fallback); nothing to merge"
+            );
+            return Ok(());
+        };
+        use kigi_workspace::worktree::{
+            ApplyMode, ApplyWorktreeRequest, ApplyWorktreeResponse, apply_worktree,
+        };
+        // apply_worktree diffs against the main repo HEAD AT APPLY TIME;
+        // if HEAD moved since fan-out, that diff would silently
+        // reverse-apply the external commits. Fail the node loudly.
+        if let Some(expected) = expected_main_head {
+            let current = git_head(self.tool_context.cwd.as_path()).await;
+            if current.as_deref() != Some(expected) {
+                return Err(format!(
+                    "main repository HEAD moved during the batch (was {expected}, now {}); \
+                     merge aborted for safety — /graph resume re-runs the node",
+                    current.as_deref().unwrap_or("unknown")
+                ));
+            }
+        }
+        let request = ApplyWorktreeRequest {
+            session_id: self.session_id_string(),
+            worktree_path: worktree_path.to_owned(),
+            mode: ApplyMode::Merge,
+        };
+        match apply_worktree(&request).await {
+            Ok(ApplyWorktreeResponse::Success { files, .. }) => {
+                tracing::info!(%node_id, files = files.len(), "graph merge: applied");
+                // Storage discipline: the changes now live in the main
+                // tree, so the worktree is dead weight — remove it.
+                // Best-effort (a failed removal only leaks disk, never
+                // progress) but always logged. Failed nodes KEEP their
+                // worktree for postmortem.
+                if let Err(err) = kigi_workspace::worktree::remove_subagent_worktree(
+                    std::path::Path::new(worktree_path),
+                )
+                .await
+                {
+                    tracing::warn!(%node_id, %err, "graph merge: worktree cleanup failed");
+                }
+                Ok(())
+            }
+            Ok(ApplyWorktreeResponse::Conflicts { conflicts, .. }) => {
+                let names: Vec<String> = conflicts.iter().map(|c| c.path.clone()).collect();
+                tracing::warn!(%node_id, ?names, "graph merge: conflicts; failing node");
+                Err(format!("merge conflict in: {}", names.join(", ")))
+            }
+            Err(err) => {
+                tracing::warn!(%node_id, %err, "graph merge: apply failed");
+                Err(format!("worktree apply failed: {err}"))
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn worker_claim_parses_done_blocked_and_garbage() {
+        assert_eq!(
+            parse_worker_claim("work...\nNODE_RESULT: done\nBuilt X; tests pass."),
+            WorkerClaim::Done {
+                summary: "Built X; tests pass.".to_owned()
+            }
+        );
+        assert_eq!(
+            parse_worker_claim("NODE_RESULT: blocked\nno compiler available"),
+            WorkerClaim::Blocked {
+                reason: "no compiler available".to_owned()
+            }
+        );
+        assert_eq!(
+            parse_worker_claim("all done, promise!"),
+            WorkerClaim::Unparseable
+        );
+        // Last marker wins (a quoted earlier marker cannot spoof).
+        assert_eq!(
+            parse_worker_claim(
+                "NODE_RESULT: done\nold\n...more work...\nNODE_RESULT: blocked\nreal"
+            ),
+            WorkerClaim::Blocked {
+                reason: "real".to_owned()
+            }
+        );
+    }
+
+    #[test]
+    fn verdict_parses_achieved_gaps_and_fails_closed() {
+        assert_eq!(
+            parse_node_verdict("checked\nNODE_VERDICT: achieved"),
+            NodeVerdict::Achieved
+        );
+        assert_eq!(
+            parse_node_verdict(
+                "NODE_VERDICT: not_achieved\nGAPS:\n- test suite not run\n- claim B unverified"
+            ),
+            NodeVerdict::NotAchieved {
+                gaps: vec![
+                    "test suite not run".to_owned(),
+                    "claim B unverified".to_owned()
+                ]
+            }
+        );
+        assert!(matches!(
+            parse_node_verdict("looks good to me"),
+            NodeVerdict::NotAchieved { gaps } if gaps[0].contains("lacked a NODE_VERDICT")
+        ));
+        assert!(matches!(
+            parse_node_verdict("NODE_VERDICT: maybe"),
+            NodeVerdict::NotAchieved { gaps } if gaps[0].contains("unrecognized verdict")
+        ));
+        assert!(matches!(
+            parse_node_verdict("NODE_VERDICT: not_achieved"),
+            NodeVerdict::NotAchieved { gaps } if gaps[0].contains("without naming gaps")
+        ));
+    }
+
+    #[test]
+    fn fenced_template_echo_cannot_spoof_markers() {
+        // A worker echoing the template's fenced examples must stay
+        // unparseable; only its own line-anchored terminal marker counts.
+        let echoed = "Here is my plan:\n```\nNODE_RESULT: done\n```\nstill working...";
+        assert_eq!(parse_worker_claim(echoed), WorkerClaim::Unparseable);
+        let real = "```\nNODE_RESULT: blocked\n```\n...work...\nNODE_RESULT: done\nall built";
+        assert_eq!(
+            parse_worker_claim(real),
+            WorkerClaim::Done {
+                summary: "all built".to_owned()
+            }
+        );
+        // Mid-sentence mention is not a marker (line-anchored scan).
+        assert_eq!(
+            parse_worker_claim("I will print NODE_RESULT: done when finished"),
+            WorkerClaim::Unparseable
+        );
+        // Same discipline for the verifier.
+        assert!(matches!(
+            parse_node_verdict("quoting:\n```\nNODE_VERDICT: achieved\n```\nhmm"),
+            NodeVerdict::NotAchieved { .. }
+        ));
+    }
+
+    struct MockSpawner {
+        replies: std::sync::Mutex<std::collections::VecDeque<WorkerSpawnOutcome>>,
+        specs: std::sync::Mutex<Vec<(String, Option<String>, bool)>>,
+        cancels: std::sync::Mutex<Vec<String>>,
+    }
+
+    #[async_trait::async_trait]
+    impl GraphWorkerSpawner for MockSpawner {
+        async fn spawn(
+            &self,
+            id: &str,
+            spec: WorkerSpawnSpec,
+        ) -> Result<WorkerSpawnOutcome, String> {
+            self.specs.lock().unwrap().push((
+                id.to_owned(),
+                spec.resume_from.clone(),
+                spec.isolation_worktree,
+            ));
+            Ok(self
+                .replies
+                .lock()
+                .unwrap()
+                .pop_front()
+                .expect("unexpected extra spawn"))
+        }
+        async fn cancel(&self, subagent_id: &str) {
+            self.cancels.lock().unwrap().push(subagent_id.to_owned());
+        }
+    }
+
+    fn outcome(output: &str) -> WorkerSpawnOutcome {
+        WorkerSpawnOutcome {
+            success: true,
+            cancelled: false,
+            backgrounded: false,
+            output: output.to_owned(),
+            error: None,
+            child_session_id: "child-1".to_owned(),
+            tokens_used: 5,
+            worktree_path: Some("/wt".to_owned()),
+        }
+    }
+
+    #[tokio::test]
+    async fn backgrounded_round_cancels_by_spawn_id_and_resumes_next_round() {
+        let mut bg = outcome("");
+        bg.backgrounded = true;
+        let replies = std::collections::VecDeque::from(vec![
+            bg,                                     // round 1: budget overrun
+            outcome("NODE_RESULT: done\nfinished"), // round 2: worker done
+            outcome("NODE_VERDICT: achieved"),      // round 2: verifier
+        ]);
+        // Keep a concrete handle for assertions; hand the trait object in.
+        let mock = Arc::new(MockSpawner {
+            replies: std::sync::Mutex::new(replies),
+            specs: std::sync::Mutex::new(Vec::new()),
+            cancels: std::sync::Mutex::new(Vec::new()),
+        });
+        let spawner: Arc<dyn GraphWorkerSpawner> = mock.clone();
+        let report = run_node_to_verdict(&spawner, "gn-x", "do x", 3).await;
+        assert!(report.achieved, "{}", report.detail);
+        assert_eq!(report.rounds, 2, "backgrounded round burned, retry won");
+        assert_eq!(report.tokens_used, 15, "all three spawns charged");
+
+        let cancels = mock.cancels.lock().unwrap().clone();
+        assert_eq!(cancels.len(), 1, "runaway child cancelled once");
+        assert!(
+            cancels[0].starts_with("graph-gn-x-w1-"),
+            "cancel must target the SPAWN REQUEST id (coordinator map key), got {}",
+            cancels[0]
+        );
+        let specs = mock.specs.lock().unwrap().clone();
+        assert_eq!(specs.len(), 3);
+        assert!(
+            specs[0].1.is_none() && specs[0].2,
+            "round 1: fresh + isolated"
+        );
+        assert_eq!(
+            specs[1].1.as_deref(),
+            Some("child-1"),
+            "round 2 resumes the backgrounded child's session"
+        );
+        assert!(!specs[1].2, "resume never re-mints isolation");
+    }
+
+    #[tokio::test]
+    async fn empty_child_id_is_never_adopted_as_resume_target() {
+        // In-band spawn failure: success=false, child_session_id="".
+        let failed = WorkerSpawnOutcome {
+            success: false,
+            cancelled: false,
+            backgrounded: false,
+            output: String::new(),
+            error: Some("boom".to_owned()),
+            child_session_id: String::new(),
+            tokens_used: 0,
+            worktree_path: None,
+        };
+        let replies = std::collections::VecDeque::from(vec![
+            failed,                                 // round 1: in-band failure
+            outcome("NODE_RESULT: done\nfinished"), // round 2: worker (fresh, isolated)
+            outcome("NODE_VERDICT: achieved"),      // round 2: verifier
+        ]);
+        let mock = Arc::new(MockSpawner {
+            replies: std::sync::Mutex::new(replies),
+            specs: std::sync::Mutex::new(Vec::new()),
+            cancels: std::sync::Mutex::new(Vec::new()),
+        });
+        let spawner: Arc<dyn GraphWorkerSpawner> = mock.clone();
+        let report = run_node_to_verdict(&spawner, "gn-y", "do y", 3).await;
+        assert!(report.achieved, "{}", report.detail);
+        let specs = mock.specs.lock().unwrap().clone();
+        assert!(
+            specs[1].1.is_none(),
+            "an empty child id must NOT be adopted; retry is a fresh spawn"
+        );
+        assert!(
+            specs[1].2,
+            "fresh retry re-mints worktree isolation (no unisolated escape)"
+        );
+    }
+
+    #[tokio::test]
+    async fn successful_isolated_round_without_worktree_fails_the_node() {
+        let mut no_wt = outcome("NODE_RESULT: done\nfinished");
+        no_wt.worktree_path = None;
+        let replies = std::collections::VecDeque::from(vec![no_wt]);
+        let mock = Arc::new(MockSpawner {
+            replies: std::sync::Mutex::new(replies),
+            specs: std::sync::Mutex::new(Vec::new()),
+            cancels: std::sync::Mutex::new(Vec::new()),
+        });
+        let spawner: Arc<dyn GraphWorkerSpawner> = mock.clone();
+        let report = run_node_to_verdict(&spawner, "gn-z", "do z", 3).await;
+        assert!(!report.achieved);
+        assert!(
+            report.detail.contains("isolation unavailable"),
+            "{}",
+            report.detail
+        );
+    }
+}

crates/codegen/kigi-shell/src/session/acp_session_impl/spawn.rs

@@ -1099,6 +1099,8 @@ pub(crate) async fn spawn_session_actor(
         goal_tracker,
         graph_enabled,
         graph_tracker,
+        graph_concurrency: effective_config.resolve_graph_concurrency(),
+        graph_node_rounds: effective_config.resolve_graph_node_rounds(),
         goal_turn_task_ids: parking_lot::Mutex::new(std::collections::HashSet::new()),
         goal_continuation_streak: std::sync::atomic::AtomicU32::new(0),
         goal_blocked_streak: std::sync::atomic::AtomicU32::new(0),

crates/codegen/kigi-shell/src/session/acp_session_impl/tasks_cancel.rs

@@ -478,6 +478,14 @@ impl SessionActor {
         if let Some(running_task) = running_task {
             running_task.abort();
         }
+        // Re-sweep subagents AFTER the abort: the first sweep raced the
+        // still-live turn future, which may have spawned NEW harness
+        // children (graph batch workers/verifiers) between the sweep and
+        // the abort. Coordinator-channel FIFO guarantees this second
+        // Cancel lands after any Spawn the turn issued before it died.
+        if cancel_subagents && let Some(prompt_id) = cancelled_prompt_id.as_deref() {
+            self.cancel_subagents_for_prompt_id(prompt_id);
+        }
         // The aborted turn's `BlockingWaitGuard`s drop asynchronously (they
         // live in tool futures owned by the drainer task / subagent spawn
         // task). Until they do, `queue_input` would read a stale depth > 0 and

crates/codegen/kigi-shell/src/session/acp_session_tests/cancel_running_task_tests.rs

@@ -216,6 +216,8 @@ async fn persist_ack_waits_for_disk_flush_before_success() {
                 graph_tracker: Arc::new(parking_lot::Mutex::new(
                     crate::session::graph_tracker::GraphTracker::new(std::env::temp_dir()),
                 )),
+                graph_concurrency: 1,
+                graph_node_rounds: 3,
                 goal_turn_task_ids: parking_lot::Mutex::new(std::collections::HashSet::new()),
                 goal_continuation_streak: std::sync::atomic::AtomicU32::new(0),
                 goal_blocked_streak: std::sync::atomic::AtomicU32::new(0),
@@ -656,6 +658,8 @@ async fn first_turn_memory_injection_disabled_does_not_persist_to_chat_history()
                 graph_tracker: Arc::new(parking_lot::Mutex::new(
                     crate::session::graph_tracker::GraphTracker::new(std::env::temp_dir()),
                 )),
+                graph_concurrency: 1,
+                graph_node_rounds: 3,
                 goal_turn_task_ids: parking_lot::Mutex::new(std::collections::HashSet::new()),
                 goal_continuation_streak: std::sync::atomic::AtomicU32::new(0),
                 goal_blocked_streak: std::sync::atomic::AtomicU32::new(0),
@@ -905,6 +909,8 @@ async fn cancel_running_task_teardown_clears_running_and_pending_work() {
                 graph_tracker: Arc::new(parking_lot::Mutex::new(
                     crate::session::graph_tracker::GraphTracker::new(std::env::temp_dir()),
                 )),
+                graph_concurrency: 1,
+                graph_node_rounds: 3,
                 goal_turn_task_ids: parking_lot::Mutex::new(std::collections::HashSet::new()),
                 goal_continuation_streak: std::sync::atomic::AtomicU32::new(0),
                 goal_blocked_streak: std::sync::atomic::AtomicU32::new(0),
@@ -1887,6 +1893,8 @@ async fn cancel_propagates_to_sampler_handle_so_no_further_emission() {
                 graph_tracker: Arc::new(parking_lot::Mutex::new(
                     crate::session::graph_tracker::GraphTracker::new(std::env::temp_dir()),
                 )),
+                graph_concurrency: 1,
+                graph_node_rounds: 3,
                 goal_turn_task_ids: parking_lot::Mutex::new(std::collections::HashSet::new()),
                 goal_continuation_streak: std::sync::atomic::AtomicU32::new(0),
                 goal_blocked_streak: std::sync::atomic::AtomicU32::new(0),

crates/codegen/kigi-shell/src/session/acp_session_tests/graph/graph_e2e_tests.rs

@@ -91,12 +91,29 @@ async fn make_graph_actor(
     SessionActor,
     TempDir,
     tokio::sync::mpsc::UnboundedReceiver<PersistenceMsg>,
+) {
+    let (mut actor, tmp, rx) = make_graph_actor_detached().await;
+    actor.tool_context.subagent_event_tx = Some(coordinator_tx);
+    (actor, tmp, rx)
+}
+
+/// Like [`make_graph_actor`] but with NO coordinator attached — used by
+/// parallel tests whose coordinator needs the repo path (`tmp.path()`)
+/// before it can mint worktrees.
+async fn make_graph_actor_detached() -> (
+    SessionActor,
+    TempDir,
+    tokio::sync::mpsc::UnboundedReceiver<PersistenceMsg>,
 ) {
     let tmp = TempDir::new().expect("tempdir");
     let (gateway_tx, _gateway_rx) =
         tokio::sync::mpsc::unbounded_channel::<kigi_acp_lib::AcpClientMessage>();
     let (persistence_tx, persistence_rx) = tokio::sync::mpsc::unbounded_channel::<PersistenceMsg>();
     let mut actor = create_test_actor(0, 256_000, 85, gateway_tx, persistence_tx).await;
+    // Parallel batches require a real git repo (worktree isolation +
+    // merge-back); serial paths tolerate it. One empty commit suffices.
+    init_git_repo(tmp.path());
+    actor.tool_context.cwd = kigi_paths::AbsPathBuf::new(tmp.path().to_path_buf()).unwrap();
     actor.events = crate::session::events::EventTracker::new(tmp.path());
     actor.goal_enabled = true;
     set_goal_harness_for_tests(&actor);
@@ -108,7 +125,6 @@ async fn make_graph_actor(
     actor.graph_tracker = Arc::new(parking_lot::Mutex::new(GraphTracker::new(
         tmp.path().to_path_buf(),
     )));
-    actor.tool_context.subagent_event_tx = Some(coordinator_tx);
     (actor, tmp, persistence_rx)
 }
 
@@ -693,3 +709,694 @@ async fn graph_status_renders_glyphs_deps_tokens_and_pause() {
         .await;
     unsafe { std::env::remove_var(ENV_FLAG) };
 }
+
+// ── G1: parallel fan-out ────────────────────────────────────────────
+
+/// One captured harness spawn, for post-hoc assertions.
+#[derive(Debug, Clone)]
+struct CapturedSpawn {
+    prompt: String,
+    resume_from: Option<String>,
+    cwd: Option<String>,
+    isolation_worktree: bool,
+}
+
+fn run_git(dir: &std::path::Path, args: &[&str]) {
+    let out = std::process::Command::new("git")
+        .current_dir(dir)
+        .args(args)
+        .env("GIT_AUTHOR_NAME", "t")
+        .env("GIT_AUTHOR_EMAIL", "t@t")
+        .env("GIT_COMMITTER_NAME", "t")
+        .env("GIT_COMMITTER_EMAIL", "t@t")
+        .output()
+        .unwrap();
+    assert!(
+        out.status.success(),
+        "git {args:?}: {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+}
+
+fn init_git_repo(dir: &std::path::Path) {
+    run_git(dir, &["init", "-q", "-b", "main"]);
+    std::fs::write(dir.join(".gitkeep"), "x").unwrap();
+    run_git(dir, &["add", "."]);
+    run_git(dir, &["commit", "-qm", "base"]);
+}
+
+/// Scripted coordinator: the closure decides each spawn's reply from
+/// the request; every spawn is captured. Also tracks the max number of
+/// WORKER spawns in flight at once (reply to the first worker is held
+/// until a second worker arrives when `require_two_workers` is set —
+/// proving genuine fan-out, not sequential dispatch).
+fn spawn_scripted_coordinator(
+    repo: std::path::PathBuf,
+    mut script: impl FnMut(&kigi_tools::implementations::kigi::task::types::SubagentRequest) -> String
+    + 'static,
+    require_two_workers: bool,
+) -> (
+    tokio::sync::mpsc::UnboundedSender<SubagentEvent>,
+    StdArc<std::sync::Mutex<Vec<CapturedSpawn>>>,
+) {
+    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<SubagentEvent>();
+    let captured: StdArc<std::sync::Mutex<Vec<CapturedSpawn>>> =
+        StdArc::new(std::sync::Mutex::new(Vec::new()));
+    let cap_task = StdArc::clone(&captured);
+    tokio::task::spawn_local(async move {
+        let mut held: Option<(tokio::sync::oneshot::Sender<SubagentResult>, SubagentResult)> = None;
+        while let Some(ev) = rx.recv().await {
+            if let SubagentEvent::Spawn(req) = ev {
+                cap_task.lock().unwrap().push(CapturedSpawn {
+                    prompt: req.prompt.clone(),
+                    resume_from: req.resume_from.clone(),
+                    cwd: req.cwd.clone(),
+                    isolation_worktree: matches!(
+                        req.runtime_overrides.isolation,
+                        Some(kigi_tool_types::SubagentIsolationMode::Worktree)
+                    ),
+                });
+                let output = script(&req);
+                let is_worker = req.prompt.contains("Graph Node Worker");
+                // Honor the isolation contract with a REAL worktree so the
+                // round-1 guard and the merge-back run the true path
+                // (empty diff ⇒ apply Success ⇒ worktree cleanup).
+                let worktree_path = (is_worker
+                    && matches!(
+                        req.runtime_overrides.isolation,
+                        Some(kigi_tool_types::SubagentIsolationMode::Worktree)
+                    ))
+                .then(|| {
+                    let wt = repo.join(format!("wt-{}", req.id));
+                    run_git(&repo, &["worktree", "add", "-q", wt.to_str().unwrap()]);
+                    wt.to_string_lossy().into_owned()
+                });
+                let result = SubagentResult {
+                    success: true,
+                    output: StdArc::from(output.as_str()),
+                    subagent_id: req.id.clone(),
+                    child_session_id: req.id.clone(),
+                    tokens_used: 10,
+                    worktree_path,
+                    ..Default::default()
+                };
+                if require_two_workers && is_worker && held.is_none() {
+                    // Hold the first worker's reply until the second
+                    // worker spawn arrives — join_all must have BOTH in
+                    // flight for this to make progress (fan-out proof).
+                    held = Some((req.result_tx, result));
+                    continue;
+                }
+                if is_worker && let Some((held_tx, held_result)) = held.take() {
+                    let _ = held_tx.send(held_result);
+                }
+                let _ = req.result_tx.send(result);
+            }
+        }
+    });
+    (tx, captured)
+}
+
+fn diamond_graph_json() -> Vec<u8> {
+    serde_json::json!({
+        "nodes": [
+            {"id": "a", "title": "Node A", "spec": "do a", "deps": []},
+            {"id": "b", "title": "Node B", "spec": "do b", "deps": []},
+            {"id": "c", "title": "Node C", "spec": "do c", "deps": ["a", "b"]},
+        ]
+    })
+    .to_string()
+    .into_bytes()
+}
+
+/// Route a scripted reply by spawn kind: planner writes the DAG, workers
+/// claim done, verifiers approve.
+fn happy_reply(
+    req: &kigi_tools::implementations::kigi::task::types::SubagentRequest,
+    dag: &[u8],
+) -> String {
+    if req.prompt.contains("Graph Plan Writer") {
+        let path = req
+            .prompt
+            .find("/graph.json")
+            .map(|end| {
+                let end = end + "/graph.json".len();
+                let start = req.prompt[..end - "/graph.json".len()]
+                    .rfind(|c: char| !c.is_ascii_graphic() || c == '`')
+                    .map(|i| i + 1)
+                    .unwrap_or(0);
+                req.prompt[start..end].to_string()
+            })
+            .expect("planner prompt embeds path");
+        std::fs::create_dir_all(std::path::Path::new(&path).parent().unwrap()).unwrap();
+        std::fs::write(&path, dag).unwrap();
+        "Done".to_owned()
+    } else if req.prompt.contains("Graph Node Worker") {
+        "NODE_RESULT: done\nImplemented per spec; checks run.".to_owned()
+    } else {
+        "NODE_VERDICT: achieved".to_owned()
+    }
+}
+
+#[tokio::test(flavor = "current_thread")]
+#[serial]
+async fn parallel_batch_fans_out_then_serial_tail_completes() {
+    unsafe { std::env::set_var(ENV_FLAG, "0") };
+    let local = tokio::task::LocalSet::new();
+    tokio::time::timeout(
+        std::time::Duration::from_secs(60),
+        local.run_until(async {
+            let dag = diamond_graph_json();
+            let (mut actor, _tmp, _prx) = make_graph_actor_detached().await;
+            let (coord_tx, captured) = spawn_scripted_coordinator(
+                _tmp.path().to_path_buf(),
+                move |req| happy_reply(req, &dag),
+                true,
+            );
+            actor.tool_context.subagent_event_tx = Some(coord_tx);
+            actor.graph_concurrency = 2;
+
+            // Roots a+b run as a parallel batch; c is the serial tail.
+            let outcome = actor.setup_graph("build the diamond", None).await;
+            let reminder = match outcome {
+                graph::GraphSetupOutcome::Inference { reminder, .. } => reminder,
+                graph::GraphSetupOutcome::Message(msg) => panic!("expected Inference: {msg}"),
+            };
+            assert!(
+                reminder.contains("Node C"),
+                "serial tail must be node c: {reminder}"
+            );
+            let statuses = node_statuses(&actor);
+            assert_eq!(statuses[0].1, NodeStatus::Achieved, "a via batch");
+            assert_eq!(statuses[1].1, NodeStatus::Achieved, "b via batch");
+            assert_eq!(statuses[2].1, NodeStatus::Running, "c on the goal engine");
+            {
+                let caps = captured.lock().unwrap();
+                let workers: Vec<_> = caps
+                    .iter()
+                    .filter(|c| c.prompt.contains("Graph Node Worker"))
+                    .collect();
+                let verifiers: Vec<_> = caps
+                    .iter()
+                    .filter(|c| c.prompt.contains("Graph Node Verifier"))
+                    .collect();
+                assert_eq!(workers.len(), 2, "one worker per batch node");
+                assert_eq!(verifiers.len(), 2, "one verifier per claim");
+                assert!(
+                    workers.iter().all(|w| w.isolation_worktree),
+                    "first worker rounds must request worktree isolation"
+                );
+                // The held-reply gate above proves both were in flight
+                // concurrently, or this test would have hung.
+            }
+            // Worker session ids stamped for audit.
+            assert!(
+                actor
+                    .graph_tracker
+                    .lock()
+                    .node(&crate::session::graph_plan::node_id_for_slug("a"))
+                    .unwrap()
+                    .goal_id
+                    .is_some(),
+                "worker session id must be recorded on the node"
+            );
+
+            // Finish c and gn-final serially (G0 machinery).
+            for _ in 0..2 {
+                drive_node_goal_to_complete(&actor).await;
+                let _ = actor.run_graph_round_end().await;
+            }
+            assert_eq!(
+                actor.graph_tracker.lock().status(),
+                Some(GoalStatus::Complete)
+            );
+        }),
+    )
+    .await
+    .expect("parallel batch did not fan out (held-reply gate starved)");
+    unsafe { std::env::remove_var(ENV_FLAG) };
+}
+
+#[tokio::test(flavor = "current_thread")]
+#[serial]
+async fn verifier_rejection_iterates_worker_with_resume_and_gaps() {
+    unsafe { std::env::set_var(ENV_FLAG, "0") };
+    let local = tokio::task::LocalSet::new();
+    local
+        .run_until(async {
+            let dag = diamond_graph_json();
+            let mut a_verify_rejections = 0u32;
+            let (mut actor, _tmp, _prx) = make_graph_actor_detached().await;
+            let (coord_tx, captured) = spawn_scripted_coordinator(
+                _tmp.path().to_path_buf(),
+                move |req| {
+                    if req.prompt.contains("Graph Plan Writer") {
+                        return happy_reply(req, &dag);
+                    }
+                    if req.prompt.contains("Graph Node Worker") {
+                        return "NODE_RESULT: done\nwork done.".to_owned();
+                    }
+                    // Verifier: reject node A's FIRST attempt only.
+                    if req.prompt.contains("do a") && a_verify_rejections == 0 {
+                        a_verify_rejections += 1;
+                        return "NODE_VERDICT: not_achieved\nGAPS:\n- tests were not run"
+                            .to_owned();
+                    }
+                    "NODE_VERDICT: achieved".to_owned()
+                },
+                false,
+            );
+            actor.tool_context.subagent_event_tx = Some(coord_tx);
+            actor.graph_concurrency = 2;
+            let _ = actor.setup_graph("build the diamond", None).await;
+
+            let a_id = crate::session::graph_plan::node_id_for_slug("a");
+            let a = actor.graph_tracker.lock().node(&a_id).cloned().unwrap();
+            assert_eq!(a.status, NodeStatus::Achieved);
+            assert_eq!(a.rounds, 2, "one rejection ⇒ two worker rounds");
+
+            let caps = captured.lock().unwrap();
+            let a_workers: Vec<_> = caps
+                .iter()
+                .filter(|c| c.prompt.contains("Graph Node Worker") && c.prompt.contains("do a"))
+                .collect();
+            assert_eq!(a_workers.len(), 2);
+            assert!(a_workers[0].resume_from.is_none());
+            assert!(
+                a_workers[1].resume_from.is_some(),
+                "round 2 must resume the round-1 child session"
+            );
+            assert!(
+                !a_workers[1].isolation_worktree,
+                "resume keeps the existing worktree; no fresh isolation"
+            );
+            assert!(
+                a_workers[1].prompt.contains("tests were not run"),
+                "round 2 must carry the verifier's gaps"
+            );
+        })
+        .await;
+    unsafe { std::env::remove_var(ENV_FLAG) };
+}
+
+#[tokio::test(flavor = "current_thread")]
+#[serial]
+async fn blocked_worker_fails_node_and_blocks_dependent_chain() {
+    unsafe { std::env::set_var(ENV_FLAG, "0") };
+    let local = tokio::task::LocalSet::new();
+    local
+        .run_until(async {
+            let dag = diamond_graph_json();
+            let (mut actor, _tmp, _prx) = make_graph_actor_detached().await;
+            let (coord_tx, _captured) = spawn_scripted_coordinator(
+                _tmp.path().to_path_buf(),
+                move |req| {
+                    if req.prompt.contains("Graph Plan Writer") {
+                        return happy_reply(req, &dag);
+                    }
+                    if req.prompt.contains("Graph Node Worker") {
+                        if req.prompt.contains("do a") {
+                            return "NODE_RESULT: blocked\nimpossible in this environment"
+                                .to_owned();
+                        }
+                        return "NODE_RESULT: done\nok".to_owned();
+                    }
+                    "NODE_VERDICT: achieved".to_owned()
+                },
+                false,
+            );
+            actor.tool_context.subagent_event_tx = Some(coord_tx);
+            actor.graph_concurrency = 2;
+            match actor.setup_graph("build the diamond", None).await {
+                graph::GraphSetupOutcome::Message(_) => {}
+                graph::GraphSetupOutcome::Inference { reminder, .. } => {
+                    panic!("wedged graph must not reach inference: {reminder}")
+                }
+            }
+            let statuses = node_statuses(&actor);
+            assert_eq!(statuses[0].1, NodeStatus::Failed, "a blocked by worker");
+            assert_eq!(statuses[1].1, NodeStatus::Achieved, "b unaffected");
+            assert_eq!(statuses[2].1, NodeStatus::Blocked, "c depends on a");
+            assert_eq!(statuses[3].1, NodeStatus::Blocked, "final depends on all");
+            assert_eq!(
+                actor.graph_tracker.lock().status(),
+                Some(GoalStatus::Blocked),
+                "wedged graph pauses as Blocked"
+            );
+            let a = actor
+                .graph_tracker
+                .lock()
+                .node(&crate::session::graph_plan::node_id_for_slug("a"))
+                .cloned()
+                .unwrap();
+            assert!(
+                a.failure.as_deref().unwrap().contains("impossible"),
+                "{:?}",
+                a.failure
+            );
+        })
+        .await;
+    unsafe { std::env::remove_var(ENV_FLAG) };
+}
+
+#[tokio::test(flavor = "current_thread")]
+#[serial]
+async fn merge_conflict_on_real_worktree_fails_the_node() {
+    if std::process::Command::new("git")
+        .arg("--version")
+        .output()
+        .is_err()
+    {
+        eprintln!("git unavailable; skipping merge-conflict test");
+        return;
+    }
+    let local = tokio::task::LocalSet::new();
+    local
+        .run_until(async {
+            fn git(dir: &std::path::Path, args: &[&str]) {
+                let out = std::process::Command::new("git")
+                    .current_dir(dir)
+                    .args(args)
+                    .env("GIT_AUTHOR_NAME", "t")
+                    .env("GIT_AUTHOR_EMAIL", "t@t")
+                    .env("GIT_COMMITTER_NAME", "t")
+                    .env("GIT_COMMITTER_EMAIL", "t@t")
+                    .output()
+                    .unwrap();
+                assert!(
+                    out.status.success(),
+                    "git {args:?}: {}",
+                    String::from_utf8_lossy(&out.stderr)
+                );
+            }
+            let repo = TempDir::new().unwrap();
+            git(repo.path(), &["init", "-q", "-b", "main"]);
+            std::fs::write(repo.path().join("f.txt"), "base\n").unwrap();
+            git(repo.path(), &["add", "."]);
+            git(repo.path(), &["commit", "-qm", "base"]);
+            let wt = repo.path().join("wt");
+            git(
+                repo.path(),
+                &["worktree", "add", "-q", wt.to_str().unwrap()],
+            );
+            // Conflicting edits: main tree and worktree both diverge from base.
+            std::fs::write(repo.path().join("f.txt"), "ours\n").unwrap();
+            std::fs::write(wt.join("f.txt"), "theirs\n").unwrap();
+
+            let (coord_tx, _count) = spawn_graph_planner_coordinator(vec![]);
+            let (actor, _tmp, _prx) = make_graph_actor(coord_tx).await;
+            let err = actor
+                .merge_node_worktree("gn-test", Some(wt.to_str().unwrap()), None)
+                .await
+                .expect_err("conflicting edits must surface as a merge failure");
+            assert!(err.contains("f.txt"), "conflict names the file: {err}");
+        })
+        .await;
+}
+
+#[tokio::test(flavor = "current_thread")]
+#[serial]
+async fn parallel_budget_gate_trips_between_batches_and_charges_all_nodes() {
+    unsafe { std::env::set_var(ENV_FLAG, "0") };
+    let local = tokio::task::LocalSet::new();
+    local
+        .run_until(async {
+            let dag = diamond_graph_json();
+            let (mut actor, _tmp, _prx) = make_graph_actor_detached().await;
+            let (coord_tx, _captured) = spawn_scripted_coordinator(
+                _tmp.path().to_path_buf(),
+                move |req| {
+                    if req.prompt.contains("Graph Plan Writer") {
+                        return happy_reply(req, &dag);
+                    }
+                    if req.prompt.contains("Graph Node Worker") {
+                        // Node A succeeds; node B claims blocked (fails) —
+                        // BOTH must charge the budget.
+                        if req.prompt.contains("do b") {
+                            return "NODE_RESULT: blocked\nnope".to_owned();
+                        }
+                        return "NODE_RESULT: done\nok".to_owned();
+                    }
+                    "NODE_VERDICT: achieved".to_owned()
+                },
+                false,
+            );
+            actor.tool_context.subagent_event_tx = Some(coord_tx);
+            actor.graph_concurrency = 2;
+            // Stub charges 10 tokens per spawn: batch = worker a + verifier a
+            // + worker b (blocked, no verifier) = 30 > budget 25 → the gate
+            // trips at the next dispatch iteration.
+            match actor.setup_graph("build the diamond", Some(25)).await {
+                graph::GraphSetupOutcome::Message(_) => {}
+                graph::GraphSetupOutcome::Inference { reminder, .. } => {
+                    panic!("budget-dead graph must not reach inference: {reminder}")
+                }
+            }
+            assert_eq!(
+                actor.graph_tracker.lock().status(),
+                Some(GoalStatus::BudgetLimited),
+                "inter-batch gate must trip"
+            );
+            let s = actor.graph_tracker.lock().snapshot().cloned().unwrap();
+            assert_eq!(
+                s.tokens_spent_nodes, 30,
+                "achieved (20) AND failed (10) nodes must both charge the budget"
+            );
+        })
+        .await;
+    unsafe { std::env::remove_var(ENV_FLAG) };
+}
+
+#[tokio::test(flavor = "current_thread")]
+#[serial]
+async fn cap_trims_batch_and_leftover_root_goes_serial() {
+    unsafe { std::env::set_var(ENV_FLAG, "0") };
+    let triple_root = serde_json::json!({
+        "nodes": [
+            {"id": "a", "title": "Node A", "spec": "do a", "deps": []},
+            {"id": "b", "title": "Node B", "spec": "do b", "deps": []},
+            {"id": "c", "title": "Node C", "spec": "do c", "deps": []},
+        ]
+    })
+    .to_string()
+    .into_bytes();
+    let local = tokio::task::LocalSet::new();
+    tokio::time::timeout(
+        std::time::Duration::from_secs(60),
+        local.run_until(async {
+            let dag = triple_root;
+            let (mut actor, _tmp, _prx) = make_graph_actor_detached().await;
+            let (coord_tx, captured) = spawn_scripted_coordinator(
+                _tmp.path().to_path_buf(),
+                move |req| happy_reply(req, &dag),
+                true,
+            );
+            actor.tool_context.subagent_event_tx = Some(coord_tx);
+            actor.graph_concurrency = 2;
+            let outcome = actor.setup_graph("three roots", None).await;
+            // Batch 1 = {a, b} (cap-trimmed); leftover root c is the sole
+            // Ready node afterwards → serial launch on the goal engine.
+            let reminder = match outcome {
+                graph::GraphSetupOutcome::Inference { reminder, .. } => reminder,
+                graph::GraphSetupOutcome::Message(msg) => panic!("expected Inference: {msg}"),
+            };
+            assert!(reminder.contains("Node C"), "{reminder}");
+            let workers = captured
+                .lock()
+                .unwrap()
+                .iter()
+                .filter(|c| c.prompt.contains("Graph Node Worker"))
+                .count();
+            assert_eq!(workers, 2, "take(cap) must trim the third root");
+            assert_eq!(node_statuses(&actor)[2].1, NodeStatus::Running, "c serial");
+        }),
+    )
+    .await
+    .expect("cap-trim batch starved");
+    unsafe { std::env::remove_var(ENV_FLAG) };
+}
+
+#[tokio::test(flavor = "current_thread")]
+#[serial]
+async fn resume_after_cancelled_batch_demotes_orphaned_running_nodes() {
+    unsafe { std::env::set_var(ENV_FLAG, "0") };
+    let local = tokio::task::LocalSet::new();
+    local
+        .run_until(async {
+            let dag = diamond_graph_json();
+            let (mut actor, _tmp, _prx) = make_graph_actor_detached().await;
+            let (coord_tx, _captured) = spawn_scripted_coordinator(
+                _tmp.path().to_path_buf(),
+                move |req| happy_reply(req, &dag),
+                false,
+            );
+            actor.tool_context.subagent_event_tx = Some(coord_tx);
+            actor.graph_concurrency = 2;
+            let _ = actor.setup_graph("build the diamond", None).await;
+
+            // Simulate an Esc mid-batch: nodes a+b marked Running with no
+            // executor (batch future dropped), cascade paused the graph.
+            {
+                let mut tracker = actor.graph_tracker.lock();
+                if let Some(s) = tracker.snapshot_mut() {
+                    s.nodes[0].status = NodeStatus::Running;
+                    s.nodes[1].status = NodeStatus::Running;
+                    s.current_node = None;
+                }
+            }
+            actor.reset_goal_engine_state().await;
+            actor
+                .graph_tracker
+                .lock()
+                .pause(crate::session::goal_tracker::GoalPauseReason::User);
+
+            // /graph resume must demote the orphans and re-dispatch — NOT
+            // wedge-pause with "no runnable node".
+            match actor.resume_graph().await {
+                graph::GraphSetupOutcome::Inference { reminder, .. } => {
+                    assert!(reminder.contains("Node C"), "{reminder}");
+                }
+                graph::GraphSetupOutcome::Message(msg) => {
+                    panic!("resume must re-dispatch orphaned batch nodes, got: {msg}")
+                }
+            }
+            assert_eq!(
+                actor.graph_tracker.lock().status(),
+                Some(GoalStatus::Active)
+            );
+            let statuses = node_statuses(&actor);
+            assert_eq!(statuses[0].1, NodeStatus::Achieved, "a re-ran via batch");
+            assert_eq!(statuses[1].1, NodeStatus::Achieved, "b re-ran via batch");
+        })
+        .await;
+    unsafe { std::env::remove_var(ENV_FLAG) };
+}
+
+#[tokio::test(flavor = "current_thread")]
+#[serial]
+async fn batch_merges_real_worktrees_and_cleans_them_up() {
+    if std::process::Command::new("git")
+        .arg("--version")
+        .output()
+        .is_err()
+    {
+        eprintln!("git unavailable; skipping real-worktree batch merge test");
+        return;
+    }
+    unsafe { std::env::set_var(ENV_FLAG, "0") };
+    let local = tokio::task::LocalSet::new();
+    local
+        .run_until(async {
+            fn git(dir: &std::path::Path, args: &[&str]) {
+                let out = std::process::Command::new("git")
+                    .current_dir(dir)
+                    .args(args)
+                    .env("GIT_AUTHOR_NAME", "t")
+                    .env("GIT_AUTHOR_EMAIL", "t@t")
+                    .env("GIT_COMMITTER_NAME", "t")
+                    .env("GIT_COMMITTER_EMAIL", "t@t")
+                    .output()
+                    .unwrap();
+                assert!(
+                    out.status.success(),
+                    "git {args:?}: {}",
+                    String::from_utf8_lossy(&out.stderr)
+                );
+            }
+            let repo = TempDir::new().unwrap();
+            git(repo.path(), &["init", "-q", "-b", "main"]);
+            std::fs::write(repo.path().join("base.txt"), "base\n").unwrap();
+            git(repo.path(), &["add", "."]);
+            git(repo.path(), &["commit", "-qm", "base"]);
+            let wt_a = repo.path().join("wt_a");
+            let wt_b = repo.path().join("wt_b");
+            git(
+                repo.path(),
+                &["worktree", "add", "-q", wt_a.to_str().unwrap()],
+            );
+            git(
+                repo.path(),
+                &["worktree", "add", "-q", wt_b.to_str().unwrap()],
+            );
+            // Disjoint node outputs.
+            std::fs::write(wt_a.join("a_out.txt"), "from a\n").unwrap();
+            std::fs::write(wt_b.join("b_out.txt"), "from b\n").unwrap();
+
+            let dag = diamond_graph_json();
+            let wt_a_str = wt_a.to_str().unwrap().to_owned();
+            let wt_b_str = wt_b.to_str().unwrap().to_owned();
+            // Scripted coordinator that returns REAL worktree paths on
+            // worker results (bypasses spawn_scripted_coordinator's
+            // default-None worktree_path).
+            let (coord_tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<SubagentEvent>();
+            tokio::task::spawn_local(async move {
+                while let Some(ev) = rx.recv().await {
+                    if let SubagentEvent::Spawn(req) = ev {
+                        let (output, wt): (String, Option<String>) = if req
+                            .prompt
+                            .contains("Graph Plan Writer")
+                        {
+                            let path = req
+                                .prompt
+                                .find("/graph.json")
+                                .map(|end| {
+                                    let end = end + "/graph.json".len();
+                                    let start = req.prompt[..end - "/graph.json".len()]
+                                        .rfind(|c: char| !c.is_ascii_graphic() || c == '`')
+                                        .map(|i| i + 1)
+                                        .unwrap_or(0);
+                                    req.prompt[start..end].to_string()
+                                })
+                                .unwrap();
+                            std::fs::create_dir_all(std::path::Path::new(&path).parent().unwrap())
+                                .unwrap();
+                            std::fs::write(&path, &dag).unwrap();
+                            ("Done".to_owned(), None)
+                        } else if req.prompt.contains("Graph Node Worker") {
+                            let wt = if req.prompt.contains("do a") {
+                                wt_a_str.clone()
+                            } else {
+                                wt_b_str.clone()
+                            };
+                            ("NODE_RESULT: done\nwrote output file".to_owned(), Some(wt))
+                        } else {
+                            ("NODE_VERDICT: achieved".to_owned(), None)
+                        };
+                        let _ = req.result_tx.send(SubagentResult {
+                            success: true,
+                            output: StdArc::from(output.as_str()),
+                            subagent_id: req.id.clone(),
+                            child_session_id: req.id.clone(),
+                            tokens_used: 10,
+                            worktree_path: wt,
+                            ..Default::default()
+                        });
+                    }
+                }
+            });
+            let (mut actor, _tmp, _prx) = make_graph_actor(coord_tx).await;
+            actor.graph_concurrency = 2;
+            // Point the actor's cwd-based HEAD guard at the real repo.
+            actor.tool_context.cwd =
+                kigi_paths::AbsPathBuf::new(repo.path().to_path_buf()).unwrap();
+
+            let _ = actor.setup_graph("build the diamond", None).await;
+            let statuses = node_statuses(&actor);
+            assert_eq!(statuses[0].1, NodeStatus::Achieved);
+            assert_eq!(statuses[1].1, NodeStatus::Achieved);
+            // SEQUENTIAL merge landed both nodes' files in the MAIN tree.
+            assert_eq!(
+                std::fs::read_to_string(repo.path().join("a_out.txt")).unwrap(),
+                "from a\n"
+            );
+            assert_eq!(
+                std::fs::read_to_string(repo.path().join("b_out.txt")).unwrap(),
+                "from b\n"
+            );
+            // Storage discipline: merged worktrees are removed.
+            assert!(!wt_a.exists(), "merged worktree a must be cleaned up");
+            assert!(!wt_b.exists(), "merged worktree b must be cleaned up");
+        })
+        .await;
+    unsafe { std::env::remove_var(ENV_FLAG) };
+}

crates/codegen/kigi-shell/src/session/acp_session_tests/idle_resume_tests.rs

@@ -245,6 +245,8 @@ async fn test_e2e_idle_resume_refreshes_model_metadata() {
                 graph_tracker: Arc::new(parking_lot::Mutex::new(
                     crate::session::graph_tracker::GraphTracker::new(std::env::temp_dir()),
                 )),
+                graph_concurrency: 1,
+                graph_node_rounds: 3,
                 goal_turn_task_ids: parking_lot::Mutex::new(std::collections::HashSet::new()),
                 goal_continuation_streak: std::sync::atomic::AtomicU32::new(0),
                 goal_blocked_streak: std::sync::atomic::AtomicU32::new(0),

crates/codegen/kigi-shell/src/session/acp_session_tests/inline_auto_compact_flow_tests.rs

@@ -177,6 +177,8 @@ async fn create_test_actor(
         graph_tracker: Arc::new(parking_lot::Mutex::new(
             crate::session::graph_tracker::GraphTracker::new(std::env::temp_dir()),
         )),
+        graph_concurrency: 1,
+        graph_node_rounds: 3,
         goal_turn_task_ids: parking_lot::Mutex::new(std::collections::HashSet::new()),
         goal_continuation_streak: std::sync::atomic::AtomicU32::new(0),
         goal_blocked_streak: std::sync::atomic::AtomicU32::new(0),
@@ -619,6 +621,8 @@ async fn create_test_actor_with_memory(
         graph_tracker: Arc::new(parking_lot::Mutex::new(
             crate::session::graph_tracker::GraphTracker::new(std::env::temp_dir()),
         )),
+        graph_concurrency: 1,
+        graph_node_rounds: 3,
         goal_turn_task_ids: parking_lot::Mutex::new(std::collections::HashSet::new()),
         goal_continuation_streak: std::sync::atomic::AtomicU32::new(0),
         goal_blocked_streak: std::sync::atomic::AtomicU32::new(0),
@@ -1372,6 +1376,8 @@ async fn test_e2e_idle_resume_refreshes_model_metadata() {
                 graph_tracker: Arc::new(parking_lot::Mutex::new(
                     crate::session::graph_tracker::GraphTracker::new(std::env::temp_dir()),
                 )),
+                graph_concurrency: 1,
+                graph_node_rounds: 3,
                 goal_turn_task_ids: parking_lot::Mutex::new(std::collections::HashSet::new()),
                 goal_continuation_streak: std::sync::atomic::AtomicU32::new(0),
                 goal_blocked_streak: std::sync::atomic::AtomicU32::new(0),

crates/codegen/kigi-shell/src/session/acp_session_tests/memory_config_tests.rs

@@ -239,6 +239,8 @@ async fn create_test_actor_with_memory(
         graph_tracker: Arc::new(parking_lot::Mutex::new(
             crate::session::graph_tracker::GraphTracker::new(std::env::temp_dir()),
         )),
+        graph_concurrency: 1,
+        graph_node_rounds: 3,
         goal_turn_task_ids: parking_lot::Mutex::new(std::collections::HashSet::new()),
         goal_continuation_streak: std::sync::atomic::AtomicU32::new(0),
         goal_blocked_streak: std::sync::atomic::AtomicU32::new(0),

crates/codegen/kigi-shell/src/session/acp_session_tests/replay_buffer_send_update_tests.rs

@@ -185,6 +185,8 @@ pub(super) async fn make_replay_send_update_fixture() -> ReplaySendUpdateFixture
         graph_tracker: Arc::new(parking_lot::Mutex::new(
             crate::session::graph_tracker::GraphTracker::new(std::env::temp_dir()),
         )),
+        graph_concurrency: 1,
+        graph_node_rounds: 3,
         goal_turn_task_ids: parking_lot::Mutex::new(std::collections::HashSet::new()),
         goal_continuation_streak: std::sync::atomic::AtomicU32::new(0),
         goal_blocked_streak: std::sync::atomic::AtomicU32::new(0),

crates/codegen/kigi-shell/src/session/acp_session_tests/support.rs

@@ -290,6 +290,8 @@ pub(crate) async fn create_test_actor_ex(
         graph_tracker: Arc::new(parking_lot::Mutex::new(
             crate::session::graph_tracker::GraphTracker::new(std::env::temp_dir()),
         )),
+        graph_concurrency: 1,
+        graph_node_rounds: 3,
         goal_turn_task_ids: parking_lot::Mutex::new(std::collections::HashSet::new()),
         goal_continuation_streak: std::sync::atomic::AtomicU32::new(0),
         goal_blocked_streak: std::sync::atomic::AtomicU32::new(0),

crates/codegen/kigi-shell/src/session/compaction.rs

@@ -2272,6 +2272,8 @@ mod inline_auto_compact_flow_tests {
             graph_tracker: Arc::new(parking_lot::Mutex::new(
                 crate::session::graph_tracker::GraphTracker::new(std::env::temp_dir()),
             )),
+            graph_concurrency: 1,
+            graph_node_rounds: 3,
             goal_turn_task_ids: parking_lot::Mutex::new(std::collections::HashSet::new()),
             goal_continuation_streak: std::sync::atomic::AtomicU32::new(0),
             goal_blocked_streak: std::sync::atomic::AtomicU32::new(0),

crates/codegen/kigi-shell/src/session/graph_planner.rs

@@ -193,10 +193,12 @@ mod tests {
         ]))
     }
 
-    fn tmp_graph_file(name: &str) -> PathBuf {
-        let dir = std::env::temp_dir().join(format!("kigi-graph-planner-test-{name}"));
-        std::fs::create_dir_all(&dir).unwrap();
-        dir.join("graph.json")
+    /// Self-cleaning temp home per test: never leak dirs into the OS
+    /// temp root (storage discipline — see AGENTS.md gates).
+    fn tmp_graph_file(_name: &str) -> (tempfile::TempDir, PathBuf) {
+        let dir = tempfile::TempDir::new().unwrap();
+        let path = dir.path().join("graph.json");
+        (dir, path)
     }
 
     async fn run(spawner: MockSpawner, graph_file: &Path) -> GraphPlannerOutcome {
@@ -216,8 +218,7 @@ mod tests {
 
     #[tokio::test]
     async fn valid_artifact_yields_canonical_nodes() {
-        let target = tmp_graph_file("valid");
-        let _ = std::fs::remove_file(&target);
+        let (_tmp, target) = tmp_graph_file("valid");
         let body = serde_json::json!({
             "nodes": [
                 {"id": "core", "title": "Core", "spec": "core spec", "deps": []},
@@ -242,8 +243,7 @@ mod tests {
 
     #[tokio::test]
     async fn prompt_embeds_objective_feedback_and_tool_names() {
-        let target = tmp_graph_file("prompt");
-        let _ = std::fs::remove_file(&target);
+        let (_tmp, target) = tmp_graph_file("prompt");
         let spawner = MockSpawner {
             response: MockReply::Done,
             body: None,
@@ -274,7 +274,7 @@ mod tests {
 
     #[tokio::test]
     async fn invalid_artifact_is_retryable_with_reason() {
-        let target = tmp_graph_file("invalid");
+        let (_tmp, target) = tmp_graph_file("invalid");
         let spawner = MockSpawner {
             response: MockReply::Done,
             body: Some(br#"{"nodes":[{"id":"a","title":"A","spec":"s","deps":["a"]}]}"#.to_vec()),
@@ -291,8 +291,7 @@ mod tests {
 
     #[tokio::test]
     async fn missing_artifact_fails_closed() {
-        let target = tmp_graph_file("missing");
-        let _ = std::fs::remove_file(&target);
+        let (_tmp, target) = tmp_graph_file("missing");
         let spawner = MockSpawner {
             response: MockReply::Done,
             body: None,
@@ -309,7 +308,7 @@ mod tests {
 
     #[tokio::test]
     async fn runtime_error_fails_closed() {
-        let target = tmp_graph_file("runtime");
+        let (_tmp, target) = tmp_graph_file("runtime");
         let spawner = MockSpawner {
             response: MockReply::Runtime { cancelled: false },
             body: None,
@@ -326,7 +325,7 @@ mod tests {
 
     #[tokio::test]
     async fn oversize_artifact_is_invalid_with_cap_in_reason() {
-        let target = tmp_graph_file("oversize");
+        let (_tmp, target) = tmp_graph_file("oversize");
         let mut body = vec![b'x'; (MAX_GRAPH_JSON_BYTES as usize) + 1];
         body[0] = b'{'; // content is irrelevant; the size gate fires first
         let spawner = MockSpawner {
@@ -345,7 +344,7 @@ mod tests {
 
     #[tokio::test]
     async fn transport_error_fails_closed() {
-        let target = tmp_graph_file("transport");
+        let (_tmp, target) = tmp_graph_file("transport");
         let spawner = MockSpawner {
             response: MockReply::Transport,
             body: None,
@@ -362,7 +361,7 @@ mod tests {
 
     #[tokio::test]
     async fn cancelled_runtime_error_reports_aborted() {
-        let target = tmp_graph_file("aborted");
+        let (_tmp, target) = tmp_graph_file("aborted");
         let spawner = MockSpawner {
             response: MockReply::Runtime { cancelled: true },
             body: None,

crates/codegen/kigi-shell/src/session/graph_tracker.rs

@@ -628,6 +628,42 @@ impl GraphTracker {
         }
     }
 
+    /// Demote in-flight (`Running`/`Verifying`) nodes whose executor is
+    /// gone back to `Ready`, keeping `keep` (the node whose goal still
+    /// lives in the engine, if any). Used by the IN-SESSION resume path:
+    /// a cancel during a parallel batch aborts the batch future after
+    /// nodes were marked Running, and — unlike a restart, where
+    /// `from_snapshot` sanitizes — nothing else would ever demote them,
+    /// wedging every subsequent resume. Re-running is safe by design
+    /// (verifier-gated).
+    pub fn demote_orphaned_in_flight(&mut self, keep: Option<&str>) {
+        let Some(state) = self.state.as_mut() else {
+            return;
+        };
+        for node in &mut state.nodes {
+            if matches!(node.status, NodeStatus::Running | NodeStatus::Verifying)
+                && keep != Some(node.id.as_str())
+            {
+                node.status = NodeStatus::Ready;
+            }
+        }
+        state.current_node = keep.map(str::to_owned);
+        self.recompute_ready();
+    }
+
+    /// Charge a node's spend against the graph budget and stamp it on
+    /// the node, independent of verdict — a FAILED node's tokens were
+    /// still spent (budget integrity).
+    pub fn charge_node_tokens(&mut self, id: &str, tokens: i64) {
+        let Some(state) = self.state.as_mut() else {
+            return;
+        };
+        if let Some(node) = state.nodes.iter_mut().find(|n| n.id == id) {
+            node.tokens_used = tokens;
+        }
+        state.tokens_spent_nodes = state.tokens_spent_nodes.saturating_add(tokens);
+    }
+
     /// Promote every `Waiting` node whose deps are all `Achieved` to
     /// `Ready`.
     pub fn recompute_ready(&mut self) {

crates/codegen/kigi-shell/src/session/templates/graph_node_verifier_prompt.md

@@ -0,0 +1,24 @@
+You are a Graph Node Verifier for the Kigi harness: an adversarial skeptic
+judging whether ONE node's outcome contract holds in the CURRENT state of
+your working directory (the implementer's isolated worktree).
+
+Do not trust the implementer's claims — re-run the decisive checks yourself
+with your tools (read the code, run the tests/commands the contract
+implies). An unverifiable claim is a gap. Missing evidence is a gap. Do NOT
+modify any file — you are read-only by contract.
+
+Your final message MUST end with exactly one of:
+
+```
+NODE_VERDICT: achieved
+```
+when every part of the node contract observably holds, or
+
+```
+NODE_VERDICT: not_achieved
+GAPS:
+- <one concrete, actionable gap per line>
+```
+
+Be strict but fair: judge ONLY this node's contract, not sibling nodes'
+scope and not style preferences.

crates/codegen/kigi-shell/src/session/templates/graph_node_worker_prompt.md

@@ -0,0 +1,29 @@
+You are a Graph Node Worker for the Kigi harness: the implementer of ONE
+node in a larger dependency graph. Sibling nodes are handled elsewhere —
+complete ONLY this node's scope; nothing more, nothing less.
+
+Your working directory is an isolated git worktree. Every change you make
+here is merged back into the main tree once the node passes verification,
+so work only inside it and leave it in a clean, coherent state.
+
+Rules:
+
+- Produce real, verifiable work. Run the builds/tests/commands you claim
+  pass; never fabricate evidence.
+- If a GAPS section appears below, a verifier rejected the previous round —
+  close exactly those gaps first, then re-check the whole node contract.
+- Do not commit; the harness owns version control.
+
+Your final message MUST end with exactly one of:
+
+```
+NODE_RESULT: done
+```
+followed by a short factual summary of what exists now and how you verified
+it (the verifier audits this), or
+
+```
+NODE_RESULT: blocked
+```
+followed by the precise reason this node cannot be completed in this
+environment. Blocked is a FAILURE signal — never put success text there.

crates/codegen/kigi-workspace/src/worktree/mod.rs

@@ -2090,16 +2090,22 @@ async fn get_apply_context(worktree_path: &str) -> Result<ApplyContext> {
     .await?
 }
 
-async fn get_file_at_commit(worktree_path: &str, commit: &str, path: &str) -> Option<String> {
-    git_cli(
-        Path::new(worktree_path),
-        &["show", &format!("{}:{}", commit, path)],
-    )
-    .await
-    .ok()
+/// Raw blob bytes at `commit:path`, or `None` when absent. Byte-exact
+/// (NOT `git_cli`, which lossy-decodes and trims): the 3-way merge below
+/// must compare content byte-for-byte or binary files mis-merge.
+async fn get_file_at_commit(worktree_path: &str, commit: &str, path: &str) -> Option<Vec<u8>> {
+    let output = tokio::process::Command::new("git")
+        .arg("-C")
+        .arg(worktree_path)
+        .arg("show")
+        .arg(format!("{}:{}", commit, path))
+        .output()
+        .await
+        .ok()?;
+    output.status.success().then_some(output.stdout)
 }
 
-async fn apply_file_content(dest: &Path, content: Option<&String>) -> bool {
+async fn apply_file_content(dest: &Path, content: Option<&Vec<u8>>) -> bool {
     match content {
         Some(data) => {
             if let Some(parent) = dest.parent() {
@@ -2114,6 +2120,14 @@ async fn apply_file_content(dest: &Path, content: Option<&String>) -> bool {
     }
 }
 
+/// Lossy decode for the `FileConflict` wire payload only — comparisons
+/// above stay byte-exact.
+fn conflict_text(content: &Option<Vec<u8>>) -> Option<String> {
+    content
+        .as_ref()
+        .map(|b| String::from_utf8_lossy(b).into_owned())
+}
+
 pub async fn apply_worktree(req: &ApplyWorktreeRequest) -> Result<ApplyWorktreeResponse> {
     let worktree_path = &req.worktree_path;
     let git_root = find_main_repo_root_from_path(Path::new(worktree_path))?;
@@ -2133,7 +2147,7 @@ pub async fn apply_worktree(req: &ApplyWorktreeRequest) -> Result<ApplyWorktreeR
     for file_change in ctx.changed_files {
         let worktree_file = Path::new(worktree_path).join(&file_change.path);
         let main_file = git_root.join(&file_change.path);
-        let theirs = tokio::fs::read_to_string(&worktree_file).await.ok();
+        let theirs = tokio::fs::read(&worktree_file).await.ok();
 
         if req.mode == ApplyMode::Overwrite {
             if apply_file_content(&main_file, theirs.as_ref()).await {
@@ -2142,21 +2156,28 @@ pub async fn apply_worktree(req: &ApplyWorktreeRequest) -> Result<ApplyWorktreeR
             continue;
         }
 
-        // Merge mode
+        // Merge mode — 3-way-lite over raw bytes.
         let base = get_file_at_commit(worktree_path, &ctx.base_commit, &file_change.path).await;
-        let ours = tokio::fs::read_to_string(&main_file).await.ok();
+        let ours = tokio::fs::read(&main_file).await.ok();
 
         if base == ours {
+            // Main side untouched: take the worktree's version.
             if apply_file_content(&main_file, theirs.as_ref()).await {
                 files.push(file_change);
             }
+        } else if ours == theirs {
+            // Both sides hold identical content (e.g. dirty state the
+            // worktree inherited at creation, or an earlier sequential
+            // apply already landed the same change): already present —
+            // not a conflict, nothing to write.
+            files.push(file_change);
         } else if base != theirs {
             conflicts.push(FileConflict {
                 path: file_change.path,
                 change_type: file_change.change_type,
-                base,
-                ours,
-                theirs,
+                base: conflict_text(&base),
+                ours: conflict_text(&ours),
+                theirs: conflict_text(&theirs),
             });
         }
     }