From bb5cbff62d2a292c02fe374ae0cfb590c4bd6d4e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Mon, 20 Jul 2026 19:17:24 -0400 Subject: [PATCH] Add /graph G3: dynamic replan from DISCOVERED work items MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Workers, verifiers, and serial node goals can now surface out-of-scope work as line-anchored 'DISCOVERED: ' markers (fence-stripped and placeholder-filtered — the templates' own examples are fenced so verbatim echoes never parse; worker summaries embedded in verifier prompts get the marker neutralized alongside NODE_RESULT/NODE_VERDICT). Discoveries queue on the orchestration as persisted state and fold into the graph at dispatch boundaries: a replanner subagent produces a strictly APPEND-ONLY appendix, validated against the live graph (existing-id deps allowed; edges onto gn-final rejected — they would cycle the moment the final-gating extension lands; Blocks deps on Failed/Blocked nodes rejected as DeadDep so the attempt-2 feedback loop repairs the artifact). Installing an appendix bumps plan_version, freezes an immutable graph.baseline.v{N}.json next to the prior versions, extends gn-final's gate (demoting a Ready final back to Waiting), and recomputes readiness. DiscoveredFrom edges are audit metadata, never scheduling gates: an origin is always terminal at replan time, so gating on it is either a no-op or a permanent wedge — and a failed node's discoveries are still real work. Replanning is bounded by KIGI_GRAPH_REPLAN_CAP (default 3; 0 disables it quietly): past the cap, after the final node has achieved, or on replan failure, discoveries drain to history only — a working graph is never paused for a failed enhancement pass, and it always converges. The budget gate now precedes the replan boundary (a budget-dead graph keeps its discoveries queued for a later --budget top-up instead of spending two replanner runs first), and both planner runners delete stale artifacts before spawning so a child that responds without writing can never get a previous pass's file validated as its own output. Tests: validate_replan unit coverage (existing-id resolution, DiscoveredFrom dedup, collisions, dead deps vs dead origins, terminal-node edges, combined-graph cycles), tracker appendix/regate/audit-edge tests, and two e2e flows — discovery → replan → appended node runs to Achieved with both baselines frozen, and cap-0 draining to history while the graph still converges. kigi-shell 4935 lib tests green; workspace clippy clean. --- AGENTS.md | 12 + crates/codegen/kigi-shell/src/agent/config.rs | 10 + .../kigi-shell/src/session/acp_session.rs | 5 + .../src/session/acp_session_impl/graph.rs | 26 +- .../session/acp_session_impl/graph_replan.rs | 233 +++++++++++ .../session/acp_session_impl/graph_workers.rs | 54 ++- .../src/session/acp_session_impl/spawn.rs | 1 + .../cancel_running_task_tests.rs | 4 + .../graph/graph_e2e_tests.rs | 181 +++++++++ .../acp_session_tests/idle_resume_tests.rs | 1 + .../inline_auto_compact_flow_tests.rs | 3 + .../acp_session_tests/memory_config_tests.rs | 1 + .../replay_buffer_send_update_tests.rs | 1 + .../src/session/acp_session_tests/support.rs | 1 + .../kigi-shell/src/session/compaction.rs | 1 + .../kigi-shell/src/session/graph_plan.rs | 376 +++++++++++++++++- .../kigi-shell/src/session/graph_planner.rs | 128 ++++++ .../kigi-shell/src/session/graph_tracker.rs | 169 +++++++- .../src/session/storage/jsonl/tests.rs | 2 + .../templates/graph_node_verifier_prompt.md | 8 + .../templates/graph_node_worker_prompt.md | 10 + .../templates/graph_replanner_prompt.md | 55 +++ 22 files changed, 1272 insertions(+), 10 deletions(-) create mode 100644 crates/codegen/kigi-shell/src/session/acp_session_impl/graph_replan.rs create mode 100644 crates/codegen/kigi-shell/src/session/templates/graph_replanner_prompt.md diff --git a/AGENTS.md b/AGENTS.md index 545af11..4d47033 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -118,6 +118,18 @@ edges stay deterministic Rust. The harness appends a terminal `persist_graph_state` chokepoint (checkpoint ⇔ badge tick); old pagers degrade via `#[serde(other)] Unknown`. PTY scenarios: `graph_slash_presession{,_disabled}.yaml`. +- G3: dynamic replan. `DISCOVERED: ` line markers (fence-stripped, + placeholder-filtered) from workers/verifiers/the serial node's final + text queue as `pending_discoveries`; at each dispatch boundary + `maybe_replan_graph` (`acp_session_impl/graph_replan.rs`) runs a + replanner subagent producing an APPEND-ONLY appendix + (`validate_replan`: existing-id deps allowed, edges onto `gn-final` + rejected — they would cycle after the final-gating extension), + bumps `plan_version`, freezes `graph.baseline.v{N}.json`, and regates + `gn-final` (Ready→Waiting). Bounded by `KIGI_GRAPH_REPLAN_CAP` + (default 3, 0 = off); past the cap — and after the final node has + achieved — discoveries drain to history only. Replan failure degrades + (history + notice); it never pauses a working graph. ## 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 bba29a9..cf4b039 100644 --- a/crates/codegen/kigi-shell/src/agent/config.rs +++ b/crates/codegen/kigi-shell/src/agent/config.rs @@ -1929,6 +1929,16 @@ impl Config { .unwrap_or(3) .clamp(1, 8) } + /// Max replan passes per graph (`KIGI_GRAPH_REPLAN_CAP`); 0 turns + /// dynamic replanning off. Past the cap, discoveries drain to + /// history only — the graph must still converge. Clamped to [0, 10]. + pub(crate) fn resolve_graph_replan_cap(&self) -> u32 { + std::env::var("KIGI_GRAPH_REPLAN_CAP") + .ok() + .and_then(|v| v.parse::().ok()) + .unwrap_or(3) + .clamp(0, 10) + } /// Max worker↔verifier rounds per parallel graph node /// (`KIGI_GRAPH_NODE_ROUNDS`); exhausting them fails the node. /// Clamped to [1, 8]. diff --git a/crates/codegen/kigi-shell/src/session/acp_session.rs b/crates/codegen/kigi-shell/src/session/acp_session.rs index a547ba4..ce8d369 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_replan.rs"] +mod graph_replan; #[path = "acp_session_impl/graph_workers.rs"] mod graph_workers; #[path = "acp_session_impl/interjection.rs"] @@ -607,6 +609,9 @@ pub(crate) struct SessionActor { /// Max worker↔verifier rounds per parallel graph node before the /// node fails. Cached at actor construction. pub(crate) graph_node_rounds: u32, + /// Max replan passes per graph (0 = replanning off). Cached at + /// actor construction. + pub(crate) graph_replan_cap: 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 6bb8bcf..f3a9d74 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 @@ -49,7 +49,10 @@ pub(super) fn node_goal_objective( {spec}\n\n\ This node is one unit of a larger graph objective:\n\ {graph_objective}\n\n\ - Complete ONLY this node's scope; other nodes cover the rest.", + Complete ONLY this node's scope; other nodes cover the rest. If you \ + find NECESSARY work outside this node's scope, do NOT do it — list \ + each item on its own line as `DISCOVERED: ` in your \ + final summary; the harness turns these into new graph nodes.", title = node.title, spec = node.spec, ) @@ -339,7 +342,7 @@ impl SessionActor { /// Write the immutable plan baseline for the current version. /// `create_new` guarantees a frozen baseline is never overwritten — /// an existing file is the infra failure it looks like. - async fn write_graph_baseline(&self, nodes: &[GraphNode]) -> std::io::Result<()> { + pub(super) async fn write_graph_baseline(&self, nodes: &[GraphNode]) -> std::io::Result<()> { let path = { let tracker = self.graph_tracker.lock(); let version = tracker.snapshot().map(|s| s.plan_version).unwrap_or(1); @@ -543,6 +546,22 @@ impl SessionActor { .snapshot() .map(|o| o.total_worker_rounds) .unwrap_or(0); + // Serial-path discovery capture: the node's final assistant text + // may carry `DISCOVERED:` items (same contract as the parallel + // workers, zero extra tool surface). + if let Some(text) = self.chat_state_handle.get_last_assistant_text().await { + let found = super::graph_workers::parse_discovered_lines(&text); + if !found.is_empty() { + let ds: Vec = found + .into_iter() + .map(|description| super::super::graph_tracker::Discovery { + from_node: node_id.clone(), + description, + }) + .collect(); + self.graph_tracker.lock().queue_discoveries(ds); + } + } self.archive_node_artifacts(&node_id).await; tracing::info!(%node_id, rounds, node_tokens, "graph: node achieved"); self.graph_tracker @@ -565,6 +584,9 @@ impl SessionActor { // serial-launch failure — the pauser already messaged. return None; } + // Replan boundary: fold queued discoveries into the graph + // (bounded; failure degrades to history-only). + self.maybe_replan_graph().await; if self.graph_tracker.lock().remaining_budget() == Some(0) { tracing::warn!("graph: budget exhausted at dispatch"); self.graph_tracker.lock().budget_limit(); diff --git a/crates/codegen/kigi-shell/src/session/acp_session_impl/graph_replan.rs b/crates/codegen/kigi-shell/src/session/acp_session_impl/graph_replan.rs new file mode 100644 index 0000000..afda5fa --- /dev/null +++ b/crates/codegen/kigi-shell/src/session/acp_session_impl/graph_replan.rs @@ -0,0 +1,233 @@ +//! Dynamic replan (G3): fold `DISCOVERED:` items surfaced during node +//! execution into the graph at dispatch boundaries. +//! +//! SGH version discipline: the running plan is immutable inside a +//! version — a replan appends new nodes (never edits existing ones), +//! bumps `plan_version`, and freezes a new immutable baseline. The pass +//! is BOUNDED by `KIGI_GRAPH_REPLAN_CAP` (default 3, 0 = off): past the +//! cap, discoveries drain to history only, so the graph always +//! converges. Replan failure DEGRADES (discoveries kept in history, the +//! graph keeps running) — unlike initial planning, a working graph is +//! never paused because an enhancement pass failed. + +use std::sync::Arc; + +use super::super::goal_planner::{ChannelSpawner, GoalPlannerSpawner}; +use super::super::graph_planner::{ + GRAPH_REPLANNER_SUBAGENT_DESCRIPTION, GraphPlannerOutcome, GraphReplannerInputs, + run_graph_replanner, +}; +use super::SessionActor; + +impl SessionActor { + /// Replan boundary, called at the top of every `drive_graph` + /// iteration. No-op without pending discoveries. + pub(super) async fn maybe_replan_graph(&self) { + let (pending, replan_runs) = { + let tracker = self.graph_tracker.lock(); + let Some(state) = tracker.snapshot() else { + return; + }; + (state.pending_discoveries.clone(), state.replan_runs) + }; + if pending.is_empty() { + return; + } + // Budget first: a budget-dead graph must not spend two replanner + // runs right before the dispatch loop trips BudgetLimited. The + // discoveries STAY QUEUED (persisted) — a later + // `/graph resume --budget` top-up re-enters and replans with + // budget actually available. + if self.graph_tracker.lock().remaining_budget() == Some(0) { + tracing::info!("graph replan: budget exhausted; keeping discoveries queued"); + return; + } + let final_achieved = self + .graph_tracker + .lock() + .node(super::super::graph_tracker::FINAL_NODE_ID) + .is_some_and(|n| n.status == super::super::graph_tracker::NodeStatus::Achieved); + if final_achieved { + // The whole-objective gate already passed; late discoveries + // (typically from the final verification itself) are + // advisory — appending nodes now would ship work the + // terminal gate never re-verified. + let n = self.graph_tracker.lock().drain_discoveries_to_history(); + self.persist_graph_state(); + tracing::info!( + drained = n, + "graph replan: final already achieved; history only" + ); + return; + } + if self.graph_replan_cap == 0 { + // Feature off: quiet drain (history keeps the audit trail). + let n = self.graph_tracker.lock().drain_discoveries_to_history(); + self.persist_graph_state(); + tracing::info!(drained = n, "graph replan: disabled (cap 0); history only"); + return; + } + if replan_runs >= self.graph_replan_cap { + let n = self.graph_tracker.lock().drain_discoveries_to_history(); + self.persist_graph_state(); + tracing::warn!( + drained = n, + replan_runs, + cap = self.graph_replan_cap, + "graph replan: cap exhausted; discoveries recorded in history only" + ); + self.send_slash_command_output(&format!( + "Graph replan cap reached ({replan_runs}/{}); {n} discover{} recorded in \ + history only — the graph will converge on the current plan.", + self.graph_replan_cap, + if n == 1 { "y" } else { "ies" }, + )) + .await; + return; + } + + let Some(event_tx) = self.tool_context.subagent_event_tx.clone() else { + tracing::warn!("graph replan: no subagent coordinator; keeping discoveries queued"); + return; + }; + let (existing, objective, current_graph, discoveries_text, graph_file, next_version) = { + let tracker = self.graph_tracker.lock(); + let Some(state) = tracker.snapshot() else { + return; + }; + let compact: Vec = state + .nodes + .iter() + .map(|n| { + serde_json::json!({ + "id": n.id, + "title": n.title, + "status": format!("{:?}", n.status), + "deps": n.deps.iter().map(|d| d.on.clone()).collect::>(), + }) + }) + .collect(); + let discoveries_text = pending + .iter() + .map(|d| format!("- (from {}) {}", d.from_node, d.description)) + .collect::>() + .join("\n"); + ( + state.nodes.clone(), + state.objective.clone(), + serde_json::to_string(&compact).unwrap_or_default(), + discoveries_text, + tracker + .artifacts_dir() + .join(format!("replan.v{}.json", state.plan_version + 1)), + state.plan_version + 1, + ) + }; + let parent_prompt_id = self + .current_prompt_id + .lock() + .expect("current_prompt_id mutex poisoned") + .clone(); + let spawner: Arc = Arc::new(ChannelSpawner { + event_tx, + parent_session_id: self.session_id_string(), + parent_prompt_id, + cwd: Some(self.tool_context.cwd.as_str().to_owned()), + role_override: Default::default(), + events: Some(self.events.writer()), + }); + let tool_names = self.resolve_inherit_role_tool_names().await; + + let mut feedback = String::new(); + for attempt in 1..=2u32 { + tracing::info!( + attempt, + next_version, + role = GRAPH_REPLANNER_SUBAGENT_DESCRIPTION, + pending = pending.len(), + "graph replan: firing" + ); + match run_graph_replanner( + spawner.clone(), + &existing, + GraphReplannerInputs { + objective: &objective, + current_graph: ¤t_graph, + discoveries: &discoveries_text, + feedback: &feedback, + graph_file: &graph_file, + tool_names: &tool_names, + inherit_tool_names: &tool_names, + }, + ) + .await + { + GraphPlannerOutcome::Planned(appendix) if appendix.is_empty() => { + // Escape hatch: everything already covered. The pass + // still counts against the cap. + tracing::info!("graph replan: empty appendix (already covered)"); + { + let mut tracker = self.graph_tracker.lock(); + tracker.drain_discoveries_to_history(); + if let Some(state) = tracker.snapshot_mut() { + state.replan_runs += 1; + } + } + self.persist_graph_state(); + return; + } + GraphPlannerOutcome::Planned(appendix) => { + let added = appendix.len(); + self.graph_tracker.lock().append_replan_nodes(appendix); + // Freeze the new version's immutable baseline (full + // node set post-append; create_new keeps v{N-1} + // byte-identical forever). + let all_nodes = self + .graph_tracker + .lock() + .snapshot() + .map(|s| s.nodes.clone()) + .unwrap_or_default(); + if let Err(err) = self.write_graph_baseline(&all_nodes).await { + tracing::warn!(%err, "graph replan: baseline write failed (audit gap only)"); + } + self.persist_graph_state(); + tracing::info!(added, next_version, "graph replan: appendix installed"); + self.send_slash_command_output(&format!( + "Graph replanned (v{next_version}): {added} node(s) added from \ + discovered work." + )) + .await; + return; + } + GraphPlannerOutcome::Invalid { reason } if attempt == 1 => { + tracing::warn!(%reason, "graph replan: invalid appendix; retrying with feedback"); + feedback = format!( + "Your previous replan JSON failed validation:\n{reason}\n\ + Rewrite the file fixing exactly this." + ); + } + GraphPlannerOutcome::Invalid { reason } + | GraphPlannerOutcome::FailClosed { reason } => { + // Degrade, never pause a working graph for a failed + // enhancement pass. The run still counts. + tracing::warn!(%reason, "graph replan: failed; draining discoveries to history"); + { + let mut tracker = self.graph_tracker.lock(); + tracker.drain_discoveries_to_history(); + if let Some(state) = tracker.snapshot_mut() { + state.replan_runs += 1; + } + } + self.persist_graph_state(); + self.send_slash_command_output(&format!( + "Graph replan failed ({reason}); discovered work recorded in \ + history only." + )) + .await; + return; + } + } + } + } +} 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 index dc52fbf..5d73f70 100644 --- 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 @@ -136,6 +136,26 @@ pub(crate) fn parse_node_verdict(output: &str) -> NodeVerdict { } } +/// Line-anchored `DISCOVERED:` items outside fenced blocks. The +/// placeholder filter (`<`) drops template echoes ("") a child may parrot back. +pub(crate) fn parse_discovered_lines(output: &str) -> Vec { + strip_fenced_blocks(output) + .lines() + .filter_map(|l| { + l.trim_start() + .trim_start_matches('`') + .strip_prefix("DISCOVERED:") + }) + .map(str::trim) + // Placeholder-echo defense: drop only the templates' literal + // "" shape, not every '<' (legit Rust + // discoveries mention generics like Vec). + .filter(|d| !d.is_empty() && !d.starts_with('<')) + .map(str::to_owned) + .collect() +} + // Spawner seam (mockable in tests) pub(crate) struct WorkerSpawnSpec { @@ -258,6 +278,8 @@ pub(crate) struct NodeRunReport { pub worktree_path: Option, /// Last worker child session id (audit link, stored on the node). pub worker_session_id: Option, + /// `DISCOVERED:` items surfaced by the workers/verifiers (deduped). + pub discoveries: Vec, } fn worker_prompt(node_objective: &str, gaps: &[String]) -> String { @@ -282,7 +304,8 @@ fn verifier_prompt(node_objective: &str, worker_summary: &str) -> String { // into the verifier's context. let safe_summary = worker_summary .replace("NODE_VERDICT", "NODE-VERDICT") - .replace("NODE_RESULT", "NODE-RESULT"); + .replace("NODE_RESULT", "NODE-RESULT") + .replace("DISCOVERED", "DISCOVERED-"); 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" @@ -298,6 +321,7 @@ pub(crate) async fn run_node_to_verdict( rounds_cap: u32, ) -> NodeRunReport { let mut tokens: i64 = 0; + let mut discoveries: Vec = Vec::new(); let mut gaps: Vec = Vec::new(); let mut resume_from: Option = None; let mut worktree_path: Option = None; @@ -326,6 +350,7 @@ pub(crate) async fn run_node_to_verdict( tokens_used: tokens, worktree_path, worker_session_id: resume_from, + discoveries: discoveries.clone(), }; } }; @@ -341,6 +366,11 @@ pub(crate) async fn run_node_to_verdict( if !outcome.child_session_id.is_empty() { resume_from = Some(outcome.child_session_id.clone()); } + for d in parse_discovered_lines(&outcome.output) { + if !discoveries.contains(&d) { + discoveries.push(d); + } + } if outcome.cancelled { return NodeRunReport { @@ -351,6 +381,7 @@ pub(crate) async fn run_node_to_verdict( tokens_used: tokens, worktree_path, worker_session_id: resume_from, + discoveries, }; } if outcome.backgrounded { @@ -393,6 +424,7 @@ pub(crate) async fn run_node_to_verdict( tokens_used: tokens, worktree_path: None, worker_session_id: resume_from, + discoveries: discoveries.clone(), }; } @@ -406,6 +438,7 @@ pub(crate) async fn run_node_to_verdict( tokens_used: tokens, worktree_path, worker_session_id: resume_from, + discoveries: discoveries.clone(), }; } WorkerClaim::Unparseable => { @@ -429,6 +462,11 @@ pub(crate) async fn run_node_to_verdict( let verdict = match spawner.spawn(&verify_id, verify_spec).await { Ok(v) => { tokens = tokens.saturating_add(v.tokens_used as i64); + for d in parse_discovered_lines(&v.output) { + if !discoveries.contains(&d) { + discoveries.push(d); + } + } if v.success { parse_node_verdict(&v.output) } else { @@ -456,6 +494,7 @@ pub(crate) async fn run_node_to_verdict( tokens_used: tokens, worktree_path, worker_session_id: resume_from, + discoveries: discoveries.clone(), }; } NodeVerdict::NotAchieved { gaps: new_gaps } => { @@ -477,6 +516,7 @@ pub(crate) async fn run_node_to_verdict( tokens_used: tokens, worktree_path, worker_session_id: resume_from, + discoveries, } } @@ -583,6 +623,18 @@ impl SessionActor { { node.goal_id = Some(worker_id.clone()); } + if !report.discoveries.is_empty() { + // A failed node's discoveries are still real work. + let ds: Vec = report + .discoveries + .iter() + .map(|d| crate::session::graph_tracker::Discovery { + from_node: report.node_id.clone(), + description: d.clone(), + }) + .collect(); + self.graph_tracker.lock().queue_discoveries(ds); + } if !report.achieved { failed += 1; { 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 798bd85..d7ecd85 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 @@ -1101,6 +1101,7 @@ pub(crate) async fn spawn_session_actor( graph_tracker, graph_concurrency: effective_config.resolve_graph_concurrency(), graph_node_rounds: effective_config.resolve_graph_node_rounds(), + graph_replan_cap: effective_config.resolve_graph_replan_cap(), 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/cancel_running_task_tests.rs b/crates/codegen/kigi-shell/src/session/acp_session_tests/cancel_running_task_tests.rs index 40cf84f..1180ea6 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 @@ -218,6 +218,7 @@ async fn persist_ack_waits_for_disk_flush_before_success() { )), graph_concurrency: 1, graph_node_rounds: 3, + graph_replan_cap: 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), @@ -660,6 +661,7 @@ async fn first_turn_memory_injection_disabled_does_not_persist_to_chat_history() )), graph_concurrency: 1, graph_node_rounds: 3, + graph_replan_cap: 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), @@ -911,6 +913,7 @@ async fn cancel_running_task_teardown_clears_running_and_pending_work() { )), graph_concurrency: 1, graph_node_rounds: 3, + graph_replan_cap: 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), @@ -1895,6 +1898,7 @@ async fn cancel_propagates_to_sampler_handle_so_no_further_emission() { )), graph_concurrency: 1, graph_node_rounds: 3, + graph_replan_cap: 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 b59fc46..a1087ba 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 @@ -1619,3 +1619,184 @@ async fn graph_slash_terminal_outcomes_through_handle_prompt() { .await; unsafe { std::env::remove_var(ENV_FLAG) }; } + +// ── G3: dynamic replan ───────────────────────────────────────────── + +/// Route replies for the replan flow: planner writes the diamond, +/// node-a's worker reports a discovery, the replanner appends a `docs` +/// node discovered_from gn-a, everything else is happy-path. +fn replan_reply( + req: &kigi_tools::implementations::kigi::task::types::SubagentRequest, + dag: &[u8], +) -> String { + if req.prompt.contains("Graph Replanner") { + let path = req + .prompt + .find("/replan.v") + .map(|start_idx| { + let end = req.prompt[start_idx..] + .find(".json") + .map(|e| start_idx + e + ".json".len()) + .unwrap(); + let start = req.prompt[..start_idx] + .rfind(|c: char| !c.is_ascii_graphic() || c == '`') + .map(|i| i + 1) + .unwrap_or(0); + req.prompt[start..end].to_string() + }) + .expect("replanner prompt embeds the artifact path"); + let a_id = crate::session::graph_plan::node_id_for_slug("a"); + let body = serde_json::json!({ + "nodes": [{ + "id": "docs", + "title": "Docs page", + "spec": "write the docs page", + "deps": [], + "discovered_from": [a_id], + }] + }) + .to_string(); + std::fs::create_dir_all(std::path::Path::new(&path).parent().unwrap()).unwrap(); + std::fs::write(&path, body).unwrap(); + return "Done".to_owned(); + } + if req.prompt.contains("Graph Node Worker") && req.prompt.contains("do a") { + return "NODE_RESULT: done\nBuilt a.\nDISCOVERED: also need a docs page".to_owned(); + } + happy_reply(req, dag) +} + +#[tokio::test(flavor = "current_thread")] +#[serial] +async fn discovery_triggers_replan_and_appended_node_runs_to_achieved() { + 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| replan_reply(req, &dag), + false, + ); + actor.tool_context.subagent_event_tx = Some(coord_tx); + actor.graph_concurrency = 2; + + let outcome = actor.setup_graph("build the diamond", None).await; + // Batch(a+b) → replan appends docs → batch/serial drains the + // rest → the serial tail is gn-final (5 nodes total now). + let reminder = match outcome { + graph::GraphSetupOutcome::Inference { reminder, .. } => reminder, + graph::GraphSetupOutcome::Message(msg) => panic!("expected Inference: {msg}"), + }; + let s = actor.graph_tracker.lock().snapshot().cloned().unwrap(); + assert_eq!(s.nodes.len(), 5, "diamond(3) + final + appended docs"); + assert_eq!(s.plan_version, 2, "replan bumped the version"); + assert_eq!(s.replan_runs, 1); + assert!(s.pending_discoveries.is_empty()); + let docs_id = crate::session::graph_plan::node_id_for_slug("docs"); + let docs = s.nodes.iter().find(|n| n.id == docs_id).expect("docs node"); + assert!( + docs.deps.iter().any(|d| { + d.on == crate::session::graph_plan::node_id_for_slug("a") + && d.kind == crate::session::graph_tracker::DepKind::DiscoveredFrom + }), + "appendix carries the DiscoveredFrom edge: {:?}", + docs.deps + ); + // gn-final gates on the appendix too. + let final_node = s + .nodes + .iter() + .find(|n| n.id == crate::session::graph_tracker::FINAL_NODE_ID) + .unwrap(); + assert!(final_node.deps.iter().any(|d| d.on == docs_id)); + // docs already ran in the second batch (parallel with c) or + // serially; by the time a serial reminder surfaces it must be + // for gn-final with everything else achieved. + assert!(reminder.contains("Final verification"), "{reminder}"); + + // Both baselines exist; v1 stayed byte-identical. + let b1 = actor.graph_tracker.lock().baseline_path(1); + let b2 = actor.graph_tracker.lock().baseline_path(2); + assert!(b1.is_file() && b2.is_file()); + let v1: Vec = + serde_json::from_slice(&std::fs::read(&b1).unwrap()).unwrap(); + assert_eq!(v1.len(), 4, "v1 baseline must not gain the appendix"); + let v2: Vec = + serde_json::from_slice(&std::fs::read(&b2).unwrap()).unwrap(); + assert_eq!(v2.len(), 5); + + // Finish gn-final serially → graph Complete. + drive_node_goal_to_complete(&actor).await; + let end = actor.run_graph_round_end().await; + assert!(end.is_none()); + assert_eq!( + actor.graph_tracker.lock().status(), + Some(GoalStatus::Complete) + ); + // Exactly one replanner spawn. + let replans = captured + .lock() + .unwrap() + .iter() + .filter(|c| c.prompt.contains("Graph Replanner")) + .count(); + assert_eq!(replans, 1); + }) + .await; + unsafe { std::env::remove_var(ENV_FLAG) }; +} + +#[tokio::test(flavor = "current_thread")] +#[serial] +async fn replan_cap_zero_drains_discoveries_to_history_and_converges() { + 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| replan_reply(req, &dag), + false, + ); + actor.tool_context.subagent_event_tx = Some(coord_tx); + actor.graph_concurrency = 2; + actor.graph_replan_cap = 0; + + let _ = actor.setup_graph("build the diamond", None).await; + let s = actor.graph_tracker.lock().snapshot().cloned().unwrap(); + assert_eq!(s.nodes.len(), 4, "no appendix with the cap at 0"); + assert_eq!(s.plan_version, 1); + assert!(s.pending_discoveries.is_empty(), "drained to history"); + assert!( + s.history.iter().any(|h| h + .detail + .as_deref() + .is_some_and(|d| d.contains("discovered: also need a docs page"))), + "discovery must survive in history" + ); + let replans = captured + .lock() + .unwrap() + .iter() + .filter(|c| c.prompt.contains("Graph Replanner")) + .count(); + assert_eq!(replans, 0, "cap 0 must never spawn the replanner"); + + // Graph still converges (serial tail: c, then gn-final). + 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; + 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 5567cba..44ae1f9 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 @@ -247,6 +247,7 @@ async fn test_e2e_idle_resume_refreshes_model_metadata() { )), graph_concurrency: 1, graph_node_rounds: 3, + graph_replan_cap: 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 09e8e6e..aa9ebde 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 @@ -179,6 +179,7 @@ async fn create_test_actor( )), graph_concurrency: 1, graph_node_rounds: 3, + graph_replan_cap: 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), @@ -623,6 +624,7 @@ async fn create_test_actor_with_memory( )), graph_concurrency: 1, graph_node_rounds: 3, + graph_replan_cap: 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), @@ -1378,6 +1380,7 @@ async fn test_e2e_idle_resume_refreshes_model_metadata() { )), graph_concurrency: 1, graph_node_rounds: 3, + graph_replan_cap: 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 95b6ee7..52cb185 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 @@ -241,6 +241,7 @@ async fn create_test_actor_with_memory( )), graph_concurrency: 1, graph_node_rounds: 3, + graph_replan_cap: 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 92e3844..f49cb02 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 @@ -187,6 +187,7 @@ pub(super) async fn make_replay_send_update_fixture() -> ReplaySendUpdateFixture )), graph_concurrency: 1, graph_node_rounds: 3, + graph_replan_cap: 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 8dc14e4..c66e38b 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 @@ -292,6 +292,7 @@ pub(crate) async fn create_test_actor_ex( )), graph_concurrency: 1, graph_node_rounds: 3, + graph_replan_cap: 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 110a51f..afaa151 100644 --- a/crates/codegen/kigi-shell/src/session/compaction.rs +++ b/crates/codegen/kigi-shell/src/session/compaction.rs @@ -2274,6 +2274,7 @@ mod inline_auto_compact_flow_tests { )), graph_concurrency: 1, graph_node_rounds: 3, + graph_replan_cap: 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_plan.rs b/crates/codegen/kigi-shell/src/session/graph_plan.rs index 69cfa4a..09f42f1 100644 --- a/crates/codegen/kigi-shell/src/session/graph_plan.rs +++ b/crates/codegen/kigi-shell/src/session/graph_plan.rs @@ -41,6 +41,10 @@ struct PlannedNode { spec: String, #[serde(default)] deps: Vec, + /// Replan artifacts only: EXISTING node ids (`gn-…`) whose execution + /// surfaced this node. Ignored by the initial-plan path. + #[serde(default)] + discovered_from: Vec, } /// Why a planner artifact was rejected. Rendered verbatim into the @@ -53,11 +57,30 @@ pub(crate) enum GraphPlanError { TooManyNodes(usize), BadSlug(String), DuplicateSlug(String), - EmptyField { slug: String, field: &'static str }, - UnknownDep { slug: String, dep: String }, + EmptyField { + slug: String, + field: &'static str, + }, + UnknownDep { + slug: String, + dep: String, + }, SelfDep(String), Cycle(Vec), IdCollision(String, String), + /// Replan: a new node's canonical id collides with an existing node. + ExistingCollision(String), + /// Replan: a `deps` entry targets a Failed/Blocked node — the new + /// node could never become Ready. + DeadDep { + slug: String, + dep: String, + }, + /// Replan: `discovered_from` references a node id not in the graph. + UnknownOrigin { + slug: String, + origin: String, + }, } impl std::fmt::Display for GraphPlanError { @@ -87,6 +110,19 @@ impl std::fmt::Display for GraphPlanError { f, "hash id collision between slugs {a:?} and {b:?}; rename one" ), + Self::ExistingCollision(s) => write!( + f, + "new node {s:?} collides with an existing graph node; rename it" + ), + Self::UnknownOrigin { slug, origin } => write!( + f, + "node {slug:?} claims discovered_from unknown node {origin:?}" + ), + Self::DeadDep { slug, dep } => write!( + f, + "node {slug:?} depends on {dep:?}, which already failed; depend on \ + live nodes only (or none)" + ), } } } @@ -295,6 +331,214 @@ fn final_verification_node(objective: &str, planner_nodes: &[GraphNode]) -> Grap } } +/// Parse and validate a REPLAN artifact against the existing graph: +/// strictly append-only. New nodes may depend on existing `gn-…` ids or +/// on each other; the combined graph must stay acyclic; existing nodes +/// are never modified. Returns the canonicalized appendix — `Waiting` +/// status, `Blocks` deps, plus one `DiscoveredFrom` edge per validated +/// `discovered_from` origin. +pub(crate) fn validate_replan( + existing: &[GraphNode], + json: &str, +) -> Result, GraphPlanError> { + let mut planned: PlannedGraph = + serde_json::from_str(json).map_err(|e| GraphPlanError::Parse(e.to_string()))?; + if planned.nodes.is_empty() { + return Err(GraphPlanError::Empty); + } + // Whole-graph cap: MAX_GRAPH_NODES planner nodes + gn-final. The + // payload excludes the final node so "the cap is N" stays truthful + // for replans too. + if planned.nodes.len() + existing.len() > MAX_GRAPH_NODES + 1 { + return Err(GraphPlanError::TooManyNodes( + planned.nodes.len() + existing.len() - 1, + )); + } + for node in &mut planned.nodes { + let mut seen_deps = std::collections::HashSet::new(); + node.deps.retain(|d| seen_deps.insert(d.clone())); + } + + let existing_ids: std::collections::HashSet<&str> = + existing.iter().map(|n| n.id.as_str()).collect(); + let mut seen = std::collections::HashSet::new(); + for node in &planned.nodes { + if !valid_slug(&node.id) { + return Err(GraphPlanError::BadSlug(node.id.clone())); + } + if !seen.insert(node.id.as_str()) { + return Err(GraphPlanError::DuplicateSlug(node.id.clone())); + } + if node.title.trim().is_empty() { + return Err(GraphPlanError::EmptyField { + slug: node.id.clone(), + field: "title", + }); + } + if node.spec.trim().is_empty() { + return Err(GraphPlanError::EmptyField { + slug: node.id.clone(), + field: "spec", + }); + } + for origin in &node.discovered_from { + if !existing_ids.contains(origin.as_str()) { + return Err(GraphPlanError::UnknownOrigin { + slug: node.id.clone(), + origin: origin.clone(), + }); + } + // Any edge onto the terminal node would cycle the moment + // append_replan_nodes gates it on the appendix. Fail fast. + if origin == FINAL_NODE_ID { + return Err(GraphPlanError::UnknownDep { + slug: node.id.clone(), + dep: FINAL_NODE_ID.to_owned(), + }); + } + } + if node.deps.iter().any(|d| d == FINAL_NODE_ID) { + return Err(GraphPlanError::UnknownDep { + slug: node.id.clone(), + dep: FINAL_NODE_ID.to_owned(), + }); + } + } + + // Canonical ids for the appendix; must not collide with anything. + let mut id_of: std::collections::HashMap<&str, String> = std::collections::HashMap::new(); + let mut owner_of_id: std::collections::HashMap = std::collections::HashMap::new(); + for node in &planned.nodes { + let id = node_id_for_slug(&node.id); + if existing_ids.contains(id.as_str()) { + return Err(GraphPlanError::ExistingCollision(node.id.clone())); + } + if let Some(prior) = owner_of_id.insert(id.clone(), node.id.as_str()) { + return Err(GraphPlanError::IdCollision( + prior.to_owned(), + node.id.clone(), + )); + } + id_of.insert(node.id.as_str(), id); + } + + // Deps resolve against existing ids (verbatim) or new slugs. + let resolve = |dep: &str| -> Option { + if existing_ids.contains(dep) { + Some(dep.to_owned()) + } else { + id_of.get(dep).cloned() + } + }; + let dead_ids: std::collections::HashSet<&str> = existing + .iter() + .filter(|n| matches!(n.status, NodeStatus::Failed | NodeStatus::Blocked)) + .map(|n| n.id.as_str()) + .collect(); + for node in &planned.nodes { + for dep in &node.deps { + if dep == &node.id { + return Err(GraphPlanError::SelfDep(node.id.clone())); + } + if resolve(dep).is_none() { + return Err(GraphPlanError::UnknownDep { + slug: node.id.clone(), + dep: dep.clone(), + }); + } + // An ordering dep on a dead node can never satisfy; fail + // fast so the attempt-2 feedback loop repairs the artifact. + // (`discovered_from` origins are exempt — audit-only edges, + // and failed origins are the NORMAL salvage case.) + if dead_ids.contains(dep.as_str()) { + return Err(GraphPlanError::DeadDep { + slug: node.id.clone(), + dep: dep.clone(), + }); + } + } + } + + // Combined-graph acyclicity (Kahn over existing edges + appendix). + // Existing nodes only ever depend on existing nodes, so seeding + // their edges verbatim is sound. + { + let mut ids: Vec = existing.iter().map(|n| n.id.clone()).collect(); + ids.extend(planned.nodes.iter().map(|n| id_of[n.id.as_str()].clone())); + let index_of: std::collections::HashMap<&str, usize> = ids + .iter() + .enumerate() + .map(|(i, id)| (id.as_str(), i)) + .collect(); + let mut edges: Vec<(usize, usize)> = Vec::new(); + for n in existing { + for d in &n.deps { + edges.push((index_of[d.on.as_str()], index_of[n.id.as_str()])); + } + } + for n in &planned.nodes { + let to = index_of[id_of[n.id.as_str()].as_str()]; + for d in &n.deps { + edges.push((index_of[resolve(d).expect("validated").as_str()], to)); + } + } + let mut indegree = vec![0usize; ids.len()]; + for (_, to) in &edges { + indegree[*to] += 1; + } + let mut done = vec![false; ids.len()]; + for _ in 0..ids.len() { + let Some(next) = (0..ids.len()).find(|&i| !done[i] && indegree[i] == 0) else { + let cycle: Vec = (0..ids.len()) + .filter(|&i| !done[i]) + .map(|i| ids[i].clone()) + .collect(); + return Err(GraphPlanError::Cycle(cycle)); + }; + done[next] = true; + for (from, to) in &edges { + if *from == next { + indegree[*to] -= 1; + } + } + } + } + + Ok(planned + .nodes + .iter() + .map(|p| { + let mut deps: Vec = p + .deps + .iter() + .map(|d| NodeDep { + on: resolve(d).expect("validated"), + kind: DepKind::Blocks, + }) + .collect(); + for origin in &p.discovered_from { + if !deps.iter().any(|d| &d.on == origin) { + deps.push(NodeDep { + on: origin.clone(), + kind: DepKind::DiscoveredFrom, + }); + } + } + GraphNode { + id: id_of[p.id.as_str()].clone(), + title: p.title.trim().to_owned(), + spec: p.spec.trim().to_owned(), + deps, + status: NodeStatus::Waiting, + goal_id: None, + rounds: 0, + tokens_used: 0, + failure: None, + } + }) + .collect()) +} + #[cfg(test)] mod tests { use super::*; @@ -424,6 +668,134 @@ mod tests { assert_eq!(nodes[1].id, node_id_for_slug("a")); } + fn existing_graph() -> Vec { + parse_and_validate(&plan_json(&[("a", &[]), ("b", &["a"])]), "objective").unwrap() + } + + #[test] + fn replan_appendix_resolves_existing_ids_and_adds_discovered_from_edges() { + let existing = existing_graph(); + let a_id = node_id_for_slug("a"); + let json = serde_json::json!({ + "nodes": [{ + "id": "docs", + "title": "Docs", + "spec": "write docs", + "deps": [a_id.clone()], + "discovered_from": [a_id.clone()], + }] + }) + .to_string(); + let appendix = validate_replan(&existing, &json).unwrap(); + assert_eq!(appendix.len(), 1); + let node = &appendix[0]; + assert_eq!(node.status, NodeStatus::Waiting); + // Blocks dep on the existing id, deduped against the + // DiscoveredFrom edge (same target keeps the Blocks edge only). + assert_eq!(node.deps.len(), 1); + assert_eq!(node.deps[0].on, a_id); + assert_eq!(node.deps[0].kind, DepKind::Blocks); + + // Distinct origin gets its own DiscoveredFrom edge. + let b_id = node_id_for_slug("b"); + let json = serde_json::json!({ + "nodes": [{ + "id": "docs2", + "title": "Docs 2", + "spec": "s", + "deps": [a_id.clone()], + "discovered_from": [b_id.clone()], + }] + }) + .to_string(); + let appendix = validate_replan(&existing, &json).unwrap(); + let node = &appendix[0]; + assert_eq!(node.deps.len(), 2); + assert!( + node.deps + .iter() + .any(|d| d.on == b_id && d.kind == DepKind::DiscoveredFrom) + ); + } + + #[test] + fn replan_rejects_blocks_deps_on_dead_nodes_but_allows_dead_origins() { + let mut existing = existing_graph(); + let a_id = node_id_for_slug("a"); + existing.iter_mut().find(|n| n.id == a_id).unwrap().status = NodeStatus::Failed; + let dead_dep = serde_json::json!({ + "nodes": [{"id": "x", "title": "T", "spec": "s", "deps": [a_id.clone()]}] + }) + .to_string(); + assert!(matches!( + validate_replan(&existing, &dead_dep).unwrap_err(), + GraphPlanError::DeadDep { .. } + )); + // A dead ORIGIN is the normal salvage case — allowed, and the + // audit-only DiscoveredFrom edge never gates scheduling. + let dead_origin = serde_json::json!({ + "nodes": [{"id": "x", "title": "T", "spec": "s", "deps": [], + "discovered_from": [a_id]}] + }) + .to_string(); + assert!(validate_replan(&existing, &dead_origin).is_ok()); + } + + #[test] + fn replan_rejects_edges_onto_the_terminal_node() { + let existing = existing_graph(); + for json in [ + serde_json::json!({"nodes": [{"id": "x", "title": "T", "spec": "s", + "deps": [FINAL_NODE_ID]}]}), + serde_json::json!({"nodes": [{"id": "x", "title": "T", "spec": "s", + "deps": [], "discovered_from": [FINAL_NODE_ID]}]}), + ] { + assert!( + matches!( + validate_replan(&existing, &json.to_string()).unwrap_err(), + GraphPlanError::UnknownDep { .. } + ), + "an edge onto gn-final would cycle after the final-gating extension" + ); + } + } + + #[test] + fn replan_rejects_collisions_unknown_origins_and_cycles() { + let existing = existing_graph(); + // Re-using an existing slug collides on the canonical id. + let dup = serde_json::json!({ + "nodes": [{"id": "a", "title": "T", "spec": "s", "deps": []}] + }) + .to_string(); + assert_eq!( + validate_replan(&existing, &dup).unwrap_err(), + GraphPlanError::ExistingCollision("a".into()) + ); + // Unknown discovered_from origin. + let bad_origin = serde_json::json!({ + "nodes": [{"id": "x", "title": "T", "spec": "s", "deps": [], + "discovered_from": ["gn-ghost"]}] + }) + .to_string(); + assert!(matches!( + validate_replan(&existing, &bad_origin).unwrap_err(), + GraphPlanError::UnknownOrigin { .. } + )); + // New-node cycle. + let cyc = serde_json::json!({ + "nodes": [ + {"id": "x", "title": "T", "spec": "s", "deps": ["y"]}, + {"id": "y", "title": "T", "spec": "s", "deps": ["x"]}, + ] + }) + .to_string(); + assert!(matches!( + validate_replan(&existing, &cyc).unwrap_err(), + GraphPlanError::Cycle(_) + )); + } + /// A repeated dep entry is harmless planner redundancy: it must be /// deduped, NOT misreported as a cycle by the indegree seed. #[test] diff --git a/crates/codegen/kigi-shell/src/session/graph_planner.rs b/crates/codegen/kigi-shell/src/session/graph_planner.rs index 0da1cb3..879106f 100644 --- a/crates/codegen/kigi-shell/src/session/graph_planner.rs +++ b/crates/codegen/kigi-shell/src/session/graph_planner.rs @@ -26,7 +26,9 @@ use super::graph_plan::{self, MAX_GRAPH_JSON_BYTES}; use super::graph_tracker::GraphNode; const GRAPH_PLANNER_PROMPT_TEMPLATE: &str = include_str!("templates/graph_planner_prompt.md"); +const GRAPH_REPLANNER_PROMPT_TEMPLATE: &str = include_str!("templates/graph_replanner_prompt.md"); pub(crate) const GRAPH_PLANNER_SUBAGENT_DESCRIPTION: &str = "graph plan writer"; +pub(crate) const GRAPH_REPLANNER_SUBAGENT_DESCRIPTION: &str = "graph replanner"; #[derive(Debug)] pub(crate) enum GraphPlannerOutcome { @@ -63,6 +65,16 @@ pub(crate) async fn run_graph_planner( }; } + // A stale artifact from a prior pass would satisfy the + // missing-file guard below and be trusted as this pass's output. + // Delete first; only NotFound is benign. + if let Err(err) = tokio::fs::remove_file(inputs.graph_file).await + && err.kind() != std::io::ErrorKind::NotFound + { + return GraphPlannerOutcome::FailClosed { + reason: format!("failed to clear stale graph artifact: {err}"), + }; + } let graph_file_str = inputs.graph_file.to_string_lossy(); let with_graph_file = GRAPH_PLANNER_PROMPT_TEMPLATE.replace("{GRAPH_FILE}", &graph_file_str); let render = |tool_names: &RoleToolNames| -> String { @@ -141,6 +153,122 @@ pub(crate) async fn run_graph_planner( } } +pub(crate) struct GraphReplannerInputs<'a> { + pub objective: &'a str, + /// Compact JSON of the existing nodes (id/title/status/deps). + pub current_graph: &'a str, + /// The queued discoveries, one per line with their origin node ids. + pub discoveries: &'a str, + /// On retry, the previous artifact's validation error. + pub feedback: &'a str, + pub graph_file: &'a Path, + pub tool_names: &'a RoleToolNames, + pub inherit_tool_names: &'a RoleToolNames, +} + +/// Run one REPLAN attempt: render, spawn, read the artifact, validate +/// against the existing graph (append-only). An empty `{"nodes": []}` +/// appendix is the sanctioned "everything already covered" escape hatch +/// and returns `Planned(vec![])`. +pub(crate) async fn run_graph_replanner( + spawner: Arc, + existing: &[GraphNode], + inputs: GraphReplannerInputs<'_>, +) -> GraphPlannerOutcome { + if let Some(parent) = inputs.graph_file.parent() + && let Err(err) = tokio::fs::create_dir_all(parent).await + { + return GraphPlannerOutcome::FailClosed { + reason: format!("failed to create graph dir {}: {err}", parent.display()), + }; + } + // A stale artifact from a prior pass would satisfy the + // missing-file guard below and be trusted as this pass's output. + // Delete first; only NotFound is benign. + if let Err(err) = tokio::fs::remove_file(inputs.graph_file).await + && err.kind() != std::io::ErrorKind::NotFound + { + return GraphPlannerOutcome::FailClosed { + reason: format!("failed to clear stale graph artifact: {err}"), + }; + } + let graph_file_str = inputs.graph_file.to_string_lossy(); + let with_graph_file = GRAPH_REPLANNER_PROMPT_TEMPLATE.replace("{GRAPH_FILE}", &graph_file_str); + let render = |tool_names: &RoleToolNames| -> String { + let rendered = tool_names.apply(&with_graph_file); + format!( + "{rendered}\n\nOBJECTIVE:\n{}\n\nCURRENT GRAPH:\n{}\n\nDISCOVERIES:\n{}\n\nCONTEXT:\n{}\n", + inputs.objective, inputs.current_graph, inputs.discoveries, inputs.feedback + ) + }; + let prompt = RoleRenderedPrompt { + primary: render(inputs.tool_names), + fallback: render(inputs.inherit_tool_names), + }; + let spawn_id = uuid::Uuid::now_v7().to_string(); + let response = match spawner.spawn_planner(&spawn_id, prompt).await { + Ok(text) => text, + Err(SpawnError::Transport(detail)) => { + return GraphPlannerOutcome::FailClosed { + reason: format!("graph replanner transport error: {detail}"), + }; + } + Err(SpawnError::Runtime { message, cancelled }) => { + return GraphPlannerOutcome::FailClosed { + reason: if cancelled { + format!("graph replanner aborted: {message}") + } else { + format!("graph replanner runtime error: {message}") + }, + }; + } + }; + match tokio::fs::metadata(inputs.graph_file).await { + Ok(meta) if meta.is_file() && meta.len() > 0 => { + if meta.len() > MAX_GRAPH_JSON_BYTES { + return GraphPlannerOutcome::Invalid { + reason: format!( + "replan JSON is {} bytes; the cap is {MAX_GRAPH_JSON_BYTES}", + meta.len() + ), + }; + } + } + _ => { + tracing::info!( + graph_file = %graph_file_str, + terminal_token_ok = parse_terminal_response(&response), + "graph replanner: artifact missing or empty; failing closed", + ); + return GraphPlannerOutcome::FailClosed { + reason: "graph replanner produced no artifact".to_owned(), + }; + } + } + let json = match tokio::fs::read_to_string(inputs.graph_file).await { + Ok(json) => json, + Err(err) => { + return GraphPlannerOutcome::FailClosed { + reason: format!("failed to read replan artifact: {err}"), + }; + } + }; + // Escape hatch: an explicitly empty appendix means "already covered". + if let Ok(v) = serde_json::from_str::(&json) + && v.get("nodes") + .and_then(|n| n.as_array()) + .is_some_and(Vec::is_empty) + { + return GraphPlannerOutcome::Planned(Vec::new()); + } + match graph_plan::validate_replan(existing, &json) { + Ok(nodes) => GraphPlannerOutcome::Planned(nodes), + Err(err) => GraphPlannerOutcome::Invalid { + reason: err.to_string(), + }, + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/codegen/kigi-shell/src/session/graph_tracker.rs b/crates/codegen/kigi-shell/src/session/graph_tracker.rs index 632c953..7070597 100644 --- a/crates/codegen/kigi-shell/src/session/graph_tracker.rs +++ b/crates/codegen/kigi-shell/src/session/graph_tracker.rs @@ -91,9 +91,13 @@ impl NodeStatus { } /// Dependency edge kind. `Blocks` is the planner-authored ordering -/// dependency; `DiscoveredFrom` marks a node appended by a replan (G3) -/// pointing back at the node whose execution surfaced it. Both gate -/// scheduling identically; the kind is audit/render metadata. +/// dependency and the ONLY kind that gates scheduling. +/// `DiscoveredFrom` marks a node appended by a replan (G3) pointing +/// back at the node whose execution surfaced it — pure audit/render +/// metadata: its origin is always terminal at replan time, so gating on +/// it would either be a no-op (origin Achieved) or a permanent wedge +/// (origin Failed — and a failed node's discoveries are still real +/// work). #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "snake_case")] pub enum DepKind { @@ -185,6 +189,16 @@ impl GraphHistoryEntry { } } +/// One piece of out-of-scope work surfaced during node execution +/// (`DISCOVERED:` marker from a worker, verifier, or the serial node's +/// final text). Queued until the next replan boundary. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct Discovery { + /// Node whose execution surfaced this work. + pub from_node: String, + pub description: String, +} + // GraphOrchestration (full persisted state) #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] @@ -220,6 +234,13 @@ pub struct GraphOrchestration { /// on resume/complete (mirrors `GoalOrchestration::pause_message`). #[serde(default, skip_serializing_if = "Option::is_none")] pub pause_message: Option, + /// Discoveries queued for the next replan boundary. Drained by a + /// replan pass; past the replan cap they drain to history only. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub pending_discoveries: Vec, + /// Replan passes consumed (bounded by `KIGI_GRAPH_REPLAN_CAP`). + #[serde(default)] + pub replan_runs: u32, } fn default_plan_version() -> u32 { @@ -366,6 +387,8 @@ impl GraphTracker { tokens_spent_nodes: 0, history: Vec::new(), pause_message: None, + pending_discoveries: Vec::new(), + replan_runs: 0, }; state .history @@ -654,6 +677,80 @@ impl GraphTracker { } } + /// Queue discoveries for the next replan boundary (records each in + /// history so the audit trail survives even past the replan cap). + pub fn queue_discoveries(&mut self, discoveries: Vec) { + let Some(state) = self.state.as_mut() else { + return; + }; + for d in discoveries { + push_history( + state, + GraphHistoryEntry::now( + GraphEvent::Unknown, + Some(d.from_node.clone()), + Some(format!("discovered: {}", d.description)), + ), + ); + state.pending_discoveries.push(d); + } + } + + /// Install a validated replan appendix: bump `plan_version`, append + /// the new nodes, gate the terminal node on them too (demoting it + /// back to `Waiting` if it was already `Ready`), consume the pending + /// discoveries, and recompute readiness. SGH discipline: versions + /// are immutable — existing nodes are never touched here (the + /// validator guarantees the appendix references them read-only). + pub fn append_replan_nodes(&mut self, new_nodes: Vec) { + let Some(state) = self.state.as_mut() else { + return; + }; + let new_ids: Vec = new_nodes.iter().map(|n| n.id.clone()).collect(); + state.nodes.extend(new_nodes); + if let Some(final_node) = state.nodes.iter_mut().find(|n| n.id == FINAL_NODE_ID) { + for id in &new_ids { + if id != FINAL_NODE_ID && !final_node.deps.iter().any(|d| &d.on == id) { + final_node.deps.push(NodeDep { + on: id.clone(), + kind: DepKind::Blocks, + }); + } + } + // A Ready terminal node whose gate just grew must wait again. + if final_node.status == NodeStatus::Ready { + final_node.status = NodeStatus::Waiting; + } + } + state.plan_version += 1; + state.replan_runs += 1; + state.pending_discoveries.clear(); + push_history( + state, + GraphHistoryEntry::now( + GraphEvent::PlanningCompleted, + None, + Some(format!( + "replan v{}: +{} node(s)", + state.plan_version, + new_ids.len() + )), + ), + ); + self.recompute_ready(); + } + + /// Drain pending discoveries WITHOUT replanning (cap exhausted): + /// they stay in history (queued there at capture time) only. + pub fn drain_discoveries_to_history(&mut self) -> usize { + let Some(state) = self.state.as_mut() else { + return 0; + }; + let n = state.pending_discoveries.len(); + state.pending_discoveries.clear(); + n + } + /// 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: @@ -704,7 +801,11 @@ impl GraphTracker { .collect(); for node in &mut state.nodes { if node.status == NodeStatus::Waiting - && node.deps.iter().all(|d| achieved.contains(&d.on)) + && node + .deps + .iter() + .filter(|d| d.kind == DepKind::Blocks) + .all(|d| achieved.contains(&d.on)) { node.status = NodeStatus::Ready; } @@ -734,7 +835,12 @@ fn block_dependents(state: &mut GraphOrchestration, failed_id: &str) { if node.status.is_terminal() || blocked.contains(&node.id) { continue; } - if node.deps.iter().any(|d| blocked.contains(&d.on)) { + if node + .deps + .iter() + .filter(|d| d.kind == DepKind::Blocks) + .any(|d| blocked.contains(&d.on)) + { node.status = NodeStatus::Blocked; node.failure = Some(format!("blocked: dependency chain failed at {failed_id}")); blocked.insert(node.id.clone()); @@ -921,6 +1027,59 @@ mod tests { assert_eq!(NodeStatus::from_wire_str("achieved"), NodeStatus::Achieved); } + #[test] + fn discovered_from_edges_never_gate_scheduling() { + let mut t = tracker_with(vec![node("a", &[]), node("b", &[])]); + t.mark_node_running("a", "g".into()); + t.mark_node_failed("a", "dead".into()); + // Appendix node whose ONLY edge is DiscoveredFrom on the failed + // origin: must become Ready (audit edge, not a gate) and must + // not be swept by block_dependents. + t.append_replan_nodes(vec![GraphNode { + id: "gn-doc".into(), + title: "Docs".into(), + spec: "s".into(), + deps: vec![NodeDep { + on: "a".into(), + kind: DepKind::DiscoveredFrom, + }], + status: NodeStatus::Waiting, + goal_id: None, + rounds: 0, + tokens_used: 0, + failure: None, + }]); + assert_eq!(t.node("gn-doc").unwrap().status, NodeStatus::Ready); + } + + #[test] + fn replan_appendix_regates_the_final_node_and_bumps_version() { + let mut t = tracker_with(vec![node("a", &[]), node(FINAL_NODE_ID, &["a"])]); + t.mark_node_running("a", "g1".into()); + t.mark_node_achieved("a", 1, 10); + // Final node unlocked… + assert_eq!(t.node(FINAL_NODE_ID).unwrap().status, NodeStatus::Ready); + assert_eq!(t.snapshot().unwrap().plan_version, 1); + // …then a replan appendix lands: final must wait again. + t.queue_discoveries(vec![Discovery { + from_node: "a".into(), + description: "docs".into(), + }]); + t.append_replan_nodes(vec![node("docs", &[])]); + let s = t.snapshot().unwrap(); + assert_eq!(s.plan_version, 2); + assert_eq!(s.replan_runs, 1); + assert!(s.pending_discoveries.is_empty()); + let final_node = t.node(FINAL_NODE_ID).unwrap(); + assert_eq!( + final_node.status, + NodeStatus::Waiting, + "a Ready terminal node whose gate grew must wait again" + ); + assert!(final_node.deps.iter().any(|d| d.on == "docs")); + assert_eq!(t.next_ready_node().unwrap().id, "docs"); + } + #[test] fn history_is_capped_dropping_the_oldest() { let mut t = tracker_with(vec![node("a", &[])]); diff --git a/crates/codegen/kigi-shell/src/session/storage/jsonl/tests.rs b/crates/codegen/kigi-shell/src/session/storage/jsonl/tests.rs index cead051..032c97c 100644 --- a/crates/codegen/kigi-shell/src/session/storage/jsonl/tests.rs +++ b/crates/codegen/kigi-shell/src/session/storage/jsonl/tests.rs @@ -2778,6 +2778,8 @@ async fn graph_mode_state_round_trips_and_tombstones() { tokens_spent_nodes: 42, history: vec![], pause_message: None, + pending_discoveries: vec![], + replan_runs: 0, }; adapter.write_graph_mode_state(&info, Some(&state)).await.unwrap(); let loaded = adapter.load_session_without_updates(&info).await.unwrap(); 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 index 06adf34..1aef814 100644 --- 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 @@ -22,3 +22,11 @@ GAPS: Be strict but fair: judge ONLY this node's contract, not sibling nodes' scope and not style preferences. + +If you notice NECESSARY work that lies OUTSIDE this node's contract, it +is NOT a gap — do not fail the node for it. Report it instead, each item +on its own line before your verdict: + +``` +DISCOVERED: +``` 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 index 9bcbe2f..d783943 100644 --- 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 @@ -13,6 +13,16 @@ Rules: - 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. +- If you find NECESSARY work outside this node's contract (a missing + prerequisite, a broken sibling area, follow-up the objective implies), + do NOT do it. Report each item on its own line, anywhere in your final + message: + + ``` + DISCOVERED: + ``` + + The harness turns these into new graph nodes. Your final message MUST end with exactly one of: diff --git a/crates/codegen/kigi-shell/src/session/templates/graph_replanner_prompt.md b/crates/codegen/kigi-shell/src/session/templates/graph_replanner_prompt.md new file mode 100644 index 0000000..b41413d --- /dev/null +++ b/crates/codegen/kigi-shell/src/session/templates/graph_replanner_prompt.md @@ -0,0 +1,55 @@ +You are the Graph Replanner for the Kigi harness. A running dependency +graph surfaced NEW out-of-scope work (DISCOVERIES below). Your job is to +extend the graph with the FEWEST additional nodes that cover exactly that +work — nothing else. + +## Inputs (below this prompt) + +- OBJECTIVE: the overall graph objective, verbatim. +- CURRENT GRAPH: the existing nodes as JSON (id, title, status, deps). + These are IMMUTABLE — you cannot edit, remove, or reorder them. +- DISCOVERIES: the queued out-of-scope items, each with the node id that + surfaced it. + +## Rules + +- Append-only: output ONLY new nodes. Merge related discoveries into one + node where a single coherent unit of work covers them. +- A discovery already covered by an existing non-terminal node's spec + needs NO new node — cover only genuine gaps. If nothing needs a new + node, still write a file with an empty check? NO — see the escape + hatch below. +- `deps` may reference EXISTING node ids (the `gn-…` strings from + CURRENT GRAPH) and/or other new nodes. Only true ordering constraints. +- Each new node MUST set `discovered_from` to the existing node id(s) + whose discoveries it covers. +- Specs are outcome contracts in the OBJECTIVE's vocabulary, sized for + one focused autonomous run. + +## Output contract — STRICT + +Use your `{WRITE_TOOL}` tool to write JSON to `{GRAPH_FILE}`: + +``` +{ + "nodes": [ + { + "id": "short-kebab-slug", + "title": "One-line human title", + "spec": "Outcome contract for this node alone.", + "deps": ["gn-existing-or-new-slug"], + "discovered_from": ["gn-originating-node"] + } + ] +} +``` + +Escape hatch: when every discovery is already covered by existing nodes, +write `{"nodes": []}` — the harness treats an empty appendix as "nothing +to add" and drains the discoveries. + +Your terminal response must be exactly: + +``` +Done +```