asepharyana
1d50b94eec
- Updated README.md to reflect the addition of 3 new built-in tools, bringing the total to 37. - Revised architecture documentation to indicate the increase in tool count. - Enhanced backend documentation with updated line counts for various modules. - Modified data documentation to change edit log format from JSON to JSONL. - Updated dependencies documentation to reflect version upgrades for several crates. - Improved prompts for auto-reviewer, division implementer, planner, tester, and quality reviewer to enforce stricter coding standards regarding linter bypasses. - Refactored code in various modules to improve clarity and performance, including updates to error handling and tool execution logic. - Added comprehensive tests for IPC frame serialization and deserialization.
635 lines
24 KiB
Rust
635 lines
24 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.
|
|
|
|
use std::collections::HashMap;
|
|
use std::sync::{Arc, Mutex, atomic::{AtomicBool, Ordering}};
|
|
use std::time::Duration;
|
|
use serde::{Deserialize, Serialize};
|
|
use super::script::{ScriptPrimitive, WorkflowScript};
|
|
|
|
/// 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. the division name in the company pipeline).
|
|
pub type LiveStateFn = Arc<dyn Fn(String, 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, 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.
|
|
#[allow(clippy::too_many_lines, clippy::too_many_arguments, clippy::ref_option)]
|
|
fn spawn_single_agent(
|
|
agent_id: &str,
|
|
agent_name: &str,
|
|
prompt: &str,
|
|
findings_snapshot: &[String],
|
|
findings: &Arc<Mutex<Vec<String>>>,
|
|
abort_flag: &Option<Arc<AtomicBool>>,
|
|
live: Option<&LiveStateFn>,
|
|
session_dir: &std::path::Path,
|
|
workspaces: &[std::path::PathBuf],
|
|
timeout_ms: Option<u64>,
|
|
) -> 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. division name).
|
|
if let Some(f) = live {
|
|
f(
|
|
agent_id.to_string(),
|
|
agent_name.to_string(),
|
|
AgentStatus {
|
|
state: AgentState::Running,
|
|
started_at: Some(started_at),
|
|
completed_at: None,
|
|
error: None,
|
|
progress: None,
|
|
},
|
|
);
|
|
}
|
|
|
|
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);
|
|
if let Some(tools) = allowed_tools {
|
|
def = def.with_allowed_tools(tools);
|
|
}
|
|
let mut ctx = build_subagent_context(&def);
|
|
ctx.session_dir = session_dir.to_path_buf();
|
|
ctx.workspaces = workspaces.to_vec();
|
|
|
|
let findings_section = if findings_snapshot.is_empty() {
|
|
String::new()
|
|
} else {
|
|
format!(
|
|
"\n\nFindings from sibling agents in this workflow run:\n{}",
|
|
findings_snapshot
|
|
.iter()
|
|
.enumerate()
|
|
.map(|(i, f)| format!("{}. {}", i + 1, f))
|
|
.collect::<Vec<_>>()
|
|
.join("\n")
|
|
)
|
|
};
|
|
|
|
ctx.system_prompt = format!("{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(findings.clone());
|
|
ctx.abort_flag.clone_from(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 = agent_id.to_string();
|
|
let drain_agent_name = agent_name.to_string();
|
|
let drain_live = 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, .. } => {
|
|
tracing::debug!("[subagent] tool call: {}", tool);
|
|
// Push intra-division progress: which tool is running
|
|
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(format!("tool: {tool}")),
|
|
},
|
|
);
|
|
}
|
|
}
|
|
SubagentEvent::ToolResult { tool, .. } => {
|
|
tracing::debug!("[subagent] tool result: {}", tool);
|
|
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(format!("done: {tool}")),
|
|
},
|
|
);
|
|
}
|
|
}
|
|
SubagentEvent::StepCompleted { .. } => {
|
|
tracing::trace!("[subagent] step completed");
|
|
}
|
|
SubagentEvent::StepFailed { step, error } => {
|
|
tracing::warn!("[subagent] step {} failed: {}", step, error);
|
|
}
|
|
SubagentEvent::Completed { .. } => {
|
|
tracing::debug!("[subagent] completed");
|
|
}
|
|
}
|
|
}
|
|
});
|
|
|
|
// Check abort before even starting the subagent.
|
|
if abort_flag.as_ref().is_some_and(|f| f.load(Ordering::SeqCst)) {
|
|
anyhow::bail!("subagent '{agent_name}' aborted before start");
|
|
}
|
|
|
|
// 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 = agent_name.to_string();
|
|
let bg_abort = 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) = 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) = live {
|
|
match &result {
|
|
Ok(_) => f(
|
|
agent_id.to_string(),
|
|
agent_name.to_string(),
|
|
AgentStatus {
|
|
state: AgentState::Completed,
|
|
started_at: Some(started_at),
|
|
completed_at: Some(completed_at),
|
|
error: None,
|
|
progress: None,
|
|
},
|
|
),
|
|
Err(e) => f(
|
|
agent_id.to_string(),
|
|
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
|
|
}
|
|
|
|
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` (semaphore-gated),
|
|
/// collect results in submission order
|
|
/// `Pipeline` → execute stages sequentially; findings flow between stages
|
|
/// `Phase` → recurse (pass-through wrapper)
|
|
///
|
|
/// 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. Findings are scoped to an
|
|
/// `Arc<Mutex<Vec<String>>>` rather than a global static, so concurrent
|
|
/// workflow runs are isolated from each other.
|
|
///
|
|
/// `timeout_ms` propagates to individual agents so that no single agent
|
|
/// can block the entire workflow beyond the configured deadline.
|
|
///
|
|
/// Return: a `Vec<String>` of all agent outputs (or error strings) in
|
|
/// the order they were submitted.
|
|
#[allow(clippy::too_many_arguments)]
|
|
#[allow(clippy::ref_option, clippy::too_many_lines)]
|
|
pub fn execute_primitive(
|
|
primitive: &ScriptPrimitive,
|
|
args: &HashMap<String, String>,
|
|
concurrency_cap: usize,
|
|
continue_on_error: bool,
|
|
abort_flag: &Option<Arc<AtomicBool>>,
|
|
live: Option<&LiveStateFn>,
|
|
session_dir: &std::path::Path,
|
|
workspaces: &[std::path::PathBuf],
|
|
findings: &Arc<Mutex<Vec<String>>>,
|
|
timeout_ms: Option<u64>,
|
|
) -> anyhow::Result<Vec<String>> {
|
|
match primitive {
|
|
ScriptPrimitive::Agent(prompt) => {
|
|
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 = 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) {
|
|
Ok(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
|
|
// independent subagents work simultaneously.
|
|
// Each branch shares the same `findings` Arc so note_finding
|
|
// calls within any branch are visible to all other branches.
|
|
let semaphore = Arc::new(Semaphore::new(concurrency_cap.max(1)));
|
|
let results: Arc<Mutex<Vec<ParallelResult>>> =
|
|
Arc::new(Mutex::new(Vec::new()));
|
|
|
|
let handles: Vec<_> = scripts
|
|
.iter()
|
|
.enumerate()
|
|
.map(|(idx, script)| {
|
|
let script = script.clone();
|
|
let args = args.clone();
|
|
let sem = Arc::clone(&semaphore);
|
|
let results = Arc::clone(&results);
|
|
let cap = concurrency_cap;
|
|
let abort = abort_flag.clone();
|
|
let live_clone = live.cloned();
|
|
let session_dir = session_dir.to_path_buf();
|
|
let workspaces = workspaces.to_vec();
|
|
let findings = Arc::clone(findings);
|
|
let to = timeout_ms;
|
|
|
|
std::thread::spawn(move || {
|
|
let _permit = sem.acquire();
|
|
let result = execute_primitive(
|
|
&script, &args, cap, continue_on_error,
|
|
&abort,
|
|
live_clone.as_ref(),
|
|
&session_dir,
|
|
&workspaces,
|
|
&findings,
|
|
to,
|
|
);
|
|
if let Ok(mut locked) = results.lock() {
|
|
locked.push((idx, result));
|
|
}
|
|
})
|
|
})
|
|
.collect();
|
|
|
|
for handle in handles {
|
|
let _ = handle.join();
|
|
}
|
|
|
|
let mut locked = results.lock().map_err(|_| anyhow::anyhow!("parallel results lock poisoned"))?;
|
|
locked.sort_by_key(|(idx, _)| *idx);
|
|
let mut all = Vec::new();
|
|
for (_, res) in locked.drain(..) {
|
|
match res {
|
|
Ok(outputs) => all.extend(outputs),
|
|
Err(e) => all.push(format!("agent error: {e}")),
|
|
}
|
|
}
|
|
Ok(all)
|
|
}
|
|
|
|
ScriptPrimitive::Pipeline(scripts) => {
|
|
// Sequential: each stage runs only after the previous completes.
|
|
//
|
|
// Abort is checked between stages so the user can cancel the
|
|
// pipeline immediately when moving to the next division, rather
|
|
// than having to wait for the current subagent to finish.
|
|
//
|
|
// 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 through the shared
|
|
// `findings` Arc (same isolation scope as parent).
|
|
let mut all = Vec::new();
|
|
for (idx, script) in scripts.iter().enumerate() {
|
|
// Check abort before each pipeline stage so we don't
|
|
// launch the next division after the user cancelled.
|
|
if abort_flag.as_ref().is_some_and(|f| f.load(Ordering::SeqCst)) {
|
|
if continue_on_error {
|
|
all.push(format!("pipeline aborted at stage {idx}"));
|
|
break;
|
|
}
|
|
anyhow::bail!("pipeline aborted by user at stage {idx}");
|
|
}
|
|
match execute_primitive(script, args, concurrency_cap, continue_on_error, abort_flag, live, session_dir, workspaces, findings, timeout_ms) {
|
|
Ok(outputs) => all.extend(outputs),
|
|
Err(e) => {
|
|
if continue_on_error {
|
|
all.push(format!("pipeline stage {idx} error: {e}"));
|
|
} else {
|
|
return Err(e);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
Ok(all)
|
|
}
|
|
|
|
ScriptPrimitive::Phase { name: _name, script } => {
|
|
execute_primitive(script, args, concurrency_cap, continue_on_error, abort_flag, live, session_dir, workspaces, findings, timeout_ms)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Run a `WorkflowScript` with the given template arguments and produce a
|
|
/// summary string. Uses no live-state callback.
|
|
///
|
|
/// Return: a human-readable summary string.
|
|
pub fn run_workflow(
|
|
script: &WorkflowScript,
|
|
args: &HashMap<String, String>,
|
|
session_dir: &std::path::Path,
|
|
workspaces: &[std::path::PathBuf],
|
|
) -> anyhow::Result<String> {
|
|
run_workflow_tracked(script, args, &None, None, session_dir, workspaces)
|
|
}
|
|
|
|
/// Run a `WorkflowScript` with real-time live-state callbacks so the TUI
|
|
/// panel updates as each agent transitions between Idle/Running/Done/Failed.
|
|
///
|
|
/// Flow: create an empty findings Arc (scoped to this invocation) → cap
|
|
/// concurrency to 8 → call `execute_primitive` with the live callback and
|
|
/// findings → format results.
|
|
///
|
|
/// Why: findings are scoped to an `Arc<Mutex<Vec<String>>>` rather than a
|
|
/// global static, so concurrent `run_workflow_tracked` calls from different
|
|
/// `spawn_agents` invocations remain fully isolated.
|
|
///
|
|
/// Return: a human-readable summary string.
|
|
#[allow(clippy::ref_option)]
|
|
pub fn run_workflow_tracked(
|
|
script: &WorkflowScript,
|
|
args: &HashMap<String, String>,
|
|
abort_flag: &Option<Arc<AtomicBool>>,
|
|
live: Option<&LiveStateFn>,
|
|
session_dir: &std::path::Path,
|
|
workspaces: &[std::path::PathBuf],
|
|
) -> anyhow::Result<String> {
|
|
let concurrency_cap = if script.options.max_concurrency > 0 {
|
|
script.options.max_concurrency.min(10) // allow up to 10 parallel agents
|
|
} else {
|
|
10
|
|
};
|
|
|
|
let findings = Arc::new(Mutex::new(Vec::new()));
|
|
let results = execute_primitive(
|
|
&script.script, args, concurrency_cap,
|
|
script.options.continue_on_error, abort_flag, live,
|
|
session_dir, workspaces, &findings,
|
|
script.options.timeout_ms,
|
|
)?;
|
|
|
|
let summary = if results.is_empty() {
|
|
"workflow completed with no output".to_string()
|
|
} else {
|
|
format!(
|
|
"workflow '{}' completed. {} agent result(s):\n{}",
|
|
script.name,
|
|
results.len(),
|
|
results
|
|
.iter()
|
|
.enumerate()
|
|
.map(|(i, r)| format!("[{}] {}", i + 1, r.lines().next().unwrap_or(r)))
|
|
.collect::<Vec<_>>()
|
|
.join("\n")
|
|
)
|
|
};
|
|
|
|
Ok(summary)
|
|
}
|
|
|
|
/// Simple template engine: replace `{{key}}` placeholders with values
|
|
/// from `args`.
|
|
///
|
|
/// 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();
|
|
for (key, value) in args {
|
|
result = result.replace(&format!("{{{{{key}}}}}"), value);
|
|
}
|
|
result
|
|
}
|
|
|
|
/// A counting semaphore built from a `Mutex` + `Condvar`.
|
|
///
|
|
/// Used by `execute_primitive` to cap concurrent parallel branches.
|
|
///
|
|
/// Panic-safety: if a thread panics while holding a permit, the Mutex
|
|
/// becomes poisoned. Both `acquire` and the `Drop` implementation recover
|
|
/// from poisoned mutexes by discarding the poison, ensuring the semaphore
|
|
/// remains usable after a thread panic.
|
|
struct Semaphore {
|
|
count: Mutex<usize>,
|
|
condvar: std::sync::Condvar,
|
|
}
|
|
|
|
impl Semaphore {
|
|
fn new(count: usize) -> Self {
|
|
Semaphore {
|
|
count: Mutex::new(count),
|
|
condvar: std::sync::Condvar::new(),
|
|
}
|
|
}
|
|
|
|
fn acquire(&self) -> SemaphoreGuard<'_> {
|
|
let mut count = self.count.lock().unwrap_or_else(|e| {
|
|
tracing::warn!("[semaphore] mutex poisoned in acquire, recovering");
|
|
e.into_inner()
|
|
});
|
|
while *count == 0 {
|
|
count = self.condvar.wait(count).unwrap_or_else(|e| {
|
|
tracing::warn!("[semaphore] mutex poisoned in wait, recovering");
|
|
e.into_inner()
|
|
});
|
|
}
|
|
*count -= 1;
|
|
SemaphoreGuard { sem: self }
|
|
}
|
|
}
|
|
|
|
struct SemaphoreGuard<'a> {
|
|
sem: &'a Semaphore,
|
|
}
|
|
|
|
impl Drop for SemaphoreGuard<'_> {
|
|
fn drop(&mut self) {
|
|
let mut count = self.sem.count.lock().unwrap_or_else(|e| {
|
|
tracing::warn!("[semaphore] mutex poisoned in drop, recovering");
|
|
e.into_inner()
|
|
});
|
|
*count += 1;
|
|
self.sem.condvar.notify_one();
|
|
}
|
|
}
|