From 28e763a695f56adfbecd4efb14edbde13bbd63dc Mon Sep 17 00:00:00 2001 From: asepharyana Date: Tue, 14 Jul 2026 10:29:54 +0700 Subject: [PATCH] fix(hive-mind): gunakan flag SessionRuntime sebagai sinyal konvergensi otoritatif MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pesan sistem bertanda [Hive-Mind Consensus] hanya di-push ke variabel lokal run_agent_turn dan diarsipkan ke SQLite, tidak pernah masuk ke rt.messages lewat TurnEvent — sehingga hive_mind_already_ran selalu memindai daftar pesan yang kosong dan gerbang "converge sekali per sesi" tidak pernah aktif. Tambahkan SessionRuntime.hive_mind_converged yang diset dari event TurnEvent::SystemNote { kind: "hive_mind_converged" } setelah konvergensi selesai, disalurkan lewat TurnCtx, dan dijadikan sinyal utama di run_agent_turn (pemindaian pesan lama tetap sebagai fallback defensif). Co-Authored-By: Claude Sonnet 5 --- src/app/runtime/actions/mod.rs | 70 +++++++++++++++++++++++++++++----- src/app/state/runtime.rs | 9 +++++ 2 files changed, 70 insertions(+), 9 deletions(-) diff --git a/src/app/runtime/actions/mod.rs b/src/app/runtime/actions/mod.rs index 3f1aeb9..c6e56b6 100644 --- a/src/app/runtime/actions/mod.rs +++ b/src/app/runtime/actions/mod.rs @@ -383,6 +383,10 @@ pub fn apply_action(state: &mut AppStateRest, action: Action) { } } else if kind == "connectivity" { state.misc.api_connected = message == "connected"; + } else if kind == "hive_mind_converged" { + if let Some(ref mut rt) = state.session_runtime { + rt.hive_mind_converged = true; + } } else if kind == "pipeline" { // Clear old workflow agents when a new pipeline starts. if message == HIVE_MIND_KICKOFF_NOTE { @@ -748,6 +752,7 @@ fn spawn_turn(state: &AppStateRest) { let workspace_roots: Vec = ctx.workspaces.clone(); let abort_flag = state.abort_flag.clone(); abort_flag.store(false, std::sync::atomic::Ordering::SeqCst); + let hive_mind_converged = state.session_runtime.as_ref().is_some_and(|rt| rt.hive_mind_converged); *in_flight_flag.lock().unwrap_or_else(|e| { tracing::error!("[spawn_turn] in_flight_flag mutex poisoned: {}", e); @@ -774,6 +779,7 @@ fn spawn_turn(state: &AppStateRest) { temperature, max_tokens, abort_flag, + hive_mind_converged, }; let result = run_agent_turn(&tc, &messages, &events_q); if let Err(e) = result { @@ -802,6 +808,10 @@ struct TurnCtx { temperature: f32, max_tokens: Option, abort_flag: std::sync::Arc, + /// Snapshot of `SessionRuntime.hive_mind_converged` taken at the start + /// of this turn — whether a hive-mind convergence already completed + /// earlier in this session. + hive_mind_converged: bool, } /// Build an ASCII tree of the workspace directory structure for the @@ -1003,15 +1013,22 @@ fn run_agent_turn( // ── AUTO CEO PIPELINE ── // Before the main agent starts working, check if the pipeline should run. // Gated on whether a hive-mind convergence has already happened earlier - // in this session (detected from message content), not an arbitrary - // message-count cutoff — a complex request in message 5 deserves the - // same treatment as one in message 1, as long as this session hasn't - // already converged once. - let already_ran_hive_mind = crate::app::workflow::hive_mind::hive_mind_already_ran( - msgs.iter() - .filter(|m| matches!(m.role, crate::dto::chat::message::Role::System)) - .filter_map(|m| m.content.as_deref()) - ); + // in this session, not an arbitrary message-count cutoff — a complex + // request in message 5 deserves the same treatment as one in message 1, + // as long as this session hasn't already converged once. + // + // `tc.hive_mind_converged` is the authoritative signal (see its doc + // comment on `SessionRuntime` for why). The message-content scan is + // kept as a defensive fallback in case a future change starts + // persisting tagged system messages into `rt.messages` (e.g. via + // compaction) — today it is a no-op since that never happens, but it's + // still correct and still tested in isolation. + let already_ran_hive_mind = tc.hive_mind_converged + || crate::app::workflow::hive_mind::hive_mind_already_ran( + msgs.iter() + .filter(|m| matches!(m.role, crate::dto::chat::message::Role::System)) + .filter_map(|m| m.content.as_deref()) + ); let should_pipeline = if already_ran_hive_mind { false } else { @@ -1141,6 +1158,12 @@ fn run_agent_turn( message: "Hive-mind convergence complete. Core Intelligence reviewing consensus...".to_string(), }); } + if let Ok(mut q) = events_q.lock() { + q.push_back(TurnEvent::SystemNote { + kind: "hive_mind_converged".to_string(), + message: String::new(), + }); + } } Err(e) => { tracing::warn!("[hive-mind] convergence failed: {}", e); @@ -1822,4 +1845,33 @@ fn rand_bytes(n: usize) -> Vec { (0..n).map(|i| ((base >> ((i as u64 % 8) * 8)) ^ (i as u64 * 2_654_435_761)) as u8).collect() } +#[cfg(test)] +mod tests { + use super::*; + use crate::app::state::rest::AppStateRest; + use crate::app::state::runtime::SessionRuntime; + + #[test] + fn hive_mind_converged_system_note_sets_session_flag() { + let tmp = std::env::temp_dir().join(format!("zesdex-actions-test-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&tmp).unwrap(); + let mut state = AppStateRest::new(vec![tmp.clone()], &tmp, tmp.join("memory")); + state.session_runtime = Some(SessionRuntime::new(tmp.clone())); + + assert!(!state.session_runtime.as_ref().unwrap().hive_mind_converged); + + if let Ok(mut q) = state.turn_events.lock() { + q.push_back(TurnEvent::SystemNote { + kind: "hive_mind_converged".to_string(), + message: String::new(), + }); + } + apply_action(&mut state, Action::Tick); + + assert!(state.session_runtime.as_ref().unwrap().hive_mind_converged); + + std::fs::remove_dir_all(&tmp).ok(); + } +} + diff --git a/src/app/state/runtime.rs b/src/app/state/runtime.rs index f1a43af..9976890 100644 --- a/src/app/state/runtime.rs +++ b/src/app/state/runtime.rs @@ -46,6 +46,14 @@ pub struct SessionRuntime { pub review_count: u32, pub session_dir: PathBuf, pub usage: UsageStats, + /// Whether a hive-mind convergence has completed at least once in this + /// session. Set by the main-thread event loop when it receives a + /// `TurnEvent::SystemNote { kind: "hive_mind_converged", .. }` — the + /// only reliable way to detect this across turns, since system messages + /// pushed mid-turn inside `run_agent_turn` are NOT persisted into + /// `rt.messages` (they stay local to that turn's background thread and + /// are only archived to `SQLite`). + pub hive_mind_converged: bool, } /// Record of one completed tool invocation, kept for transcript/history. @@ -139,6 +147,7 @@ impl SessionRuntime { review_count: 0, session_dir, usage: UsageStats::default(), + hive_mind_converged: false, } }