Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7070c96460 | ||
|
|
28e763a695 | ||
|
|
3e6f9a6a5f | ||
|
|
e023f2c5a8 | ||
|
|
1039f67c12 | ||
|
|
fdd62f8303 | ||
|
|
5498088532 | ||
|
|
a125f5d440 | ||
|
|
79e2bfcc9c | ||
|
|
b1c0265e8c | ||
|
|
e2878a3d83 | ||
|
|
6790fe481b |
@@ -1,3 +1,22 @@
|
|||||||
|
# [1.5.0](https://github.com/asepharyana/zesdex/compare/v1.4.0...v1.5.0) (2026-07-14)
|
||||||
|
|
||||||
|
|
||||||
|
### Bug Fixes
|
||||||
|
|
||||||
|
* **hive-mind:** ganti gerbang pipeline berbasis jumlah pesan dengan deteksi konvergensi sebelumnya ([5498088](https://github.com/asepharyana/zesdex/commit/5498088532314f8dbc005d5e8058c0170ca92320))
|
||||||
|
* **hive-mind:** gunakan flag SessionRuntime sebagai sinyal konvergensi otoritatif ([28e763a](https://github.com/asepharyana/zesdex/commit/28e763a695f56adfbecd4efb14edbde13bbd63dc))
|
||||||
|
* **hive-mind:** hapus penulisan docs/runs ganda dan sambungkan abort_flag ke tool hive_mind manual ([a125f5d](https://github.com/asepharyana/zesdex/commit/a125f5d4400b0c417ba4049e67480bd18479e90b))
|
||||||
|
* **hive-mind:** tambah timeout per-node dan jamin dokumentasi convergence tetap tertulis saat sintesis gagal ([b1c0265](https://github.com/asepharyana/zesdex/commit/b1c0265e8cdf9278664e64f77fbde4ec8c22fcfd))
|
||||||
|
* **subagent:** panic-proof overlap guards and update stale docs ([e023f2c](https://github.com/asepharyana/zesdex/commit/e023f2c5a8f036d89e925bb9ceec253343476a74))
|
||||||
|
* **subagent:** perbaiki filter is_production_code berbasis substring dan tambah pembatalan/anti-tumpang-tindih pada background review ([1039f67](https://github.com/asepharyana/zesdex/commit/1039f67c12749c6b2c93e3ab7037feded8ab01c6))
|
||||||
|
* **tui:** perbaiki roster workflow yang tidak pernah ter-reset karena substring "started" tidak pernah cocok ([fdd62f8](https://github.com/asepharyana/zesdex/commit/fdd62f830330b5b3e2b4f9fcc7274daf7f7842a5))
|
||||||
|
|
||||||
|
|
||||||
|
### Features
|
||||||
|
|
||||||
|
* **settings:** tambah hive_mind_node_timeout_ms dengan fallback serde default ([e2878a3](https://github.com/asepharyana/zesdex/commit/e2878a3d83f171aa181ac29ca689828e0cb1408f))
|
||||||
|
* **tool:** tambah abort_flag ke ToolCtx dan sambungkan dari session state ([79e2bfc](https://github.com/asepharyana/zesdex/commit/79e2bfcc9ca67424ca2b34a5652d3c2bb93291bf))
|
||||||
|
|
||||||
# [1.4.0](https://github.com/asepharyana/zesdex/compare/v1.3.0...v1.4.0) (2026-07-14)
|
# [1.4.0](https://github.com/asepharyana/zesdex/compare/v1.3.0...v1.4.0) (2026-07-14)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Generated
+1
-1
@@ -4436,7 +4436,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "zesdex"
|
name = "zesdex"
|
||||||
version = "1.4.0"
|
version = "1.5.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"anyhow",
|
"anyhow",
|
||||||
"base64",
|
"base64",
|
||||||
|
|||||||
+1
-1
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "zesdex"
|
name = "zesdex"
|
||||||
version = "1.4.0"
|
version = "1.5.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
authors = ["asepharyana <superaseph@gmail.com>"]
|
authors = ["asepharyana <superaseph@gmail.com>"]
|
||||||
|
|
||||||
|
|||||||
File diff suppressed because it is too large
Load Diff
@@ -383,9 +383,13 @@ pub fn apply_action(state: &mut AppStateRest, action: Action) {
|
|||||||
}
|
}
|
||||||
} else if kind == "connectivity" {
|
} else if kind == "connectivity" {
|
||||||
state.misc.api_connected = message == "connected";
|
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" {
|
} else if kind == "pipeline" {
|
||||||
// Clear old workflow agents when a new pipeline starts.
|
// Clear old workflow agents when a new pipeline starts.
|
||||||
if message.contains("started") {
|
if message == HIVE_MIND_KICKOFF_NOTE {
|
||||||
state.workflow_engine.agents.clear();
|
state.workflow_engine.agents.clear();
|
||||||
state.workflow_engine.findings.clear();
|
state.workflow_engine.findings.clear();
|
||||||
}
|
}
|
||||||
@@ -748,6 +752,7 @@ fn spawn_turn(state: &AppStateRest) {
|
|||||||
let workspace_roots: Vec<std::path::PathBuf> = ctx.workspaces.clone();
|
let workspace_roots: Vec<std::path::PathBuf> = ctx.workspaces.clone();
|
||||||
let abort_flag = state.abort_flag.clone();
|
let abort_flag = state.abort_flag.clone();
|
||||||
abort_flag.store(false, std::sync::atomic::Ordering::SeqCst);
|
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| {
|
*in_flight_flag.lock().unwrap_or_else(|e| {
|
||||||
tracing::error!("[spawn_turn] in_flight_flag mutex poisoned: {}", e);
|
tracing::error!("[spawn_turn] in_flight_flag mutex poisoned: {}", e);
|
||||||
@@ -774,6 +779,7 @@ fn spawn_turn(state: &AppStateRest) {
|
|||||||
temperature,
|
temperature,
|
||||||
max_tokens,
|
max_tokens,
|
||||||
abort_flag,
|
abort_flag,
|
||||||
|
hive_mind_converged,
|
||||||
};
|
};
|
||||||
let result = run_agent_turn(&tc, &messages, &events_q);
|
let result = run_agent_turn(&tc, &messages, &events_q);
|
||||||
if let Err(e) = result {
|
if let Err(e) = result {
|
||||||
@@ -802,6 +808,10 @@ struct TurnCtx {
|
|||||||
temperature: f32,
|
temperature: f32,
|
||||||
max_tokens: Option<u32>,
|
max_tokens: Option<u32>,
|
||||||
abort_flag: std::sync::Arc<std::sync::atomic::AtomicBool>,
|
abort_flag: std::sync::Arc<std::sync::atomic::AtomicBool>,
|
||||||
|
/// 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
|
/// Build an ASCII tree of the workspace directory structure for the
|
||||||
@@ -937,6 +947,16 @@ fn archive_message(db: Option<&std::sync::Arc<std::sync::Mutex<rusqlite::Connect
|
|||||||
/// background subagents still fire at the end of the turn.
|
/// background subagents still fire at the end of the turn.
|
||||||
const MAX_AUTO_REVIEWS_PER_TURN: usize = 2;
|
const MAX_AUTO_REVIEWS_PER_TURN: usize = 2;
|
||||||
|
|
||||||
|
/// Exact text of the "pipeline started" `SystemNote` pushed once per
|
||||||
|
/// hive-mind kickoff. Matched by exact equality (not a loose substring)
|
||||||
|
/// when deciding whether to reset the workflow panel's agent roster —
|
||||||
|
/// shared between the push site and the check site so they cannot drift
|
||||||
|
/// out of sync the way the previous `.contains("started")` check did
|
||||||
|
/// (no real pipeline message ever contained that word, so the roster
|
||||||
|
/// never cleared and agent cards accumulated across every hive-mind run
|
||||||
|
/// in a session).
|
||||||
|
const HIVE_MIND_KICKOFF_NOTE: &str = "Core Intelligence is compiling a cognitive cycle plan...";
|
||||||
|
|
||||||
/// Execute one full agent turn: stream the conversation to the LLM,
|
/// Execute one full agent turn: stream the conversation to the LLM,
|
||||||
/// handle tool calls, and loop until the LLM produces a non-tool response
|
/// handle tool calls, and loop until the LLM produces a non-tool response
|
||||||
/// or runs out of unfinished todo items.
|
/// or runs out of unfinished todo items.
|
||||||
@@ -992,10 +1012,26 @@ fn run_agent_turn(
|
|||||||
|
|
||||||
// ── AUTO CEO PIPELINE ──
|
// ── AUTO CEO PIPELINE ──
|
||||||
// Before the main agent starts working, check if the pipeline should run.
|
// Before the main agent starts working, check if the pipeline should run.
|
||||||
let user_msg_count = msgs.iter()
|
// Gated on whether a hive-mind convergence has already happened earlier
|
||||||
.filter(|m| matches!(m.role, crate::dto::chat::message::Role::User))
|
// in this session, not an arbitrary message-count cutoff — a complex
|
||||||
.count();
|
// request in message 5 deserves the same treatment as one in message 1,
|
||||||
let should_pipeline = if user_msg_count <= 2 {
|
// 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 {
|
||||||
let user_request = msgs.iter()
|
let user_request = msgs.iter()
|
||||||
.rev().find(|m| matches!(m.role, crate::dto::chat::message::Role::User))
|
.rev().find(|m| matches!(m.role, crate::dto::chat::message::Role::User))
|
||||||
.and_then(|m| m.content.as_deref())
|
.and_then(|m| m.content.as_deref())
|
||||||
@@ -1006,8 +1042,6 @@ fn run_agent_turn(
|
|||||||
} else {
|
} else {
|
||||||
crate::app::workflow::hive_mind::is_complex_request(user_request)
|
crate::app::workflow::hive_mind::is_complex_request(user_request)
|
||||||
}
|
}
|
||||||
} else {
|
|
||||||
false
|
|
||||||
};
|
};
|
||||||
|
|
||||||
if should_pipeline {
|
if should_pipeline {
|
||||||
@@ -1021,7 +1055,7 @@ fn run_agent_turn(
|
|||||||
if let Ok(mut q) = events_q.lock() {
|
if let Ok(mut q) = events_q.lock() {
|
||||||
q.push_back(TurnEvent::SystemNote {
|
q.push_back(TurnEvent::SystemNote {
|
||||||
kind: "pipeline".to_string(),
|
kind: "pipeline".to_string(),
|
||||||
message: "Core Intelligence is compiling a cognitive cycle plan...".to_string(),
|
message: HIVE_MIND_KICKOFF_NOTE.to_string(),
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1105,18 +1139,15 @@ fn run_agent_turn(
|
|||||||
};
|
};
|
||||||
|
|
||||||
match pipeline_result {
|
match pipeline_result {
|
||||||
Ok((consensus, reports)) => {
|
Ok((consensus, _reports)) => {
|
||||||
|
// run_hive_mind already wrote docs/runs/*.md internally
|
||||||
|
// (guaranteed, even on synthesis failure) — nothing to do
|
||||||
|
// here besides feeding the consensus back to the LLM.
|
||||||
tracing::info!("[hive-mind] convergence completed successfully");
|
tracing::info!("[hive-mind] convergence completed successfully");
|
||||||
|
|
||||||
if let Some(workspace_root) = tc.workspace_roots.first() {
|
|
||||||
match crate::app::workflow::docs::write_hive_mind_convergence(workspace_root, user_request, &reports, &consensus) {
|
|
||||||
Ok(path) => tracing::info!("[hive-mind] convergence documented at {}", path.display()),
|
|
||||||
Err(e) => tracing::warn!("[hive-mind] failed to write docs/runs report: {e}"),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
let pipeline_msg = ChatMessage::system(format!(
|
let pipeline_msg = ChatMessage::system(format!(
|
||||||
"[Hive-Mind Consensus]\n{consensus}",
|
"{}\n{consensus}",
|
||||||
|
crate::app::workflow::hive_mind::HIVE_MIND_CONSENSUS_TAG,
|
||||||
));
|
));
|
||||||
archive_message(tc.db.as_ref(), &tc.session_id, &pipeline_msg);
|
archive_message(tc.db.as_ref(), &tc.session_id, &pipeline_msg);
|
||||||
msgs.push(pipeline_msg);
|
msgs.push(pipeline_msg);
|
||||||
@@ -1127,6 +1158,12 @@ fn run_agent_turn(
|
|||||||
message: "Hive-mind convergence complete. Core Intelligence reviewing consensus...".to_string(),
|
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) => {
|
Err(e) => {
|
||||||
tracing::warn!("[hive-mind] convergence failed: {}", e);
|
tracing::warn!("[hive-mind] convergence failed: {}", e);
|
||||||
@@ -1495,12 +1532,14 @@ fn run_agent_turn(
|
|||||||
let bg_session_dir = tc.edit_log_session_dir.clone();
|
let bg_session_dir = tc.edit_log_session_dir.clone();
|
||||||
let bg_workspaces = tc.workspace_roots.clone();
|
let bg_workspaces = tc.workspace_roots.clone();
|
||||||
let bg_events = events_q.clone();
|
let bg_events = events_q.clone();
|
||||||
|
let bg_abort = tc.abort_flag.clone();
|
||||||
std::thread::spawn(move || {
|
std::thread::spawn(move || {
|
||||||
crate::app::subagent::auto::spawn_all_background(
|
crate::app::subagent::auto::spawn_all_background(
|
||||||
&bg_paths,
|
&bg_paths,
|
||||||
&bg_session_dir,
|
&bg_session_dir,
|
||||||
&bg_workspaces,
|
&bg_workspaces,
|
||||||
&bg_events,
|
&bg_events,
|
||||||
|
bg_abort,
|
||||||
);
|
);
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
@@ -1806,4 +1845,33 @@ fn rand_bytes(n: usize) -> Vec<u8> {
|
|||||||
(0..n).map(|i| ((base >> ((i as u64 % 8) * 8)) ^ (i as u64 * 2_654_435_761)) as u8).collect()
|
(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();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -283,6 +283,29 @@ impl AppStateRest {
|
|||||||
lsp_manager: self.lsp_manager.clone(),
|
lsp_manager: self.lsp_manager.clone(),
|
||||||
turn_events: Some(self.turn_events.clone()),
|
turn_events: Some(self.turn_events.clone()),
|
||||||
workflow_findings: None,
|
workflow_findings: None,
|
||||||
|
abort_flag: Some(self.abort_flag.clone()),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn tool_ctx_for_shares_the_session_abort_flag() {
|
||||||
|
let tmp = std::env::temp_dir().join(format!("zesdex-rest-test-{}", uuid::Uuid::new_v4()));
|
||||||
|
std::fs::create_dir_all(&tmp).unwrap();
|
||||||
|
let state = AppStateRest::new(vec![tmp.clone()], &tmp, tmp.join("memory"));
|
||||||
|
|
||||||
|
let ctx = state.tool_ctx_for(Origin::Main);
|
||||||
|
|
||||||
|
assert!(ctx.abort_flag.is_some());
|
||||||
|
assert!(std::sync::Arc::ptr_eq(
|
||||||
|
ctx.abort_flag.as_ref().unwrap(),
|
||||||
|
&state.abort_flag,
|
||||||
|
));
|
||||||
|
|
||||||
|
std::fs::remove_dir_all(&tmp).ok();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -46,6 +46,14 @@ pub struct SessionRuntime {
|
|||||||
pub review_count: u32,
|
pub review_count: u32,
|
||||||
pub session_dir: PathBuf,
|
pub session_dir: PathBuf,
|
||||||
pub usage: UsageStats,
|
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.
|
/// Record of one completed tool invocation, kept for transcript/history.
|
||||||
@@ -139,6 +147,7 @@ impl SessionRuntime {
|
|||||||
review_count: 0,
|
review_count: 0,
|
||||||
session_dir,
|
session_dir,
|
||||||
usage: UsageStats::default(),
|
usage: UsageStats::default(),
|
||||||
|
hive_mind_converged: false,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+186
-13
@@ -18,6 +18,7 @@
|
|||||||
|
|
||||||
use std::path::Path;
|
use std::path::Path;
|
||||||
use std::sync::{Arc, Mutex};
|
use std::sync::{Arc, Mutex};
|
||||||
|
use std::sync::atomic::{AtomicBool, Ordering};
|
||||||
use std::collections::VecDeque;
|
use std::collections::VecDeque;
|
||||||
use crate::app::state::runtime::TurnEvent;
|
use crate::app::state::runtime::TurnEvent;
|
||||||
use crate::app::subagent::context::build_subagent_context;
|
use crate::app::subagent::context::build_subagent_context;
|
||||||
@@ -37,9 +38,37 @@ const SKIP_REVIEW_FILES: &[&str] = &[
|
|||||||
".gitignore", ".env", ".env.example",
|
".gitignore", ".env", ".env.example",
|
||||||
];
|
];
|
||||||
|
|
||||||
|
/// Prevents a second background subagent of the same kind from spawning
|
||||||
|
/// while one is already in flight. Without this, a chatty multi-turn edit
|
||||||
|
/// session could stack overlapping test-gen/arch/security reviews of
|
||||||
|
/// overlapping file sets, none of which could be told apart in the
|
||||||
|
/// `SystemNote` toast stream.
|
||||||
|
static TEST_GEN_RUNNING: AtomicBool = AtomicBool::new(false);
|
||||||
|
static ARCH_REVIEW_RUNNING: AtomicBool = AtomicBool::new(false);
|
||||||
|
static SECURITY_REVIEW_RUNNING: AtomicBool = AtomicBool::new(false);
|
||||||
|
|
||||||
|
/// RAII guard that resets a per-kind overlap flag back to `false` on drop —
|
||||||
|
/// including during a panic-triggered unwind inside the spawned thread — so
|
||||||
|
/// a background review can never wedge itself permanently disabled for the
|
||||||
|
/// rest of the process if the subagent run panics before reaching its
|
||||||
|
/// normal completion path.
|
||||||
|
struct RunningGuard(&'static AtomicBool);
|
||||||
|
|
||||||
|
impl Drop for RunningGuard {
|
||||||
|
fn drop(&mut self) {
|
||||||
|
self.0.store(false, Ordering::SeqCst);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// ─── Helpers ───
|
/// ─── Helpers ───
|
||||||
///
|
///
|
||||||
/// Check whether a file path is worth auto-reviewing (not config/lock/data).
|
/// Check whether a file path is worth auto-reviewing (not config/lock/data).
|
||||||
|
///
|
||||||
|
/// Vendored/generated directories are matched by path *segment* rather than
|
||||||
|
/// a `/target/`-style substring check — the substring form misses paths
|
||||||
|
/// where the directory is the first component (e.g. `target/debug/build.rs`,
|
||||||
|
/// which has no leading slash), the same class of bug fixed in
|
||||||
|
/// `is_production_code` below.
|
||||||
pub fn is_reviewable_path(path: &str) -> bool {
|
pub fn is_reviewable_path(path: &str) -> bool {
|
||||||
let lower = path.to_lowercase();
|
let lower = path.to_lowercase();
|
||||||
if SKIP_REVIEW_FILES.iter().any(|f| lower.ends_with(f)) {
|
if SKIP_REVIEW_FILES.iter().any(|f| lower.ends_with(f)) {
|
||||||
@@ -49,9 +78,14 @@ pub fn is_reviewable_path(path: &str) -> bool {
|
|||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
// Skip paths that are clearly generated or vendored
|
// Skip paths that are clearly generated or vendored
|
||||||
if lower.contains("/target/") || lower.contains("/node_modules/")
|
let in_vendored_dir = std::path::Path::new(&lower).components().any(|c| {
|
||||||
|| lower.contains("/.git/") || lower.contains("/vendor/")
|
matches!(
|
||||||
{
|
c,
|
||||||
|
std::path::Component::Normal(seg)
|
||||||
|
if matches!(seg.to_str(), Some("target" | "node_modules" | ".git" | "vendor"))
|
||||||
|
)
|
||||||
|
});
|
||||||
|
if in_vendored_dir {
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
true
|
true
|
||||||
@@ -60,15 +94,40 @@ pub fn is_reviewable_path(path: &str) -> bool {
|
|||||||
/// Determine whether a file change looks like it modifies production logic
|
/// Determine whether a file change looks like it modifies production logic
|
||||||
/// (vs. tests, config, or documentation) — used to decide if a test-gen
|
/// (vs. tests, config, or documentation) — used to decide if a test-gen
|
||||||
/// or security-review background subagent should fire.
|
/// or security-review background subagent should fire.
|
||||||
|
///
|
||||||
|
/// Matches test-ness by path *segment* (a directory literally named
|
||||||
|
/// "test"/"tests"/"__tests__") or by filename convention
|
||||||
|
/// (`foo_test.rs`, `foo.test.ts`, `test_foo.py`, `foo_spec.rb`), not by a
|
||||||
|
/// raw substring check — a plain `.contains("test")` would wrongly exclude
|
||||||
|
/// legitimate production files like `src/attestation.rs` or
|
||||||
|
/// `src/latest/foo.rs`.
|
||||||
fn is_production_code(path: &str) -> bool {
|
fn is_production_code(path: &str) -> bool {
|
||||||
let lower = path.to_lowercase();
|
let lower = path.to_lowercase();
|
||||||
// Skip test files — they don't need test-gen from another agent
|
let path_obj = std::path::Path::new(&lower);
|
||||||
if lower.contains("test") || lower.contains("spec") || lower.contains("_test.") {
|
|
||||||
|
let in_test_dir = path_obj.components().any(|c| {
|
||||||
|
matches!(
|
||||||
|
c,
|
||||||
|
std::path::Component::Normal(seg)
|
||||||
|
if matches!(seg.to_str(), Some("test") | Some("tests") | Some("__tests__"))
|
||||||
|
)
|
||||||
|
});
|
||||||
|
|
||||||
|
let file_stem = path_obj.file_stem().and_then(|s| s.to_str()).unwrap_or("");
|
||||||
|
let is_test_filename = file_stem.starts_with("test_")
|
||||||
|
|| file_stem.ends_with("_test")
|
||||||
|
|| file_stem.ends_with(".test")
|
||||||
|
|| file_stem == "spec"
|
||||||
|
|| file_stem.ends_with("_spec")
|
||||||
|
|| file_stem.ends_with(".spec");
|
||||||
|
|
||||||
|
if in_test_dir || is_test_filename {
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Only source files — use Path::extension() to avoid clippy
|
// Only source files — use Path::extension() to avoid clippy
|
||||||
// case_sensitive_file_extension_comparisons lint
|
// case_sensitive_file_extension_comparisons lint
|
||||||
std::path::Path::new(&lower)
|
path_obj
|
||||||
.extension()
|
.extension()
|
||||||
.and_then(|ext| ext.to_str())
|
.and_then(|ext| ext.to_str())
|
||||||
.is_some_and(|ext| {
|
.is_some_and(|ext| {
|
||||||
@@ -148,19 +207,30 @@ pub fn spawn_quick_review(
|
|||||||
/// silently swallowing the error into a note string, so a single transient
|
/// silently swallowing the error into a note string, so a single transient
|
||||||
/// LLM/tool failure doesn't just disappear.
|
/// LLM/tool failure doesn't just disappear.
|
||||||
///
|
///
|
||||||
|
/// `abort_flag` is checked before every attempt (including the first) and
|
||||||
|
/// forwarded into the subagent's own context, so a cancelled turn stops
|
||||||
|
/// retrying immediately instead of burning a second attempt.
|
||||||
|
///
|
||||||
/// Return: `Ok(output)` if either attempt succeeded, `Err(message)`
|
/// Return: `Ok(output)` if either attempt succeeded, `Err(message)`
|
||||||
/// describing the final failure if both attempts failed.
|
/// describing the final failure if both attempts failed, or the literal
|
||||||
|
/// message `"aborted by user"` if `abort_flag` was already set before an
|
||||||
|
/// attempt could start.
|
||||||
fn run_subagent_with_retry(
|
fn run_subagent_with_retry(
|
||||||
def: &AgentDefinition,
|
def: &AgentDefinition,
|
||||||
session_dir: &Path,
|
session_dir: &Path,
|
||||||
workspaces: &[std::path::PathBuf],
|
workspaces: &[std::path::PathBuf],
|
||||||
label: &str,
|
label: &str,
|
||||||
|
abort_flag: Option<&Arc<AtomicBool>>,
|
||||||
) -> Result<String, String> {
|
) -> Result<String, String> {
|
||||||
let mut last_err = String::new();
|
let mut last_err = String::new();
|
||||||
for attempt in 1..=2 {
|
for attempt in 1..=2 {
|
||||||
|
if abort_flag.is_some_and(|f| f.load(Ordering::SeqCst)) {
|
||||||
|
return Err("aborted by user".to_string());
|
||||||
|
}
|
||||||
let mut ctx = build_subagent_context(def);
|
let mut ctx = build_subagent_context(def);
|
||||||
ctx.session_dir = session_dir.to_path_buf();
|
ctx.session_dir = session_dir.to_path_buf();
|
||||||
ctx.workspaces = workspaces.to_vec();
|
ctx.workspaces = workspaces.to_vec();
|
||||||
|
ctx.abort_flag = abort_flag.cloned();
|
||||||
|
|
||||||
let (tx, mut rx) = tokio::sync::mpsc::channel(32);
|
let (tx, mut rx) = tokio::sync::mpsc::channel(32);
|
||||||
let drain_label = label.to_string();
|
let drain_label = label.to_string();
|
||||||
@@ -188,15 +258,25 @@ fn run_subagent_with_retry(
|
|||||||
/// Uses the test-generator prompt and has read-write access so it can
|
/// Uses the test-generator prompt and has read-write access so it can
|
||||||
/// create test files. Runs in a separate OS thread and reports completion
|
/// create test files. Runs in a separate OS thread and reports completion
|
||||||
/// via `TurnEvent::SystemNote { kind: "bg-test-gen" }`.
|
/// via `TurnEvent::SystemNote { kind: "bg-test-gen" }`.
|
||||||
|
///
|
||||||
|
/// Skipped (no-op) if a test-gen run is already in flight (guarded by
|
||||||
|
/// `TEST_GEN_RUNNING`) — prevents a chatty multi-turn edit session from
|
||||||
|
/// stacking overlapping runs. `abort_flag` is forwarded to
|
||||||
|
/// `run_subagent_with_retry` so the run can be cancelled if the turn aborts.
|
||||||
pub fn spawn_background_test_gen(
|
pub fn spawn_background_test_gen(
|
||||||
file_paths: &[String],
|
file_paths: &[String],
|
||||||
session_dir: &Path,
|
session_dir: &Path,
|
||||||
workspaces: &[std::path::PathBuf],
|
workspaces: &[std::path::PathBuf],
|
||||||
turn_events: &Arc<Mutex<VecDeque<TurnEvent>>>,
|
turn_events: &Arc<Mutex<VecDeque<TurnEvent>>>,
|
||||||
|
abort_flag: Arc<AtomicBool>,
|
||||||
) {
|
) {
|
||||||
if file_paths.is_empty() {
|
if file_paths.is_empty() {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
if TEST_GEN_RUNNING.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst).is_err() {
|
||||||
|
tracing::debug!("[bg-test-gen] skipped — a test-gen run is already in flight");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
let paths = file_paths.to_vec();
|
let paths = file_paths.to_vec();
|
||||||
let sd = session_dir.to_path_buf();
|
let sd = session_dir.to_path_buf();
|
||||||
@@ -204,6 +284,7 @@ pub fn spawn_background_test_gen(
|
|||||||
let events = turn_events.clone();
|
let events = turn_events.clone();
|
||||||
|
|
||||||
std::thread::spawn(move || {
|
std::thread::spawn(move || {
|
||||||
|
let _running_guard = RunningGuard(&TEST_GEN_RUNNING);
|
||||||
tracing::info!(
|
tracing::info!(
|
||||||
"[bg-test-gen] spawning for {} file(s): {:?}",
|
"[bg-test-gen] spawning for {} file(s): {:?}",
|
||||||
paths.len(),
|
paths.len(),
|
||||||
@@ -224,12 +305,13 @@ pub fn spawn_background_test_gen(
|
|||||||
.with_system_prompt(prompt)
|
.with_system_prompt(prompt)
|
||||||
;
|
;
|
||||||
|
|
||||||
let result = run_subagent_with_retry(&def, &sd, &ws, "bg-test-gen");
|
let result = run_subagent_with_retry(&def, &sd, &ws, "bg-test-gen", Some(&abort_flag));
|
||||||
let message = match &result {
|
let message = match &result {
|
||||||
Ok(output) => {
|
Ok(output) => {
|
||||||
let first = output.lines().next().unwrap_or(output);
|
let first = output.lines().next().unwrap_or(output);
|
||||||
format!("Auto test-gen: {first}")
|
format!("Auto test-gen: {first}")
|
||||||
}
|
}
|
||||||
|
Err(e) if e.contains("aborted") => format!("Auto test-gen cancelled: {e}"),
|
||||||
Err(e) => format!("ESCALATED: Auto test-gen {e}"),
|
Err(e) => format!("ESCALATED: Auto test-gen {e}"),
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -247,15 +329,24 @@ pub fn spawn_background_test_gen(
|
|||||||
/// Inspects the modified files for architectural consistency (layering,
|
/// Inspects the modified files for architectural consistency (layering,
|
||||||
/// coupling, module boundaries). Reports via
|
/// coupling, module boundaries). Reports via
|
||||||
/// `TurnEvent::SystemNote { kind: "bg-arch-review" }`.
|
/// `TurnEvent::SystemNote { kind: "bg-arch-review" }`.
|
||||||
|
///
|
||||||
|
/// Skipped (no-op) if an arch-review run is already in flight (guarded by
|
||||||
|
/// `ARCH_REVIEW_RUNNING`). `abort_flag` is forwarded to
|
||||||
|
/// `run_subagent_with_retry` so the run can be cancelled if the turn aborts.
|
||||||
pub fn spawn_background_arch_review(
|
pub fn spawn_background_arch_review(
|
||||||
file_paths: &[String],
|
file_paths: &[String],
|
||||||
session_dir: &Path,
|
session_dir: &Path,
|
||||||
workspaces: &[std::path::PathBuf],
|
workspaces: &[std::path::PathBuf],
|
||||||
turn_events: &Arc<Mutex<VecDeque<TurnEvent>>>,
|
turn_events: &Arc<Mutex<VecDeque<TurnEvent>>>,
|
||||||
|
abort_flag: Arc<AtomicBool>,
|
||||||
) {
|
) {
|
||||||
if file_paths.is_empty() {
|
if file_paths.is_empty() {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
if ARCH_REVIEW_RUNNING.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst).is_err() {
|
||||||
|
tracing::debug!("[bg-arch-review] skipped — an arch-review run is already in flight");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
let paths = file_paths.to_vec();
|
let paths = file_paths.to_vec();
|
||||||
let sd = session_dir.to_path_buf();
|
let sd = session_dir.to_path_buf();
|
||||||
@@ -263,6 +354,7 @@ pub fn spawn_background_arch_review(
|
|||||||
let events = turn_events.clone();
|
let events = turn_events.clone();
|
||||||
|
|
||||||
std::thread::spawn(move || {
|
std::thread::spawn(move || {
|
||||||
|
let _running_guard = RunningGuard(&ARCH_REVIEW_RUNNING);
|
||||||
let file_list = paths.join("\n");
|
let file_list = paths.join("\n");
|
||||||
let prompt = format!(
|
let prompt = format!(
|
||||||
"{}\n\nModified files for architecture review:\n{}",
|
"{}\n\nModified files for architecture review:\n{}",
|
||||||
@@ -277,12 +369,13 @@ pub fn spawn_background_arch_review(
|
|||||||
.with_system_prompt(prompt)
|
.with_system_prompt(prompt)
|
||||||
;
|
;
|
||||||
|
|
||||||
let result = run_subagent_with_retry(&def, &sd, &ws, "bg-arch-review");
|
let result = run_subagent_with_retry(&def, &sd, &ws, "bg-arch-review", Some(&abort_flag));
|
||||||
let message = match &result {
|
let message = match &result {
|
||||||
Ok(output) => {
|
Ok(output) => {
|
||||||
let first = output.lines().next().unwrap_or(output);
|
let first = output.lines().next().unwrap_or(output);
|
||||||
format!("Architecture review: {first}")
|
format!("Architecture review: {first}")
|
||||||
}
|
}
|
||||||
|
Err(e) if e.contains("aborted") => format!("Architecture review cancelled: {e}"),
|
||||||
Err(e) => format!("ESCALATED: Architecture review {e}"),
|
Err(e) => format!("ESCALATED: Architecture review {e}"),
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -299,11 +392,16 @@ pub fn spawn_background_arch_review(
|
|||||||
///
|
///
|
||||||
/// Checks modified files for security vulnerabilities. Reports via
|
/// Checks modified files for security vulnerabilities. Reports via
|
||||||
/// `TurnEvent::SystemNote { kind: "bg-security-review" }`.
|
/// `TurnEvent::SystemNote { kind: "bg-security-review" }`.
|
||||||
|
///
|
||||||
|
/// Skipped (no-op) if a security-review run is already in flight (guarded by
|
||||||
|
/// `SECURITY_REVIEW_RUNNING`). `abort_flag` is forwarded to
|
||||||
|
/// `run_subagent_with_retry` so the run can be cancelled if the turn aborts.
|
||||||
pub fn spawn_background_security_review(
|
pub fn spawn_background_security_review(
|
||||||
file_paths: &[String],
|
file_paths: &[String],
|
||||||
session_dir: &Path,
|
session_dir: &Path,
|
||||||
workspaces: &[std::path::PathBuf],
|
workspaces: &[std::path::PathBuf],
|
||||||
turn_events: &Arc<Mutex<VecDeque<TurnEvent>>>,
|
turn_events: &Arc<Mutex<VecDeque<TurnEvent>>>,
|
||||||
|
abort_flag: Arc<AtomicBool>,
|
||||||
) {
|
) {
|
||||||
if file_paths.is_empty() {
|
if file_paths.is_empty() {
|
||||||
return;
|
return;
|
||||||
@@ -320,6 +418,10 @@ pub fn spawn_background_security_review(
|
|||||||
if prod_paths.is_empty() {
|
if prod_paths.is_empty() {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
if SECURITY_REVIEW_RUNNING.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst).is_err() {
|
||||||
|
tracing::debug!("[bg-security-review] skipped — a security-review run is already in flight");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
let paths = prod_paths;
|
let paths = prod_paths;
|
||||||
let sd = session_dir.to_path_buf();
|
let sd = session_dir.to_path_buf();
|
||||||
@@ -327,6 +429,7 @@ pub fn spawn_background_security_review(
|
|||||||
let events = turn_events.clone();
|
let events = turn_events.clone();
|
||||||
|
|
||||||
std::thread::spawn(move || {
|
std::thread::spawn(move || {
|
||||||
|
let _running_guard = RunningGuard(&SECURITY_REVIEW_RUNNING);
|
||||||
let file_list = paths.join("\n");
|
let file_list = paths.join("\n");
|
||||||
let prompt = format!(
|
let prompt = format!(
|
||||||
"{}\n\nModified files for security review:\n{}",
|
"{}\n\nModified files for security review:\n{}",
|
||||||
@@ -341,12 +444,13 @@ pub fn spawn_background_security_review(
|
|||||||
.with_system_prompt(prompt)
|
.with_system_prompt(prompt)
|
||||||
;
|
;
|
||||||
|
|
||||||
let result = run_subagent_with_retry(&def, &sd, &ws, "bg-security-review");
|
let result = run_subagent_with_retry(&def, &sd, &ws, "bg-security-review", Some(&abort_flag));
|
||||||
let message = match &result {
|
let message = match &result {
|
||||||
Ok(output) => {
|
Ok(output) => {
|
||||||
let first = output.lines().next().unwrap_or(output);
|
let first = output.lines().next().unwrap_or(output);
|
||||||
format!("Security review: {first}")
|
format!("Security review: {first}")
|
||||||
}
|
}
|
||||||
|
Err(e) if e.contains("aborted") => format!("Security review cancelled: {e}"),
|
||||||
Err(e) => format!("ESCALATED: Security review {e}"),
|
Err(e) => format!("ESCALATED: Security review {e}"),
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -365,11 +469,15 @@ pub fn spawn_background_security_review(
|
|||||||
/// Flow: always spawns arch-review and security-review if there are
|
/// Flow: always spawns arch-review and security-review if there are
|
||||||
/// reviewable production files → spawns test-gen only if there are source
|
/// reviewable production files → spawns test-gen only if there are source
|
||||||
/// files that aren't already tests.
|
/// files that aren't already tests.
|
||||||
|
///
|
||||||
|
/// `abort_flag` is cloned and forwarded to all three spawn calls so a
|
||||||
|
/// single cancellation source stops every kind of background review.
|
||||||
pub fn spawn_all_background(
|
pub fn spawn_all_background(
|
||||||
file_paths: &[String],
|
file_paths: &[String],
|
||||||
session_dir: &Path,
|
session_dir: &Path,
|
||||||
workspaces: &[std::path::PathBuf],
|
workspaces: &[std::path::PathBuf],
|
||||||
turn_events: &Arc<Mutex<VecDeque<TurnEvent>>>,
|
turn_events: &Arc<Mutex<VecDeque<TurnEvent>>>,
|
||||||
|
abort_flag: Arc<AtomicBool>,
|
||||||
) {
|
) {
|
||||||
if file_paths.is_empty() {
|
if file_paths.is_empty() {
|
||||||
return;
|
return;
|
||||||
@@ -381,7 +489,7 @@ pub fn spawn_all_background(
|
|||||||
.filter(|p| is_production_code(p))
|
.filter(|p| is_production_code(p))
|
||||||
.cloned()
|
.cloned()
|
||||||
.collect();
|
.collect();
|
||||||
spawn_background_test_gen(&source_paths, session_dir, workspaces, turn_events);
|
spawn_background_test_gen(&source_paths, session_dir, workspaces, turn_events, abort_flag.clone());
|
||||||
|
|
||||||
// Background arch review: for all files that are reviewable
|
// Background arch review: for all files that are reviewable
|
||||||
let reviewable: Vec<String> = file_paths
|
let reviewable: Vec<String> = file_paths
|
||||||
@@ -389,8 +497,73 @@ pub fn spawn_all_background(
|
|||||||
.filter(|p| is_reviewable_path(p))
|
.filter(|p| is_reviewable_path(p))
|
||||||
.cloned()
|
.cloned()
|
||||||
.collect();
|
.collect();
|
||||||
spawn_background_arch_review(&reviewable, session_dir, workspaces, turn_events);
|
spawn_background_arch_review(&reviewable, session_dir, workspaces, turn_events, abort_flag.clone());
|
||||||
|
|
||||||
// Background security review: only production source files
|
// Background security review: only production source files
|
||||||
spawn_background_security_review(&source_paths, session_dir, workspaces, turn_events);
|
spawn_background_security_review(&source_paths, session_dir, workspaces, turn_events, abort_flag);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn reviewable_path_skips_lockfiles_and_known_extensions() {
|
||||||
|
assert!(!is_reviewable_path("Cargo.lock"));
|
||||||
|
assert!(!is_reviewable_path("package.json"));
|
||||||
|
assert!(!is_reviewable_path("logo.svg"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn reviewable_path_skips_vendored_and_generated_dirs() {
|
||||||
|
assert!(!is_reviewable_path("target/debug/build.rs"));
|
||||||
|
assert!(!is_reviewable_path("node_modules/foo/index.js"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn reviewable_path_accepts_ordinary_source_files() {
|
||||||
|
assert!(is_reviewable_path("src/main.rs"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn production_code_excludes_dedicated_test_directories() {
|
||||||
|
assert!(!is_production_code("src/tests/foo.rs"));
|
||||||
|
assert!(!is_production_code("__tests__/baz.test.ts"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn production_code_excludes_test_filename_conventions() {
|
||||||
|
assert!(!is_production_code("src/foo_test.rs"));
|
||||||
|
assert!(!is_production_code("src/test_foo.py"));
|
||||||
|
assert!(!is_production_code("src/foo.spec.ts"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn production_code_does_not_false_positive_on_substring_test() {
|
||||||
|
// Regression: a plain `.contains("test")` would wrongly exclude
|
||||||
|
// these legitimate production files.
|
||||||
|
assert!(is_production_code("src/attestation.rs"));
|
||||||
|
assert!(is_production_code("src/latest/foo.rs"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn production_code_requires_known_source_extension() {
|
||||||
|
assert!(!is_production_code("README.md"));
|
||||||
|
assert!(is_production_code("src/main.rs"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn running_guard_resets_flag_on_drop_even_after_panic() {
|
||||||
|
static TEST_FLAG: AtomicBool = AtomicBool::new(false);
|
||||||
|
TEST_FLAG.store(true, Ordering::SeqCst);
|
||||||
|
let result = std::panic::catch_unwind(|| {
|
||||||
|
let _guard = RunningGuard(&TEST_FLAG);
|
||||||
|
panic!("simulated failure inside guarded region");
|
||||||
|
});
|
||||||
|
assert!(result.is_err());
|
||||||
|
assert!(
|
||||||
|
!TEST_FLAG.load(Ordering::SeqCst),
|
||||||
|
"guard must reset the flag even when the guarded closure panics"
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -88,4 +88,14 @@ mod tests {
|
|||||||
assert!(!tools.contains(&"write".to_string()));
|
assert!(!tools.contains(&"write".to_string()));
|
||||||
assert!(!tools.contains(&"delete".to_string()));
|
assert!(!tools.contains(&"delete".to_string()));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn read_tier_is_subset_of_write_tier_and_write_is_subset_of_full() {
|
||||||
|
use std::collections::HashSet;
|
||||||
|
let read: HashSet<_> = tools_for(READ).into_iter().collect();
|
||||||
|
let write: HashSet<_> = tools_for(WRITE).into_iter().collect();
|
||||||
|
let full: HashSet<_> = tools_for(FULL).into_iter().collect();
|
||||||
|
assert!(read.is_subset(&write), "read tier must be a subset of write tier");
|
||||||
|
assert!(write.is_subset(&full), "write tier must be a subset of full tier");
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -69,6 +69,27 @@ pub struct NodeReport {
|
|||||||
pub output: String,
|
pub output: String,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Tag prefixing the system message `run_hive_mind`'s caller pushes into
|
||||||
|
/// the conversation after a successful convergence. Shared between the
|
||||||
|
/// push site (`actions/mod.rs`) and `hive_mind_already_ran` below so the
|
||||||
|
/// two can never drift out of sync.
|
||||||
|
pub const HIVE_MIND_CONSENSUS_TAG: &str = "[Hive-Mind Consensus]";
|
||||||
|
|
||||||
|
/// Detect whether a hive-mind convergence has already run earlier in this
|
||||||
|
/// conversation, by checking prior system-message bodies for the
|
||||||
|
/// consensus tag.
|
||||||
|
///
|
||||||
|
/// Why: gates re-triggering the Core Intelligence pipeline more than once
|
||||||
|
/// per session on message *content* actually observed, rather than an
|
||||||
|
/// arbitrary "first two user messages" cutoff that silently disabled the
|
||||||
|
/// pipeline for any complex request phrased later in a long conversation.
|
||||||
|
///
|
||||||
|
/// Return: `true` if any prior system message starts with
|
||||||
|
/// `HIVE_MIND_CONSENSUS_TAG`.
|
||||||
|
pub fn hive_mind_already_ran<'a>(system_message_bodies: impl Iterator<Item = &'a str>) -> bool {
|
||||||
|
system_message_bodies.into_iter().any(|body| body.starts_with(HIVE_MIND_CONSENSUS_TAG))
|
||||||
|
}
|
||||||
|
|
||||||
/// Build the live-state callback that forwards node status updates to the
|
/// Build the live-state callback that forwards node status updates to the
|
||||||
/// TUI's workflow panel.
|
/// TUI's workflow panel.
|
||||||
fn build_live(
|
fn build_live(
|
||||||
@@ -105,10 +126,19 @@ fn build_live(
|
|||||||
/// directive is to reconcile the complete collective state into a single
|
/// directive is to reconcile the complete collective state into a single
|
||||||
/// consensus, not list what each node said.
|
/// consensus, not list what each node said.
|
||||||
///
|
///
|
||||||
/// Return: `(consensus, all_node_reports)`. `consensus` is the synthesis
|
/// Concurrency per cycle and the per-node timeout both come from
|
||||||
/// node's reconciled output — what the Core Intelligence actually
|
/// `Settings::load()` (`workflow_max_concurrency`, `hive_mind_node_timeout_ms`)
|
||||||
/// receives. `all_node_reports` is the complete per-node record,
|
/// rather than a hardcoded cap/no-timeout — a stuck node can no longer hang
|
||||||
/// persisted verbatim to `docs/runs/*.md`.
|
/// the whole convergence forever.
|
||||||
|
///
|
||||||
|
/// Return: `(consensus, all_node_reports)` on success. `consensus` is the
|
||||||
|
/// synthesis node's reconciled output — what the Core Intelligence
|
||||||
|
/// actually receives. `all_node_reports` is the complete per-node record.
|
||||||
|
///
|
||||||
|
/// The convergence doc under `docs/runs/*.md` is written unconditionally
|
||||||
|
/// before this function returns — even when synthesis itself fails — so a
|
||||||
|
/// synthesis-node error never discards the work already done by cycle
|
||||||
|
/// nodes. Callers must not write their own copy of this doc.
|
||||||
pub fn run_hive_mind(
|
pub fn run_hive_mind(
|
||||||
user_request: &str,
|
user_request: &str,
|
||||||
plan: &CognitiveCyclePlan,
|
plan: &CognitiveCyclePlan,
|
||||||
@@ -121,6 +151,10 @@ pub fn run_hive_mind(
|
|||||||
anyhow::bail!("cognitive cycle plan has no cycles");
|
anyhow::bail!("cognitive cycle plan has no cycles");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
let settings = crate::model::settings::Settings::load();
|
||||||
|
let node_timeout_ms = Some(settings.hive_mind_node_timeout_ms);
|
||||||
|
let max_cycle_concurrency = settings.workflow_max_concurrency.max(1);
|
||||||
|
|
||||||
let live = build_live(turn_events);
|
let live = build_live(turn_events);
|
||||||
let collective_state: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
|
let collective_state: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
|
||||||
let args: HashMap<String, String> = HashMap::new();
|
let args: HashMap<String, String> = HashMap::new();
|
||||||
@@ -162,14 +196,14 @@ pub fn run_hive_mind(
|
|||||||
let results = execute_primitive(
|
let results = execute_primitive(
|
||||||
&cycle_primitive,
|
&cycle_primitive,
|
||||||
&args,
|
&args,
|
||||||
directives.len().clamp(1, 10),
|
directives.len().clamp(1, max_cycle_concurrency),
|
||||||
true,
|
true,
|
||||||
&abort_owned,
|
&abort_owned,
|
||||||
live.as_ref(),
|
live.as_ref(),
|
||||||
session_dir,
|
session_dir,
|
||||||
workspaces,
|
workspaces,
|
||||||
&collective_state,
|
&collective_state,
|
||||||
None,
|
node_timeout_ms,
|
||||||
)?;
|
)?;
|
||||||
|
|
||||||
// engine::execute_primitive's ScopedAgent arm already merged each
|
// engine::execute_primitive's ScopedAgent arm already merged each
|
||||||
@@ -185,9 +219,29 @@ pub fn run_hive_mind(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
let consensus = synthesize_consensus(
|
let consensus_result = synthesize_consensus(
|
||||||
user_request, session_dir, workspaces, &collective_state, live.as_ref(), abort_flag,
|
user_request, session_dir, workspaces, &collective_state, live.as_ref(), abort_flag, node_timeout_ms,
|
||||||
)?;
|
);
|
||||||
|
|
||||||
|
// Guaranteed documentation: write the convergence doc for whatever
|
||||||
|
// reports/consensus we actually have, whether synthesis succeeded or
|
||||||
|
// failed. A synthesis-node failure must not silently discard every
|
||||||
|
// completed cycle node's work — this is the durable audit trail
|
||||||
|
// CLAUDE.md promises for every convergence.
|
||||||
|
let doc_consensus = match &consensus_result {
|
||||||
|
Ok(c) => c.clone(),
|
||||||
|
Err(e) => format!(
|
||||||
|
"Synthesis failed: {e}. See individual node reports above for partial results.",
|
||||||
|
),
|
||||||
|
};
|
||||||
|
if let Some(workspace_root) = workspaces.first() {
|
||||||
|
match crate::app::workflow::docs::write_hive_mind_convergence(workspace_root, user_request, &reports, &doc_consensus) {
|
||||||
|
Ok(path) => tracing::info!("[hive-mind] convergence documented at {}", path.display()),
|
||||||
|
Err(e) => tracing::warn!("[hive-mind] failed to write docs/runs report: {e}"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let consensus = consensus_result?;
|
||||||
Ok((consensus, reports))
|
Ok((consensus, reports))
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -200,6 +254,9 @@ pub fn run_hive_mind(
|
|||||||
/// reasoning can reconcile that into a coherent answer; deterministic
|
/// reasoning can reconcile that into a coherent answer; deterministic
|
||||||
/// formatting can only concatenate, not resolve conflicts.
|
/// formatting can only concatenate, not resolve conflicts.
|
||||||
///
|
///
|
||||||
|
/// `node_timeout_ms` is forwarded from `run_hive_mind`'s `Settings::load()`
|
||||||
|
/// read so the synthesis node is bound by the same deadline as cycle nodes.
|
||||||
|
///
|
||||||
/// Return: the synthesis node's reconciled consensus text.
|
/// Return: the synthesis node's reconciled consensus text.
|
||||||
fn synthesize_consensus(
|
fn synthesize_consensus(
|
||||||
user_request: &str,
|
user_request: &str,
|
||||||
@@ -208,6 +265,7 @@ fn synthesize_consensus(
|
|||||||
collective_state: &Arc<Mutex<Vec<String>>>,
|
collective_state: &Arc<Mutex<Vec<String>>>,
|
||||||
live: Option<&LiveStateFn>,
|
live: Option<&LiveStateFn>,
|
||||||
abort_flag: Option<&Arc<AtomicBool>>,
|
abort_flag: Option<&Arc<AtomicBool>>,
|
||||||
|
node_timeout_ms: Option<u64>,
|
||||||
) -> anyhow::Result<String> {
|
) -> anyhow::Result<String> {
|
||||||
let synthesis = ScriptPrimitive::ScopedAgent {
|
let synthesis = ScriptPrimitive::ScopedAgent {
|
||||||
prompt: format!(
|
prompt: format!(
|
||||||
@@ -227,7 +285,7 @@ fn synthesize_consensus(
|
|||||||
let args: HashMap<String, String> = HashMap::new();
|
let args: HashMap<String, String> = HashMap::new();
|
||||||
let abort_owned: Option<Arc<AtomicBool>> = abort_flag.cloned();
|
let abort_owned: Option<Arc<AtomicBool>> = abort_flag.cloned();
|
||||||
let results = execute_primitive(
|
let results = execute_primitive(
|
||||||
&synthesis, &args, 1, false, &abort_owned, live, session_dir, workspaces, collective_state, None,
|
&synthesis, &args, 1, false, &abort_owned, live, session_dir, workspaces, collective_state, node_timeout_ms,
|
||||||
)?;
|
)?;
|
||||||
Ok(results.into_iter().next().unwrap_or_default())
|
Ok(results.into_iter().next().unwrap_or_default())
|
||||||
}
|
}
|
||||||
@@ -371,4 +429,19 @@ mod tests {
|
|||||||
let node_id = format!("Node-{}-{}", 2, 1);
|
let node_id = format!("Node-{}-{}", 2, 1);
|
||||||
assert_eq!(node_id, "Node-2-1");
|
assert_eq!(node_id, "Node-2-1");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn hive_mind_already_ran_detects_prior_consensus_tag() {
|
||||||
|
let bodies = vec![
|
||||||
|
"you are a helpful assistant".to_string(),
|
||||||
|
format!("{HIVE_MIND_CONSENSUS_TAG}\nthe bug is a null check"),
|
||||||
|
];
|
||||||
|
assert!(hive_mind_already_ran(bodies.iter().map(std::string::String::as_str)));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn hive_mind_already_ran_false_when_no_prior_convergence() {
|
||||||
|
let bodies = vec!["you are a helpful assistant".to_string()];
|
||||||
|
assert!(!hive_mind_already_ran(bodies.iter().map(std::string::String::as_str)));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+49
-1
@@ -19,7 +19,10 @@ pub enum InternetMode {
|
|||||||
Full,
|
Full,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Default per-node timeout for hive-mind nodes: 10 minutes.
|
||||||
|
fn default_hive_mind_node_timeout_ms() -> u64 {
|
||||||
|
600_000
|
||||||
|
}
|
||||||
|
|
||||||
/// Top-level application settings, serialized to `settings.json` in the store dir.
|
/// Top-level application settings, serialized to `settings.json` in the store dir.
|
||||||
///
|
///
|
||||||
@@ -42,6 +45,11 @@ pub struct Settings {
|
|||||||
pub session_archive_enabled: bool,
|
pub session_archive_enabled: bool,
|
||||||
pub lsp_auto_provision: bool,
|
pub lsp_auto_provision: bool,
|
||||||
pub lsp_languages: Vec<String>,
|
pub lsp_languages: Vec<String>,
|
||||||
|
/// Wall-clock deadline for a single hive-mind processing node (cycle
|
||||||
|
/// node or synthesis node). Prevents one stuck node from hanging an
|
||||||
|
/// entire hive-mind convergence forever.
|
||||||
|
#[serde(default = "default_hive_mind_node_timeout_ms")]
|
||||||
|
pub hive_mind_node_timeout_ms: u64,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Default for Settings {
|
impl Default for Settings {
|
||||||
@@ -62,6 +70,7 @@ impl Default for Settings {
|
|||||||
session_archive_enabled: true,
|
session_archive_enabled: true,
|
||||||
lsp_auto_provision: true,
|
lsp_auto_provision: true,
|
||||||
lsp_languages: Vec::new(),
|
lsp_languages: Vec::new(),
|
||||||
|
hive_mind_node_timeout_ms: default_hive_mind_node_timeout_ms(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -103,3 +112,42 @@ impl Settings {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn default_hive_mind_node_timeout_is_ten_minutes() {
|
||||||
|
let settings = Settings::default();
|
||||||
|
assert_eq!(settings.hive_mind_node_timeout_ms, 600_000);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn missing_hive_mind_node_timeout_field_falls_back_to_default() {
|
||||||
|
// Simulates loading a settings.json written before this field
|
||||||
|
// existed — #[serde(default = ...)] must fill it in rather than
|
||||||
|
// failing the whole parse (which would silently reset every
|
||||||
|
// other saved setting to default too).
|
||||||
|
let old_json = r#"{
|
||||||
|
"internet_mode": "Off",
|
||||||
|
"provider": "zen",
|
||||||
|
"model": "deepseek-v4-flash-free",
|
||||||
|
"api_keys": {},
|
||||||
|
"max_tokens": null,
|
||||||
|
"temperature": null,
|
||||||
|
"review_enabled": true,
|
||||||
|
"review_max_lessons_per_run": 5,
|
||||||
|
"adaptive_review_max_skip": 3,
|
||||||
|
"verify_command": null,
|
||||||
|
"verify_timeout_ms": 30000,
|
||||||
|
"workflow_max_concurrency": 5,
|
||||||
|
"session_archive_enabled": true,
|
||||||
|
"lsp_auto_provision": true,
|
||||||
|
"lsp_languages": []
|
||||||
|
}"#;
|
||||||
|
let parsed: Settings = serde_json::from_str(old_json)
|
||||||
|
.expect("must parse even without the new field present");
|
||||||
|
assert_eq!(parsed.hive_mind_node_timeout_ms, 600_000);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -2,6 +2,7 @@
|
|||||||
|
|
||||||
use std::path::PathBuf;
|
use std::path::PathBuf;
|
||||||
use std::sync::{Arc, Mutex};
|
use std::sync::{Arc, Mutex};
|
||||||
|
use std::sync::atomic::AtomicBool;
|
||||||
use serde_json::Value;
|
use serde_json::Value;
|
||||||
use anyhow::Result;
|
use anyhow::Result;
|
||||||
|
|
||||||
@@ -56,6 +57,11 @@ pub struct ToolCtx {
|
|||||||
/// reference earlier results. `None` means "not inside a workflow" —
|
/// reference earlier results. `None` means "not inside a workflow" —
|
||||||
/// `note_finding` becomes a no-op.
|
/// `note_finding` becomes a no-op.
|
||||||
pub workflow_findings: Option<Arc<Mutex<Vec<String>>>>,
|
pub workflow_findings: Option<Arc<Mutex<Vec<String>>>>,
|
||||||
|
/// The current turn's abort flag, threaded through so tools that
|
||||||
|
/// delegate to long-running orchestration (e.g. the `hive_mind` tool)
|
||||||
|
/// can be cancelled the same way the main agent loop is. `None` when
|
||||||
|
/// no turn-level abort flag is available.
|
||||||
|
pub abort_flag: Option<Arc<AtomicBool>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Find which graduated checks apply to a given file path/content pair.
|
/// Find which graduated checks apply to a given file path/content pair.
|
||||||
@@ -93,6 +99,7 @@ pub struct ToolCtxBuilder {
|
|||||||
pub lsp_manager: Arc<Mutex<crate::app::lsp::LspManager>>,
|
pub lsp_manager: Arc<Mutex<crate::app::lsp::LspManager>>,
|
||||||
pub turn_events: Option<Arc<Mutex<std::collections::VecDeque<crate::app::state::runtime::TurnEvent>>>>,
|
pub turn_events: Option<Arc<Mutex<std::collections::VecDeque<crate::app::state::runtime::TurnEvent>>>>,
|
||||||
pub workflow_findings: Option<Arc<Mutex<Vec<String>>>>,
|
pub workflow_findings: Option<Arc<Mutex<Vec<String>>>>,
|
||||||
|
pub abort_flag: Option<Arc<AtomicBool>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Default for ToolCtxBuilder {
|
impl Default for ToolCtxBuilder {
|
||||||
@@ -108,6 +115,7 @@ impl Default for ToolCtxBuilder {
|
|||||||
lsp_manager: Arc::new(Mutex::new(crate::app::lsp::LspManager::new())),
|
lsp_manager: Arc::new(Mutex::new(crate::app::lsp::LspManager::new())),
|
||||||
turn_events: None,
|
turn_events: None,
|
||||||
workflow_findings: None,
|
workflow_findings: None,
|
||||||
|
abort_flag: None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -135,6 +143,7 @@ impl ToolCtxBuilder {
|
|||||||
lsp_manager: self.lsp_manager,
|
lsp_manager: self.lsp_manager,
|
||||||
turn_events: self.turn_events,
|
turn_events: self.turn_events,
|
||||||
workflow_findings: self.workflow_findings,
|
workflow_findings: self.workflow_findings,
|
||||||
|
abort_flag: self.abort_flag,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -264,3 +273,14 @@ pub fn resolve_path(workspaces: &[PathBuf], rel: &str) -> Result<PathBuf> {
|
|||||||
anyhow::bail!("path '{rel}' is outside all workspace roots")
|
anyhow::bail!("path '{rel}' is outside all workspace roots")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn tool_ctx_builder_defaults_abort_flag_to_none() {
|
||||||
|
let ctx = ToolCtx::builder().build();
|
||||||
|
assert!(ctx.abort_flag.is_none());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -219,21 +219,18 @@ impl Tool for HiveMind {
|
|||||||
json!({ "cycles": cycles_value })
|
json!({ "cycles": cycles_value })
|
||||||
).map_err(|e| anyhow!("failed to parse cycles: {e}"))?;
|
).map_err(|e| anyhow!("failed to parse cycles: {e}"))?;
|
||||||
|
|
||||||
let (consensus, reports) = crate::app::workflow::hive_mind::run_hive_mind(
|
// run_hive_mind now writes the docs/runs/*.md convergence report
|
||||||
|
// itself (guaranteed, even if synthesis fails) — do not write it
|
||||||
|
// again here.
|
||||||
|
let (consensus, _reports) = crate::app::workflow::hive_mind::run_hive_mind(
|
||||||
request,
|
request,
|
||||||
&plan,
|
&plan,
|
||||||
&ctx.session_dir,
|
&ctx.session_dir,
|
||||||
&ctx.workspaces,
|
&ctx.workspaces,
|
||||||
ctx.turn_events.as_ref(),
|
ctx.turn_events.as_ref(),
|
||||||
None,
|
ctx.abort_flag.as_ref(),
|
||||||
)?;
|
)?;
|
||||||
|
|
||||||
if let Some(workspace_root) = ctx.workspaces.first() {
|
|
||||||
if let Err(e) = crate::app::workflow::docs::write_hive_mind_convergence(workspace_root, request, &reports, &consensus) {
|
|
||||||
tracing::warn!("[hive_mind] failed to write docs/runs report: {e}");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
Ok(consensus)
|
Ok(consensus)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user