fix(hive-mind): tambah timeout per-node dan jamin dokumentasi convergence tetap tertulis saat sintesis gagal

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
asepharyana
2026-07-14 10:55:05 +07:00
co-authored by Claude Sonnet 5
parent e2878a3d83
commit b1c0265e8c
+47 -10
View File
@@ -105,10 +105,19 @@ fn build_live(
/// directive is to reconcile the complete collective state into a single
/// consensus, not list what each node said.
///
/// Return: `(consensus, all_node_reports)`. `consensus` is the synthesis
/// node's reconciled output — what the Core Intelligence actually
/// receives. `all_node_reports` is the complete per-node record,
/// persisted verbatim to `docs/runs/*.md`.
/// Concurrency per cycle and the per-node timeout both come from
/// `Settings::load()` (`workflow_max_concurrency`, `hive_mind_node_timeout_ms`)
/// rather than a hardcoded cap/no-timeout — a stuck node can no longer hang
/// 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(
user_request: &str,
plan: &CognitiveCyclePlan,
@@ -121,6 +130,10 @@ pub fn run_hive_mind(
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 collective_state: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
let args: HashMap<String, String> = HashMap::new();
@@ -162,14 +175,14 @@ pub fn run_hive_mind(
let results = execute_primitive(
&cycle_primitive,
&args,
directives.len().clamp(1, 10),
directives.len().clamp(1, max_cycle_concurrency),
true,
&abort_owned,
live.as_ref(),
session_dir,
workspaces,
&collective_state,
None,
node_timeout_ms,
)?;
// engine::execute_primitive's ScopedAgent arm already merged each
@@ -185,9 +198,29 @@ pub fn run_hive_mind(
}
}
let consensus = synthesize_consensus(
user_request, session_dir, workspaces, &collective_state, live.as_ref(), abort_flag,
)?;
let consensus_result = synthesize_consensus(
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))
}
@@ -200,6 +233,9 @@ pub fn run_hive_mind(
/// reasoning can reconcile that into a coherent answer; deterministic
/// 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.
fn synthesize_consensus(
user_request: &str,
@@ -208,6 +244,7 @@ fn synthesize_consensus(
collective_state: &Arc<Mutex<Vec<String>>>,
live: Option<&LiveStateFn>,
abort_flag: Option<&Arc<AtomicBool>>,
node_timeout_ms: Option<u64>,
) -> anyhow::Result<String> {
let synthesis = ScriptPrimitive::ScopedAgent {
prompt: format!(
@@ -227,7 +264,7 @@ fn synthesize_consensus(
let args: HashMap<String, String> = HashMap::new();
let abort_owned: Option<Arc<AtomicBool>> = abort_flag.cloned();
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())
}