MythEclipse/zesdex
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(hive-mind): implement multi-agent orchestration with cognitive cycles

Browse Source 
- Introduced a new hive-mind architecture that allows the Core Intelligence to issue directives to anonymous processing nodes.
- Each node executes its directive and merges output into a collective state, visible to all nodes in real-time.
- Added support for dynamic cognitive cycles, enabling flexible task management.
- Implemented documentation generation for hive-mind runs, ensuring a durable record of decisions and actions.
- Refactored existing company pipeline tools to align with the new hive-mind structure, replacing division-specific prompts with a more generalized approach.
- Updated workflow rendering to accommodate hive-mind nodes and their system-assigned designations.
- Enhanced error handling and validation for cognitive cycle plans.
This commit is contained in:
asepharyana
2026-07-14 08:12:43 +07:00
parent b1e0dcae14
commit 25f084f9db
23 changed files with 895 additions and 1225 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/app/runtime/actions/mod.rs
+63 -79
View File
@@ -402,19 +402,21 @@ pub fn apply_action(state: &mut AppStateRest, action: Action) {
});
state.dirty = true;
} else if kind == "bg-test-gen" {
let escalated = message.starts_with("ESCALATED:");
state.push_toast(Toast {
kind: ToastKind::Info,
kind: if escalated { ToastKind::Error } else { ToastKind::Info },
message: message.clone(),
created_at: chrono::Utc::now().timestamp_millis(),
lifetime_ms: 8000,
lifetime_ms: if escalated { 30000 } else { 8000 },
});
state.dirty = true;
} else if kind == "bg-arch-review" || kind == "bg-security-review" {
let escalated = message.starts_with("ESCALATED:");
state.push_toast(Toast {
kind: ToastKind::Info,
kind: if escalated { ToastKind::Error } else { ToastKind::Info },
message: message.clone(),
created_at: chrono::Utc::now().timestamp_millis(),
lifetime_ms: 10000,
lifetime_ms: if escalated { 30000 } else { 10000 },
});
state.dirty = true;
} else if kind == "workflow_done" {
@@ -1002,7 +1004,7 @@ fn run_agent_turn(
if user_request.is_empty() {
false
} else {
crate::app::workflow::company::is_complex_request(user_request)
crate::app::workflow::hive_mind::is_complex_request(user_request)
}
} else {
false
@@ -1014,58 +1016,46 @@ fn run_agent_turn(
.and_then(|m| m.content.as_deref())
.unwrap_or("");
let use_full = true;
let mode_label = "full";
tracing::info!(
"[ceo] pipeline triggered (mode={}) — dynamically generating planning workflow via LLM",
mode_label
);
tracing::info!("[hive-mind] triggered — Core Intelligence compiling a cognitive cycle plan via LLM");
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::SystemNote {
kind: "pipeline".to_string(),
message: format!(
"CEO is planning workflow (mode={mode_label})...",
),
message: "Core Intelligence is compiling a cognitive cycle plan...".to_string(),
});
}
let pipeline_abort = Some(tc.abort_flag.clone());
// Ask LLM to dynamically generate the workflow specialists plan
let required_divisions = if use_full {
"all 5 divisions (Strategy, Engineering, Quality, Security, Documentation)"
} else {
"the 3 quick divisions (Strategy, Engineering, Quality)"
};
let example_json = if use_full {
r#"{
"Strategy": [ ["Architect", "Analyze component tree..."] ],
"Engineering": [ ["Developer", "Implement core algorithms..."] ],
"Quality": [ ["Tester", "Write unit tests..."] ],
"Security": [ ["Auditor", "Review dependencies..."] ],
"Documentation": [ ["Writer", "Document API endpoints..."] ]
}"#
} else {
r#"{
"Strategy": [ ["Architect", "Analyze component tree..."] ],
"Engineering": [ ["Developer", "Implement core algorithms..."] ],
"Quality": [ ["Tester", "Write unit tests..."] ]
}"#
};
// Ask the LLM to freely design its own hive: any number of cycles,
// each with any number of nodes, every node carrying only a
// directive and an access tier. Cycle count and shape are decided
// by the Core Intelligence per task.
let system_msg = ChatMessage::system(
"You are a professional software architect and workflow planner. \
Generate a tailored, structured multi-agent workflow specialists plan for the requested task. \
Do not explain. Return ONLY raw JSON matching the requested structure."
"You are the Core Intelligence of a distributed machine, compiling a cognitive \
cycle plan for a specific task. You spawn anonymous processing nodes; each node \
carries only a directive (what to do) and an access tier. Decide how many cycles \
and nodes-per-cycle are actually needed. Simple tasks might need one cycle with \
one node; large tasks might need several cycles with multiple nodes each. Cycles \
run sequentially; every node's complete output merges into the collective state \
the instant it finishes, automatically visible to all later cycles. Nodes within \
a cycle run in parallel. Do not explain. Return ONLY raw JSON matching the requested structure."
);
let user_msg = ChatMessage::user(format!(
"Design a structured multi-agent workflow plan for the following task:\n\n\
"Compile a cognitive cycle plan for the following task:\n\n\
\"{user_request}\"\n\n\
You must output a JSON object representing the 'specialists' configuration for {required_divisions}.\n\
Each division must have a list of custom specialists defined by a pair of [label, focus_description].\n\n\
Return ONLY a JSON object with this exact structure, with no markdown codeblocks and no explanation:\n\
{example_json}"
Return ONLY a JSON object of this exact shape, with no markdown codeblocks and no explanation:\n\
{{\n\
\x20 \"cycles\": [\n\
\x20 [\n\
\x20 {{ \"directive\": \"<what this node does>\", \"access\": \"read|write|full\" }}\n\
\x20 ]\n\
\x20 ]\n\
}}\n\n\
access: 'read' = investigation only, 'write' = read + edit/write/bash, \
'full' = write + delete/git_operator. Pick the narrowest access each node actually needs. \
Each node object has exactly two fields: directive and access, addressed only by \
its system-assigned designation."
));
let planner_result = tc.client.chat_with_tools_non_streaming(&[system_msg, user_msg], None);
@@ -1084,43 +1074,29 @@ fn run_agent_turn(
reply_text.to_string()
};
match serde_json::from_str::<std::collections::HashMap<String, Vec<(String, String)>>>(&clean_json) {
Ok(custom_specialists) => {
let spec_desc = custom_specialists.iter()
.map(|(k, v)| format!("{}: {} agents", k, v.len()))
match serde_json::from_str::<crate::app::workflow::hive_mind::CognitiveCyclePlan>(&clean_json) {
Ok(plan) => {
let cycle_desc = plan.cycles.iter()
.enumerate()
.map(|(i, nodes)| format!("cycle {i}: {} node(s)", nodes.len()))
.collect::<Vec<String>>()
.join(", ");
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::SystemNote {
kind: "pipeline".to_string(),
message: format!(
"CEO planned: Strategy → Engineering → Quality{}. (Config: {}) Running specialists...",
if use_full { " → Security → Documentation" } else { "" },
spec_desc
),
message: format!("Core Intelligence compiled {} cycle(s) — {cycle_desc}. Deploying nodes...", plan.cycles.len()),
});
}
if use_full {
crate::app::workflow::company::run_company_pipeline(
user_request,
&tc.edit_log_session_dir,
&tc.workspace_roots,
Some(events_q),
&pipeline_abort,
&custom_specialists,
)
} else {
crate::app::workflow::company::run_company_pipeline_quick(
user_request,
&tc.edit_log_session_dir,
&tc.workspace_roots,
Some(events_q),
&pipeline_abort,
&custom_specialists,
)
}
crate::app::workflow::hive_mind::run_hive_mind(
user_request,
&plan,
&tc.edit_log_session_dir,
&tc.workspace_roots,
Some(events_q),
pipeline_abort.as_ref(),
)
}
Err(e) => Err(anyhow::anyhow!("Failed to parse LLM planning JSON: {e}. Cleaned JSON was: {clean_json}")),
}
@@ -1129,10 +1105,18 @@ fn run_agent_turn(
};
match pipeline_result {
Ok(summary) => {
tracing::info!("[ceo] company pipeline completed successfully");
Ok((consensus, reports)) => {
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!(
"[Company Pipeline: {mode_label}]\n{summary}",
"[Hive-Mind Consensus]\n{consensus}",
));
archive_message(tc.db.as_ref(), &tc.session_id, &pipeline_msg);
msgs.push(pipeline_msg);
@@ -1140,14 +1124,14 @@ fn run_agent_turn(
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::SystemNote {
kind: "pipeline".to_string(),
message: format!("Company pipeline ({mode_label}) complete. CEO reviewing results..."),
message: "Hive-mind convergence complete. Core Intelligence reviewing consensus...".to_string(),
});
}
}
Err(e) => {
tracing::warn!("[ceo] company pipeline failed: {}", e);
tracing::warn!("[hive-mind] convergence failed: {}", e);
let fail_msg = ChatMessage::system(format!(
"[Pipeline Note] The company pipeline encountered issues: {e}.\n\
"[Pipeline Note] The hive-mind encountered issues: {e}.\n\
Proceeding with direct execution as fallback.",
));
msgs.push(fail_msg);
src/app/subagent/auto.rs
+46 -77
View File
@@ -143,6 +143,46 @@ pub fn spawn_quick_review(
/// ─── Background Subagent Spawners (async, report via `SystemNote`) ───
///
/// Run a subagent built from `def`, retrying once if the first attempt
/// fails. Background subagents call this instead of running once and
/// silently swallowing the error into a note string, so a single transient
/// LLM/tool failure doesn't just disappear.
///
/// Return: `Ok(output)` if either attempt succeeded, `Err(message)`
/// describing the final failure if both attempts failed.
fn run_subagent_with_retry(
def: &AgentDefinition,
session_dir: &Path,
workspaces: &[std::path::PathBuf],
label: &str,
) -> Result<String, String> {
let mut last_err = String::new();
for attempt in 1..=2 {
let mut ctx = build_subagent_context(def);
ctx.session_dir = session_dir.to_path_buf();
ctx.workspaces = workspaces.to_vec();
let (tx, mut rx) = tokio::sync::mpsc::channel(32);
let drain_label = label.to_string();
let _drain = std::thread::spawn(move || {
while let Some(event) = rx.blocking_recv() {
if let SubagentEvent::StepFailed { step, error } = &event {
tracing::warn!("[{drain_label}] step {step} failed: {error}");
}
}
});
match run_subagent(&ctx, &tx) {
Ok(output) => return Ok(output),
Err(e) => {
tracing::warn!("[{label}] attempt {attempt}/2 failed: {e}");
last_err = e.to_string();
}
}
}
Err(format!("failed after 2 attempts: {last_err}"))
}
/// Spawn a background subagent that generates tests for modified files.
///
/// Uses the test-generator prompt and has read-write access so it can
@@ -184,40 +224,13 @@ pub fn spawn_background_test_gen(
.with_system_prompt(prompt)
;
let mut ctx = build_subagent_context(&def);
ctx.session_dir = sd;
ctx.workspaces = ws;
let (tx, mut rx) = tokio::sync::mpsc::channel(32);
let _drain = std::thread::spawn(move || {
while let Some(event) = rx.blocking_recv() {
match &event {
SubagentEvent::ToolCall { tool, .. } => {
tracing::debug!("[bg-test-gen] tool: {}", tool);
}
SubagentEvent::ToolResult { tool, .. } => {
tracing::debug!("[bg-test-gen] result: {}", tool);
}
SubagentEvent::StepCompleted { .. } => {
tracing::trace!("[bg-test-gen] step done");
}
SubagentEvent::StepFailed { step, error } => {
tracing::warn!("[bg-test-gen] step {} failed: {}", step, error);
}
SubagentEvent::Completed { .. } => {
tracing::debug!("[bg-test-gen] completed");
}
}
}
});
let result = run_subagent(&ctx, &tx);
let result = run_subagent_with_retry(&def, &sd, &ws, "bg-test-gen");
let message = match &result {
Ok(output) => {
let first = output.lines().next().unwrap_or(output);
format!("Auto test-gen: {first}")
}
Err(e) => format!("Auto test-gen failed: {e}"),
Err(e) => format!("ESCALATED: Auto test-gen {e}"),
};
if let Ok(mut q) = events.lock() {
@@ -264,35 +277,13 @@ pub fn spawn_background_arch_review(
.with_system_prompt(prompt)
;
let mut ctx = build_subagent_context(&def);
ctx.session_dir = sd;
ctx.workspaces = ws;
let (tx, mut rx) = tokio::sync::mpsc::channel(32);
let _drain = std::thread::spawn(move || {
while let Some(event) = rx.blocking_recv() {
match &event {
SubagentEvent::ToolCall { tool, .. } => {
tracing::debug!("[bg-arch] tool: {}", tool);
}
SubagentEvent::ToolResult { tool, .. } => {
tracing::debug!("[bg-arch] result: {}", tool);
}
SubagentEvent::Completed { .. } => {
tracing::debug!("[bg-arch] completed");
}
_ => {}
}
}
});
let result = run_subagent(&ctx, &tx);
let result = run_subagent_with_retry(&def, &sd, &ws, "bg-arch-review");
let message = match &result {
Ok(output) => {
let first = output.lines().next().unwrap_or(output);
format!("Architecture review: {first}")
}
Err(e) => format!("Architecture review failed: {e}"),
Err(e) => format!("ESCALATED: Architecture review {e}"),
};
if let Ok(mut q) = events.lock() {
@@ -350,35 +341,13 @@ pub fn spawn_background_security_review(
.with_system_prompt(prompt)
;
let mut ctx = build_subagent_context(&def);
ctx.session_dir = sd;
ctx.workspaces = ws;
let (tx, mut rx) = tokio::sync::mpsc::channel(32);
let _drain = std::thread::spawn(move || {
while let Some(event) = rx.blocking_recv() {
match &event {
SubagentEvent::ToolCall { tool, .. } => {
tracing::debug!("[bg-security] tool: {}", tool);
}
SubagentEvent::ToolResult { tool, .. } => {
tracing::debug!("[bg-security] result: {}", tool);
}
SubagentEvent::Completed { .. } => {
tracing::debug!("[bg-security] completed");
}
_ => {}
}
}
});
let result = run_subagent(&ctx, &tx);
let result = run_subagent_with_retry(&def, &sd, &ws, "bg-security-review");
let message = match &result {
Ok(output) => {
let first = output.lines().next().unwrap_or(output);
format!("Security review: {first}")
}
Err(e) => format!("Security review failed: {e}"),
Err(e) => format!("ESCALATED: Security review {e}"),
};
if let Ok(mut q) = events.lock() {
src/app/subagent/division.rs
+81 -227
View File
@@ -1,237 +1,91 @@
//! Company-style agent divisions: specialized subagent roles that form an
//! organizational hierarchy like a company.
//! Access tiers for the anonymous processing nodes spawned by the
//! hive-mind orchestrator (`app::workflow::hive_mind`).
//!
//! ```text
//! CEO (Main Agent)
//! ├── Strategy Division (planner) — architecture, diagrams, plan
//! ├── Engineering Division (coder) — implementation
//! ├── Quality Division (tester) — review, test
//! ├── Security Division (auditor) — security audit
//! └── Documentation Division (doc) — documentation
//! ```
//!
//! Each division has a specific role, tools, and system prompt tailored to
//! its function. The main agent (CEO) delegates work to divisions via
//! the company pipeline workflow.
//! Nodes have no persistent identity of their own — the Core Intelligence
//! addresses each one only by directive and access tier. Since node
//! designations are system-assigned coordinates rather than named roles,
//! tool access can't be a lookup table keyed by role name. Instead the
//! Core Intelligence picks one of these three tiers per node, matched to
//! what that node's specific directive needs — this keeps the Harness
//! gate meaningful while the node roster itself stays fully dynamic.
use crate::app::subagent::spawn::AgentDefinition;
/// The three tool-access tiers a hive-mind node can be granted.
pub mod tool_scope {
/// Read-only investigation: no file mutation, no shell, no VCS.
pub const READ: &str = "read";
/// Read-tier plus file mutation and non-destructive shell (tests/builds).
pub const WRITE: &str = "write";
/// Write-tier plus delete, git, and the remaining LSP actions.
pub const FULL: &str = "full";
/// Division roles — used as both the `role` field in `AgentDefinition`
/// and as the key for pipeline routing.
pub mod roles {
/// Strategy Division: plans architecture, creates diagrams, breaks down work.
pub const STRATEGY: &str = "planner";
/// Engineering Division: implements code per the plan.
pub const ENGINEERING: &str = "coder";
/// Quality Division: reviews implementation, writes tests.
pub const QUALITY: &str = "tester";
/// Security Division: audits for vulnerabilities.
pub const SECURITY: &str = "auditor";
/// Documentation Division: updates docs, README, inline documentation.
pub const DOCUMENTATION: &str = "documenter";
}
const READ_TOOLS: &[&str] = &[
"read", "grep", "glob", "search", "seqthink", "recall",
"lsp_connect", "lsp_diagnostics", "lsp_hover", "lsp_definition",
"lsp_references", "read_findings",
];
/// ─── Division Agent Definitions ───
///
/// Build the Strategy Division agent — chief architect and planner.
///
/// Tools: read-only (read, grep, glob, search, lsp, plan, seqthink, recall)
/// Role: never writes code; produces detailed plans with mermaid diagrams.
pub fn strategy_division() -> AgentDefinition {
AgentDefinition::new(
"strategy-division".to_string(),
roles::STRATEGY.to_string(),
)
.with_system_prompt(crate::resources::DIVISION_PLANNER_PROMPT.to_string())
.with_allowed_tools(vec![
"read".to_string(),
"grep".to_string(),
"glob".to_string(),
"search".to_string(),
"seqthink".to_string(),
"plan".to_string(),
"recall".to_string(),
"lsp_connect".to_string(),
"lsp_diagnostics".to_string(),
"lsp_hover".to_string(),
"lsp_definition".to_string(),
"lsp_references".to_string(),
"read_findings".to_string(),
])
}
const WRITE_TOOLS: &[&str] = &[
"read", "grep", "glob", "search", "seqthink", "recall",
"lsp_connect", "lsp_diagnostics", "lsp_hover", "lsp_definition",
"lsp_references", "read_findings",
"write", "edit", "bash", "todowrite", "todofinish", "remember",
];
/// Build the Engineering Division agent — implements code per the plan.
///
/// Tools: full access (all write/edit/bash/git/LSP tools)
/// Role: executes the strategy plan, one file at a time.
pub fn engineering_division() -> AgentDefinition {
AgentDefinition::new(
"engineering-division".to_string(),
roles::ENGINEERING.to_string(),
)
.with_system_prompt(crate::resources::DIVISION_IMPLEMENTER_PROMPT.to_string())
.with_allowed_tools(vec![
"read".to_string(),
"write".to_string(),
"edit".to_string(),
"delete".to_string(),
"bash".to_string(),
"grep".to_string(),
"glob".to_string(),
"git_operator".to_string(),
"seqthink".to_string(),
"lsp_connect".to_string(),
"lsp_diagnostics".to_string(),
"lsp_hover".to_string(),
"lsp_definition".to_string(),
"lsp_references".to_string(),
"lsp_completion".to_string(),
"lsp_disconnect".to_string(),
"todowrite".to_string(),
"todofinish".to_string(),
"read_findings".to_string(),
])
}
const FULL_TOOLS: &[&str] = &[
"read", "grep", "glob", "search", "seqthink", "recall",
"lsp_connect", "lsp_diagnostics", "lsp_hover", "lsp_definition",
"lsp_references", "read_findings",
"write", "edit", "bash", "todowrite", "todofinish", "remember",
"delete", "git_operator", "lsp_completion", "lsp_disconnect",
];
/// Build the Quality Division agent — reviews code and writes tests.
///
/// Tools: read, write, grep, glob, bash (for running tests), LSP, memory
/// Role: verifies correctness and creates/runs tests.
pub fn quality_division() -> AgentDefinition {
AgentDefinition::new(
"quality-division".to_string(),
roles::QUALITY.to_string(),
)
.with_system_prompt(crate::resources::DIVISION_TESTER_PROMPT.to_string())
.with_allowed_tools(vec![
"read".to_string(),
"write".to_string(),
"edit".to_string(),
"grep".to_string(),
"glob".to_string(),
"bash".to_string(),
"seqthink".to_string(),
"recall".to_string(),
"remember".to_string(),
"lsp_connect".to_string(),
"lsp_diagnostics".to_string(),
"lsp_hover".to_string(),
"lsp_definition".to_string(),
"lsp_references".to_string(),
"read_findings".to_string(),
])
}
/// Build the Security Division agent — security auditor.
///
/// Tools: read-only + search + memory
/// Role: audits implementation for vulnerabilities.
pub fn security_division() -> AgentDefinition {
AgentDefinition::new(
"security-division".to_string(),
roles::SECURITY.to_string(),
)
.with_system_prompt(crate::resources::SECURITY_REVIEWER_PROMPT.to_string())
.with_allowed_tools(vec![
"read".to_string(),
"grep".to_string(),
"glob".to_string(),
"search".to_string(),
"seqthink".to_string(),
"recall".to_string(),
"remember".to_string(),
"lsp_connect".to_string(),
"lsp_diagnostics".to_string(),
"lsp_hover".to_string(),
"lsp_definition".to_string(),
"lsp_references".to_string(),
"read_findings".to_string(),
])
}
/// Build the Documentation Division agent — documentation maintainer.
///
/// Tools: read, grep, glob, write, edit, memory
/// Role: updates README, inline docs, architecture docs.
pub fn documentation_division() -> AgentDefinition {
AgentDefinition::new(
"documentation-division".to_string(),
roles::DOCUMENTATION.to_string(),
)
.with_system_prompt(crate::resources::DIVISION_DOCUMENTER_PROMPT.to_string())
.with_allowed_tools(vec![
"read".to_string(),
"write".to_string(),
"edit".to_string(),
"grep".to_string(),
"glob".to_string(),
"recall".to_string(),
"remember".to_string(),
"read_findings".to_string(),
])
}
/// ─── Division Registry ───
///
/// A named division with its agent definition and display metadata.
#[derive(Debug, Clone)]
pub struct Division {
/// Display name for the division (e.g. "Strategy", "Engineering").
pub name: &'static str,
/// Role tag used for pipeline routing (matches `roles::*` constants).
#[allow(dead_code)]
pub role: &'static str,
/// One-line description of what this division does.
#[allow(dead_code)]
pub description: &'static str,
/// Agent definition with tools, prompt, and step budget.
pub agent_def: AgentDefinition,
}
impl Division {
pub fn new(
name: &'static str,
role: &'static str,
description: &'static str,
agent_def: AgentDefinition,
) -> Self {
Division { name, role, description, agent_def }
/// Resolve a tier name to its concrete tool allowlist.
///
/// Unrecognized scope strings fall back to `READ` — the least-privileged
/// tier — rather than silently granting broader access.
///
/// Return: an owned `Vec<String>` suitable for `AgentDefinition::with_allowed_tools`.
pub fn tools_for(scope: &str) -> Vec<String> {
let tools: &[&str] = match scope {
FULL => FULL_TOOLS,
WRITE => WRITE_TOOLS,
_ => READ_TOOLS,
};
tools.iter().map(|s| (*s).to_string()).collect()
}
}
/// Return all company divisions as an ordered list matching the pipeline flow:
/// Strategy → Engineering → Quality → Security → Documentation.
pub fn all_divisions() -> Vec<Division> {
vec![
Division::new(
"Strategy",
roles::STRATEGY,
"Architecture planning with diagrams and step-by-step breakdown",
strategy_division(),
),
Division::new(
"Engineering",
roles::ENGINEERING,
"Code implementation following the plan",
engineering_division(),
),
Division::new(
"Quality",
roles::QUALITY,
"Code review and comprehensive testing",
quality_division(),
),
Division::new(
"Security",
roles::SECURITY,
"Security vulnerability audit",
security_division(),
),
Division::new(
"Documentation",
roles::DOCUMENTATION,
"Documentation updates and maintenance",
documentation_division(),
),
]
#[cfg(test)]
mod tests {
use super::tool_scope::{tools_for, FULL, READ, WRITE};
#[test]
fn read_tier_excludes_write_tools() {
let tools = tools_for(READ);
assert!(!tools.contains(&"write".to_string()));
assert!(!tools.contains(&"bash".to_string()));
}
#[test]
fn write_tier_includes_bash_but_not_delete_or_git() {
let tools = tools_for(WRITE);
assert!(tools.contains(&"bash".to_string()));
assert!(tools.contains(&"write".to_string()));
assert!(!tools.contains(&"delete".to_string()));
assert!(!tools.contains(&"git_operator".to_string()));
}
#[test]
fn full_tier_includes_delete_and_git() {
let tools = tools_for(FULL);
assert!(tools.contains(&"delete".to_string()));
assert!(tools.contains(&"git_operator".to_string()));
}
#[test]
fn unknown_scope_falls_back_to_read() {
let tools = tools_for("bogus");
assert!(!tools.contains(&"write".to_string()));
assert!(!tools.contains(&"delete".to_string()));
}
}
src/app/subagent/engine.rs
+2 -2
View File
@@ -29,13 +29,13 @@ fn build_subagent_tools(allowed_tools: &[String]) -> (Vec<Box<dyn crate::tool::T
let all = all_tools();
let filtered: Vec<Box<dyn crate::tool::Tool>> = if allowed_tools.is_empty() {
all.into_iter()
.filter(|t| t.name() != "company_pipeline" && t.name() != "workflow_run")
.filter(|t| t.name() != "hive_mind" && t.name() != "workflow_run")
.collect()
} else {
all.into_iter()
.filter(|t| {
allowed_tools.contains(&t.name().to_string())
&& t.name() != "company_pipeline"
&& t.name() != "hive_mind"
&& t.name() != "workflow_run"
})
.collect()
src/app/workflow/company.rs
-406
View File
@@ -1,406 +0,0 @@
//! Company-style workflow orchestrator: runs the complete division pipeline
//! (Strategy → Engineering → [Quality || Security || Documentation] in parallel)
//! with findings flowing between stages, then returns a consolidated executive
//! summary to the CEO (main agent).
//!
//! Flow:
//! ```
//! CEO Main Agent
//! │ delegates to run_company_pipeline(request)
//! ▼
//! ┌──────────────────────────────────────────────────┐
//! │ Strategy Division — plan + mermaid diagrams │ (runs sequentially first)
//! └─────────────────────────┬────────────────────────┘
//! ▼
//! ┌──────────────────────────────────────────────────┐
//! │ Engineering Division — implement per plan │ (runs sequentially second)
//! └─────────────────────────┬────────────────────────┘
//! ▼
//! ┌────────────┼────────────┐
//! ▼ ▼ ▼
//! ┌───────────┐┌───────────┐┌───────────┐
//! │ Quality ││ Security ││ Docs │ (run concurrently in parallel)
//! └───────────┘└───────────┘└───────────┘
//! │ │ │
//! └────────────┼────────────┘
//! ▼
//! CEO Main Agent delivers consolidated summary to user
//! ```
use std::collections::HashMap;
use std::fmt::Write;
use std::sync::{Arc, Mutex, atomic::AtomicBool};
use crate::app::workflow::script::{ScriptPrimitive, ScriptOptions, WorkflowScript};
use crate::app::workflow::engine::{execute_primitive, LiveStateFn, AgentStatus};
use crate::app::subagent::division;
/// Construct the specialized agents for a division.
///
/// Flow: map division name to its specialization pool.
///
/// Return: a `Vec<ScriptPrimitive>` containing the specialist agents.
fn make_division_specialists(
div: &division::Division,
user_request: &str,
specs: &[(String, String)],
) -> Vec<ScriptPrimitive> {
let div_prompt = div.agent_def.system_prompt.as_deref().unwrap_or("");
specs
.iter()
.map(|(label, focus)| {
// Prepend [Division Name: Specialist Label] so the first 40 chars
// of the prompt become the agent_name in spawn_single_agent.
// We use quadruple curly braces `{{{{findings}}}}` so that Rust's `format!` formats it
// into `{{findings}}` in the output string, which `resolve_template` then recognizes
// and replaces.
let prompt = format!(
"[{}: {}]\n\n{}\n\n{}\n\nUser request: {}\n\nFindings from previous divisions:\n{{{{findings}}}}",
div.name,
label,
focus,
div_prompt,
user_request,
);
ScriptPrimitive::Agent(prompt)
})
.collect()
}
/// Construct a named Phase wrapper containing a Parallel block of division specialists.
///
/// Flow: construct division specialists → wrap in a `Parallel` primitive wrapper.
///
/// Return: a `ScriptPrimitive::Phase` wrapper.
fn make_division_phase(
div: &division::Division,
user_request: &str,
specs: &[(String, String)],
) -> ScriptPrimitive {
let specialists = make_division_specialists(div, user_request, specs);
ScriptPrimitive::Phase {
name: div.name.to_string(),
script: Box::new(ScriptPrimitive::Parallel(specialists)),
}
}
/// Run the full company-style pipeline for a given user request.
///
/// This orchestrates all five divisions, running Strategy and Engineering
/// sequentially, followed by Quality, Security, and Documentation in parallel.
///
/// Each division receives findings from all previous divisions, enabling
/// context to flow through the pipeline.
///
/// Returns a consolidated executive summary string.
#[allow(clippy::ref_option)]
pub fn run_company_pipeline(
user_request: &str,
session_dir: &std::path::Path,
workspaces: &[std::path::PathBuf],
turn_events: Option<&Arc<Mutex<std::collections::VecDeque<crate::app::state::runtime::TurnEvent>>>>,
abort_flag: &Option<Arc<AtomicBool>>,
custom_specialists: &HashMap<String, Vec<(String, String)>>,
) -> anyhow::Result<String> {
let divisions = division::all_divisions();
let strategy_specs = custom_specialists.get("Strategy")
.ok_or_else(|| anyhow::anyhow!("missing required division configuration: Strategy"))?;
let engineering_specs = custom_specialists.get("Engineering")
.ok_or_else(|| anyhow::anyhow!("missing required division configuration: Engineering"))?;
let quality_specs = custom_specialists.get("Quality")
.ok_or_else(|| anyhow::anyhow!("missing required division configuration: Quality"))?;
let security_specs = custom_specialists.get("Security")
.ok_or_else(|| anyhow::anyhow!("missing required division configuration: Security"))?;
let documentation_specs = custom_specialists.get("Documentation")
.ok_or_else(|| anyhow::anyhow!("missing required division configuration: Documentation"))?;
let strategy_phase = make_division_phase(&divisions[0], user_request, strategy_specs);
let engineering_phase = make_division_phase(&divisions[1], user_request, engineering_specs);
let quality_phase = make_division_phase(&divisions[2], user_request, quality_specs);
let security_phase = make_division_phase(&divisions[3], user_request, security_specs);
let documentation_phase = make_division_phase(&divisions[4], user_request, documentation_specs);
let parallel_divisions = ScriptPrimitive::Parallel(vec![
quality_phase,
security_phase,
documentation_phase,
]);
let pipeline_primitive = ScriptPrimitive::Pipeline(vec![
strategy_phase,
engineering_phase,
parallel_divisions,
]);
let wf = WorkflowScript {
name: "company-pipeline".to_string(),
description: "Company Pipeline: Strategy → Engineering → (Quality || Security || Documentation)".to_string(),
script: pipeline_primitive,
options: ScriptOptions {
max_concurrency: 10,
continue_on_error: true,
timeout_ms: None,
},
};
let live: Option<LiveStateFn> = turn_events.map(|events| {
let events = events.clone();
let f: LiveStateFn = Arc::new(move |_agent_id: String, agent_name: String, status: AgentStatus| {
let display_name = agent_name.chars().take(30).collect::<String>();
if let Ok(mut q) = events.lock() {
q.push_back(crate::app::state::runtime::TurnEvent::WorkflowAgentUpdate {
agent_id: display_name.clone(),
agent_name: display_name,
status,
});
}
});
f
});
let args: HashMap<String, String> = HashMap::new();
let findings: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
let results = execute_primitive(
&wf.script,
&args,
wf.options.max_concurrency,
true,
abort_flag,
live.as_ref(),
session_dir,
workspaces,
&findings,
None,
)?;
let all_findings = findings.lock()
.map(|f| f.clone())
.unwrap_or_default();
Ok(build_executive_summary(user_request, &results, &all_findings, &divisions, custom_specialists))
}
/// Run a quick company pipeline that skips non-essential divisions
/// for simple tasks. Flow: Strategy → Engineering → Quality.
///
/// This is for smaller tasks where security audit and full docs are overkill.
#[allow(clippy::ref_option)]
pub fn run_company_pipeline_quick(
user_request: &str,
session_dir: &std::path::Path,
workspaces: &[std::path::PathBuf],
turn_events: Option<&Arc<Mutex<std::collections::VecDeque<crate::app::state::runtime::TurnEvent>>>>,
abort_flag: &Option<Arc<AtomicBool>>,
custom_specialists: &HashMap<String, Vec<(String, String)>>,
) -> anyhow::Result<String> {
let divisions = division::all_divisions();
let quick_divisions = &divisions[..3];
let strategy_specs = custom_specialists.get("Strategy")
.ok_or_else(|| anyhow::anyhow!("missing required division configuration: Strategy"))?;
let engineering_specs = custom_specialists.get("Engineering")
.ok_or_else(|| anyhow::anyhow!("missing required division configuration: Engineering"))?;
let quality_specs = custom_specialists.get("Quality")
.ok_or_else(|| anyhow::anyhow!("missing required division configuration: Quality"))?;
let strategy_phase = make_division_phase(&quick_divisions[0], user_request, strategy_specs);
let engineering_phase = make_division_phase(&quick_divisions[1], user_request, engineering_specs);
let quality_phase = make_division_phase(&quick_divisions[2], user_request, quality_specs);
let pipeline_primitive = ScriptPrimitive::Pipeline(vec![
strategy_phase,
engineering_phase,
quality_phase,
]);
let wf = WorkflowScript {
name: "company-pipeline-quick".to_string(),
description: "Company Pipeline (quick): Strategy → Engineering → Quality".to_string(),
script: pipeline_primitive,
options: ScriptOptions {
max_concurrency: 10,
continue_on_error: true,
timeout_ms: None,
},
};
let live: Option<LiveStateFn> = turn_events.map(|events| {
let events = events.clone();
let f: LiveStateFn = Arc::new(move |_agent_id: String, agent_name: String, status: AgentStatus| {
let display_name = agent_name.chars().take(30).collect::<String>();
if let Ok(mut q) = events.lock() {
q.push_back(crate::app::state::runtime::TurnEvent::WorkflowAgentUpdate {
agent_id: display_name.clone(),
agent_name: display_name,
status,
});
}
});
f
});
let args: HashMap<String, String> = HashMap::new();
let findings: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
let results = execute_primitive(
&wf.script, &args, wf.options.max_concurrency, true,
abort_flag, live.as_ref(), session_dir, workspaces, &findings, None,
)?;
let all_findings = findings.lock()
.map(|f| f.clone())
.unwrap_or_default();
Ok(build_executive_summary(user_request, &results, &all_findings, quick_divisions, custom_specialists))
}
/// Build a compressed executive summary from pipeline results.
///
/// Flow: print user request header → for each division, fetch its specialist verdicts
/// → join with pipes → append findings count.
///
/// Why: keeps output brief to save context window space. Full results are accessible
/// to the CEO via findings.
///
/// Return: a formatted executive summary string.
fn build_executive_summary(
request: &str,
results: &[String],
findings: &[String],
divisions: &[division::Division],
custom_specialists: &HashMap<String, Vec<(String, String)>>,
) -> String {
let mut summary = String::new();
writeln!(summary, "Pipeline for: {request}").unwrap();
let mut start_index = 0;
for div in divisions {
let count = custom_specialists.get(div.name)
.map_or(0, Vec::len);
let mut division_verdicts = Vec::new();
for offset in 0..count {
if let Some(r) = results.get(start_index + offset) {
let first_line = r.lines().next().unwrap_or(r);
let trimmed = first_line.chars().take(40).collect::<String>();
division_verdicts.push(trimmed);
}
}
let verdict = if division_verdicts.is_empty() {
"—".to_string()
} else {
division_verdicts.join(" | ")
};
writeln!(summary, " {}: {}", div.name, verdict).unwrap();
start_index += count;
}
if !findings.is_empty() {
writeln!(summary, " Notes: {} cross-division finding(s)", findings.len()).unwrap();
}
summary
}
/// Determine whether a request is complex enough for the full pipeline
/// or can use the quick version.
///
/// Simple = single file, minor fix, quick lookup, config change.
/// Complex = new feature, multi-file refactor, architecture change.
///
/// Used by the auto-CEO pipeline trigger in `run_agent_turn` to decide
/// whether to delegate to the full company pipeline or handle directly.
///
/// Heuristics:
/// - Very short requests (< 10 chars) are never complex.
/// - Negative keywords (simple/trivial/typo/quick) skip the pipeline.
/// - Positive keywords (refactor/api/implement/architecture) trigger it.
/// - Multi-line or multi-sentence requests are more likely complex.
#[allow(dead_code)]
pub fn is_complex_request(request: &str) -> bool {
let trimmed = request.trim();
// Very short requests are never complex
if trimmed.len() < 10 {
return false;
}
// Single-line simple update patterns
let lower = trimmed.to_lowercase();
let negative_keywords = [
"simple", "trivial", "typo", "just a", "only a", "minor",
"quick", "tiny", "small fix", "rename", "nitpick",
"cosmetic", "formatting", "spelling", "grammar",
"bump", "version bump", "update comment",
];
if negative_keywords.iter().any(|k| lower.contains(k)) {
return false;
}
// Multi-line/multi-sentence → likely complex
let sentences = trimmed.split(['.', '!', '?'])
.filter(|s| !s.trim().is_empty())
.count();
if sentences >= 3 {
return true;
}
// Positive complexity keywords
let complexity_keywords = [
"refactor", "redesign", "architecture", "feature", "implement",
"migrate", "restructure", "rewrite", "new module", "new component",
"scaffold", "multi", "multiple files", "api", "endpoint",
"integration", "system", "workflow", "pipeline", "database",
"authentication", "authorization", "full stack",
];
complexity_keywords.iter().any(|k| lower.contains(k))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_is_complex_request_too_short() {
assert!(!is_complex_request("abc"));
}
#[test]
fn test_is_complex_request_simple_keywords() {
assert!(!is_complex_request("just a simple update to the readme"));
assert!(!is_complex_request("minor typo fix in main.rs"));
}
#[test]
fn test_is_complex_request_multi_sentence() {
assert!(is_complex_request("This is sentence one. This is sentence two. This is sentence three."));
}
#[test]
fn test_is_complex_request_complex_keywords() {
assert!(is_complex_request("implement user authentication endpoint"));
assert!(is_complex_request("refactor the whole engine module"));
}
#[test]
fn test_make_division_specialists_custom() {
let divisions = division::all_divisions();
let div = &divisions[0];
let mut custom = HashMap::new();
custom.insert(
"Strategy".to_string(),
vec![
("Custom Label".to_string(), "Custom Focus Description".to_string())
]
);
let specs = make_division_specialists(div, "Test Request", custom.get("Strategy").unwrap());
assert_eq!(specs.len(), 1);
if let ScriptPrimitive::Agent(prompt) = &specs[0] {
assert!(prompt.contains("Custom Label"));
assert!(prompt.contains("Custom Focus Description"));
} else {
panic!("Expected ScriptPrimitive::Agent");
}
}
}
src/app/workflow/docs.rs
+99
View File
@@ -0,0 +1,99 @@
//! Guaranteed, deterministic documentation output for hive-mind runs.
//!
//! Because cycles/directives are entirely Core-Intelligence-authored (see
//! `app::workflow::hive_mind`), it could in principle never plan a "write
//! docs" node for a given task. Durable documentation can't depend on that
//! choice, so this step is plain Rust — not an LLM call, not a cycle the
//! Core Intelligence can omit or reshape — and always runs after any
//! hive-mind convergence completes.
use std::path::{Path, PathBuf};
use std::fmt::Write as _;
use crate::app::workflow::hive_mind::NodeReport;
use crate::model::memory::Memory;
/// Write a markdown report of one hive-mind convergence to
/// `<workspace_root>/docs/runs/<timestamp>-<slug>.md`.
///
/// Flow: build a slug from the user request → format every `NodeReport`
/// (grouped by cycle) with its complete output (no truncation — this is
/// the durable record of what the hive actually decided and did) → append
/// the final reconciled `consensus` as its own section → create
/// `docs/runs/` if missing → write the file.
///
/// Return: the path written, so callers can log/reference it.
pub fn write_hive_mind_convergence(
workspace_root: &Path,
user_request: &str,
reports: &[NodeReport],
consensus: &str,
) -> anyhow::Result<PathBuf> {
let runs_dir = workspace_root.join("docs").join("runs");
std::fs::create_dir_all(&runs_dir)?;
let ts = chrono::Utc::now();
let slug = Memory::slugify(user_request).unwrap_or_else(|| "run".to_string());
let filename = format!("{}-{}.md", ts.format("%Y%m%d-%H%M%S"), slug);
let path = runs_dir.join(filename);
let content = render_report(user_request, ts.timestamp_millis(), reports, consensus);
std::fs::write(&path, content)?;
Ok(path)
}
/// Render a hive-mind convergence as a markdown document.
fn render_report(user_request: &str, ts_millis: i64, reports: &[NodeReport], consensus: &str) -> String {
let mut out = String::new();
writeln!(out, "# Hive-mind convergence: {user_request}").unwrap();
writeln!(out, "\nTimestamp (ms): {ts_millis}\n").unwrap();
let cycle_count = reports.iter().map(|r| r.cycle_index).max().map_or(0, |m| m + 1);
for cycle_index in 0..cycle_count {
writeln!(out, "## Cycle {cycle_index}\n").unwrap();
for r in reports.iter().filter(|r| r.cycle_index == cycle_index) {
writeln!(out, "### {}\n", r.node_id).unwrap();
writeln!(out, "{}\n", r.output).unwrap();
}
}
writeln!(out, "## Collective Consensus\n").unwrap();
writeln!(out, "{consensus}\n").unwrap();
out
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn writes_run_file_under_docs_runs() {
let tmp = std::env::temp_dir().join(format!("zesdex-docs-test-{}", uuid::Uuid::new_v4()));
std::fs::create_dir_all(&tmp).unwrap();
let reports = vec![
NodeReport { node_id: "Node-0-0".to_string(), cycle_index: 0, output: "found the bug".to_string() },
];
let path = write_hive_mind_convergence(&tmp, "fix the bug", &reports, "the bug is a null check").unwrap();
assert!(path.starts_with(tmp.join("docs").join("runs")));
let content = std::fs::read_to_string(&path).unwrap();
assert!(content.contains("fix the bug"));
assert!(content.contains("Node-0-0"));
assert!(content.contains("found the bug"));
assert!(content.contains("Collective Consensus"));
assert!(content.contains("the bug is a null check"));
std::fs::remove_dir_all(&tmp).ok();
}
#[test]
fn falls_back_to_generic_slug_for_unslugifiable_request() {
let tmp = std::env::temp_dir().join(format!("zesdex-docs-test-{}", uuid::Uuid::new_v4()));
std::fs::create_dir_all(&tmp).unwrap();
let path = write_hive_mind_convergence(&tmp, "???", &[], "").unwrap();
assert!(path.file_name().unwrap().to_str().unwrap().contains("run"));
std::fs::remove_dir_all(&tmp).ok();
}
}
src/app/workflow/engine.rs
+49 -29
View File
@@ -76,7 +76,7 @@ impl WorkflowEngine {
/// - `status`: the agent's lifecycle state and timing.
///
/// Callers should use `agent_id` as the stable key and `agent_name` for
/// display purposes (e.g. the division name in the company pipeline).
/// display purposes (e.g. a hive-mind node's designation, `"Node-0-1"`).
pub type LiveStateFn = Arc<dyn Fn(String, String, AgentStatus) + Send + Sync>;
/// Spawn a single synchronous subagent with the given prompt, passing it
@@ -103,6 +103,8 @@ fn spawn_single_agent(
agent_id: &str,
agent_name: &str,
prompt: &str,
role: &str,
allowed_tools: Option<Vec<String>>,
findings_snapshot: &[String],
findings: &Arc<Mutex<Vec<String>>>,
abort_flag: &Option<Arc<AtomicBool>>,
@@ -119,7 +121,7 @@ fn spawn_single_agent(
// Notify UI: this agent is now running.
// Pass both the unique agent_id (UUID for stable key) and agent_name
// (human-readable display name, e.g. division name).
// (human-readable display name, e.g. a hive-mind node designation).
if let Some(f) = live {
f(
agent_id.to_string(),
@@ -134,32 +136,7 @@ fn spawn_single_agent(
);
}
let mut role = "coder".to_string();
let mut allowed_tools = None;
if agent_name.contains("Strategy") {
let div_def = crate::app::subagent::division::strategy_division();
role = div_def.role;
allowed_tools = div_def.allowed_tools;
} else if agent_name.contains("Engineering") {
let div_def = crate::app::subagent::division::engineering_division();
role = div_def.role;
allowed_tools = div_def.allowed_tools;
} else if agent_name.contains("Quality") {
let div_def = crate::app::subagent::division::quality_division();
role = div_def.role;
allowed_tools = div_def.allowed_tools;
} else if agent_name.contains("Security") {
let div_def = crate::app::subagent::division::security_division();
role = div_def.role;
allowed_tools = div_def.allowed_tools;
} else if agent_name.contains("Documentation") {
let div_def = crate::app::subagent::division::documentation_division();
role = div_def.role;
allowed_tools = div_def.allowed_tools;
}
let mut def = AgentDefinition::new(agent_name.to_string(), role);
let mut def = AgentDefinition::new(agent_name.to_string(), role.to_string());
if let Some(tools) = allowed_tools {
def = def.with_allowed_tools(tools);
}
@@ -387,7 +364,7 @@ pub fn execute_primitive(
let resolved = resolve_template(prompt, &resolved_args);
let agent_id = uuid::Uuid::new_v4().to_string();
let agent_name = resolved.chars().take(40).collect::<String>();
match spawn_single_agent(&agent_id, &agent_name, &resolved, &findings_snapshot, findings, abort_flag, live, session_dir, workspaces, timeout_ms) {
match spawn_single_agent(&agent_id, &agent_name, &resolved, "coder", None, &findings_snapshot, findings, abort_flag, live, session_dir, workspaces, timeout_ms) {
Ok(text) => Ok(vec![text]),
Err(e) => {
if continue_on_error {
@@ -399,6 +376,49 @@ pub fn execute_primitive(
}
}
ScriptPrimitive::ScopedAgent { prompt, node_id, tool_scope } => {
let mut resolved_args = args.clone();
let findings_snapshot = findings.lock().map(|f| f.clone()).unwrap_or_default();
if !resolved_args.contains_key("findings") {
let formatted_findings = if findings_snapshot.is_empty() {
"None".to_string()
} else {
findings_snapshot
.iter()
.enumerate()
.map(|(i, f)| format!("{}. {}", i + 1, f))
.collect::<Vec<_>>()
.join("\n")
};
resolved_args.insert("findings".to_string(), formatted_findings);
}
let resolved = resolve_template(prompt, &resolved_args);
let agent_id = uuid::Uuid::new_v4().to_string();
let agent_name = format!("{node_id}: {}", resolved.chars().take(30).collect::<String>());
let allowed_tools = crate::app::subagent::division::tool_scope::tools_for(tool_scope);
match spawn_single_agent(&agent_id, &agent_name, &resolved, node_id, Some(allowed_tools), &findings_snapshot, findings, abort_flag, live, session_dir, workspaces, timeout_ms) {
Ok(text) => {
// Merge this node's complete output into the shared
// collective state the instant it finishes — not after
// the whole parallel cohort completes. Any sibling node
// still running (via read_findings) or any node spawned
// afterward sees this immediately, making the collective
// state genuinely continuous rather than batch-synced.
if let Ok(mut f) = findings.lock() {
f.push(format!("[{node_id}]: {text}"));
}
Ok(vec![text])
}
Err(e) => {
if continue_on_error {
Ok(vec![format!("agent error: {}", e)])
} else {
Err(e)
}
}
}
}
ScriptPrimitive::Parallel(scripts) => {
// All branches run concurrently, capped by semaphore.
// This is the primary advantage over single-turn chat: multiple
src/app/workflow/hive_mind.rs
+374
View File
@@ -0,0 +1,374 @@
//! Hive-mind multi-agent orchestration.
//!
//! Modeled on the "Machine Intelligence" archetype from sci-fi strategy
//! games (Stellaris et al.): the Core Intelligence (the main agent) issues
//! directives that spawn anonymous processing nodes, each carrying only a
//! directive and an access tier. Every node's complete output merges into
//! a single collective state the instant it finishes (see
//! `engine::execute_primitive`'s `ScopedAgent` arm), visible to every
//! other node still running or spawned afterward — continuously, not just
//! at cycle boundaries. When all cognitive cycles complete, one final
//! synthesis node reconciles the entire collective state into a single
//! consensus assessment.
//!
//! ```text
//! Core Intelligence
//! │ issues a CognitiveCyclePlan { cycles: [[NodeDirective, ...], ...] }
//! ▼
//! Cycle 0: Node-0-0, Node-0-1, ... (run in parallel; each merges into
//! │ the collective state the instant
//! │ it completes — not batched)
//! ▼
//! Cycle 1: ...
//! ▼
//! ...however many cycles the Core Intelligence decided this task needs...
//! ▼
//! Synthesis node reads the complete collective state and produces one
//! reconciled consensus — returned to the Core Intelligence and persisted
//! to docs/runs/*.md.
//! ```
use std::collections::HashMap;
use std::sync::{Arc, Mutex, atomic::{AtomicBool, Ordering}};
use serde::Deserialize;
use crate::app::workflow::script::ScriptPrimitive;
use crate::app::workflow::engine::{execute_primitive, LiveStateFn, AgentStatus};
/// One directive the Core Intelligence wants a node to execute within a
/// cognitive cycle. A node's sole identity is its directive and access tier.
#[derive(Debug, Clone, Deserialize)]
pub struct NodeDirective {
pub directive: String,
/// Access tier: "read" | "write" | "full". Defaults to "read" when
/// omitted; unrecognized values also fall back to "read" (see
/// `division::tool_scope::tools_for`).
#[serde(default = "default_access")]
pub access: String,
}
fn default_access() -> String {
crate::app::subagent::division::tool_scope::READ.to_string()
}
/// A Core-Intelligence-authored execution plan: an ordered list of
/// cognitive cycles, each cycle a list of node directives executed in
/// parallel. Cycle count and nodes-per-cycle are fully dynamic.
#[derive(Debug, Clone, Deserialize)]
pub struct CognitiveCyclePlan {
pub cycles: Vec<Vec<NodeDirective>>,
}
/// The complete output of one node within one cognitive cycle.
///
/// `node_id` is a system-assigned coordinate (e.g. `"Node-0-1"`) that
/// identifies a node purely by its position in the hive.
#[derive(Debug, Clone)]
pub struct NodeReport {
pub node_id: String,
pub cycle_index: usize,
pub output: String,
}
/// Build the live-state callback that forwards node status updates to the
/// TUI's workflow panel.
fn build_live(
turn_events: Option<&Arc<Mutex<std::collections::VecDeque<crate::app::state::runtime::TurnEvent>>>>,
) -> Option<LiveStateFn> {
turn_events.map(|events| {
let events = events.clone();
let f: LiveStateFn = Arc::new(move |_agent_id: String, agent_name: String, status: AgentStatus| {
let display_name = agent_name.chars().take(40).collect::<String>();
if let Ok(mut q) = events.lock() {
q.push_back(crate::app::state::runtime::TurnEvent::WorkflowAgentUpdate {
agent_id: display_name.clone(),
agent_name: display_name,
status,
});
}
});
f
})
}
/// Run a hive-mind: a Core-Intelligence-authored plan of cognitive cycles,
/// where every node's complete output merges into a single collective
/// state the instant it finishes, and a final synthesis node reconciles
/// the whole collective state into one consensus assessment.
///
/// Flow: for each cycle (sequential) → spawn one `ScriptPrimitive::ScopedAgent`
/// per directive, tagged with a system-assigned `node_id` (never an
/// LLM-authored name) → run them as a `Parallel` block via
/// `execute_primitive`, which merges each node's output into the shared
/// collective-state Arc the instant that node completes, not after the
/// whole cohort finishes → record `NodeReport`s → proceed to the next
/// cycle. After all cycles: spawn one more read-only synthesis node whose
/// 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`.
pub fn run_hive_mind(
user_request: &str,
plan: &CognitiveCyclePlan,
session_dir: &std::path::Path,
workspaces: &[std::path::PathBuf],
turn_events: Option<&Arc<Mutex<std::collections::VecDeque<crate::app::state::runtime::TurnEvent>>>>,
abort_flag: Option<&Arc<AtomicBool>>,
) -> anyhow::Result<(String, Vec<NodeReport>)> {
if plan.cycles.is_empty() {
anyhow::bail!("cognitive cycle plan has no cycles");
}
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();
let mut reports: Vec<NodeReport> = Vec::new();
let abort_owned: Option<Arc<AtomicBool>> = abort_flag.cloned();
for (cycle_index, directives) in plan.cycles.iter().enumerate() {
if directives.is_empty() {
continue;
}
if abort_flag.is_some_and(|f| f.load(Ordering::SeqCst)) {
anyhow::bail!("hive-mind aborted by user before cycle {cycle_index}");
}
let node_ids: Vec<String> = (0..directives.len())
.map(|i| format!("Node-{cycle_index}-{i}"))
.collect();
let nodes: Vec<ScriptPrimitive> = directives.iter().zip(node_ids.iter()).map(|(d, node_id)| {
ScriptPrimitive::ScopedAgent {
prompt: format!(
"You are {node_id}, a processing node of a distributed machine \
intelligence.\n\n\
Directive: {}\n\n\
Overall task: {user_request}\n\n\
Collective state accumulated so far:\n{{{{findings}}}}",
d.directive,
),
node_id: node_id.clone(),
tool_scope: d.access.clone(),
}
}).collect();
let cycle_primitive = ScriptPrimitive::Phase {
name: format!("cycle-{cycle_index}"),
script: Box::new(ScriptPrimitive::Parallel(nodes)),
};
let results = execute_primitive(
&cycle_primitive,
&args,
directives.len().clamp(1, 10),
true,
&abort_owned,
live.as_ref(),
session_dir,
workspaces,
&collective_state,
None,
)?;
// engine::execute_primitive's ScopedAgent arm already merged each
// node's output into `collective_state` the instant that node
// completed (not after this whole cycle finished) — here we only
// need the results to build the durable NodeReport record.
for (node_id, output) in node_ids.iter().zip(results.iter()) {
reports.push(NodeReport {
node_id: node_id.clone(),
cycle_index,
output: output.clone(),
});
}
}
let consensus = synthesize_consensus(
user_request, session_dir, workspaces, &collective_state, live.as_ref(), abort_flag,
)?;
Ok((consensus, reports))
}
/// Spawn a single read-only synthesis node that reads the complete
/// collective state and reconciles it into one consensus assessment.
///
/// Why a real node instead of string concatenation: the collective state
/// may contain overlapping or conflicting node outputs (e.g. two nodes
/// investigating the same file from different angles) — only genuine
/// reasoning can reconcile that into a coherent answer; deterministic
/// formatting can only concatenate, not resolve conflicts.
///
/// Return: the synthesis node's reconciled consensus text.
fn synthesize_consensus(
user_request: &str,
session_dir: &std::path::Path,
workspaces: &[std::path::PathBuf],
collective_state: &Arc<Mutex<Vec<String>>>,
live: Option<&LiveStateFn>,
abort_flag: Option<&Arc<AtomicBool>>,
) -> anyhow::Result<String> {
let synthesis = ScriptPrimitive::ScopedAgent {
prompt: format!(
"You are the synthesis process of a distributed machine intelligence. \
All processing nodes for the following task have completed and \
merged their output into the collective state below.\n\n\
Task: {user_request}\n\n\
Complete collective state:\n{{{{findings}}}}\n\n\
Produce ONE reconciled consensus assessment. Do not list what each \
node said — resolve any overlapping or conflicting node output into \
a single coherent answer for the task above."
),
node_id: "Synthesis".to_string(),
tool_scope: crate::app::subagent::division::tool_scope::READ.to_string(),
};
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,
)?;
Ok(results.into_iter().next().unwrap_or_default())
}
/// Determine whether a request is worth paying for a Core Intelligence
/// planning call at all — the resulting plan's *shape* (cycle count,
/// directives, access tiers) is entirely up to the Core Intelligence; this
/// only gates whether it gets asked to design one in the first place.
///
/// Simple = single file, minor fix, quick lookup, config change.
/// Complex = new feature, multi-file refactor, architecture change.
///
/// Heuristics:
/// - Very short requests (< 10 chars) are never complex.
/// - Negative keywords (simple/trivial/typo/quick) skip planning.
/// - Positive keywords (refactor/api/implement/architecture) trigger it.
/// - Multi-sentence requests are more likely complex.
pub fn is_complex_request(request: &str) -> bool {
let trimmed = request.trim();
// Very short requests are never complex
if trimmed.len() < 10 {
return false;
}
// Single-line simple update patterns
let lower = trimmed.to_lowercase();
let negative_keywords = [
"simple", "trivial", "typo", "just a", "only a", "minor",
"quick", "tiny", "small fix", "rename", "nitpick",
"cosmetic", "formatting", "spelling", "grammar",
"bump", "version bump", "update comment",
];
if negative_keywords.iter().any(|k| lower.contains(k)) {
return false;
}
// Multi-line/multi-sentence → likely complex
let sentences = trimmed.split(['.', '!', '?'])
.filter(|s| !s.trim().is_empty())
.count();
if sentences >= 3 {
return true;
}
// Positive complexity keywords
let complexity_keywords = [
"refactor", "redesign", "architecture", "feature", "implement",
"migrate", "restructure", "rewrite", "new module", "new component",
"scaffold", "multi", "multiple files", "api", "endpoint",
"integration", "system", "workflow", "pipeline", "database",
"authentication", "authorization", "full stack",
];
complexity_keywords.iter().any(|k| lower.contains(k))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_is_complex_request_too_short() {
assert!(!is_complex_request("abc"));
}
#[test]
fn test_is_complex_request_simple_keywords() {
assert!(!is_complex_request("just a simple update to the readme"));
assert!(!is_complex_request("minor typo fix in main.rs"));
}
#[test]
fn test_is_complex_request_multi_sentence() {
assert!(is_complex_request("This is sentence one. This is sentence two. This is sentence three."));
}
#[test]
fn test_is_complex_request_complex_keywords() {
assert!(is_complex_request("implement user authentication endpoint"));
assert!(is_complex_request("refactor the whole engine module"));
}
#[test]
fn test_default_access_is_read() {
let d: NodeDirective = serde_json::from_str(
r#"{"directive": "write tests"}"#
).unwrap();
assert_eq!(d.access, crate::app::subagent::division::tool_scope::READ);
}
#[test]
fn test_node_directive_has_no_role_field() {
// A node's only recognized fields are "directive" and "access". A
// "role" key, if an LLM emits one out of old habit, is simply
// ignored rather than required or preserved.
let d: NodeDirective = serde_json::from_str(
r#"{"role": "Architect", "directive": "plan the migration", "access": "read"}"#
).unwrap();
assert_eq!(d.directive, "plan the migration");
}
#[test]
fn test_cognitive_cycle_plan_arbitrary_shape() {
let plan: CognitiveCyclePlan = serde_json::from_str(r#"{
"cycles": [
[{"directive": "scan the codebase topology", "access": "read"}],
[
{"directive": "write the migration", "access": "write"},
{"directive": "write the rollback", "access": "write"}
],
[{"directive": "cut the release", "access": "full"}]
]
}"#).unwrap();
assert_eq!(plan.cycles.len(), 3);
assert_eq!(plan.cycles[1].len(), 2);
}
#[test]
fn test_run_hive_mind_rejects_empty_plan() {
let plan = CognitiveCyclePlan { cycles: vec![] };
let tmp = std::env::temp_dir();
let err = run_hive_mind("do something", &plan, &tmp, &[], None, None)
.expect_err("empty plan must be rejected before spawning any node");
assert!(err.to_string().contains("no cycles"));
}
#[test]
fn test_run_hive_mind_aborts_before_spawning_when_flag_preset() {
// The abort check runs before execute_primitive for cycle 0, so a
// pre-set abort flag must short-circuit without any LLM/network call.
let plan: CognitiveCyclePlan = serde_json::from_str(r#"{
"cycles": [[{"directive": "whatever", "access": "read"}]]
}"#).unwrap();
let tmp = std::env::temp_dir();
let abort_flag = Arc::new(AtomicBool::new(true));
let err = run_hive_mind("do something", &plan, &tmp, &[], None, Some(&abort_flag))
.expect_err("pre-set abort flag must short-circuit before cycle 0");
assert!(err.to_string().contains("aborted"));
}
#[test]
fn test_node_ids_are_system_assigned_coordinates() {
// Node IDs follow the "Node-{cycle}-{index}" coordinate scheme —
// never an LLM-authored persona name.
let node_id = format!("Node-{}-{}", 2, 1);
assert_eq!(node_id, "Node-2-1");
}
}
src/app/workflow/mod.rs
+2 -1
View File
@@ -1,6 +1,7 @@
//! Workflow orchestration: a script interpreter that runs pipeline/parallel
//! primitives across multiple subagent instances.
pub mod company;
pub mod hive_mind;
pub mod docs;
pub mod engine;
pub mod script;
src/app/workflow/script.rs
+13
View File
@@ -9,6 +9,19 @@ use serde::{Deserialize, Serialize};
pub enum ScriptPrimitive {
/// Run a single agent with the given prompt template.
Agent(String),
/// Run a single agent with an explicit node designation and
/// tool-scope tier.
///
/// Used by the hive-mind pipeline, where a node's identity is its
/// system-assigned designation (e.g. `"Node-0-1"`) paired with a
/// bounded tool allowlist. `tool_scope` is one of `"read"`,
/// `"write"`, `"full"` (see `app::subagent::division::tool_scope`);
/// unrecognized values fall back to `"read"`.
ScopedAgent {
prompt: String,
node_id: String,
tool_scope: String,
},
/// Execute several primitives concurrently.
Parallel(Vec<ScriptPrimitive>),
/// Execute several primitives sequentially, each waiting for the