473 lines
17 KiB
Rust
473 lines
17 KiB
Rust
|
//! Workflow engine: interprets `ScriptPrimitive` values (agent, parallel,
|
||
|
|
//! pipeline, phase) by spawning subagents, collecting results, and
|
||
|
|
//! managing concurrency.
|
||
|
|
//!
|
||
|
|
//! Key design points:
|
||
|
|
//! - `Parallel` branches run concurrently (capped by semaphore) — this is
|
||
|
|
//! the main advantage over single-turn chat.
|
||
|
|
//! - `Pipeline` branches run sequentially so each stage sees findings from
|
||
|
|
//! the previous one.
|
||
|
|
//! - `run_workflow_tracked` accepts a `LiveState` callback that receives
|
||
|
|
//! real-time agent status updates for the TUI panel.
|
||
|
|
//! - Findings (inter-agent notes) are scoped per invocation via an
|
||
|
|
//! `Arc<Mutex<Vec<String>>>` threaded through `execute_primitive` and
|
||
|
|
//! `spawn_single_agent` rather than a global static, preventing data
|
||
|
|
//! leaks between concurrent workflow runs.
|
||
|
|
|
||
|
|
pub mod primitives;
|
||
|
|
pub mod phases;
|
||
|
|
pub mod execution;
|
||
|
|
|
||
|
|
// Re-exports so existing `crate::app::workflow::engine::*` paths continue to work.
|
||
|
|
pub use execution::run_workflow;
|
||
|
|
pub use primitives::execute_primitive;
|
||
|
|
pub(crate) use primitives::PrimitiveCtx;
|
||
|
|
|
||
|
|
use serde::{Deserialize, Serialize};
|
||
|
|
use std::sync::{
|
||
|
|
atomic::{AtomicBool, Ordering},
|
||
|
|
Arc, Mutex,
|
||
|
|
};
|
||
|
|
use std::time::Duration;
|
||
|
|
|
||
|
|
/// The lifecycle state of an agent within a workflow run.
|
||
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||
|
|
pub enum AgentState {
|
||
|
|
Idle,
|
||
|
|
Running,
|
||
|
|
Completed,
|
||
|
|
Failed,
|
||
|
|
}
|
||
|
|
|
||
|
|
/// Timestamped status of one workflow agent.
|
||
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||
|
|
pub struct AgentStatus {
|
||
|
|
pub state: AgentState,
|
||
|
|
pub started_at: Option<i64>,
|
||
|
|
pub completed_at: Option<i64>,
|
||
|
|
pub error: Option<String>,
|
||
|
|
/// Human-readable progress message (e.g. "editing src/main.rs",
|
||
|
|
/// "running cargo test"). Shown in the TUI panel alongside the state.
|
||
|
|
pub progress: Option<String>,
|
||
|
|
}
|
||
|
|
|
||
|
|
/// A single agent tracked within a workflow run.
|
||
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||
|
|
pub struct WorkflowAgent {
|
||
|
|
pub id: String,
|
||
|
|
pub name: String,
|
||
|
|
pub status: AgentStatus,
|
||
|
|
}
|
||
|
|
|
||
|
|
/// Orchestrator for running workflow scripts: holds agent roster and a
|
||
|
|
/// shared finding accumulator visible to all pipeline stages.
|
||
|
|
#[derive(Debug, Clone)]
|
||
|
|
pub struct WorkflowEngine {
|
||
|
|
pub agents: Vec<WorkflowAgent>,
|
||
|
|
pub findings: Vec<String>,
|
||
|
|
}
|
||
|
|
|
||
|
|
impl WorkflowEngine {
|
||
|
|
/// Create an empty workflow engine with no agents or findings.
|
||
|
|
pub fn new() -> Self {
|
||
|
|
WorkflowEngine {
|
||
|
|
agents: Vec::new(),
|
||
|
|
findings: Vec::new(),
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
/// Shared live state used by `run_workflow_tracked` to push real-time
|
||
|
|
/// agent status updates into the TUI's `WorkflowEngine`.
|
||
|
|
///
|
||
|
|
/// The closure receives `(agent_id, agent_name, new_status)`:
|
||
|
|
/// - `agent_id`: unique identifier (UUID) for upserting the agent.
|
||
|
|
/// - `agent_name`: human-readable display name for the TUI panel.
|
||
|
|
/// - `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. a hive-mind node's designation, `"Node-0-1"`).
|
||
|
|
pub type LiveStateFn = Arc<dyn Fn(String, String, AgentStatus) + Send + Sync>;
|
||
|
|
|
||
|
|
/// Bundled context for spawning a single subagent.
|
||
|
|
pub(crate) struct SpawnCtx<'a> {
|
||
|
|
pub agent_id: &'a str,
|
||
|
|
pub agent_name: &'a str,
|
||
|
|
pub prompt: &'a str,
|
||
|
|
pub role: &'a str,
|
||
|
|
pub allowed_tools: Option<Vec<String>>,
|
||
|
|
pub findings_snapshot: &'a [String],
|
||
|
|
pub findings: &'a Arc<Mutex<Vec<String>>>,
|
||
|
|
pub abort_flag: &'a Option<Arc<AtomicBool>>,
|
||
|
|
pub live: Option<&'a LiveStateFn>,
|
||
|
|
pub session_dir: &'a std::path::Path,
|
||
|
|
pub workspaces: &'a [std::path::PathBuf],
|
||
|
|
pub timeout_ms: Option<u64>,
|
||
|
|
}
|
||
|
|
|
||
|
|
// ---------------------------------------------------------------------------
|
||
|
|
// Helpers
|
||
|
|
// ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
fn format_tool_call_progress(prefix: &str, tool: &str, args: &serde_json::Value) -> String {
|
||
|
|
let details = match tool {
|
||
|
|
"read"
|
||
|
|
| "view_file"
|
||
|
|
| "write"
|
||
|
|
| "write_to_file"
|
||
|
|
| "edit"
|
||
|
|
| "replace_file_content"
|
||
|
|
| "multi_replace_file_content"
|
||
|
|
| "delete" => args
|
||
|
|
.get("path")
|
||
|
|
.or_else(|| args.get("TargetFile"))
|
||
|
|
.or_else(|| args.get("AbsolutePath"))
|
||
|
|
.and_then(|v| v.as_str())
|
||
|
|
.unwrap_or("")
|
||
|
|
.to_string(),
|
||
|
|
"grep" | "grep_search" => {
|
||
|
|
let pattern = args
|
||
|
|
.get("pattern")
|
||
|
|
.or_else(|| args.get("Query"))
|
||
|
|
.and_then(|v| v.as_str())
|
||
|
|
.unwrap_or("");
|
||
|
|
let path = args
|
||
|
|
.get("path")
|
||
|
|
.or_else(|| args.get("SearchPath"))
|
||
|
|
.and_then(|v| v.as_str())
|
||
|
|
.unwrap_or("");
|
||
|
|
if path.is_empty() {
|
||
|
|
format!("\"{pattern}\"")
|
||
|
|
} else {
|
||
|
|
format!("\"{pattern}\" in {path}")
|
||
|
|
}
|
||
|
|
}
|
||
|
|
"glob" => {
|
||
|
|
let pattern = args.get("pattern").and_then(|v| v.as_str()).unwrap_or("");
|
||
|
|
let path = args.get("path").and_then(|v| v.as_str()).unwrap_or("");
|
||
|
|
if path.is_empty() {
|
||
|
|
pattern.to_string()
|
||
|
|
} else {
|
||
|
|
format!("{pattern} in {path}")
|
||
|
|
}
|
||
|
|
}
|
||
|
|
"bash" | "run_command" => {
|
||
|
|
let cmd = args
|
||
|
|
.get("command")
|
||
|
|
.or_else(|| args.get("CommandLine"))
|
||
|
|
.and_then(|v| v.as_str())
|
||
|
|
.unwrap_or("");
|
||
|
|
if cmd.len() > 60 {
|
||
|
|
format!("\"{}...\"", &cmd[..57])
|
||
|
|
} else {
|
||
|
|
format!("\"{cmd}\"")
|
||
|
|
}
|
||
|
|
}
|
||
|
|
"recall" => args
|
||
|
|
.get("query")
|
||
|
|
.and_then(|v| v.as_str())
|
||
|
|
.unwrap_or("")
|
||
|
|
.to_string(),
|
||
|
|
"remember" => args
|
||
|
|
.get("name")
|
||
|
|
.and_then(|v| v.as_str())
|
||
|
|
.unwrap_or("")
|
||
|
|
.to_string(),
|
||
|
|
"dir_list" | "list_dir" => args
|
||
|
|
.get("DirectoryPath")
|
||
|
|
.or_else(|| args.get("path"))
|
||
|
|
.and_then(|v| v.as_str())
|
||
|
|
.unwrap_or("")
|
||
|
|
.to_string(),
|
||
|
|
_ => {
|
||
|
|
if let Some(obj) = args.as_object() {
|
||
|
|
if !obj.is_empty() {
|
||
|
|
return obj
|
||
|
|
.values()
|
||
|
|
.find_map(|v| v.as_str())
|
||
|
|
.unwrap_or("")
|
||
|
|
.to_string();
|
||
|
|
}
|
||
|
|
}
|
||
|
|
String::new()
|
||
|
|
}
|
||
|
|
};
|
||
|
|
|
||
|
|
if details.is_empty() {
|
||
|
|
format!("{prefix}: {tool}")
|
||
|
|
} else {
|
||
|
|
format!("{prefix}: {tool} {details}")
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
/// Spawn a single synchronous subagent with the given prompt, passing it
|
||
|
|
/// any findings from earlier sibling agents. Updates live state before and
|
||
|
|
/// after to reflect Running → Completed/Failed transitions.
|
||
|
|
///
|
||
|
|
/// Flow: push agent as `Running` → build `SubagentContext` with prompt +
|
||
|
|
/// findings preamble, linking the `workflow_findings` Arc so the subagent's
|
||
|
|
/// `note_finding` tool pushes into the same vec → call `run_subagent`
|
||
|
|
/// (draining the event channel into a consumer so events are not blocked)
|
||
|
|
/// → push `Completed` or `Failed`.
|
||
|
|
///
|
||
|
|
/// Why: the `workflow_findings` Arc is shared by all agents within the same
|
||
|
|
/// `execute_primitive` scope, so pipeline stages can pass data between each
|
||
|
|
/// other while different workflow invocations remain isolated.
|
||
|
|
///
|
||
|
|
/// When `timeout_ms` is `Some`, the subagent is killed (abandoned on a
|
||
|
|
/// separate thread) if it does not complete within the deadline, preventing
|
||
|
|
/// a stuck stage from blocking the entire pipeline forever.
|
||
|
|
///
|
||
|
|
/// Return: the agent's text output, or an error on failure.
|
||
|
|
fn spawn_single_agent(sp: SpawnCtx<'_>) -> anyhow::Result<String> {
|
||
|
|
use crate::app::subagent::context::build_subagent_context;
|
||
|
|
use crate::app::subagent::engine::run_subagent;
|
||
|
|
use crate::app::subagent::spawn::AgentDefinition;
|
||
|
|
|
||
|
|
let started_at = chrono::Utc::now().timestamp_millis();
|
||
|
|
|
||
|
|
// 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. a hive-mind node designation).
|
||
|
|
if let Some(f) = &sp.live {
|
||
|
|
f(
|
||
|
|
sp.agent_id.to_string(),
|
||
|
|
sp.agent_name.to_string(),
|
||
|
|
AgentStatus {
|
||
|
|
state: AgentState::Running,
|
||
|
|
started_at: Some(started_at),
|
||
|
|
completed_at: None,
|
||
|
|
error: None,
|
||
|
|
progress: None,
|
||
|
|
},
|
||
|
|
);
|
||
|
|
}
|
||
|
|
|
||
|
|
let mut def = AgentDefinition::new(sp.agent_name.to_string(), sp.role.to_string());
|
||
|
|
if let Some(tools) = &sp.allowed_tools {
|
||
|
|
def = def.with_allowed_tools(tools.clone());
|
||
|
|
}
|
||
|
|
let mut ctx = build_subagent_context(&def);
|
||
|
|
ctx.session_dir = sp.session_dir.to_path_buf();
|
||
|
|
ctx.workspaces = sp.workspaces.to_vec();
|
||
|
|
|
||
|
|
let findings_section = if sp.findings_snapshot.is_empty() {
|
||
|
|
String::new()
|
||
|
|
} else {
|
||
|
|
format!(
|
||
|
|
"\n\nFindings from sibling drones in this Hive run:\n{}",
|
||
|
|
sp.findings_snapshot
|
||
|
|
.iter()
|
||
|
|
.enumerate()
|
||
|
|
.map(|(i, f)| format!("{}. {}", i + 1, f))
|
||
|
|
.collect::<Vec<_>>()
|
||
|
|
.join("\n")
|
||
|
|
)
|
||
|
|
};
|
||
|
|
|
||
|
|
ctx.system_prompt = format!("{}{}", sp.prompt, findings_section);
|
||
|
|
// Link the shared findings Arc so note_finding calls within this
|
||
|
|
// subagent write into the same vec visible to sibling agents.
|
||
|
|
ctx.workflow_findings = Some(sp.findings.clone());
|
||
|
|
ctx.abort_flag.clone_from(sp.abort_flag);
|
||
|
|
|
||
|
|
// Create an mpsc channel and drain events in a background thread.
|
||
|
|
// The drain thread also pushes intra-division progress updates to the
|
||
|
|
// live callback (current tool being executed), so the TUI panel shows
|
||
|
|
// real-time "editing X" or "running build" instead of just "Running…".
|
||
|
|
let (tx, rx) = tokio::sync::mpsc::channel(64);
|
||
|
|
let drain_agent_id = sp.agent_id.to_string();
|
||
|
|
let drain_agent_name = sp.agent_name.to_string();
|
||
|
|
let drain_live = sp.live.cloned();
|
||
|
|
let drain_started_at = started_at;
|
||
|
|
let _drain_thread = std::thread::spawn(move || {
|
||
|
|
use crate::app::subagent::event::SubagentEvent;
|
||
|
|
let mut rx = rx;
|
||
|
|
while let Some(event) = rx.blocking_recv() {
|
||
|
|
match &event {
|
||
|
|
SubagentEvent::ToolCall { tool, args } => {
|
||
|
|
tracing::debug!("[subagent] tool call: {}", tool);
|
||
|
|
// Push intra-division progress: which tool is running
|
||
|
|
if let Some(ref f) = drain_live {
|
||
|
|
let formatted = format_tool_call_progress("tool", tool, args);
|
||
|
|
f(
|
||
|
|
drain_agent_id.clone(),
|
||
|
|
drain_agent_name.clone(),
|
||
|
|
AgentStatus {
|
||
|
|
state: AgentState::Running,
|
||
|
|
started_at: Some(drain_started_at),
|
||
|
|
completed_at: None,
|
||
|
|
error: None,
|
||
|
|
progress: Some(formatted),
|
||
|
|
},
|
||
|
|
);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
SubagentEvent::ToolResult { tool, args, .. } => {
|
||
|
|
tracing::debug!("[subagent] tool result: {}", tool);
|
||
|
|
if let Some(ref f) = drain_live {
|
||
|
|
let formatted = format_tool_call_progress("done", tool, args);
|
||
|
|
f(
|
||
|
|
drain_agent_id.clone(),
|
||
|
|
drain_agent_name.clone(),
|
||
|
|
AgentStatus {
|
||
|
|
state: AgentState::Running,
|
||
|
|
started_at: Some(drain_started_at),
|
||
|
|
completed_at: None,
|
||
|
|
error: None,
|
||
|
|
progress: Some(formatted),
|
||
|
|
},
|
||
|
|
);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
SubagentEvent::StepCompleted { output, .. } => {
|
||
|
|
// Show the agent's thinking/reasoning text as progress
|
||
|
|
// instead of just the tool name — first line, truncated.
|
||
|
|
if let Some(ref f) = drain_live {
|
||
|
|
let summary = output
|
||
|
|
.lines()
|
||
|
|
.next()
|
||
|
|
.unwrap_or(output)
|
||
|
|
.chars()
|
||
|
|
.take(80)
|
||
|
|
.collect::<String>();
|
||
|
|
f(
|
||
|
|
drain_agent_id.clone(),
|
||
|
|
drain_agent_name.clone(),
|
||
|
|
AgentStatus {
|
||
|
|
state: AgentState::Running,
|
||
|
|
started_at: Some(drain_started_at),
|
||
|
|
completed_at: None,
|
||
|
|
error: None,
|
||
|
|
progress: Some(summary),
|
||
|
|
},
|
||
|
|
);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
SubagentEvent::StepFailed { step, error } => {
|
||
|
|
tracing::warn!("[subagent] step {} failed: {}", step, error);
|
||
|
|
}
|
||
|
|
SubagentEvent::Progress(prog) => {
|
||
|
|
if let Some(ref f) = drain_live {
|
||
|
|
f(
|
||
|
|
drain_agent_id.clone(),
|
||
|
|
drain_agent_name.clone(),
|
||
|
|
AgentStatus {
|
||
|
|
state: AgentState::Running,
|
||
|
|
started_at: Some(drain_started_at),
|
||
|
|
completed_at: None,
|
||
|
|
error: None,
|
||
|
|
progress: Some(prog.clone()),
|
||
|
|
},
|
||
|
|
);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
SubagentEvent::Completed => {
|
||
|
|
tracing::debug!("[subagent] completed");
|
||
|
|
}
|
||
|
|
SubagentEvent::Usage {
|
||
|
|
tokens_in,
|
||
|
|
tokens_out,
|
||
|
|
} => {
|
||
|
|
tracing::debug!("[subagent] usage: {} in, {} out", tokens_in, tokens_out);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
});
|
||
|
|
|
||
|
|
// Check abort before even starting the subagent.
|
||
|
|
if sp
|
||
|
|
.abort_flag
|
||
|
|
.as_ref()
|
||
|
|
.is_some_and(|f| f.load(Ordering::SeqCst))
|
||
|
|
{
|
||
|
|
anyhow::bail!("subagent '{}' aborted before start", sp.agent_name);
|
||
|
|
}
|
||
|
|
|
||
|
|
// Run subagent on a separate thread so the abort flag can be polled.
|
||
|
|
// If abort is requested while the subagent is running, we abandon the
|
||
|
|
// thread (Rust threads cannot be forcibly killed) and return early.
|
||
|
|
let (done_tx, done_rx) = std::sync::mpsc::channel::<anyhow::Result<String>>();
|
||
|
|
let bg_ctx = ctx;
|
||
|
|
let bg_tx = tx;
|
||
|
|
let bg_name = sp.agent_name.to_string();
|
||
|
|
let bg_abort = sp.abort_flag.clone();
|
||
|
|
std::thread::spawn(move || {
|
||
|
|
let _ = done_tx.send(run_subagent(&bg_ctx, &bg_tx));
|
||
|
|
});
|
||
|
|
|
||
|
|
let poll_interval = Duration::from_millis(200);
|
||
|
|
let result = if let Some(timeout) = sp.timeout_ms {
|
||
|
|
let deadline = Duration::from_millis(timeout);
|
||
|
|
let mut elapsed = Duration::ZERO;
|
||
|
|
loop {
|
||
|
|
if let Ok(r) = done_rx.recv_timeout(poll_interval) {
|
||
|
|
break r;
|
||
|
|
}
|
||
|
|
elapsed += poll_interval;
|
||
|
|
if elapsed >= deadline {
|
||
|
|
break Err(anyhow::anyhow!(
|
||
|
|
"subagent '{bg_name}' timed out after {timeout}ms",
|
||
|
|
));
|
||
|
|
}
|
||
|
|
if bg_abort.as_ref().is_some_and(|f| f.load(Ordering::SeqCst)) {
|
||
|
|
break Err(anyhow::anyhow!("subagent '{bg_name}' aborted by user"));
|
||
|
|
}
|
||
|
|
}
|
||
|
|
} else {
|
||
|
|
loop {
|
||
|
|
if let Ok(r) = done_rx.recv_timeout(poll_interval) {
|
||
|
|
break r;
|
||
|
|
}
|
||
|
|
if bg_abort.as_ref().is_some_and(|f| f.load(Ordering::SeqCst)) {
|
||
|
|
break Err(anyhow::anyhow!("subagent '{bg_name}' aborted by user"));
|
||
|
|
}
|
||
|
|
}
|
||
|
|
};
|
||
|
|
|
||
|
|
let completed_at = chrono::Utc::now().timestamp_millis();
|
||
|
|
|
||
|
|
// Notify UI: agent completed or failed
|
||
|
|
if let Some(f) = &sp.live {
|
||
|
|
let summary_from = |text: &str| {
|
||
|
|
text.lines()
|
||
|
|
.next()
|
||
|
|
.unwrap_or(text)
|
||
|
|
.chars()
|
||
|
|
.take(80)
|
||
|
|
.collect::<String>()
|
||
|
|
};
|
||
|
|
match &result {
|
||
|
|
Ok(text) => {
|
||
|
|
let summary = summary_from(text);
|
||
|
|
f(
|
||
|
|
sp.agent_id.to_string(),
|
||
|
|
sp.agent_name.to_string(),
|
||
|
|
AgentStatus {
|
||
|
|
state: AgentState::Completed,
|
||
|
|
started_at: Some(started_at),
|
||
|
|
completed_at: Some(completed_at),
|
||
|
|
error: None,
|
||
|
|
progress: Some(summary),
|
||
|
|
},
|
||
|
|
);
|
||
|
|
}
|
||
|
|
Err(e) => {
|
||
|
|
f(
|
||
|
|
sp.agent_id.to_string(),
|
||
|
|
sp.agent_name.to_string(),
|
||
|
|
AgentStatus {
|
||
|
|
state: AgentState::Failed,
|
||
|
|
started_at: Some(started_at),
|
||
|
|
completed_at: Some(completed_at),
|
||
|
|
error: Some(e.to_string()),
|
||
|
|
progress: None,
|
||
|
|
},
|
||
|
|
);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
result
|
||
|
|
}
|