diff --git a/AGENTS.md b/AGENTS.md index b675467..9d68db0 100644 --- a/AGENTS.md +++ b/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) diff --git a/crates/codegen/kigi-shell/src/agent/config.rs b/crates/codegen/kigi-shell/src/agent/config.rs index 19190c2..bba29a9 100644 --- a/crates/codegen/kigi-shell/src/agent/config.rs +++ b/crates/codegen/kigi-shell/src/agent/config.rs @@ -1919,6 +1919,26 @@ impl Config { pub(crate) fn resolve_graph(&self) -> Resolved { 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::().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::().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 diff --git a/crates/codegen/kigi-shell/src/session/acp_session.rs b/crates/codegen/kigi-shell/src/session/acp_session.rs index 6dd2f6c..a547ba4 100644 --- a/crates/codegen/kigi-shell/src/session/acp_session.rs +++ b/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>, + /// 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 diff --git a/crates/codegen/kigi-shell/src/session/acp_session_impl/graph.rs b/crates/codegen/kigi-shell/src/session/acp_session_impl/graph.rs index b61151a..8e6e131 100644 --- a/crates/codegen/kigi-shell/src/session/acp_session_impl/graph.rs +++ b/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 { + 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 .", ) - }; - 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 = 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 = 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 .".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) => { { diff --git a/crates/codegen/kigi-shell/src/session/acp_session_impl/graph_workers.rs b/crates/codegen/kigi-shell/src/session/acp_session_impl/graph_workers.rs new file mode 100644 index 0000000..dc52fbf --- /dev/null +++ b/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 }, +} + +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 = 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, + /// 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, +} + +#[derive(Debug, Clone)] +pub(crate) struct WorkerSpawnOutcome { + pub success: bool, + pub cancelled: bool, + pub backgrounded: bool, + pub output: String, + pub error: Option, + pub child_session_id: String, + pub tokens_used: u64, + pub worktree_path: Option, +} + +#[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; + /// 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, + pub parent_session_id: String, + pub parent_prompt_id: Option, +} + +#[async_trait::async_trait] +impl GraphWorkerSpawner for GraphWorkerChannelSpawner { + async fn spawn(&self, id: &str, spec: WorkerSpawnSpec) -> Result { + 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, + /// Last worker child session id (audit link, stored on the node). + pub worker_session_id: Option, +} + +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, + node_id: &str, + node_objective: &str, + rounds_cap: u32, +) -> NodeRunReport { + let mut tokens: i64 = 0; + let mut gaps: Vec = Vec::new(); + let mut resume_from: Option = None; + let mut worktree_path: Option = 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 { + 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> { + 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) { + 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 = 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>, + specs: std::sync::Mutex, bool)>>, + cancels: std::sync::Mutex>, + } + + #[async_trait::async_trait] + impl GraphWorkerSpawner for MockSpawner { + async fn spawn( + &self, + id: &str, + spec: WorkerSpawnSpec, + ) -> Result { + 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 = 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 = 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 = 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 + ); + } +} diff --git a/crates/codegen/kigi-shell/src/session/acp_session_impl/spawn.rs b/crates/codegen/kigi-shell/src/session/acp_session_impl/spawn.rs index 360d3ed..d1d0cf3 100644 --- a/crates/codegen/kigi-shell/src/session/acp_session_impl/spawn.rs +++ b/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), diff --git a/crates/codegen/kigi-shell/src/session/acp_session_impl/tasks_cancel.rs b/crates/codegen/kigi-shell/src/session/acp_session_impl/tasks_cancel.rs index ee09aab..a775842 100644 --- a/crates/codegen/kigi-shell/src/session/acp_session_impl/tasks_cancel.rs +++ b/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 diff --git a/crates/codegen/kigi-shell/src/session/acp_session_tests/cancel_running_task_tests.rs b/crates/codegen/kigi-shell/src/session/acp_session_tests/cancel_running_task_tests.rs index f4f5d00..40cf84f 100644 --- a/crates/codegen/kigi-shell/src/session/acp_session_tests/cancel_running_task_tests.rs +++ b/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), diff --git a/crates/codegen/kigi-shell/src/session/acp_session_tests/graph/graph_e2e_tests.rs b/crates/codegen/kigi-shell/src/session/acp_session_tests/graph/graph_e2e_tests.rs index da6e521..50061f5 100644 --- a/crates/codegen/kigi-shell/src/session/acp_session_tests/graph/graph_e2e_tests.rs +++ b/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, +) { + 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, ) { let tmp = TempDir::new().expect("tempdir"); let (gateway_tx, _gateway_rx) = tokio::sync::mpsc::unbounded_channel::(); let (persistence_tx, persistence_rx) = tokio::sync::mpsc::unbounded_channel::(); 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, + cwd: Option, + 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, + StdArc>>, +) { + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::(); + let captured: StdArc>> = + 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)> = 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 { + 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::(); + tokio::task::spawn_local(async move { + while let Some(ev) = rx.recv().await { + if let SubagentEvent::Spawn(req) = ev { + let (output, wt): (String, Option) = 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) }; +} diff --git a/crates/codegen/kigi-shell/src/session/acp_session_tests/idle_resume_tests.rs b/crates/codegen/kigi-shell/src/session/acp_session_tests/idle_resume_tests.rs index 5222d2f..5567cba 100644 --- a/crates/codegen/kigi-shell/src/session/acp_session_tests/idle_resume_tests.rs +++ b/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), diff --git a/crates/codegen/kigi-shell/src/session/acp_session_tests/inline_auto_compact_flow_tests.rs b/crates/codegen/kigi-shell/src/session/acp_session_tests/inline_auto_compact_flow_tests.rs index 148ed98..09e8e6e 100644 --- a/crates/codegen/kigi-shell/src/session/acp_session_tests/inline_auto_compact_flow_tests.rs +++ b/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), diff --git a/crates/codegen/kigi-shell/src/session/acp_session_tests/memory_config_tests.rs b/crates/codegen/kigi-shell/src/session/acp_session_tests/memory_config_tests.rs index 63e45f5..95b6ee7 100644 --- a/crates/codegen/kigi-shell/src/session/acp_session_tests/memory_config_tests.rs +++ b/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), diff --git a/crates/codegen/kigi-shell/src/session/acp_session_tests/replay_buffer_send_update_tests.rs b/crates/codegen/kigi-shell/src/session/acp_session_tests/replay_buffer_send_update_tests.rs index f78ff0f..92e3844 100644 --- a/crates/codegen/kigi-shell/src/session/acp_session_tests/replay_buffer_send_update_tests.rs +++ b/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), diff --git a/crates/codegen/kigi-shell/src/session/acp_session_tests/support.rs b/crates/codegen/kigi-shell/src/session/acp_session_tests/support.rs index ce8b108..8dc14e4 100644 --- a/crates/codegen/kigi-shell/src/session/acp_session_tests/support.rs +++ b/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), diff --git a/crates/codegen/kigi-shell/src/session/compaction.rs b/crates/codegen/kigi-shell/src/session/compaction.rs index ab87d05..110a51f 100644 --- a/crates/codegen/kigi-shell/src/session/compaction.rs +++ b/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), diff --git a/crates/codegen/kigi-shell/src/session/graph_planner.rs b/crates/codegen/kigi-shell/src/session/graph_planner.rs index edb1091..0da1cb3 100644 --- a/crates/codegen/kigi-shell/src/session/graph_planner.rs +++ b/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, diff --git a/crates/codegen/kigi-shell/src/session/graph_tracker.rs b/crates/codegen/kigi-shell/src/session/graph_tracker.rs index d969e79..44ed117 100644 --- a/crates/codegen/kigi-shell/src/session/graph_tracker.rs +++ b/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) { diff --git a/crates/codegen/kigi-shell/src/session/templates/graph_node_verifier_prompt.md b/crates/codegen/kigi-shell/src/session/templates/graph_node_verifier_prompt.md new file mode 100644 index 0000000..06adf34 --- /dev/null +++ b/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: +- +``` + +Be strict but fair: judge ONLY this node's contract, not sibling nodes' +scope and not style preferences. diff --git a/crates/codegen/kigi-shell/src/session/templates/graph_node_worker_prompt.md b/crates/codegen/kigi-shell/src/session/templates/graph_node_worker_prompt.md new file mode 100644 index 0000000..9bcbe2f --- /dev/null +++ b/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. diff --git a/crates/codegen/kigi-workspace/src/worktree/mod.rs b/crates/codegen/kigi-workspace/src/worktree/mod.rs index 3a36928..c5598c2 100644 --- a/crates/codegen/kigi-workspace/src/worktree/mod.rs +++ b/crates/codegen/kigi-workspace/src/worktree/mod.rs @@ -2090,16 +2090,22 @@ async fn get_apply_context(worktree_path: &str) -> Result { .await? } -async fn get_file_at_commit(worktree_path: &str, commit: &str, path: &str) -> Option { - 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> { + 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>) -> 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>) -> Option { + content + .as_ref() + .map(|b| String::from_utf8_lossy(b).into_owned()) +} + pub async fn apply_worktree(req: &ApplyWorktreeRequest) -> Result { 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 Result