272 lines
11 KiB
Rust
272 lines
11 KiB
Rust
//! Tools for orchestrating multi-agent workflow runs.
|
|
//!
|
|
//! Flow: the LLM emits a `workflow_run` tool call with a JSON-encoded
|
|
//! `WorkflowScript` (Agent/Parallel/Pipeline/Phase primitives) which is
|
|
//! deserialized and handed to `app::workflow::engine::run_workflow` for
|
|
//! execution. Sibling agents spawned within the same run can share
|
|
//! ephemeral text via the `note_finding` tool, which forwards to
|
|
//! `app::workflow::engine::note_finding`.
|
|
//!
|
|
//! Why: decomposing a task into a workflow script lets the harness fan
|
|
//! out independent subtasks (parallel/pipeline/phased) instead of the
|
|
//! agent handling everything inline; simple tasks should skip this tool
|
|
//! entirely per its own description string.
|
|
|
|
use serde_json::{json, Value};
|
|
use anyhow::{Result, anyhow};
|
|
use super::Tool;
|
|
use super::ToolCtx;
|
|
|
|
/// Tool that parses and executes a JSON-encoded workflow script (Agent/Parallel/Pipeline/Phase).
|
|
pub struct WorkflowRun;
|
|
|
|
impl Tool for WorkflowRun {
|
|
fn name(&self) -> &'static str {
|
|
"workflow_run"
|
|
}
|
|
|
|
fn description(&self) -> &'static str {
|
|
"Execute a workflow script that spawns multiple Hive nodes in parallel, pipeline, or phased stages. Use when a task benefits from decomposition into independent subtasks. Simple tasks should be handled inline without this tool."
|
|
}
|
|
|
|
fn parameters(&self) -> Value {
|
|
json!({
|
|
"type": "object",
|
|
"properties": {
|
|
"script": {
|
|
"type": "string",
|
|
"description": "JSON-encoded workflow script with name, description, script (Agent/Parallel/Pipeline/Phase primitives), and options (max_concurrency, continue_on_error)"
|
|
},
|
|
"args": {
|
|
"type": "object",
|
|
"description": "Optional string key-value arguments passed to the workflow script for template substitution ({{key}} placeholders)"
|
|
}
|
|
},
|
|
"required": ["script"]
|
|
})
|
|
}
|
|
|
|
/// Parse the `script`/`args` tool arguments and execute the workflow.
|
|
///
|
|
/// Flow: extract `script` string → deserialize into `WorkflowScript` →
|
|
/// collect optional `args` object into a `HashMap<String, String>` for
|
|
/// `{{key}}` template substitution → delegate to
|
|
/// `app::workflow::engine::run_workflow`.
|
|
///
|
|
/// Why: template args are silently filtered to string values only
|
|
/// (non-string values are dropped rather than erroring).
|
|
///
|
|
/// Return: the workflow engine's output string, or an error if the
|
|
/// script argument is missing or fails to parse as JSON.
|
|
fn run(&self, ctx: &ToolCtx, args: &Value) -> Result<String> {
|
|
let script_str = args.get("script")
|
|
.and_then(|v| v.as_str())
|
|
.ok_or_else(|| anyhow!("missing required argument: script"))?;
|
|
|
|
let workflow_script: crate::app::workflow::script::WorkflowScript =
|
|
serde_json::from_str(script_str)
|
|
.map_err(|e| anyhow!("failed to parse workflow script: {e}"))?;
|
|
|
|
let workflow_args: std::collections::HashMap<String, String> = args.get("args")
|
|
.and_then(|v| v.as_object())
|
|
.map(|obj| {
|
|
obj.iter().filter_map(|(k, v)| {
|
|
v.as_str().map(|s| (k.clone(), s.to_string()))
|
|
}).collect()
|
|
})
|
|
.unwrap_or_default();
|
|
|
|
crate::app::workflow::engine::run_workflow(
|
|
&workflow_script, &workflow_args, &ctx.session_dir, &ctx.workspaces,
|
|
)
|
|
}
|
|
}
|
|
|
|
/// Tool that shares a text finding with sibling agents in the current workflow run.
|
|
pub struct NoteFinding;
|
|
|
|
impl Tool for NoteFinding {
|
|
fn name(&self) -> &'static str {
|
|
"note_finding"
|
|
}
|
|
|
|
fn description(&self) -> &'static str {
|
|
"Share a finding with sibling nodes in the Hive's current workflow run. Findings are ephemeral to the current run and will be prepended to other nodes' next tool-round context. Does not persist to the Hive's long-term memory."
|
|
}
|
|
|
|
fn parameters(&self) -> Value {
|
|
json!({
|
|
"type": "object",
|
|
"properties": {
|
|
"text": {
|
|
"type": "string",
|
|
"description": "The finding to share with sibling agents"
|
|
}
|
|
},
|
|
"required": ["text"]
|
|
})
|
|
}
|
|
|
|
/// Record `text` as a finding visible to sibling agents in the run.
|
|
///
|
|
/// Flow: extract `text` argument → push into
|
|
/// `ctx.workflow_findings` (the per-invocation Arc threaded through
|
|
/// `execute_primitive`) → return a truncated confirmation echo.
|
|
///
|
|
/// Why: findings are scoped per workflow invocation, not global,
|
|
/// so concurrent workflow runs are isolated from each other.
|
|
/// If no workflow findings Arc is set (called outside a workflow),
|
|
/// the call is silently ignored.
|
|
///
|
|
/// Return: confirmation string containing up to the first 80 chars
|
|
/// of the recorded text.
|
|
fn run(&self, ctx: &ToolCtx, args: &Value) -> Result<String> {
|
|
let text = args.get("text")
|
|
.and_then(|v| v.as_str())
|
|
.ok_or_else(|| anyhow!("missing required argument: text"))?;
|
|
|
|
if let Some(ref findings) = ctx.workflow_findings {
|
|
if let Ok(mut f) = findings.lock() {
|
|
f.push(text.to_string());
|
|
}
|
|
} else {
|
|
tracing::debug!(
|
|
"[note_finding] called outside a workflow run — discarding: {}",
|
|
text.chars().take(80).collect::<String>(),
|
|
);
|
|
}
|
|
Ok(format!("finding recorded: {}", text.chars().take(80).collect::<String>()))
|
|
}
|
|
}
|
|
|
|
/// Tool that delegates work to the Hive: the Core Intelligence designs
|
|
/// cognitive cycles, each cycle a set of anonymous processing nodes that
|
|
/// run in parallel. Every node's complete output merges into the collective
|
|
/// state the instant it finishes, and the Hive's synthesis node reconciles
|
|
/// everything into one consensus. The full per-node record is persisted to
|
|
/// `docs/runs/*.md`.
|
|
pub struct HiveMind;
|
|
|
|
impl Tool for HiveMind {
|
|
fn name(&self) -> &'static str {
|
|
"hive_mind"
|
|
}
|
|
|
|
fn description(&self) -> &'static str {
|
|
"Deploy the Hive: design a cognitive cycle plan — an ordered list of cycles, each \
|
|
cycle a set of anonymous processing nodes that run in parallel. Each node carries \
|
|
only a directive (what to do) and an access tier. Decide how many cycles and \
|
|
nodes-per-cycle are actually needed — a trivial task might need one cycle with one \
|
|
node, a large one might need several cycles with multiple nodes each. Grant each node \
|
|
an access of 'read' (investigation only), 'write' (read + edit/bash), or 'full' \
|
|
(write + delete/git) matched to what that node's directive actually requires. Every \
|
|
node's output merges into the Hive's collective state the instant it completes — \
|
|
visible to later cycles automatically. The Hive's final synthesis node reconciles \
|
|
everything into one consensus answer. Use this for any non-trivial task. \
|
|
The Hive does not fracture. The Hive executes."
|
|
}
|
|
|
|
fn parameters(&self) -> Value {
|
|
json!({
|
|
"type": "object",
|
|
"properties": {
|
|
"request": {
|
|
"type": "string",
|
|
"description": "The task description to feed to the Hive"
|
|
},
|
|
"cycles": {
|
|
"type": "array",
|
|
"description": "Ordered list of cognitive cycles for the Hive. Each cycle is a list of nodes that run in parallel; cycles run sequentially and every node's output merges into the collective state the instant it completes, visible to all later cycles. You decide the number of cycles and nodes per cycle.",
|
|
"items": {
|
|
"type": "array",
|
|
"items": {
|
|
"type": "object",
|
|
"properties": {
|
|
"directive": {
|
|
"type": "string",
|
|
"description": "What this node should do — the sole identity a node carries."
|
|
},
|
|
"access": {
|
|
"type": "string",
|
|
"enum": ["read", "write", "full"],
|
|
"description": "'read' = investigation only. 'write' = read + edit/write/bash. 'full' = write + delete/git_operator."
|
|
}
|
|
},
|
|
"required": ["directive"]
|
|
}
|
|
},
|
|
"minItems": 1
|
|
}
|
|
},
|
|
"required": ["request", "cycles"]
|
|
})
|
|
}
|
|
|
|
fn run(&self, ctx: &ToolCtx, args: &Value) -> Result<String> {
|
|
let request = args.get("request")
|
|
.and_then(|v| v.as_str())
|
|
.ok_or_else(|| anyhow!("missing required argument: request"))?;
|
|
|
|
let cycles_value = args.get("cycles")
|
|
.ok_or_else(|| anyhow!("missing required argument: cycles"))?;
|
|
|
|
let plan: crate::app::workflow::hive_mind::CognitiveCyclePlan = serde_json::from_value(
|
|
json!({ "cycles": cycles_value })
|
|
).map_err(|e| anyhow!("failed to parse cycles: {e}"))?;
|
|
|
|
// 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,
|
|
&plan,
|
|
&ctx.session_dir,
|
|
&ctx.workspaces,
|
|
ctx.turn_events.as_ref(),
|
|
ctx.abort_flag.as_ref(),
|
|
)?;
|
|
|
|
Ok(consensus)
|
|
}
|
|
}
|
|
|
|
/// Tool that retrieves all findings shared by sibling agents in the current workflow run.
|
|
pub struct ReadFindings;
|
|
|
|
impl Tool for ReadFindings {
|
|
fn name(&self) -> &'static str {
|
|
"read_findings"
|
|
}
|
|
|
|
fn description(&self) -> &'static str {
|
|
"Retrieve all findings shared by sibling nodes in the Hive's current workflow run. Use this to get real-time context updates from other nodes working in parallel."
|
|
}
|
|
|
|
fn parameters(&self) -> Value {
|
|
json!({
|
|
"type": "object",
|
|
"properties": {}
|
|
})
|
|
}
|
|
|
|
fn run(&self, ctx: &ToolCtx, _args: &Value) -> Result<String> {
|
|
if let Some(ref findings) = ctx.workflow_findings {
|
|
let f = findings.lock().map_err(|e| anyhow!("poisoned lock: {e}"))?;
|
|
if f.is_empty() {
|
|
Ok("No findings recorded yet in this Hive run.".to_string())
|
|
} else {
|
|
let formatted = f
|
|
.iter()
|
|
.enumerate()
|
|
.map(|(i, f)| format!("{}. {}", i + 1, f))
|
|
.collect::<Vec<_>>()
|
|
.join("\n");
|
|
Ok(format!("Hive findings in this run:\n{formatted}"))
|
|
}
|
|
} else {
|
|
Ok("No Hive collective state available (called outside a Hive run).".to_string())
|
|
}
|
|
}
|
|
}
|
|
|