diff --git a/crates/codegen/kigi-shell/src/session/acp_session.rs b/crates/codegen/kigi-shell/src/session/acp_session.rs index ce8d369..2c38d5f 100644 --- a/crates/codegen/kigi-shell/src/session/acp_session.rs +++ b/crates/codegen/kigi-shell/src/session/acp_session.rs @@ -612,6 +612,14 @@ pub(crate) struct SessionActor { /// Max replan passes per graph (0 = replanning off). Cached at /// actor construction. pub(crate) graph_replan_cap: u32, + /// `.kigi` dir at the git root, when the session cwd is in a git + /// repo — home of the project-level shared graph projection. + pub(crate) graph_project_dir: Option, + /// Held single-writer lock on the project graph. `Some` while this + /// session owns the graph (created or resumed it); dropped on + /// `/graph clear`. + pub(crate) graph_project_lock: + std::cell::RefCell>, /// `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 f3a9d74..8bb80fb 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 @@ -167,6 +167,21 @@ impl SessionActor { Some(state) => build_graph_updated(state), None => build_graph_cleared(), }; + // Project-level shared file (G4): refreshed at every checkpoint + // while this session holds the writer lock. Failures are loud in + // logs but never block progress — the session state below is the + // durable source of truth. + if let Some(dir) = &self.graph_project_dir + && self.graph_project_lock.borrow().is_some() + { + let result = match &snapshot { + Some(state) => crate::session::graph_project::project(dir, state), + None => crate::session::graph_project::remove(dir), + }; + if let Err(err) = result { + tracing::warn!(%err, dir = %dir.display(), "graph: project file sync failed"); + } + } let _ = self .notifications .persistence_tx @@ -174,6 +189,78 @@ impl SessionActor { self.goal_notify_sender().send_update(update); } + /// Best-effort peek at the projected graph's identity, for the + /// lock-then-mutate sites: holding the flock proves nobody writes + /// NOW, not that the file's content belongs to THIS session's graph. + /// `None` = no file / unreadable (warned). + pub(super) fn projected_graph_id(&self) -> Option { + let dir = self.graph_project_dir.as_deref()?; + match crate::session::graph_project::load(dir) { + Ok(state) => state.map(|s| s.graph_id), + Err(err) => { + tracing::warn!(%err, "graph: projected file unreadable during identity check"); + None + } + } + } + + /// Claim project writership for a SESSION-owned graph about to + /// resume: acquire the lock and verify the projected file (if any) + /// belongs to this graph — a foreign projection must never be + /// clobbered by our next checkpoint. `Some(msg)` = refuse with msg + /// (lock released). + pub(super) fn claim_project_graph_for_resume(&self) -> Option { + match self.acquire_project_graph_writer() { + Ok(true) => {} + Ok(false) => { + return Some( + "Another kigi instance holds this project's graph (single-writer \ + lock); resume it there, or /graph clear here to drop this \ + session's copy." + .to_owned(), + ); + } + Err(err) => { + return Some(format!("Failed to acquire the project graph lock: {err}")); + } + } + let session_graph_id = self + .graph_tracker + .lock() + .snapshot() + .map(|s| s.graph_id.clone()); + if let (Some(projected), Some(ours)) = (self.projected_graph_id(), session_graph_id) + && projected != ours + { + self.graph_project_lock.borrow_mut().take(); + return Some(format!( + "The project graph file belongs to a different graph ({projected}); \ + this session's graph is {ours}. Use /graph resume in a fresh session \ + to revive the project graph, or /graph clear here first." + )); + } + None + } + + /// Become the project graph's single writer. `Ok(false)` = another + /// kigi instance owns it (caller degrades read-only); `Ok(true)` = + /// acquired or no project dir (feature off outside git). + pub(super) fn acquire_project_graph_writer(&self) -> std::io::Result { + let Some(dir) = &self.graph_project_dir else { + return Ok(true); + }; + if self.graph_project_lock.borrow().is_some() { + return Ok(true); + } + match crate::session::graph_project::try_acquire_writer(dir)? { + crate::session::graph_project::LockOutcome::Acquired(lock) => { + *self.graph_project_lock.borrow_mut() = Some(lock); + Ok(true) + } + crate::session::graph_project::LockOutcome::Busy => Ok(false), + } + } + /// True when the graph occupies the goal engine (any non-terminal /// state). `/goal` commands are refused in this window. pub(super) fn graph_owns_goal_engine(&self) -> bool { @@ -211,6 +298,45 @@ impl SessionActor { } } + match self.acquire_project_graph_writer() { + Ok(true) => {} + Ok(false) => { + return GraphSetupOutcome::Message( + "Another kigi instance holds this project's graph (single-writer \ + lock). Use /graph status here for a read-only view, or run the \ + graph from that instance." + .to_owned(), + ); + } + Err(err) => { + return GraphSetupOutcome::Message(format!( + "Failed to acquire the project graph lock: {err}" + )); + } + } + // The project file may hold a dead session's REVIVABLE graph — + // the same never-silently-replace rule as the session guard + // above applies to it (a Complete one may be overwritten). + if let Some(dir) = self.graph_project_dir.as_deref() { + match crate::session::graph_project::load(dir) { + Ok(Some(existing)) if existing.status != GoalStatus::Complete => { + self.graph_project_lock.borrow_mut().take(); + return GraphSetupOutcome::Message(format!( + "A project graph exists in .kigi/graph.jsonl ({}: {}). Use \ + /graph resume to revive it or /graph clear to discard it.", + existing.graph_id, existing.objective, + )); + } + Ok(_) => {} + Err(err) => { + self.graph_project_lock.borrow_mut().take(); + return GraphSetupOutcome::Message(format!( + "Project graph file is unreadable: {err}\n\ + Fix or delete .kigi/graph.jsonl, then retry." + )); + } + } + } let graph_id = uuid::Uuid::new_v4().to_string(); tracing::info!(%graph_id, "graph: created, planning started"); self.graph_tracker.lock().create_graph( @@ -757,9 +883,74 @@ impl SessionActor { use super::goal_support::GoalResumeOutcome; let status = self.graph_tracker.lock().status(); match status { - None => GraphSetupOutcome::Message( - "No graph is set. Use /graph to start one.".to_owned(), - ), + None => { + // Cross-session revive: the graph follows the REPO. A + // fresh session in the same project can pick it up from + // .kigi/graph.jsonl (single-writer lock required). + let Some(dir) = self.graph_project_dir.clone() else { + return GraphSetupOutcome::Message( + "No graph is set. Use /graph to start one.".to_owned(), + ); + }; + // Lock BEFORE load: reading first races the owner's final + // checkpoint/clear and could resurrect a just-cleared + // graph from the pre-release copy. + match self.acquire_project_graph_writer() { + Ok(true) => {} + Ok(false) => { + return GraphSetupOutcome::Message( + "Another kigi instance holds this project's graph \ + (single-writer lock); /graph status shows a read-only view." + .to_owned(), + ); + } + Err(err) => { + return GraphSetupOutcome::Message(format!( + "Failed to acquire the project graph lock: {err}" + )); + } + } + let release_and = |msg: String| { + self.graph_project_lock.borrow_mut().take(); + GraphSetupOutcome::Message(msg) + }; + let loaded = match crate::session::graph_project::load(&dir) { + Ok(Some(state)) => state, + Ok(None) => { + return release_and( + "No graph is set. Use /graph to start one.".to_owned(), + ); + } + Err(err) => { + return release_and(format!( + "Project graph file is unreadable: {err}\n\ + Fix or delete .kigi/graph.jsonl, then retry." + )); + } + }; + tracing::info!( + graph_id = %loaded.graph_id, + "graph: revived from project file" + ); + // from_snapshot sanitization (Active→UserPaused, + // Running→Ready) applies to the revived state too. + { + let session_dir = + crate::session::persistence::session_dir(&crate::session::info::Info { + id: self.session_info.id.clone(), + cwd: self.session_info.cwd.clone(), + }); + *self.graph_tracker.lock() = + crate::session::graph_tracker::GraphTracker::from_snapshot( + session_dir, + loaded, + ); + } + self.persist_graph_state(); + // Recurse exactly once: the tracker now holds a paused + // graph, so the paused-family arm below handles it. + Box::pin(self.resume_graph(extra_budget)).await + } Some(GoalStatus::Active) => { GraphSetupOutcome::Message("Graph is already running.".to_owned()) } @@ -774,6 +965,14 @@ impl SessionActor { .to_owned(), ); }; + if extra <= 0 { + return GraphSetupOutcome::Message( + "Budget top-up must be a positive token count.".to_owned(), + ); + } + if let Some(msg) = self.claim_project_graph_for_resume() { + return GraphSetupOutcome::Message(msg); + } if !self.graph_tracker.lock().resume_budget_limited(extra) { return GraphSetupOutcome::Message( "Budget top-up must be a positive token count.".to_owned(), @@ -803,6 +1002,13 @@ impl SessionActor { .to_owned(), ); } + // A session-snapshot-restored graph resumes here WITHOUT + // ever having passed the create/revive lock sites; claim + // the project writer (with identity check) now or the + // projection silently diverges forever. + if let Some(msg) = self.claim_project_graph_for_resume() { + return GraphSetupOutcome::Message(msg); + } { let mut tracker = self.graph_tracker.lock(); tracker.resume(); @@ -924,8 +1130,32 @@ impl SessionActor { let current_tokens = self.chat_state_handle.get_total_tokens().await as i64; let node_goal_tokens = self.goal_tokens_used(current_tokens); let tracker = self.graph_tracker.lock(); - let Some(s) = tracker.snapshot() else { - return "No graph is set. Use /graph to start one.".to_owned(); + let project_fallback; + let s = match tracker.snapshot() { + Some(s) => s, + None => { + // Read-only project view (e.g. a second kigi instance + // that does not hold the writer lock). + match self + .graph_project_dir + .as_deref() + .map(crate::session::graph_project::load) + { + Some(Ok(Some(state))) => { + project_fallback = state; + &project_fallback + } + Some(Err(err)) => { + return format!( + "Project graph file is unreadable: {err}\n\ + Fix or delete .kigi/graph.jsonl." + ); + } + _ => { + return "No graph is set. Use /graph to start one.".to_owned(); + } + } + } }; let achieved = s .nodes diff --git a/crates/codegen/kigi-shell/src/session/acp_session_impl/slash_exec.rs b/crates/codegen/kigi-shell/src/session/acp_session_impl/slash_exec.rs index 07f040e..2acf403 100644 --- a/crates/codegen/kigi-shell/src/session/acp_session_impl/slash_exec.rs +++ b/crates/codegen/kigi-shell/src/session/acp_session_impl/slash_exec.rs @@ -857,8 +857,57 @@ impl SessionActor { if self.graph_owns_goal_engine() { self.reset_goal_engine_state().await; } + // Projection teardown only when there is something of + // OURS to un-project: with no session graph, deleting + // .kigi/graph.jsonl would destroy another session's + // revivable graph while replying "No graph is set". + let session_graph_id = self + .graph_tracker + .lock() + .snapshot() + .map(|s| s.graph_id.clone()); self.graph_tracker.lock().clear(); + if had_graph { + // Take the writer lock if we don't hold it (e.g. a + // session-snapshot-restored graph cleared before any + // resume). Busy = another instance owns the project + // graph; local-only clear is then correct. + match self.acquire_project_graph_writer() { + Ok(true) => { + // Identity check: only remove a projection + // that belongs to the graph being cleared. + let foreign = match (self.projected_graph_id(), &session_graph_id) { + (Some(projected), Some(ours)) => projected != *ours, + _ => false, + }; + if foreign { + tracing::info!( + "graph clear: projection belongs to a different \ + graph; leaving .kigi/graph.jsonl in place" + ); + self.graph_project_lock.borrow_mut().take(); + } + } + Ok(false) => { + tracing::info!( + "graph clear: another instance holds the project \ + graph; local session state cleared only" + ); + } + Err(err) => { + tracing::warn!( + %err, + "graph clear: project lock acquisition failed; \ + .kigi/graph.jsonl may survive as stale" + ); + } + } + } + // persist runs BEFORE the lock drops so the projection + // removal (when we hold writer rights on OUR graph) + // executes; without the lock it is a session-only clear. self.persist_graph_state(); + self.graph_project_lock.borrow_mut().take(); self.send_slash_command_output(if had_graph { "Graph cleared." } else { 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 d7ecd85..f372a6b 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 @@ -431,6 +431,8 @@ pub(crate) async fn spawn_session_actor( }; Arc::new(parking_lot::Mutex::new(tracker)) }; + let graph_project_dir = + crate::session::graph_project::project_graph_dir(tool_context.cwd.as_path()); let graph_tracker = { let session_dir = crate::session::persistence::session_dir(&session_info); let tracker = if let Some(snapshot) = persisted_graph_mode { @@ -1102,6 +1104,8 @@ pub(crate) async fn spawn_session_actor( 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(), + graph_project_dir, + graph_project_lock: std::cell::RefCell::new(None), 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), @@ -1248,8 +1252,13 @@ pub(crate) async fn spawn_session_actor( // A restored graph was demoted (Active→UserPaused, Running→Ready) IN // MEMORY after the updates-log replay, whose last GraphUpdated still // shows the pre-shutdown Active state. Re-emit truth once so a - // reattached pager never renders a stale self-driving chip. + // reattached pager never renders a stale self-driving chip — and + // best-effort reclaim project writership so the shared file gets the + // demoted truth too (Busy = another instance owns it; skip quietly). if session.graph_tracker.lock().snapshot().is_some() { + if let Some(msg) = session.claim_project_graph_for_resume() { + tracing::info!(%msg, "graph restore: project writership not reclaimed"); + } session.persist_graph_state(); } if let Some(ref display_cwd) = prompt_display_cwd { 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 1180ea6..77081b3 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 @@ -219,6 +219,8 @@ async fn persist_ack_waits_for_disk_flush_before_success() { graph_concurrency: 1, graph_node_rounds: 3, graph_replan_cap: 3, + graph_project_dir: None, + graph_project_lock: std::cell::RefCell::new(None), 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), @@ -662,6 +664,8 @@ async fn first_turn_memory_injection_disabled_does_not_persist_to_chat_history() graph_concurrency: 1, graph_node_rounds: 3, graph_replan_cap: 3, + graph_project_dir: None, + graph_project_lock: std::cell::RefCell::new(None), 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), @@ -914,6 +918,8 @@ async fn cancel_running_task_teardown_clears_running_and_pending_work() { graph_concurrency: 1, graph_node_rounds: 3, graph_replan_cap: 3, + graph_project_dir: None, + graph_project_lock: std::cell::RefCell::new(None), 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), @@ -1899,6 +1905,8 @@ async fn cancel_propagates_to_sampler_handle_so_no_further_emission() { graph_concurrency: 1, graph_node_rounds: 3, graph_replan_cap: 3, + graph_project_dir: None, + graph_project_lock: std::cell::RefCell::new(None), 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 a1087ba..d8e343d 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 @@ -1800,3 +1800,95 @@ async fn replan_cap_zero_drains_discoveries_to_history_and_converges() { .await; unsafe { std::env::remove_var(ENV_FLAG) }; } + +// ── G4: project-level shared graph ───────────────────────────────── + +#[tokio::test(flavor = "current_thread")] +#[serial] +async fn project_graph_revives_in_a_fresh_session_and_write_lock_is_exclusive() { + unsafe { std::env::set_var(ENV_FLAG, "0") }; + let local = tokio::task::LocalSet::new(); + local + .run_until(async { + let dag = chain_graph_json(); + let (mut actor_a, tmp, _prx) = make_graph_actor_detached().await; + let (coord_tx, _count) = spawn_graph_planner_coordinator(vec![dag]); + actor_a.tool_context.subagent_event_tx = Some(coord_tx); + // The fixture repo is the project root: point the projection + // there explicitly (find_git_root works too, but this keeps + // the test hermetic against outer repos). + let project_dir = tmp.path().join(".kigi"); + actor_a.graph_project_dir = Some(project_dir.clone()); + + let _ = actor_a.setup_graph("ship the widget", None).await; + // Node a achieved, node b running; projection follows. + drive_node_goal_to_complete(&actor_a).await; + let _ = actor_a.run_graph_round_end().await; + let projected = crate::session::graph_project::load(&project_dir) + .unwrap() + .expect("checkpoint must project to .kigi/graph.jsonl"); + assert_eq!(projected.nodes.len(), 4); + assert_eq!(projected.nodes[0].status, NodeStatus::Achieved); + assert_eq!(projected.nodes[1].status, NodeStatus::Running); + + // While A holds the writer lock, a second instance is + // read-only: status renders from the file, resume refuses. + let (coord_b, _cb) = spawn_graph_planner_coordinator(vec![]); + let (mut actor_b, _tmp_b, _prx_b) = make_graph_actor_detached().await; + actor_b.tool_context.subagent_event_tx = Some(coord_b); + actor_b.graph_project_dir = Some(project_dir.clone()); + let status = actor_b.graph_status_message().await; + assert!( + status.contains("ship the widget"), + "read-only view: {status}" + ); + match actor_b.resume_graph(None).await { + graph::GraphSetupOutcome::Message(msg) => { + assert!(msg.contains("single-writer"), "{msg}"); + } + graph::GraphSetupOutcome::Inference { .. } => { + panic!("second instance must not steal the writer lock") + } + } + + // A dies (lock released); a FRESH session revives the graph + // from the project file: b demotes to Ready and relaunches. + actor_a.graph_project_lock.borrow_mut().take(); + drop(actor_a); + let (coord_c, _cc) = spawn_graph_planner_coordinator(vec![]); + let (mut actor_c, _tmp_c, _prx_c) = make_graph_actor_detached().await; + actor_c.tool_context.subagent_event_tx = Some(coord_c); + actor_c.graph_project_dir = Some(project_dir.clone()); + match actor_c.resume_graph(None).await { + graph::GraphSetupOutcome::Inference { reminder, .. } => { + assert!( + reminder.contains("Graph node 2/4"), + "revive must relaunch node b: {reminder}" + ); + } + graph::GraphSetupOutcome::Message(msg) => { + panic!("fresh session must revive from the project file: {msg}") + } + } + assert_eq!( + actor_c.graph_tracker.lock().status(), + Some(GoalStatus::Active) + ); + + // /graph clear removes the projection and releases the lock. + drive_node_goal_to_complete(&actor_c).await; + let actor_c = StdArc::new(actor_c); + let _ = actor_c + .execute_builtin_slash_command(BuiltinAction::GraphClear) + .await; + assert!( + crate::session::graph_project::load(&project_dir) + .unwrap() + .is_none(), + "clear must remove .kigi/graph.jsonl" + ); + assert!(actor_c.graph_project_lock.borrow().is_none()); + }) + .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 44ae1f9..da8146e 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 @@ -248,6 +248,8 @@ async fn test_e2e_idle_resume_refreshes_model_metadata() { graph_concurrency: 1, graph_node_rounds: 3, graph_replan_cap: 3, + graph_project_dir: None, + graph_project_lock: std::cell::RefCell::new(None), 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 aa9ebde..c7b2a4f 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 @@ -180,6 +180,8 @@ async fn create_test_actor( graph_concurrency: 1, graph_node_rounds: 3, graph_replan_cap: 3, + graph_project_dir: None, + graph_project_lock: std::cell::RefCell::new(None), 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), @@ -625,6 +627,8 @@ async fn create_test_actor_with_memory( graph_concurrency: 1, graph_node_rounds: 3, graph_replan_cap: 3, + graph_project_dir: None, + graph_project_lock: std::cell::RefCell::new(None), 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), @@ -1381,6 +1385,8 @@ async fn test_e2e_idle_resume_refreshes_model_metadata() { graph_concurrency: 1, graph_node_rounds: 3, graph_replan_cap: 3, + graph_project_dir: None, + graph_project_lock: std::cell::RefCell::new(None), 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 52cb185..0f82987 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 @@ -242,6 +242,8 @@ async fn create_test_actor_with_memory( graph_concurrency: 1, graph_node_rounds: 3, graph_replan_cap: 3, + graph_project_dir: None, + graph_project_lock: std::cell::RefCell::new(None), 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 f49cb02..b28101d 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 @@ -188,6 +188,8 @@ pub(super) async fn make_replay_send_update_fixture() -> ReplaySendUpdateFixture graph_concurrency: 1, graph_node_rounds: 3, graph_replan_cap: 3, + graph_project_dir: None, + graph_project_lock: std::cell::RefCell::new(None), 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 c66e38b..637a1a8 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 @@ -293,6 +293,8 @@ pub(crate) async fn create_test_actor_ex( graph_concurrency: 1, graph_node_rounds: 3, graph_replan_cap: 3, + graph_project_dir: None, + graph_project_lock: std::cell::RefCell::new(None), 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 afaa151..cecd878 100644 --- a/crates/codegen/kigi-shell/src/session/compaction.rs +++ b/crates/codegen/kigi-shell/src/session/compaction.rs @@ -2275,6 +2275,8 @@ mod inline_auto_compact_flow_tests { graph_concurrency: 1, graph_node_rounds: 3, graph_replan_cap: 3, + graph_project_dir: None, + graph_project_lock: std::cell::RefCell::new(None), 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_project.rs b/crates/codegen/kigi-shell/src/session/graph_project.rs new file mode 100644 index 0000000..9549a42 --- /dev/null +++ b/crates/codegen/kigi-shell/src/session/graph_project.rs @@ -0,0 +1,300 @@ +//! Project-level shared graph file (G4): `.kigi/graph.jsonl` at the git +//! root, so a graph follows the REPOSITORY, not the session. +//! +//! The session tracker remains the single source of truth; this file is +//! a PROJECTION refreshed at every checkpoint. Format (beads-style, +//! line-mergeable thanks to content-hash node ids): +//! +//! - line 1: the orchestration header (everything except `nodes`) +//! - lines 2..: one `GraphNode` per line +//! +//! Concurrency: an advisory `flock` on a sidecar `.lock` file makes the +//! session that CREATED or RESUMED the graph the single writer; other +//! kigi instances get a read-only view (`/graph status`) with an +//! explicit notice. The lock is held for the graph's lifetime in that +//! session and released on `/graph clear` (or process exit). +//! +//! Git discipline: kigi only WRITES the file — committing it is the +//! user's decision, never automated. + +use std::io::Write; +use std::path::{Path, PathBuf}; + +use fs2::FileExt; + +use super::graph_tracker::{GraphNode, GraphOrchestration}; + +/// Header line: the orchestration minus its nodes (which follow one per +/// line). The shadow `nodes` field is skipped on write and REJECTED on +/// read when non-empty — nodes embedded in the header would silently +/// duplicate the per-line entries. +#[derive(serde::Serialize, serde::Deserialize)] +struct ProjectGraphHeader { + #[serde(flatten)] + orchestration: GraphOrchestration, +} + +fn header_has_inline_nodes(header: &ProjectGraphHeader) -> bool { + !header.orchestration.nodes.is_empty() +} + +/// Held exclusive advisory lock on the project graph. Dropping releases. +#[derive(Debug)] +pub struct ProjectGraphLock { + _file: std::fs::File, +} + +#[derive(Debug)] +pub enum LockOutcome { + Acquired(ProjectGraphLock), + /// Another kigi instance holds the lock. + Busy, +} + +/// `.kigi` dir under the git root of `cwd`; `None` outside a git repo +/// (the project-graph feature is git-scoped by design). +pub fn project_graph_dir(cwd: &Path) -> Option { + kigi_workspace::session::git::find_git_root_from_path(cwd) + .ok() + .map(|root| root.join(".kigi")) +} + +pub fn graph_file_path(dir: &Path) -> PathBuf { + dir.join("graph.jsonl") +} + +fn lock_file_path(dir: &Path) -> PathBuf { + dir.join("graph.jsonl.lock") +} + +/// Try to become the project graph's single writer. Fail-fast: any I/O +/// error other than "already locked" propagates. +pub fn try_acquire_writer(dir: &Path) -> std::io::Result { + std::fs::create_dir_all(dir)?; + let file = std::fs::OpenOptions::new() + .create(true) + .truncate(false) + .write(true) + .open(lock_file_path(dir))?; + match file.try_lock_exclusive() { + Ok(()) => Ok(LockOutcome::Acquired(ProjectGraphLock { _file: file })), + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => Ok(LockOutcome::Busy), + // fs2 maps "already locked" differently per platform; treat the + // documented contention errno as Busy too. + Err(err) if err.raw_os_error() == Some(libc::EWOULDBLOCK) => Ok(LockOutcome::Busy), + Err(err) => Err(err), + } +} + +/// Atomically project the orchestration to `.kigi/graph.jsonl` +/// (tmp + rename, same discipline as the session state file). +pub fn project(dir: &Path, state: &GraphOrchestration) -> std::io::Result<()> { + std::fs::create_dir_all(dir)?; + let mut header_state = state.clone(); + let nodes = std::mem::take(&mut header_state.nodes); + let mut header_value = serde_json::to_value(&header_state) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; + if let Some(obj) = header_value.as_object_mut() { + // The contract says "minus nodes"; drop the empty vec the + // struct serializer would otherwise emit. + obj.remove("nodes"); + } + let mut buf = Vec::with_capacity(4096); + serde_json::to_writer(&mut buf, &header_value) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; + buf.push(b'\n'); + for node in &nodes { + serde_json::to_writer(&mut buf, node) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; + buf.push(b'\n'); + } + let target = graph_file_path(dir); + let tmp = target.with_extension("jsonl.tmp"); + { + let mut f = std::fs::File::create(&tmp)?; + f.write_all(&buf)?; + f.sync_all()?; + } + std::fs::rename(&tmp, &target) +} + +/// Load the projected graph, `Ok(None)` when absent. Malformed content +/// is an ERROR (never silently treated as "no graph") — the file is +/// user-visible, git-merged state; corruption must surface. +pub fn load(dir: &Path) -> std::io::Result> { + let path = graph_file_path(dir); + let raw = match std::fs::read_to_string(&path) { + Ok(raw) => raw, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(err) => return Err(err), + }; + let mut lines = raw.lines().filter(|l| !l.trim().is_empty()); + let Some(header_line) = lines.next() else { + return Ok(None); + }; + let header: ProjectGraphHeader = serde_json::from_str(header_line).map_err(|e| { + std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("{} line 1: {e}", path.display()), + ) + })?; + if header_has_inline_nodes(&header) { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!( + "{} line 1 embeds nodes inline; nodes belong one per line", + path.display() + ), + )); + } + let mut state = header.orchestration; + for (idx, line) in lines.enumerate() { + let node: GraphNode = serde_json::from_str(line).map_err(|e| { + std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("{} node line {}: {e}", path.display(), idx + 2), + ) + })?; + state.nodes.push(node); + } + Ok(Some(state)) +} + +/// Remove the projection (on `/graph clear`). Missing file is fine. +pub fn remove(dir: &Path) -> std::io::Result<()> { + match std::fs::remove_file(graph_file_path(dir)) { + Ok(()) => Ok(()), + Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(err) => Err(err), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::session::goal_tracker::{GoalPhase, GoalStatus}; + use crate::session::graph_tracker::{DepKind, NodeDep, NodeStatus}; + use tempfile::TempDir; + + fn sample_state() -> GraphOrchestration { + GraphOrchestration { + graph_id: "g-1".into(), + objective: "ship it".into(), + status: GoalStatus::Active, + phase: GoalPhase::Executing, + plan_version: 2, + nodes: vec![ + GraphNode { + id: "gn-aaaa".into(), + title: "A".into(), + spec: "do a".into(), + deps: vec![], + status: NodeStatus::Achieved, + goal_id: Some("goal-1".into()), + rounds: 2, + tokens_used: 100, + failure: None, + }, + GraphNode { + id: "gn-bbbb".into(), + title: "B".into(), + spec: "do b".into(), + deps: vec![NodeDep { + on: "gn-aaaa".into(), + kind: DepKind::DiscoveredFrom, + }], + status: NodeStatus::Running, + goal_id: None, + rounds: 0, + tokens_used: 0, + failure: None, + }, + ], + current_node: Some("gn-bbbb".into()), + created_at: "2026-07-20T00:00:00Z".into(), + elapsed_ms: 12, + token_budget: Some(1_000), + tokens_spent_nodes: 100, + history: vec![], + pause_message: None, + pending_discoveries: vec![], + replan_runs: 1, + } + } + + #[test] + fn project_load_round_trip_is_line_per_node() { + let tmp = TempDir::new().unwrap(); + let state = sample_state(); + project(tmp.path(), &state).unwrap(); + let raw = std::fs::read_to_string(graph_file_path(tmp.path())).unwrap(); + assert_eq!(raw.lines().count(), 3, "header + one line per node"); + assert!( + !raw.lines().next().unwrap().contains("\"nodes\""), + "header must omit nodes entirely" + ); + assert!(raw.lines().nth(1).unwrap().contains("gn-aaaa")); + let loaded = load(tmp.path()).unwrap().expect("present"); + assert_eq!(loaded.graph_id, state.graph_id); + assert_eq!(loaded.plan_version, 2); + assert_eq!(loaded.nodes.len(), 2); + assert_eq!(loaded.nodes[1].deps[0].kind, DepKind::DiscoveredFrom); + assert_eq!(loaded.current_node.as_deref(), Some("gn-bbbb")); + } + + #[test] + fn load_absent_is_none_and_remove_is_idempotent() { + let tmp = TempDir::new().unwrap(); + assert!(load(tmp.path()).unwrap().is_none()); + remove(tmp.path()).unwrap(); + project(tmp.path(), &sample_state()).unwrap(); + remove(tmp.path()).unwrap(); + assert!(load(tmp.path()).unwrap().is_none()); + remove(tmp.path()).unwrap(); + } + + #[test] + fn header_with_inline_nodes_is_rejected() { + let tmp = TempDir::new().unwrap(); + let mut bad = serde_json::to_value(sample_state()).unwrap(); + // Keep nodes inline in the header — a hand-edited/merged file. + bad.as_object_mut().unwrap().remove("current_node"); + std::fs::write(graph_file_path(tmp.path()), format!("{bad}\n")).unwrap(); + let err = load(tmp.path()).unwrap_err(); + assert!(err.to_string().contains("inline"), "{err}"); + } + + #[test] + fn malformed_content_is_a_loud_error_not_a_missing_graph() { + let tmp = TempDir::new().unwrap(); + std::fs::write(graph_file_path(tmp.path()), "not json\n").unwrap(); + let err = load(tmp.path()).unwrap_err(); + assert_eq!(err.kind(), std::io::ErrorKind::InvalidData); + assert!(err.to_string().contains("line 1"), "{err}"); + } + + #[test] + fn writer_lock_is_exclusive_within_and_across_handles() { + let tmp = TempDir::new().unwrap(); + let first = try_acquire_writer(tmp.path()).unwrap(); + let LockOutcome::Acquired(_guard) = first else { + panic!("first acquire must win"); + }; + match try_acquire_writer(tmp.path()).unwrap() { + LockOutcome::Busy => {} + LockOutcome::Acquired(_) => { + // flock is per-fd on some platforms within one process; + // if this arm is reached the platform lets the same + // process re-lock, which is still safe for our + // cross-INSTANCE contract — but on macOS/Linux flock + // between distinct fds does contend, so treat as bug. + panic!("second handle must observe Busy"); + } + } + drop(_guard); + assert!(matches!( + try_acquire_writer(tmp.path()).unwrap(), + LockOutcome::Acquired(_) + )); + } +} diff --git a/crates/codegen/kigi-shell/src/session/graph_tracker.rs b/crates/codegen/kigi-shell/src/session/graph_tracker.rs index 7070597..13cae72 100644 --- a/crates/codegen/kigi-shell/src/session/graph_tracker.rs +++ b/crates/codegen/kigi-shell/src/session/graph_tracker.rs @@ -213,7 +213,9 @@ pub struct GraphOrchestration { pub plan_version: u32, /// Topological-friendly storage order (planner order + harness /// appendix). The scheduler picks the first `Ready` node in this - /// order, so execution is deterministic. + /// order, so execution is deterministic. `default` so the project + /// header line (which omits `nodes`) deserializes. + #[serde(default)] pub nodes: Vec, /// Node currently running as the active goal, if any. #[serde(default, skip_serializing_if = "Option::is_none")] diff --git a/crates/codegen/kigi-shell/src/session/mod.rs b/crates/codegen/kigi-shell/src/session/mod.rs index 079de8c..e6fadae 100644 --- a/crates/codegen/kigi-shell/src/session/mod.rs +++ b/crates/codegen/kigi-shell/src/session/mod.rs @@ -303,6 +303,7 @@ pub(crate) mod goal_summarizer; pub mod goal_tracker; pub(crate) mod graph_plan; pub(crate) mod graph_planner; +pub(crate) mod graph_project; pub mod graph_tracker; pub mod helpers; pub(crate) mod image_describe;