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

feat: introduce workflow management tools and commands

Browse Source 
- Added new workflow commands: `/workflow` to open the workflow panel and `/workflow run <prompt>` to execute workflows.
- Implemented `spawn_agents` and `spawn_pipeline` tools for parallel and sequential task execution, respectively.
- Enhanced workflow engine to handle real-time agent status updates and display in the UI.
- Updated workflow panel to show agent statuses, findings count, and session counters.
- Refactored existing code to integrate new workflow functionalities and improve overall structure.
This commit is contained in:
asepharyana
2026-07-12 17:49:34 +07:00
parent 7bdfd9c4c4
commit 53b0cb271f
14 changed files with 668 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/app/workflow/engine.rs
+141 -58
View File
@@ -1,6 +1,14 @@
//! 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.
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
@@ -53,21 +61,48 @@ impl WorkflowEngine {
}
}
/// Spawn a single synchronous subagent with the given prompt, passing it
/// any findings from earlier sibling agents.
/// Shared live state used by `run_workflow_tracked` to push real-time
/// agent status updates into the TUI's `WorkflowEngine`.
///
/// Flow: build an `AgentDefinition` -> build a `SubagentContext` ->
/// inject findings into the system prompt -> call `run_subagent` on a
/// dedicated mpsc channel.
/// The closure receives `(agent_id, new_status)` and should update the
/// corresponding agent in `AppStateRest::workflow_engine`.
pub type LiveStateFn = Arc<dyn Fn(String, AgentStatus) + Send + Sync>;
/// 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 → call `run_subagent` (draining the event channel into
/// a throwaway consumer so events are not blocked) → push `Completed` or
/// `Failed`.
///
/// Return: the agent's text output, or an error on failure.
fn spawn_single_agent(prompt: &str, findings_snapshot: Vec<String>) -> anyhow::Result<String> {
fn spawn_single_agent(
agent_id: &str,
agent_name: &str,
prompt: &str,
findings_snapshot: Vec<String>,
live: Option<&LiveStateFn>,
) -> 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 def = AgentDefinition::new("workflow-agent".to_string(), "coder".to_string())
.with_max_steps(usize::MAX);
let started_at = chrono::Utc::now().timestamp_millis();
// Notify UI: this agent is now running
if let Some(f) = live {
f(agent_id.to_string(), AgentStatus {
state: AgentState::Running,
started_at: Some(started_at),
completed_at: None,
error: None,
});
}
let def = AgentDefinition::new(agent_name.to_string(), "coder".to_string())
.with_max_steps(50);
let mut ctx = build_subagent_context(def);
let findings_section = if findings_snapshot.is_empty() {
@@ -86,8 +121,41 @@ fn spawn_single_agent(prompt: &str, findings_snapshot: Vec<String>) -> anyhow::R
ctx.system_prompt = format!("{}{}", prompt, findings_section);
let (tx, _rx) = tokio::sync::mpsc::channel(32);
run_subagent(ctx, tx)
// Create an mpsc channel and drain events in a background thread so
// run_subagent's blocking_send never blocks (previously the _rx was
// dropped immediately, which would cause blocking_send to panic/fail
// on a closed channel).
let (tx, rx) = tokio::sync::mpsc::channel(64);
let _drain_thread = std::thread::spawn(move || {
// Drain all events; we don't surface them individually to the UI
// (the live state callbacks handle coarse-grained status).
let mut rx = rx;
while rx.blocking_recv().is_some() {}
});
let result = run_subagent(ctx, tx);
let completed_at = chrono::Utc::now().timestamp_millis();
// Notify UI: agent completed or failed
if let Some(f) = live {
match &result {
Ok(_) => f(agent_id.to_string(), AgentStatus {
state: AgentState::Completed,
started_at: Some(started_at),
completed_at: Some(completed_at),
error: None,
}),
Err(e) => f(agent_id.to_string(), AgentStatus {
state: AgentState::Failed,
started_at: Some(started_at),
completed_at: Some(completed_at),
error: Some(e.to_string()),
}),
}
}
result
}
type ParallelResult = (usize, anyhow::Result<Vec<String>>);
@@ -95,14 +163,16 @@ type ParallelResult = (usize, anyhow::Result<Vec<String>>);
/// Recursively execute a `ScriptPrimitive` tree, respecting an overall
/// concurrency cap for parallel branches.
///
/// Flow: match the primitive ->
/// `Agent` -> `spawn_single_agent`
/// `Parallel` -> spawn threads up to `concurrency_cap`, join
/// `Pipeline` -> spawn threads sequentially, collect in order
/// `Phase` -> recurse (pass-through wrapper)
/// Flow: match the primitive →
/// `Agent` → `spawn_single_agent`
/// `Parallel` → spawn threads up to `concurrency_cap` (semaphore-gated),
/// collect results in submission order
/// `Pipeline` → execute stages sequentially; findings flow between stages
/// `Phase` → recurse (pass-through wrapper)
///
/// Why: parallelism is implemented with `std::thread::spawn` and a
/// counting semaphore so the main async event loop remains unblocked.
/// Why: `Parallel` uses OS threads + a semaphore so the main async event
/// loop remains responsive. `Pipeline` is sequential so each stage sees
/// findings deposited by the previous one.
///
/// Return: a `Vec<String>` of all agent outputs (or error strings) in
/// the order they were submitted.
@@ -111,12 +181,15 @@ pub fn execute_primitive(
args: &HashMap<String, String>,
concurrency_cap: usize,
continue_on_error: bool,
live: Option<&LiveStateFn>,
) -> anyhow::Result<Vec<String>> {
match primitive {
ScriptPrimitive::Agent(prompt) => {
let resolved = resolve_template(prompt, args);
let findings_snapshot = FINDINGS.lock().map(|f| f.clone()).unwrap_or_default();
match spawn_single_agent(&resolved, findings_snapshot) {
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, live) {
Ok(text) => Ok(vec![text]),
Err(e) => {
if continue_on_error {
@@ -129,6 +202,9 @@ pub fn execute_primitive(
}
ScriptPrimitive::Parallel(scripts) => {
// All branches run concurrently, capped by semaphore.
// This is the primary advantage over single-turn chat: multiple
// independent subagents work simultaneously.
let semaphore = Arc::new(Semaphore::new(concurrency_cap.max(1)));
let results: Arc<Mutex<Vec<ParallelResult>>> =
Arc::new(Mutex::new(Vec::new()));
@@ -142,10 +218,14 @@ pub fn execute_primitive(
let sem = Arc::clone(&semaphore);
let results = Arc::clone(&results);
let cap = concurrency_cap;
let live_clone = live.cloned();
std::thread::spawn(move || {
let _permit = sem.acquire();
let result = execute_primitive(&script, &args, cap, continue_on_error);
let result = execute_primitive(
&script, &args, cap, continue_on_error,
live_clone.as_ref(),
);
if let Ok(mut locked) = results.lock() {
locked.push((idx, result));
}
@@ -170,66 +250,69 @@ pub fn execute_primitive(
}
ScriptPrimitive::Pipeline(scripts) => {
let results_store: Arc<Mutex<Vec<Option<Vec<String>>>>> =
Arc::new(Mutex::new(vec![None; scripts.len()]));
let args_arc = Arc::new(args.clone());
let handles: Vec<_> = scripts
.iter()
.enumerate()
.map(|(idx, script)| {
let script = script.clone();
let args = Arc::clone(&args_arc);
let store = Arc::clone(&results_store);
let cap = concurrency_cap;
std::thread::spawn(move || {
let result = execute_primitive(&script, &args, cap, continue_on_error);
if let Ok(mut locked) = store.lock() {
locked[idx] = Some(result.unwrap_or_else(|e| vec![format!("pipeline stage {} error: {}", idx, e)]));
}
})
})
.collect();
for handle in handles {
let _ = handle.join();
}
let locked = results_store.lock().map_err(|_| anyhow::anyhow!("pipeline results lock poisoned"))?;
// Sequential: each stage runs only after the previous completes.
//
// Why: parallel execution defeats the purpose of a pipeline whose
// stages are supposed to build on each other's output. Findings
// written by stage N are visible to stage N+1 because we share
// the global FINDINGS mutex.
let mut all = Vec::new();
for outputs in locked.iter().flatten() {
all.extend(outputs.iter().cloned());
for (idx, script) in scripts.iter().enumerate() {
match execute_primitive(script, args, concurrency_cap, continue_on_error, live) {
Ok(outputs) => all.extend(outputs),
Err(e) => {
if continue_on_error {
all.push(format!("pipeline stage {} error: {}", idx, e));
} else {
return Err(e);
}
}
}
}
Ok(all)
}
ScriptPrimitive::Phase { name: _name, script } => {
execute_primitive(script, args, concurrency_cap, continue_on_error)
execute_primitive(script, args, concurrency_cap, continue_on_error, live)
}
}
}
/// Run a `WorkflowScript` with the given template arguments and produce a
/// summary string.
///
/// Flow: clear the global finding store -> cap concurrency to 5 -> call
/// `execute_primitive` on the script's root primitive -> format results
/// into a one-line-per-agent summary.
/// summary string. Uses no live-state callback.
///
/// Return: a human-readable summary string.
pub fn run_workflow(script: &WorkflowScript, args: &HashMap<String, String>) -> anyhow::Result<String> {
run_workflow_tracked(script, args, None)
}
/// Run a `WorkflowScript` with real-time live-state callbacks so the TUI
/// panel updates as each agent transitions between Idle/Running/Done/Failed.
///
/// Flow: clear the global finding store → cap concurrency to 8 → call
/// `execute_primitive` with the live callback → format results.
///
/// Return: a human-readable summary string.
pub fn run_workflow_tracked(
script: &WorkflowScript,
args: &HashMap<String, String>,
live: Option<LiveStateFn>,
) -> anyhow::Result<String> {
if let Ok(mut findings) = FINDINGS.lock() {
findings.clear();
}
let concurrency_cap = if script.options.max_concurrency > 0 {
script.options.max_concurrency.min(5)
script.options.max_concurrency.min(8) // allow up to 8 parallel agents
} else {
5
4
};
let results = execute_primitive(&script.script, args, concurrency_cap, script.options.continue_on_error)?;
let live_ref = live.as_ref();
let results = execute_primitive(
&script.script, args, concurrency_cap,
script.options.continue_on_error, live_ref,
)?;
let summary = if results.is_empty() {
"workflow completed with no output".to_string()
@@ -261,7 +344,7 @@ pub fn note_finding(text: &str) {
/// Simple template engine: replace `{{key}}` placeholders with values
/// from `args`.
///
/// Why: a structed template engine is unnecessary for the limited
/// Why: a structured template engine is unnecessary for the limited
/// use-case; this is intentionally simple and safe.
fn resolve_template(template: &str, args: &HashMap<String, String>) -> String {
let mut result = template.to_string();