scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-05 08:10:39 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

Rename Pipeline to Workflow/Run in arc-workflows engine internals

Browse source 
- pipeline.rs → workflow.rs: PipelineBuilder → WorkflowBuilder,
  prepare_pipeline() → prepare_workflow()
- event.rs: PipelineEvent → WorkflowRunEvent, variant renames
  PipelineStarted/Completed/Failed → WorkflowRunStarted/Completed/Failed
- engine.rs: PipelineEngine → WorkflowRunEngine, pipeline_name → workflow_name
- retro.rs: pipeline_name → workflow_name, pipeline_failed → run_failed
- All handler, CLI, and test files updated to match

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-03-01 23:58:42 -05:00
parent 4c971d81b1
commit 212de7a7ea
17 changed files with 566 additions and 566 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
crates/arc-workflows/src/cli/backend.rs
View file

@ -231,7 +231,7 @@ impl CodergenBackend for AgentApiBackend {
let pipeline_emitter = Arc::clone(emitter);
let mut rx = session.subscribe();
tokio::spawn(async move {
use crate::event::PipelineEvent;
use crate::event::WorkflowRunEvent;
while let Ok(event) = rx.recv().await {
// Track file changes from tool calls
match &event.event {
@ -276,7 +276,7 @@ impl CodergenBackend for AgentApiBackend {
| AgentEvent::ToolCallOutputDelta { .. }
| AgentEvent::SkillExpanded { .. }
) {
pipeline_emitter.emit(&PipelineEvent::Agent {
pipeline_emitter.emit(&WorkflowRunEvent::Agent {
stage: node_id.clone(),
event: event.event.clone(),
});
@ -329,7 +329,7 @@ impl CodergenBackend for AgentApiBackend {
});
// Emit Prompt event before processing
emitter.emit(&crate::event::PipelineEvent::Prompt {
emitter.emit(&crate::event::WorkflowRunEvent::Prompt {
stage: node.id.clone(),
text: prompt.to_string(),
});

174
crates/arc-workflows/src/cli/mod.rs
View file

@ -13,7 +13,7 @@ use std::fmt;
use std::path::PathBuf;
use std::str::FromStr;
use crate::event::PipelineEvent;
use crate::event::WorkflowRunEvent;
use crate::outcome::StageUsage;
use crate::validation::{Diagnostic, Severity};
use arc_agent::AgentEvent;
@ -57,7 +57,7 @@ impl FromStr for ExecutionEnvKind {
#[command(
name = "arc-workflows",
version,
about = "DOT-based pipeline runner for AI workflows"
about = "DOT-based workflow runner for AI workflows"
)]
pub struct Cli {
#[command(subcommand)]
@ -66,17 +66,17 @@ pub struct Cli {
#[derive(Subcommand)]
pub enum Command {
/// Launch a pipeline from a .dot or .toml task file
/// Launch a workflow from a .dot or .toml task file
Run(RunArgs),
/// Parse and validate a pipeline without executing
/// Parse and validate a workflow without executing
Validate(ValidateArgs),
}
#[derive(Args)]
pub struct RunArgs {
/// Path to a .dot pipeline file or .toml task config (not required with --run-branch)
/// Path to a .dot workflow file or .toml task config (not required with --run-branch)
#[arg(required_unless_present = "run_branch")]
pub pipeline: Option<PathBuf>,
pub workflow: Option<PathBuf>,
/// Log/artifact directory
#[arg(long)]
@ -121,8 +121,8 @@ pub struct RunArgs {
#[derive(Args)]
pub struct ValidateArgs {
/// Path to the .dot pipeline file
pub pipeline: PathBuf,
/// Path to the .dot workflow file
pub workflow: PathBuf,
}
/// Read a .dot file from disk.
@ -191,32 +191,32 @@ pub fn format_duration_human(ms: u64) -> String {
}
}
/// One-line summary of a pipeline event for `-v` output (dimmed).
/// One-line summary of a workflow run event for `-v` output (dimmed).
#[must_use]
pub fn format_event_summary(event: &PipelineEvent, styles: &Styles) -> String {
pub fn format_event_summary(event: &WorkflowRunEvent, styles: &Styles) -> String {
let body = match event {
PipelineEvent::PipelineStarted { name, run_id, .. } => {
format!("[PIPELINE_STARTED] name={name} id={run_id}")
WorkflowRunEvent::WorkflowRunStarted { name, run_id, .. } => {
format!("[WORKFLOW_RUN_STARTED] name={name} id={run_id}")
}
PipelineEvent::PipelineCompleted {
WorkflowRunEvent::WorkflowRunCompleted {
duration_ms,
artifact_count,
total_cost,
..
} => {
let mut s =
format!("[PIPELINE_COMPLETED] duration={duration_ms}ms artifacts={artifact_count}");
format!("[WORKFLOW_RUN_COMPLETED] duration={duration_ms}ms artifacts={artifact_count}");
if let Some(cost) = total_cost {
s.push_str(&format!(" total_cost={}", format_cost(*cost)));
}
s
}
PipelineEvent::PipelineFailed {
WorkflowRunEvent::WorkflowRunFailed {
error, duration_ms, ..
} => {
format!("[PIPELINE_FAILED] error=\"{error}\" duration={duration_ms}ms")
format!("[WORKFLOW_RUN_FAILED] error=\"{error}\" duration={duration_ms}ms")
}
PipelineEvent::StageStarted {
WorkflowRunEvent::StageStarted {
name,
index,
handler_type,
@ -230,7 +230,7 @@ pub fn format_event_summary(event: &PipelineEvent, styles: &Styles) -> String {
s.push_str(&format!(" attempt={attempt}/{max_attempts}"));
s
}
PipelineEvent::StageCompleted {
WorkflowRunEvent::StageCompleted {
name,
index,
duration_ms,
@ -279,7 +279,7 @@ pub fn format_event_summary(event: &PipelineEvent, styles: &Styles) -> String {
}
s
}
PipelineEvent::StageFailed {
WorkflowRunEvent::StageFailed {
name,
index,
error,
@ -298,7 +298,7 @@ pub fn format_event_summary(event: &PipelineEvent, styles: &Styles) -> String {
}
s
}
PipelineEvent::StageRetrying {
WorkflowRunEvent::StageRetrying {
name,
index,
attempt,
@ -309,17 +309,17 @@ pub fn format_event_summary(event: &PipelineEvent, styles: &Styles) -> String {
"[STAGE_RETRYING] name={name} index={index} attempt={attempt}/{max_attempts} delay={delay_ms}ms"
)
}
PipelineEvent::ParallelStarted {
WorkflowRunEvent::ParallelStarted {
branch_count,
join_policy,
error_policy,
} => {
format!("[PARALLEL_STARTED] branches={branch_count} join_policy={join_policy} error_policy={error_policy}")
}
PipelineEvent::ParallelBranchStarted { branch, index } => {
WorkflowRunEvent::ParallelBranchStarted { branch, index } => {
format!("[PARALLEL_BRANCH_STARTED] branch={branch} index={index}")
}
PipelineEvent::ParallelBranchCompleted {
WorkflowRunEvent::ParallelBranchCompleted {
branch,
index,
duration_ms,
@ -327,21 +327,21 @@ pub fn format_event_summary(event: &PipelineEvent, styles: &Styles) -> String {
} => {
format!("[PARALLEL_BRANCH_COMPLETED] branch={branch} index={index} duration={duration_ms}ms status={status}")
}
PipelineEvent::ParallelCompleted {
WorkflowRunEvent::ParallelCompleted {
duration_ms,
success_count,
failure_count,
} => {
format!("[PARALLEL_COMPLETED] duration={duration_ms}ms succeeded={success_count} failed={failure_count}")
}
PipelineEvent::InterviewStarted {
WorkflowRunEvent::InterviewStarted {
question,
stage,
question_type,
} => {
format!("[INTERVIEW_STARTED] stage={stage} question=\"{question}\" question_type={question_type}")
}
PipelineEvent::InterviewCompleted {
WorkflowRunEvent::InterviewCompleted {
question,
answer,
duration_ms,
@ -350,15 +350,15 @@ pub fn format_event_summary(event: &PipelineEvent, styles: &Styles) -> String {
"[INTERVIEW_COMPLETED] question=\"{question}\" answer=\"{answer}\" duration={duration_ms}ms"
)
}
PipelineEvent::InterviewTimeout {
WorkflowRunEvent::InterviewTimeout {
stage, duration_ms, ..
} => {
format!("[INTERVIEW_TIMEOUT] stage={stage} duration={duration_ms}ms")
}
PipelineEvent::CheckpointSaved { node_id } => {
WorkflowRunEvent::CheckpointSaved { node_id } => {
format!("[CHECKPOINT_SAVED] node={node_id}")
}
PipelineEvent::GitCheckpoint {
WorkflowRunEvent::GitCheckpoint {
node_id,
git_commit_sha,
status,
@ -366,7 +366,7 @@ pub fn format_event_summary(event: &PipelineEvent, styles: &Styles) -> String {
} => {
format!("[GIT_CHECKPOINT] node={node_id} sha={git_commit_sha} status={status}")
}
PipelineEvent::EdgeSelected {
WorkflowRunEvent::EdgeSelected {
from_node,
to_node,
label,
@ -381,14 +381,14 @@ pub fn format_event_summary(event: &PipelineEvent, styles: &Styles) -> String {
}
s
}
PipelineEvent::LoopRestart { from_node, to_node } => {
WorkflowRunEvent::LoopRestart { from_node, to_node } => {
format!("[LOOP_RESTART] from={from_node} to={to_node}")
}
PipelineEvent::Prompt { stage, text } => {
WorkflowRunEvent::Prompt { stage, text } => {
let truncated = if text.len() > 80 { &text[..80] } else { text };
format!("[PROMPT] stage={stage} text=\"{truncated}\"")
}
PipelineEvent::Agent { stage, event } => match event {
WorkflowRunEvent::Agent { stage, event } => match event {
AgentEvent::AssistantMessage {
model,
usage,
@ -490,20 +490,20 @@ pub fn format_event_summary(event: &PipelineEvent, styles: &Styles) -> String {
}
other => format!("[AGENT] stage={stage} event={other:?}"),
},
PipelineEvent::ParallelEarlyTermination {
WorkflowRunEvent::ParallelEarlyTermination {
reason,
completed_count,
pending_count,
} => {
format!("[PARALLEL_EARLY_TERMINATION] reason={reason} completed={completed_count} pending={pending_count}")
}
PipelineEvent::SubgraphStarted {
WorkflowRunEvent::SubgraphStarted {
node_id,
start_node,
} => {
format!("[SUBGRAPH_STARTED] node={node_id} start_node={start_node}")
}
PipelineEvent::SubgraphCompleted {
WorkflowRunEvent::SubgraphCompleted {
node_id,
steps_executed,
status,
@ -511,7 +511,7 @@ pub fn format_event_summary(event: &PipelineEvent, styles: &Styles) -> String {
} => {
format!("[SUBGRAPH_COMPLETED] node={node_id} steps={steps_executed} status={status} duration={duration_ms}ms")
}
PipelineEvent::ExecutionEnv { event } => {
WorkflowRunEvent::ExecutionEnv { event } => {
use arc_agent::ExecutionEnvEvent;
match event {
ExecutionEnvEvent::Initializing { env_type } => format!("[EXEC_ENV_INITIALIZING] env_type={env_type}"),
@ -534,13 +534,13 @@ pub fn format_event_summary(event: &PipelineEvent, styles: &Styles) -> String {
ExecutionEnvEvent::GitCloneFailed { url, error } => format!("[EXEC_ENV_GIT_CLONE_FAILED] url={url} error=\"{error}\""),
}
}
PipelineEvent::SetupStarted { command_count } => {
WorkflowRunEvent::SetupStarted { command_count } => {
format!("[SETUP_STARTED] command_count={command_count}")
}
PipelineEvent::SetupCommandStarted { command, index } => {
WorkflowRunEvent::SetupCommandStarted { command, index } => {
format!("[SETUP_COMMAND_STARTED] index={index} command=\"{command}\"")
}
PipelineEvent::SetupCommandCompleted {
WorkflowRunEvent::SetupCommandCompleted {
command,
index,
exit_code,
@ -548,10 +548,10 @@ pub fn format_event_summary(event: &PipelineEvent, styles: &Styles) -> String {
} => {
format!("[SETUP_COMMAND_COMPLETED] index={index} command=\"{command}\" exit_code={exit_code} duration={duration_ms}ms")
}
PipelineEvent::SetupCompleted { duration_ms } => {
WorkflowRunEvent::SetupCompleted { duration_ms } => {
format!("[SETUP_COMPLETED] duration={duration_ms}ms")
}
PipelineEvent::SetupFailed {
WorkflowRunEvent::SetupFailed {
command,
index,
exit_code,
@ -564,42 +564,42 @@ pub fn format_event_summary(event: &PipelineEvent, styles: &Styles) -> String {
};
format!("[SETUP_FAILED] index={index} command=\"{command}\" exit_code={exit_code} stderr=\"{truncated}\"")
}
PipelineEvent::StallWatchdogTimeout { node, idle_seconds } => {
WorkflowRunEvent::StallWatchdogTimeout { node, idle_seconds } => {
format!("[STALL_WATCHDOG_TIMEOUT] node={node} idle_seconds={idle_seconds}")
}
};
format!("{dim}{body}{reset}", dim = styles.dim, reset = styles.reset)
}
/// Multi-line detail view of a pipeline event for `-vv` output.
/// Multi-line detail view of a workflow run event for `-vv` output.
/// Box-drawing is dimmed; values are normal.
#[must_use]
pub fn format_event_detail(event: &PipelineEvent, styles: &Styles) -> String {
pub fn format_event_detail(event: &WorkflowRunEvent, styles: &Styles) -> String {
let d = styles.dim;
let r = styles.reset;
match event {
PipelineEvent::PipelineStarted { name, run_id, .. } => {
WorkflowRunEvent::WorkflowRunStarted { name, run_id, .. } => {
format!(
"{d}── PIPELINE_STARTED ─────────────────────────{r}\n {d}name:{r} {name}\n {d}id:{r} {run_id}\n"
"{d}── WORKFLOW_RUN_STARTED ─────────────────────────{r}\n {d}name:{r} {name}\n {d}id:{r} {run_id}\n"
)
}
PipelineEvent::PipelineCompleted {
WorkflowRunEvent::WorkflowRunCompleted {
duration_ms,
artifact_count,
total_cost,
..
} => {
let mut s = format!("{d}── PIPELINE_COMPLETED ───────────────────────{r}\n {d}duration_ms:{r} {duration_ms}\n {d}artifact_count:{r} {artifact_count}\n");
let mut s = format!("{d}── WORKFLOW_RUN_COMPLETED ───────────────────────{r}\n {d}duration_ms:{r} {duration_ms}\n {d}artifact_count:{r} {artifact_count}\n");
if let Some(cost) = total_cost {
s.push_str(&format!(" {d}total_cost:{r} {}\n", format_cost(*cost)));
}
s
}
PipelineEvent::PipelineFailed { error, duration_ms, .. } => {
format!("{d}── PIPELINE_FAILED ──────────────────────────{r}\n {d}error:{r} {error}\n {d}duration_ms:{r} {duration_ms}\n")
WorkflowRunEvent::WorkflowRunFailed { error, duration_ms, .. } => {
format!("{d}── WORKFLOW_RUN_FAILED ──────────────────────────{r}\n {d}error:{r} {error}\n {d}duration_ms:{r} {duration_ms}\n")
}
PipelineEvent::StageStarted { name, index, handler_type, attempt, max_attempts } => {
WorkflowRunEvent::StageStarted { name, index, handler_type, attempt, max_attempts } => {
let mut s = format!(
"{d}── STAGE_STARTED ────────────────────────────{r}\n {d}name:{r} {name}\n {d}index:{r} {index}\n"
);
@ -609,7 +609,7 @@ pub fn format_event_detail(event: &PipelineEvent, styles: &Styles) -> String {
s.push_str(&format!(" {d}attempt:{r} {attempt}/{max_attempts}\n"));
s
}
PipelineEvent::StageCompleted {
WorkflowRunEvent::StageCompleted {
name,
index,
duration_ms,
@ -667,7 +667,7 @@ pub fn format_event_detail(event: &PipelineEvent, styles: &Styles) -> String {
}
s
}
PipelineEvent::StageFailed {
WorkflowRunEvent::StageFailed {
name,
index,
error,
@ -684,7 +684,7 @@ pub fn format_event_detail(event: &PipelineEvent, styles: &Styles) -> String {
}
s
}
PipelineEvent::StageRetrying {
WorkflowRunEvent::StageRetrying {
name,
index,
attempt,
@ -693,13 +693,13 @@ pub fn format_event_detail(event: &PipelineEvent, styles: &Styles) -> String {
} => {
format!("{d}── STAGE_RETRYING ───────────────────────────{r}\n {d}name:{r} {name}\n {d}index:{r} {index}\n {d}attempt:{r} {attempt}/{max_attempts}\n {d}delay_ms:{r} {delay_ms}\n")
}
PipelineEvent::ParallelStarted { branch_count, join_policy, error_policy } => {
WorkflowRunEvent::ParallelStarted { branch_count, join_policy, error_policy } => {
format!("{d}── PARALLEL_STARTED ─────────────────────────{r}\n {d}branch_count:{r} {branch_count}\n {d}join_policy:{r} {join_policy}\n {d}error_policy:{r} {error_policy}\n")
}
PipelineEvent::ParallelBranchStarted { branch, index } => {
WorkflowRunEvent::ParallelBranchStarted { branch, index } => {
format!("{d}── PARALLEL_BRANCH_STARTED ──────────────────{r}\n {d}branch:{r} {branch}\n {d}index:{r} {index}\n")
}
PipelineEvent::ParallelBranchCompleted {
WorkflowRunEvent::ParallelBranchCompleted {
branch,
index,
duration_ms,
@ -707,41 +707,41 @@ pub fn format_event_detail(event: &PipelineEvent, styles: &Styles) -> String {
} => {
format!("{d}── PARALLEL_BRANCH_COMPLETED ────────────────{r}\n {d}branch:{r} {branch}\n {d}index:{r} {index}\n {d}duration_ms:{r} {duration_ms}\n {d}status:{r} {status}\n")
}
PipelineEvent::ParallelCompleted {
WorkflowRunEvent::ParallelCompleted {
duration_ms,
success_count,
failure_count,
} => {
format!("{d}── PARALLEL_COMPLETED ───────────────────────{r}\n {d}duration_ms:{r} {duration_ms}\n {d}success_count:{r} {success_count}\n {d}failure_count:{r} {failure_count}\n")
}
PipelineEvent::InterviewStarted { question, stage, question_type } => {
WorkflowRunEvent::InterviewStarted { question, stage, question_type } => {
format!("{d}── INTERVIEW_STARTED ────────────────────────{r}\n {d}stage:{r} {stage}\n {d}question:{r} {question}\n {d}question_type:{r} {question_type}\n")
}
PipelineEvent::InterviewCompleted {
WorkflowRunEvent::InterviewCompleted {
question,
answer,
duration_ms,
} => {
format!("{d}── INTERVIEW_COMPLETED ──────────────────────{r}\n {d}question:{r} {question}\n {d}answer:{r} {answer}\n {d}duration_ms:{r} {duration_ms}\n")
}
PipelineEvent::InterviewTimeout {
WorkflowRunEvent::InterviewTimeout {
question,
stage,
duration_ms,
} => {
format!("{d}── INTERVIEW_TIMEOUT ────────────────────────{r}\n {d}question:{r} {question}\n {d}stage:{r} {stage}\n {d}duration_ms:{r} {duration_ms}\n")
}
PipelineEvent::CheckpointSaved { node_id } => {
WorkflowRunEvent::CheckpointSaved { node_id } => {
format!(
"{d}── CHECKPOINT_SAVED ─────────────────────────{r}\n {d}node_id:{r} {node_id}\n"
)
}
PipelineEvent::GitCheckpoint { run_id, node_id, status, git_commit_sha } => {
WorkflowRunEvent::GitCheckpoint { run_id, node_id, status, git_commit_sha } => {
format!(
"{d}── GIT_CHECKPOINT ───────────────────────────{r}\n {d}run_id:{r} {run_id}\n {d}node_id:{r} {node_id}\n {d}status:{r} {status}\n {d}sha:{r} {git_commit_sha}\n"
)
}
PipelineEvent::EdgeSelected { from_node, to_node, label, condition } => {
WorkflowRunEvent::EdgeSelected { from_node, to_node, label, condition } => {
let mut s = format!("{d}── EDGE_SELECTED ────────────────────────────{r}\n {d}from:{r} {from_node}\n {d}to:{r} {to_node}\n");
if let Some(l) = label {
s.push_str(&format!(" {d}label:{r} {l}\n"));
@ -751,13 +751,13 @@ pub fn format_event_detail(event: &PipelineEvent, styles: &Styles) -> String {
}
s
}
PipelineEvent::LoopRestart { from_node, to_node } => {
WorkflowRunEvent::LoopRestart { from_node, to_node } => {
format!("{d}── LOOP_RESTART ─────────────────────────────{r}\n {d}from:{r} {from_node}\n {d}to:{r} {to_node}\n")
}
PipelineEvent::Prompt { stage, text } => {
WorkflowRunEvent::Prompt { stage, text } => {
format!("{d}── PROMPT ───────────────────────────────────{r}\n {d}stage:{r} {stage}\n {d}text:{r}\n{text}\n")
}
PipelineEvent::Agent { stage, event } => match event {
WorkflowRunEvent::Agent { stage, event } => match event {
AgentEvent::AssistantMessage { text, model, usage, tool_call_count } => {
let total = usage.input_tokens + usage.output_tokens;
let truncated = if text.len() > 200 { &text[..200] } else { text.as_str() };
@ -828,17 +828,17 @@ pub fn format_event_detail(event: &PipelineEvent, styles: &Styles) -> String {
}
other => format!("{d}── AGENT ────────────────────────────────────{r}\n {d}stage:{r} {stage}\n {d}event:{r} {other:?}\n"),
}
PipelineEvent::ParallelEarlyTermination {
WorkflowRunEvent::ParallelEarlyTermination {
reason,
completed_count,
pending_count,
} => {
format!("{d}── PARALLEL_EARLY_TERMINATION ───────────────{r}\n {d}reason:{r} {reason}\n {d}completed_count:{r} {completed_count}\n {d}pending_count:{r} {pending_count}\n")
}
PipelineEvent::SubgraphStarted { node_id, start_node } => {
WorkflowRunEvent::SubgraphStarted { node_id, start_node } => {
format!("{d}── SUBGRAPH_STARTED ─────────────────────────{r}\n {d}node_id:{r} {node_id}\n {d}start_node:{r} {start_node}\n")
}
PipelineEvent::SubgraphCompleted {
WorkflowRunEvent::SubgraphCompleted {
node_id,
steps_executed,
status,
@ -846,7 +846,7 @@ pub fn format_event_detail(event: &PipelineEvent, styles: &Styles) -> String {
} => {
format!("{d}── SUBGRAPH_COMPLETED ───────────────────────{r}\n {d}node_id:{r} {node_id}\n {d}steps_executed:{r} {steps_executed}\n {d}status:{r} {status}\n {d}duration_ms:{r} {duration_ms}\n")
}
PipelineEvent::ExecutionEnv { event } => {
WorkflowRunEvent::ExecutionEnv { event } => {
use arc_agent::ExecutionEnvEvent;
match event {
ExecutionEnvEvent::Initializing { env_type } => {
@ -897,22 +897,22 @@ pub fn format_event_detail(event: &PipelineEvent, styles: &Styles) -> String {
}
}
}
PipelineEvent::SetupStarted { command_count } => {
WorkflowRunEvent::SetupStarted { command_count } => {
format!("{d}── SETUP_STARTED ────────────────────────────{r}\n {d}command_count:{r} {command_count}\n")
}
PipelineEvent::SetupCommandStarted { command, index } => {
WorkflowRunEvent::SetupCommandStarted { command, index } => {
format!("{d}── SETUP_COMMAND_STARTED ────────────────────{r}\n {d}index:{r} {index}\n {d}command:{r} {command}\n")
}
PipelineEvent::SetupCommandCompleted { command, index, exit_code, duration_ms } => {
WorkflowRunEvent::SetupCommandCompleted { command, index, exit_code, duration_ms } => {
format!("{d}── SETUP_COMMAND_COMPLETED ──────────────────{r}\n {d}index:{r} {index}\n {d}command:{r} {command}\n {d}exit_code:{r} {exit_code}\n {d}duration_ms:{r} {duration_ms}\n")
}
PipelineEvent::SetupCompleted { duration_ms } => {
WorkflowRunEvent::SetupCompleted { duration_ms } => {
format!("{d}── SETUP_COMPLETED ──────────────────────────{r}\n {d}duration_ms:{r} {duration_ms}\n")
}
PipelineEvent::SetupFailed { command, index, exit_code, stderr } => {
WorkflowRunEvent::SetupFailed { command, index, exit_code, stderr } => {
format!("{d}── SETUP_FAILED ─────────────────────────────{r}\n {d}index:{r} {index}\n {d}command:{r} {command}\n {d}exit_code:{r} {exit_code}\n {d}stderr:{r} {stderr}\n")
}
PipelineEvent::StallWatchdogTimeout { node, idle_seconds } => {
WorkflowRunEvent::StallWatchdogTimeout { node, idle_seconds } => {
format!("{d}── STALL_WATCHDOG_TIMEOUT ────────────────────{r}\n {d}node:{r} {node}\n {d}idle_seconds:{r} {idle_seconds}\n")
}
}
@ -989,7 +989,7 @@ mod tests {
#[test]
fn format_summary_execution_env_initializing() {
let event = PipelineEvent::ExecutionEnv {
let event = WorkflowRunEvent::ExecutionEnv {
event: arc_agent::ExecutionEnvEvent::Initializing {
env_type: "docker".into(),
},
@ -1001,7 +1001,7 @@ mod tests {
#[test]
fn format_summary_setup_started() {
let event = PipelineEvent::SetupStarted { command_count: 3 };
let event = WorkflowRunEvent::SetupStarted { command_count: 3 };
let s = format_event_summary(&event, test_styles());
assert!(s.contains("[SETUP_STARTED]"));
assert!(s.contains("3"));
@ -1009,7 +1009,7 @@ mod tests {
#[test]
fn format_detail_execution_env_ready() {
let event = PipelineEvent::ExecutionEnv {
let event = WorkflowRunEvent::ExecutionEnv {
event: arc_agent::ExecutionEnvEvent::Ready {
env_type: "local".into(),
duration_ms: 42,
@ -1023,7 +1023,7 @@ mod tests {
#[test]
fn format_summary_subagent_spawned() {
let event = PipelineEvent::Agent {
let event = WorkflowRunEvent::Agent {
stage: "code".into(),
event: AgentEvent::SubAgentSpawned {
agent_id: "abcdef12-3456-7890-abcd-ef1234567890".into(),
@ -1039,7 +1039,7 @@ mod tests {
#[test]
fn format_summary_subagent_completed() {
let event = PipelineEvent::Agent {
let event = WorkflowRunEvent::Agent {
stage: "code".into(),
event: AgentEvent::SubAgentCompleted {
agent_id: "abcdef12-xxxx".into(),
@ -1056,7 +1056,7 @@ mod tests {
#[test]
fn format_detail_subagent_failed() {
let event = PipelineEvent::Agent {
let event = WorkflowRunEvent::Agent {
stage: "code".into(),
event: AgentEvent::SubAgentFailed {
agent_id: "abcdef12-xxxx".into(),
@ -1072,7 +1072,7 @@ mod tests {
#[test]
fn format_detail_subagent_closed() {
let event = PipelineEvent::Agent {
let event = WorkflowRunEvent::Agent {
stage: "code".into(),
event: AgentEvent::SubAgentClosed {
agent_id: "abcdef12-xxxx".into(),
@ -1086,7 +1086,7 @@ mod tests {
#[test]
fn format_summary_subagent_event() {
let event = PipelineEvent::Agent {
let event = WorkflowRunEvent::Agent {
stage: "code".into(),
event: AgentEvent::SubAgentEvent {
agent_id: "abcdef12-xxxx".into(),
@ -1101,7 +1101,7 @@ mod tests {
#[test]
fn format_detail_setup_command_completed() {
let event = PipelineEvent::SetupCommandCompleted {
let event = WorkflowRunEvent::SetupCommandCompleted {
command: "npm install".into(),
index: 0,
exit_code: 0,

90
crates/arc-workflows/src/cli/run.rs
View file

@ -11,14 +11,14 @@ use arc_util::terminal::Styles;
use chrono::{Local, Utc};
use crate::checkpoint::Checkpoint;
use crate::engine::{GitCheckpointMode, PipelineEngine, RunConfig};
use crate::engine::{GitCheckpointMode, WorkflowRunEngine, RunConfig};
use crate::event::EventEmitter;
use crate::handler::default_registry;
use crate::interviewer::auto_approve::AutoApproveInterviewer;
use crate::interviewer::console::ConsoleInterviewer;
use crate::interviewer::Interviewer;
use crate::outcome::StageStatus;
use crate::pipeline::PipelineBuilder;
use crate::workflow::WorkflowBuilder;
use crate::validation::Severity;
use arc_llm::provider::Provider;
@ -32,7 +32,7 @@ use super::{
RunArgs,
};
/// Accumulates token usage and cost across all pipeline stages.
/// Accumulates token usage and cost across all workflow stages.
#[derive(Default)]
struct CostAccumulator {
total_input_tokens: i64,
@ -44,29 +44,29 @@ struct CostAccumulator {
has_pricing: bool,
}
/// Execute a full pipeline run.
/// Execute a full workflow run.
///
/// # Errors
///
/// Returns an error if the pipeline cannot be read, parsed, validated, or executed.
/// Returns an error if the workflow cannot be read, parsed, validated, or executed.
pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Result<()> {
// Handle --run-branch resume: read everything from git metadata
if let Some(branch) = args.run_branch.clone() {
return run_from_branch(args, &branch, styles).await;
}
let pipeline_path = args
.pipeline
let workflow_path = args
.workflow
.as_ref()
.ok_or_else(|| anyhow::anyhow!("--pipeline is required unless --run-branch is provided"))?;
.ok_or_else(|| anyhow::anyhow!("--workflow is required unless --run-branch is provided"))?;
// 0. Load task config if TOML, resolve DOT path, run setup
let (dot_path, task_cfg) = if pipeline_path.extension().is_some_and(|ext| ext == "toml") {
let cfg = task_config::load_task_config(pipeline_path)?;
let dot = task_config::resolve_graph_path(pipeline_path, &cfg.graph);
let (dot_path, task_cfg) = if workflow_path.extension().is_some_and(|ext| ext == "toml") {
let cfg = task_config::load_task_config(workflow_path)?;
let dot = task_config::resolve_graph_path(workflow_path, &cfg.graph);
(dot, Some(cfg))
} else {
(pipeline_path.clone(), None)
(workflow_path.clone(), None)
};
if let Some(ref cfg) = task_cfg {
@ -83,16 +83,16 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu
.map(|s| s.commands.clone())
.unwrap_or_default();
// 1. Parse and validate pipeline
// 1. Parse and validate workflow
let source = read_dot_file(&dot_path)?;
let source = match task_cfg.as_ref().and_then(|c| c.vars.as_ref()) {
Some(vars) => task_config::expand_vars(&source, vars)?,
None => source,
};
let (graph, diagnostics) = PipelineBuilder::new().prepare(&source)?;
let (graph, diagnostics) = WorkflowBuilder::new().prepare(&source)?;
eprintln!(
"{bold}Parsed pipeline:{reset} {} ({dim}{} nodes, {} edges{reset})",
"{bold}Parsed workflow:{reset} {} ({dim}{} nodes, {} edges{reset})",
graph.name,
graph.nodes.len(),
graph.edges.len(),
@ -148,8 +148,8 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu
tokio::fs::create_dir_all(&logs_dir).await?;
tokio::fs::write(logs_dir.join("graph.dot"), &source).await?;
tokio::fs::write(logs_dir.join("run.pid"), std::process::id().to_string()).await?;
if pipeline_path.extension().is_some_and(|ext| ext == "toml") {
if let Ok(toml_contents) = tokio::fs::read(pipeline_path).await {
if workflow_path.extension().is_some_and(|ext| ext == "toml") {
if let Ok(toml_contents) = tokio::fs::read(workflow_path).await {
tokio::fs::write(logs_dir.join("task.toml"), toml_contents).await?;
}
}
@ -171,7 +171,7 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu
{
let sha_clone = Arc::clone(&last_git_sha);
emitter.on_event(move |event| {
if let crate::event::PipelineEvent::GitCheckpoint { git_commit_sha, .. } = event {
if let crate::event::WorkflowRunEvent::GitCheckpoint { git_commit_sha, .. } = event {
*sha_clone.lock().unwrap() = Some(git_commit_sha.clone());
}
});
@ -181,7 +181,7 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu
let accumulator = Arc::new(Mutex::new(CostAccumulator::default()));
let acc_clone = Arc::clone(&accumulator);
emitter.on_event(move |event| {
if let crate::event::PipelineEvent::StageCompleted { usage: Some(u), .. } = event {
if let crate::event::WorkflowRunEvent::StageCompleted { usage: Some(u), .. } = event {
let mut acc = acc_clone.lock().unwrap();
acc.total_input_tokens += u.input_tokens;
acc.total_output_tokens += u.output_tokens;
@ -202,7 +202,7 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu
let run_id = Arc::new(Mutex::new(String::new()));
let run_id_clone = Arc::clone(&run_id);
emitter.on_event(move |event| {
if let crate::event::PipelineEvent::PipelineStarted { run_id, .. } = event {
if let crate::event::WorkflowRunEvent::WorkflowRunStarted { run_id, .. } = event {
*run_id_clone.lock().unwrap() = run_id.clone();
}
let envelope = serde_json::json!({
@ -240,7 +240,7 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu
});
} else {
emitter.on_event(move |event| match event {
crate::event::PipelineEvent::StageCompleted {
crate::event::WorkflowRunEvent::StageCompleted {
name,
duration_ms,
status,
@ -266,7 +266,7 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu
}
eprintln!("{line}{reset}", reset = styles.reset);
}
crate::event::PipelineEvent::StageFailed { name, .. } => {
crate::event::WorkflowRunEvent::StageFailed { name, .. } => {
eprintln!(
"{dim}Stage \"{name}\" failed{reset}",
dim = styles.dim,
@ -335,7 +335,7 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu
.map_err(|e| anyhow::anyhow!("Failed to create Docker environment: {e}"))?;
let emitter_cb = Arc::clone(&emitter);
env.set_event_callback(Arc::new(move |event| {
emitter_cb.emit(&crate::event::PipelineEvent::ExecutionEnv { event });
emitter_cb.emit(&crate::event::WorkflowRunEvent::ExecutionEnv { event });
}));
Arc::new(env)
}
@ -348,7 +348,7 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu
crate::daytona_env::DaytonaExecutionEnvironment::new(daytona_client, config);
let emitter_cb = Arc::clone(&emitter);
env.set_event_callback(Arc::new(move |event| {
emitter_cb.emit(&crate::event::PipelineEvent::ExecutionEnv { event });
emitter_cb.emit(&crate::event::WorkflowRunEvent::ExecutionEnv { event });
}));
Arc::new(env)
}
@ -356,13 +356,13 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu
let mut env = LocalExecutionEnvironment::new(cwd);
let emitter_cb = Arc::clone(&emitter);
env.set_event_callback(Arc::new(move |event| {
emitter_cb.emit(&crate::event::PipelineEvent::ExecutionEnv { event });
emitter_cb.emit(&crate::event::WorkflowRunEvent::ExecutionEnv { event });
}));
Arc::new(env)
}
};
// Initialize execution environment (creates sandbox/container once for the whole pipeline)
// Initialize execution environment (creates sandbox/container once for the whole run)
execution_env
.initialize()
.await
@ -403,12 +403,12 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu
// Run setup commands inside the execution environment (once, not per-stage)
if !setup_commands.is_empty() {
emitter.emit(&crate::event::PipelineEvent::SetupStarted {
emitter.emit(&crate::event::WorkflowRunEvent::SetupStarted {
command_count: setup_commands.len(),
});
let setup_start = Instant::now();
for (index, cmd) in setup_commands.iter().enumerate() {
emitter.emit(&crate::event::PipelineEvent::SetupCommandStarted {
emitter.emit(&crate::event::WorkflowRunEvent::SetupCommandStarted {
command: cmd.clone(),
index,
});
@ -419,7 +419,7 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu
.map_err(|e| anyhow::anyhow!("Setup command failed: {e}"))?;
let cmd_duration = u64::try_from(cmd_start.elapsed().as_millis()).unwrap_or(u64::MAX);
if result.exit_code != 0 {
emitter.emit(&crate::event::PipelineEvent::SetupFailed {
emitter.emit(&crate::event::WorkflowRunEvent::SetupFailed {
command: cmd.clone(),
index,
exit_code: result.exit_code,
@ -431,7 +431,7 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu
result.stderr,
);
}
emitter.emit(&crate::event::PipelineEvent::SetupCommandCompleted {
emitter.emit(&crate::event::WorkflowRunEvent::SetupCommandCompleted {
command: cmd.clone(),
index,
exit_code: result.exit_code,
@ -439,7 +439,7 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu
});
}
let setup_duration = u64::try_from(setup_start.elapsed().as_millis()).unwrap_or(u64::MAX);
emitter.emit(&crate::event::PipelineEvent::SetupCompleted {
emitter.emit(&crate::event::WorkflowRunEvent::SetupCompleted {
duration_ms: setup_duration,
});
}
@ -529,7 +529,7 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu
Some(Box::new(BackendRouter::new(Box::new(api), cli)))
}
});
let engine = PipelineEngine::with_interviewer(
let engine = WorkflowRunEngine::with_interviewer(
registry,
Arc::clone(&emitter),
interviewer,
@ -634,7 +634,7 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu
// 8. Print result
eprintln!(
"\n{bold}=== Pipeline Result ==={reset}",
"\n{bold}=== Run Result ==={reset}",
bold = styles.bold,
reset = styles.reset,
);
@ -708,7 +708,7 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu
}
}
/// Set up a git worktree for an isolated pipeline run.
/// Set up a git worktree for an isolated workflow run.
/// Caller must have already verified the repo is clean via `git::ensure_clean`.
/// Returns (run_id, work_dir, worktree_path, branch_name, base_sha) on success.
fn setup_worktree(
@ -774,7 +774,7 @@ async fn setup_daytona_git(
Ok((run_id, base_sha, branch_name))
}
/// Resume a pipeline run from a git run branch.
/// Resume a workflow run from a git run branch.
///
/// Reads the checkpoint, manifest, and graph DOT from the metadata branch
/// (`refs/arc/{run_id}`), re-attaches a worktree to the existing run branch,
@ -807,16 +807,16 @@ async fn run_from_branch(
.ok_or_else(|| anyhow::anyhow!("no graph.dot found on metadata branch for run {run_id}"))?;
// If --pipeline was also provided, use it instead (allows overriding)
let source = if let Some(ref pipeline_path) = args.pipeline {
super::read_dot_file(pipeline_path)?
let source = if let Some(ref workflow_path) = args.workflow {
super::read_dot_file(workflow_path)?
} else {
source
};
let (graph, diagnostics) = crate::pipeline::PipelineBuilder::new().prepare(&source)?;
let (graph, diagnostics) = crate::workflow::WorkflowBuilder::new().prepare(&source)?;
eprintln!(
"{bold}Resuming pipeline:{reset} {} from branch {dim}{run_branch}{reset}",
"{bold}Resuming workflow:{reset} {} from branch {dim}{run_branch}{reset}",
graph.name,
bold = styles.bold,
dim = styles.dim,
@ -860,7 +860,7 @@ async fn run_from_branch(
let mut env = arc_agent::LocalExecutionEnvironment::new(worktree_path.clone());
let emitter_cb = Arc::clone(&emitter);
env.set_event_callback(Arc::new(move |event| {
emitter_cb.emit(&crate::event::PipelineEvent::ExecutionEnv { event });
emitter_cb.emit(&crate::event::WorkflowRunEvent::ExecutionEnv { event });
}));
Arc::new(env)
};
@ -897,7 +897,7 @@ async fn run_from_branch(
Some(Box::new(BackendRouter::new(Box::new(api), cli)))
}
});
let engine = crate::engine::PipelineEngine::with_interviewer(
let engine = crate::engine::WorkflowRunEngine::with_interviewer(
registry,
Arc::clone(&emitter),
interviewer,
@ -961,7 +961,7 @@ async fn run_from_branch(
let outcome = engine_result?;
eprintln!(
"\n{bold}=== Pipeline Result ==={reset}",
"\n{bold}=== Run Result ==={reset}",
bold = styles.bold,
reset = styles.reset,
);
@ -991,14 +991,14 @@ async fn run_from_branch(
}
}
/// Generate a retro report for a completed pipeline run.
/// Generate a retro report for a completed workflow run.
///
/// Derives a basic retro from the checkpoint, then optionally runs the retro agent
/// for a richer narrative. Errors are logged as warnings rather than propagated.
#[allow(clippy::too_many_arguments)]
async fn generate_retro(
run_id: &str,
pipeline_name: &str,
workflow_name: &str,
goal: &str,
logs_dir: &std::path::Path,
failed: bool,
@ -1026,7 +1026,7 @@ async fn generate_retro(
let stage_durations = crate::retro::extract_stage_durations(logs_dir);
let mut retro = crate::retro::derive_retro(
run_id,
pipeline_name,
workflow_name,
goal,
&cp,
failed,

68
crates/arc-workflows/src/cli/runs.rs
View file

@ -12,9 +12,9 @@ pub struct RunsListArgs {
#[arg(long)]
pub before: Option<String>,
/// Filter by pipeline name (substring match)
/// Filter by workflow name (substring match)
#[arg(long)]
pub pipeline: Option<String>,
pub workflow: Option<String>,
/// Filter by label (KEY=VALUE, repeatable, AND semantics)
#[arg(long = "label", value_name = "KEY=VALUE")]
@ -35,9 +35,9 @@ pub struct RunsPruneArgs {
#[arg(long)]
pub before: Option<String>,
/// Filter by pipeline name (substring match)
/// Filter by workflow name (substring match)
#[arg(long)]
pub pipeline: Option<String>,
pub workflow: Option<String>,
/// Filter by label (KEY=VALUE, repeatable, AND semantics)
#[arg(long = "label", value_name = "KEY=VALUE")]
@ -56,7 +56,7 @@ pub struct RunsPruneArgs {
pub struct RunInfo {
pub run_id: String,
pub dir_name: String,
pub pipeline_name: String,
pub workflow_name: String,
pub status: String,
pub start_time: String,
pub labels: HashMap<String, String>,
@ -99,7 +99,7 @@ pub fn scan_runs(base: &Path) -> Result<Vec<RunInfo>> {
.as_str()
.unwrap_or(&dir_name)
.to_string();
let pipeline_name = manifest["pipeline_name"]
let workflow_name = manifest["workflow_name"]
.as_str()
.unwrap_or("unknown")
.to_string();
@ -117,7 +117,7 @@ pub fn scan_runs(base: &Path) -> Result<Vec<RunInfo>> {
runs.push(RunInfo {
run_id,
dir_name,
pipeline_name,
workflow_name,
status,
start_time,
labels,
@ -139,7 +139,7 @@ pub fn scan_runs(base: &Path) -> Result<Vec<RunInfo>> {
runs.push(RunInfo {
run_id: dir_name.clone(),
dir_name,
pipeline_name: "[no manifest]".to_string(),
workflow_name: "[no manifest]".to_string(),
status: "unknown".to_string(),
start_time: mtime,
labels: HashMap::new(),
@ -176,7 +176,7 @@ fn read_status(run_dir: &Path) -> String {
pub fn filter_runs(
runs: &[RunInfo],
before: Option<&str>,
pipeline: Option<&str>,
workflow: Option<&str>,
labels: &[(String, String)],
include_orphans: bool,
) -> Vec<RunInfo> {
@ -190,8 +190,8 @@ pub fn filter_runs(
return false;
}
}
if let Some(pat) = pipeline {
if !r.pipeline_name.contains(pat) {
if let Some(pat) = workflow {
if !r.workflow_name.contains(pat) {
return false;
}
}
@ -229,7 +229,7 @@ pub fn list_command(args: &RunsListArgs) -> Result<()> {
let filtered = filter_runs(
&runs,
args.before.as_deref(),
args.pipeline.as_deref(),
args.workflow.as_deref(),
&label_filters,
args.orphans,
);
@ -247,7 +247,7 @@ pub fn list_command(args: &RunsListArgs) -> Result<()> {
// Print table header
let header = format!(
"{:<30} {:<25} {:<10} {:<25} LABELS",
"RUN ID", "PIPELINE", "STATUS", "STARTED"
"RUN ID", "WORKFLOW", "STATUS", "STARTED"
);
println!("{header}");
println!("{}", "-".repeat(100));
@ -271,7 +271,7 @@ pub fn list_command(args: &RunsListArgs) -> Result<()> {
};
println!(
"{:<30} {:<25} {:<10} {:<25} {}",
run_id_display, run.pipeline_name, run.status, start_display, labels_str
run_id_display, run.workflow_name, run.status, start_display, labels_str
);
}
eprintln!("\n{} run(s) listed.", filtered.len());
@ -289,7 +289,7 @@ pub fn prune_from(args: &RunsPruneArgs, base: &Path) -> Result<()> {
let filtered = filter_runs(
&runs,
args.before.as_deref(),
args.pipeline.as_deref(),
args.workflow.as_deref(),
&label_filters,
args.orphans,
);
@ -310,7 +310,7 @@ pub fn prune_from(args: &RunsPruneArgs, base: &Path) -> Result<()> {
debug!(run_id = %run.run_id, "would delete run (dry-run)");
println!(
"would delete: {} ({})",
run.dir_name, run.pipeline_name
run.dir_name, run.workflow_name
);
}
eprintln!(
@ -359,7 +359,7 @@ mod tests {
"arc-run-20260101-120000",
Some(serde_json::json!({
"run_id": "abc123",
"pipeline_name": "my-pipeline",
"workflow_name": "my-pipeline",
"start_time": "2026-01-01T12:00:00Z",
"labels": { "env": "prod" }
})),
@ -373,13 +373,13 @@ mod tests {
assert_eq!(runs.len(), 2);
let completed = runs.iter().find(|r| r.run_id == "abc123").unwrap();
assert_eq!(completed.pipeline_name, "my-pipeline");
assert_eq!(completed.workflow_name, "my-pipeline");
assert_eq!(completed.status, "success");
assert_eq!(completed.labels.get("env").unwrap(), "prod");
assert!(!completed.is_orphan);
let orphan = runs.iter().find(|r| r.is_orphan).unwrap();
assert_eq!(orphan.pipeline_name, "[no manifest]");
assert_eq!(orphan.workflow_name, "[no manifest]");
assert_eq!(orphan.status, "unknown");
}
@ -393,7 +393,7 @@ mod tests {
"arc-run-running",
Some(serde_json::json!({
"run_id": "running-1",
"pipeline_name": "pipeline-a",
"workflow_name": "pipeline-a",
"start_time": "2026-01-15T10:00:00Z"
})),
None,
@ -424,7 +424,7 @@ mod tests {
RunInfo {
run_id: "old".into(),
dir_name: "d1".into(),
pipeline_name: "p".into(),
workflow_name: "p".into(),
status: "success".into(),
start_time: "2025-06-01T00:00:00Z".into(),
labels: HashMap::new(),
@ -434,7 +434,7 @@ mod tests {
RunInfo {
run_id: "new".into(),
dir_name: "d2".into(),
pipeline_name: "p".into(),
workflow_name: "p".into(),
status: "success".into(),
start_time: "2026-03-01T00:00:00Z".into(),
labels: HashMap::new(),
@ -448,12 +448,12 @@ mod tests {
}
#[test]
fn filter_runs_pipeline() {
fn filter_runs_workflow() {
let runs = vec![
RunInfo {
run_id: "a".into(),
dir_name: "d1".into(),
pipeline_name: "deploy-prod".into(),
workflow_name: "deploy-prod".into(),
status: "success".into(),
start_time: "2026-01-01T00:00:00Z".into(),
labels: HashMap::new(),
@ -463,7 +463,7 @@ mod tests {
RunInfo {
run_id: "b".into(),
dir_name: "d2".into(),
pipeline_name: "test-suite".into(),
workflow_name: "test-suite".into(),
status: "success".into(),
start_time: "2026-01-01T00:00:00Z".into(),
labels: HashMap::new(),
@ -482,7 +482,7 @@ mod tests {
RunInfo {
run_id: "a".into(),
dir_name: "d1".into(),
pipeline_name: "p".into(),
workflow_name: "p".into(),
status: "success".into(),
start_time: "2026-01-01T00:00:00Z".into(),
labels: HashMap::from([("env".into(), "prod".into())]),
@ -492,7 +492,7 @@ mod tests {
RunInfo {
run_id: "b".into(),
dir_name: "d2".into(),
pipeline_name: "p".into(),
workflow_name: "p".into(),
status: "success".into(),
start_time: "2026-01-01T00:00:00Z".into(),
labels: HashMap::from([("env".into(), "staging".into())]),
@ -516,7 +516,7 @@ mod tests {
let runs = vec![RunInfo {
run_id: "orphan".into(),
dir_name: "d1".into(),
pipeline_name: "[no manifest]".into(),
workflow_name: "[no manifest]".into(),
status: "unknown".into(),
start_time: "".into(),
labels: HashMap::new(),
@ -540,7 +540,7 @@ mod tests {
"arc-run-20250101-120000",
Some(serde_json::json!({
"run_id": "to-prune",
"pipeline_name": "old-pipeline",
"workflow_name": "old-pipeline",
"start_time": "2025-01-01T12:00:00Z"
})),
Some(serde_json::json!({ "status": "success" })),
@ -549,7 +549,7 @@ mod tests {
let args = RunsPruneArgs {
before: Some("2026-01-01".into()),
pipeline: None,
workflow: None,
label: Vec::new(),
orphans: false,
yes: false,
@ -569,7 +569,7 @@ mod tests {
"arc-run-20250101-120000",
Some(serde_json::json!({
"run_id": "to-prune",
"pipeline_name": "old-pipeline",
"workflow_name": "old-pipeline",
"start_time": "2025-01-01T12:00:00Z"
})),
Some(serde_json::json!({ "status": "success" })),
@ -582,7 +582,7 @@ mod tests {
"arc-run-20260301-120000",
Some(serde_json::json!({
"run_id": "keep-this",
"pipeline_name": "new-pipeline",
"workflow_name": "new-pipeline",
"start_time": "2026-03-01T12:00:00Z"
})),
Some(serde_json::json!({ "status": "success" })),
@ -591,7 +591,7 @@ mod tests {
let args = RunsPruneArgs {
before: Some("2026-01-01".into()),
pipeline: None,
workflow: None,
label: Vec::new(),
orphans: false,
yes: true,
@ -611,7 +611,7 @@ mod tests {
let args = RunsPruneArgs {
before: None,
pipeline: None,
workflow: None,
label: Vec::new(),
orphans: true,
yes: true,

18
crates/arc-workflows/src/cli/task_config.rs
View file

@ -164,7 +164,7 @@ mod tests {
let toml = r#"
version = 1
task = "Run tests"
graph = "pipeline.dot"
graph = "workflow.dot"
[vars]
repo_url = "https://github.com/org/repo"
@ -224,7 +224,7 @@ language = "python"
let toml = r#"
version = 1
task = "Run tests"
graph = "pipeline.dot"
graph = "workflow.dot"
[execution]
environment = "daytona"
@ -240,7 +240,7 @@ environment = "daytona"
let toml = r#"
version = 1
task = "Run tests"
graph = "pipeline.dot"
graph = "workflow.dot"
[execution]
environment = "daytona"
@ -283,7 +283,7 @@ dockerfile = "FROM rust:1.85-slim-bookworm\nRUN apt-get update"
let toml = r#"
version = 1
task = "Run tests"
graph = "pipeline.dot"
graph = "workflow.dot"
[execution]
environment = "daytona"
@ -302,12 +302,12 @@ auto_stop_interval = 30
let toml = r#"
version = 1
task = "Run tests"
graph = "pipeline.dot"
graph = "workflow.dot"
"#;
let config = parse_task_config(toml).unwrap();
assert_eq!(config.version, 1);
assert_eq!(config.task, "Run tests");
assert_eq!(config.graph, "pipeline.dot");
assert_eq!(config.graph, "workflow.dot");
assert!(config.directory.is_none());
assert!(config.llm.is_none());
assert!(config.setup.is_none());
@ -318,7 +318,7 @@ graph = "pipeline.dot"
let toml = r#"
version = 1
task = "Full workflow"
graph = "pipeline.dot"
graph = "workflow.dot"
directory = "/tmp/repo"
[llm]
@ -367,8 +367,8 @@ graph = "p.dot"
#[test]
fn graph_path_absolute_unchanged() {
let toml_path = Path::new("/tmp/sub/task.toml");
let resolved = resolve_graph_path(toml_path, "/other/pipeline.dot");
assert_eq!(resolved, PathBuf::from("/other/pipeline.dot"));
let resolved = resolve_graph_path(toml_path, "/other/workflow.dot");
assert_eq!(resolved, PathBuf::from("/other/workflow.dot"));
}
#[test]

10
crates/arc-workflows/src/cli/validate.rs
View file

@ -1,22 +1,22 @@
use anyhow::bail;
use arc_util::terminal::Styles;
use crate::pipeline::PipelineBuilder;
use crate::workflow::WorkflowBuilder;
use crate::validation::Severity;
use super::{print_diagnostics, read_dot_file, ValidateArgs};
/// Parse and validate a pipeline file without executing it.
/// Parse and validate a workflow file without executing it.
///
/// # Errors
///
/// Returns an error if the file cannot be read, parsed, or has validation errors.
pub fn validate_command(args: &ValidateArgs, styles: &Styles) -> anyhow::Result<()> {
let source = read_dot_file(&args.pipeline)?;
let (graph, diagnostics) = PipelineBuilder::new().prepare(&source)?;
let source = read_dot_file(&args.workflow)?;
let (graph, diagnostics) = WorkflowBuilder::new().prepare(&source)?;
eprintln!(
"{bold}Parsed pipeline:{reset} {} ({dim}{} nodes, {} edges{reset})",
"{bold}Parsed workflow:{reset} {} ({dim}{} nodes, {} edges{reset})",
graph.name,
graph.nodes.len(),
graph.edges.len(),

164
crates/arc-workflows/src/engine.rs
View file

@ -18,7 +18,7 @@ use crate::checkpoint::Checkpoint;
use crate::condition::evaluate_condition;
use crate::context::Context;
use crate::error::{classify_failure_reason, ArcError, FailureClass, FailureSignature, Result};
use crate::event::{EventEmitter, PipelineEvent};
use crate::event::{EventEmitter, WorkflowRunEvent};
use crate::graph::{Edge, Graph, Node};
use crate::handler::{EngineServices, HandlerRegistry};
use crate::interviewer::Interviewer;
@ -283,16 +283,16 @@ pub fn resolve_thread_id(
// --- Run directory helpers (spec 5.6) ---
/// Write manifest.json at the start of a pipeline run. Returns the manifest value.
/// Write manifest.json at the start of a workflow run. Returns the manifest value.
fn write_manifest(logs_root: &Path, graph: &Graph, config: &RunConfig) -> serde_json::Value {
let pipeline_name = if graph.name.is_empty() {
let workflow_name = if graph.name.is_empty() {
"unnamed"
} else {
&graph.name
};
let mut manifest = serde_json::json!({
"run_id": config.run_id,
"pipeline_name": pipeline_name,
"workflow_name": workflow_name,
"goal": graph.goal(),
"start_time": Utc::now().to_rfc3339(),
"node_count": graph.nodes.len(),
@ -514,9 +514,9 @@ fn is_terminal(node: &Node) -> bool {
node.shape() == "Msquare" || node.handler_type() == Some("exit")
}
// --- Pipeline engine ---
// --- Workflow run engine ---
/// Captured git state for a pipeline run, shared with handlers.
/// Captured git state for a workflow run, shared with handlers.
#[derive(Debug, Clone)]
pub struct GitState {
pub mode: GitCheckpointMode,
@ -526,7 +526,7 @@ pub struct GitState {
pub meta_branch: Option<String>,
}
/// How git checkpointing should be performed for a pipeline run.
/// How git checkpointing should be performed for a workflow run.
#[derive(Debug, Clone)]
pub enum GitCheckpointMode {
/// Run git commands on the host filesystem (local & Docker bind-mount).
@ -724,12 +724,12 @@ pub async fn git_replace_worktree_remote(
git_add_worktree_remote(exec_env, path, branch).await
}
/// Configuration for a pipeline run.
/// Configuration for a workflow run.
pub struct RunConfig {
pub logs_root: PathBuf,
pub cancel_token: Option<Arc<AtomicBool>>,
pub dry_run: bool,
/// Unique identifier for this pipeline run.
/// Unique identifier for this workflow run.
pub run_id: String,
/// Git checkpoint mode (None = no checkpointing).
pub git_checkpoint: Option<GitCheckpointMode>,
@ -743,13 +743,13 @@ pub struct RunConfig {
pub labels: HashMap<String, String>,
}
/// The pipeline execution engine.
pub struct PipelineEngine {
/// The workflow run execution engine.
pub struct WorkflowRunEngine {
services: EngineServices,
pub interviewer: Option<Arc<dyn Interviewer>>,
}
impl PipelineEngine {
impl WorkflowRunEngine {
#[must_use]
pub fn new(
registry: HandlerRegistry,
@ -884,7 +884,7 @@ impl PipelineEngine {
// Gap #7: Check should_retry predicate before retrying
if attempt < policy.max_attempts && handler.should_retry(&e) {
let delay = policy.backoff.delay_for_attempt(attempt);
self.services.emitter.emit(&PipelineEvent::StageFailed {
self.services.emitter.emit(&WorkflowRunEvent::StageFailed {
name: node.label().to_string(),
index: stage_index,
error: e.to_string(),
@ -892,7 +892,7 @@ impl PipelineEngine {
failure_reason: None,
failure_class: Some(e.failure_class().to_string()),
});
self.services.emitter.emit(&PipelineEvent::StageRetrying {
self.services.emitter.emit(&WorkflowRunEvent::StageRetrying {
name: node.label().to_string(),
index: stage_index,
attempt: usize::try_from(attempt).unwrap_or(usize::MAX),
@ -917,7 +917,7 @@ impl PipelineEngine {
StageStatus::Retry => {
if attempt < policy.max_attempts {
let delay = policy.backoff.delay_for_attempt(attempt);
self.services.emitter.emit(&PipelineEvent::StageRetrying {
self.services.emitter.emit(&WorkflowRunEvent::StageRetrying {
name: node.label().to_string(),
index: stage_index,
attempt: usize::try_from(attempt).unwrap_or(usize::MAX),
@ -946,7 +946,7 @@ impl PipelineEngine {
Ok((Outcome::fail("max retries exceeded"), policy.max_attempts))
}
/// Run the pipeline. Returns the final outcome.
/// Run the workflow. Returns the final outcome.
///
/// # Errors
///
@ -959,7 +959,7 @@ impl PipelineEngine {
Ok(outcome)
}
/// Run a pipeline seeded with an existing context. Returns both the outcome
/// Run a workflow seeded with an existing context. Returns both the outcome
/// and the final context so the caller can diff changes.
pub async fn run_with_context(
&self,
@ -1028,7 +1028,7 @@ impl PipelineEngine {
};
self.services.set_git_state(git_state);
self.services.emitter.emit(&PipelineEvent::PipelineStarted {
self.services.emitter.emit(&WorkflowRunEvent::WorkflowRunStarted {
name: graph.name.clone(),
run_id: run_id.clone(),
base_sha: config.base_sha.clone(),
@ -1038,7 +1038,7 @@ impl PipelineEngine {
_ => None,
},
});
self.inform(&format!("Pipeline started: {}", graph.name), "pipeline");
self.inform(&format!("Run started: {}", graph.name), "run");
// Write manifest.json (spec 5.6)
let manifest = write_manifest(&config.logs_root, graph, config);
@ -1205,9 +1205,9 @@ impl PipelineEngine {
.or_insert(0);
*count += 1;
if max_node_visits > 0 && *count >= max_node_visits {
tracing::warn!(node = %current_node_id, visits = *count, limit = max_node_visits, "Node visit limit exceeded, pipeline stuck in cycle");
tracing::warn!(node = %current_node_id, visits = *count, limit = max_node_visits, "Node visit limit exceeded, run stuck in cycle");
return Err(ArcError::Engine(format!(
"node \"{}\" visited {count} times (limit {max_node_visits}); pipeline is stuck in a cycle",
"node \"{}\" visited {count} times (limit {max_node_visits}); run is stuck in a cycle",
current_node_id
)));
}
@ -1225,7 +1225,7 @@ impl PipelineEngine {
let error_msg = format!(
"goal gate unsatisfied for node {failed_node_id} and no retry target"
);
self.services.emitter.emit(&PipelineEvent::PipelineFailed {
self.services.emitter.emit(&WorkflowRunEvent::WorkflowRunFailed {
error: error_msg.clone(),
duration_ms,
git_commit_sha: last_git_sha.clone(),
@ -1273,7 +1273,7 @@ impl PipelineEngine {
context.set("current_node", serde_json::json!(&node.id));
let retry_policy = build_retry_policy(node, graph);
self.services.emitter.emit(&PipelineEvent::StageStarted {
self.services.emitter.emit(&WorkflowRunEvent::StageStarted {
name: node.label().to_string(),
index: stage_index,
handler_type: node.handler_type().map(String::from),
@ -1292,7 +1292,7 @@ impl PipelineEngine {
) => result?,
() = token.cancelled() => {
let idle_secs = graph.stall_timeout().map_or(0, |d| d.as_secs());
self.services.emitter.emit(&PipelineEvent::StallWatchdogTimeout {
self.services.emitter.emit(&WorkflowRunEvent::StallWatchdogTimeout {
node: node.id.clone(),
idle_seconds: idle_secs,
});
@ -1361,7 +1361,7 @@ impl PipelineEngine {
};
if outcome.status == StageStatus::Fail {
self.services.emitter.emit(&PipelineEvent::StageFailed {
self.services.emitter.emit(&WorkflowRunEvent::StageFailed {
name: node.label().to_string(),
index: stage_index,
error: outcome
@ -1374,7 +1374,7 @@ impl PipelineEngine {
failure_class: outcome_failure_class.map(|fc| fc.to_string()),
});
} else {
self.services.emitter.emit(&PipelineEvent::StageCompleted {
self.services.emitter.emit(&WorkflowRunEvent::StageCompleted {
name: node.label().to_string(),
index: stage_index,
duration_ms: stage_duration_ms,
@ -1434,7 +1434,7 @@ impl PipelineEngine {
// Step 5: Select next edge (done before checkpoint so we can store next_node_id)
let next_edge = select_edge(&node.id, &outcome, &context, graph);
if let Some(edge) = next_edge {
self.services.emitter.emit(&PipelineEvent::EdgeSelected {
self.services.emitter.emit(&WorkflowRunEvent::EdgeSelected {
from_node: node.id.clone(),
to_node: edge.to.clone(),
label: edge.label().map(String::from),
@ -1458,7 +1458,7 @@ impl PipelineEngine {
if let Err(e) = checkpoint.save(&checkpoint_path) {
context.append_log(format!("checkpoint save failed: {e}"));
} else {
self.services.emitter.emit(&PipelineEvent::CheckpointSaved {
self.services.emitter.emit(&WorkflowRunEvent::CheckpointSaved {
node_id: node.id.clone(),
});
}
@ -1539,7 +1539,7 @@ impl PipelineEngine {
if let Err(e) = checkpoint.save(&checkpoint_path) {
context.append_log(format!("checkpoint re-save with SHA failed: {e}"));
}
self.services.emitter.emit(&PipelineEvent::GitCheckpoint {
self.services.emitter.emit(&WorkflowRunEvent::GitCheckpoint {
run_id: run_id.clone(),
node_id: node.id.clone(),
status: outcome.status.to_string(),
@ -1587,7 +1587,7 @@ impl PipelineEngine {
let duration_ms = millis_u64(run_start.elapsed());
let error_msg =
format!("stage {} failed with no outgoing fail edge", node.id);
self.services.emitter.emit(&PipelineEvent::PipelineFailed {
self.services.emitter.emit(&WorkflowRunEvent::WorkflowRunFailed {
error: error_msg.clone(),
duration_ms,
git_commit_sha: last_git_sha.clone(),
@ -1625,7 +1625,7 @@ impl PipelineEngine {
)));
}
}
self.services.emitter.emit(&PipelineEvent::LoopRestart {
self.services.emitter.emit(&WorkflowRunEvent::LoopRestart {
from_node: node.id.clone(),
to_node: edge.to.clone(),
});
@ -1663,7 +1663,7 @@ impl PipelineEngine {
};
self.services
.emitter
.emit(&PipelineEvent::PipelineCompleted {
.emit(&WorkflowRunEvent::WorkflowRunCompleted {
duration_ms,
artifact_count: artifact_store.list().len(),
total_cost,
@ -2300,7 +2300,7 @@ mod tests {
assert!(!is_terminal(&n));
}
// --- PipelineEngine integration tests ---
// --- WorkflowRunEngine integration tests ---
fn simple_graph() -> Graph {
let mut g = Graph::new("test_pipeline");
@ -2336,11 +2336,11 @@ mod tests {
}
#[tokio::test]
async fn engine_runs_simple_pipeline() {
async fn engine_runs_simple_workflow() {
let dir = tempfile::tempdir().unwrap();
let g = simple_graph();
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -2361,7 +2361,7 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let g = simple_graph();
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -2390,7 +2390,7 @@ mod tests {
events_clone.lock().unwrap().push(format!("{event:?}"));
});
let engine = PipelineEngine::new(make_registry(), Arc::new(emitter), local_env());
let engine = WorkflowRunEngine::new(make_registry(), Arc::new(emitter), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -2405,8 +2405,8 @@ mod tests {
engine.run(&g, &config).await.unwrap();
let collected = events.lock().unwrap();
// Should have: PipelineStarted, StageStarted (start), StageCompleted (start),
// CheckpointSaved, PipelineCompleted
// Should have: RunStarted, StageStarted (start), StageCompleted (start),
// CheckpointSaved, RunCompleted
assert!(collected.len() >= 4);
}
@ -2415,7 +2415,7 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let g = Graph::new("empty");
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -2436,7 +2436,7 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let g = simple_graph();
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -2459,7 +2459,7 @@ mod tests {
}
#[tokio::test]
async fn engine_multi_node_pipeline() {
async fn engine_multi_node_workflow() {
let dir = tempfile::tempdir().unwrap();
let mut g = simple_graph();
// Insert a work node between start and exit
@ -2470,7 +2470,7 @@ mod tests {
g.edges.push(Edge::new("work", "exit"));
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -2528,7 +2528,7 @@ mod tests {
g.edges.push(Edge::new("path_b", "exit"));
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -2606,7 +2606,7 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let g = simple_graph();
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -2624,7 +2624,7 @@ mod tests {
assert!(manifest_path.exists());
let manifest: serde_json::Value =
serde_json::from_str(&std::fs::read_to_string(&manifest_path).unwrap()).unwrap();
assert_eq!(manifest["pipeline_name"], "test_pipeline");
assert_eq!(manifest["workflow_name"], "test_pipeline");
assert_eq!(manifest["goal"], "Run tests");
assert!(manifest["start_time"].is_string());
assert!(manifest["node_count"].is_number());
@ -2636,7 +2636,7 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let g = simple_graph();
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -2661,7 +2661,7 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let g = simple_graph();
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -2686,7 +2686,7 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let g = simple_graph();
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -2713,7 +2713,7 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let g = simple_graph();
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -2868,7 +2868,7 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let g = simple_graph();
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -2907,7 +2907,7 @@ mod tests {
g.edges.push(Edge::new("start", "exit"));
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -2964,7 +2964,7 @@ mod tests {
let mut registry = make_registry();
registry.register("always_fail", Box::new(AlwaysFailHandler));
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -3023,7 +3023,7 @@ mod tests {
let mut registry = make_registry();
registry.register("always_fail", Box::new(AlwaysFailHandler));
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -3086,7 +3086,7 @@ mod tests {
let mut registry = make_registry();
registry.register("slow", Box::new(SlowHandler { sleep_ms: 500 }));
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -3138,7 +3138,7 @@ mod tests {
let mut registry = make_registry();
registry.register("slow", Box::new(SlowHandler { sleep_ms: 10 }));
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -3191,7 +3191,7 @@ mod tests {
let mut registry = make_registry();
registry.register("slow", Box::new(SlowHandler { sleep_ms: 500 }));
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -3242,11 +3242,11 @@ mod tests {
}
#[tokio::test]
async fn engine_calls_inform_on_pipeline_start() {
async fn engine_calls_inform_on_run_start() {
let dir = tempfile::tempdir().unwrap();
let g = simple_graph();
let informer = Arc::new(RecordingInformer::new());
let engine = PipelineEngine::with_interviewer(
let engine = WorkflowRunEngine::with_interviewer(
make_registry(),
Arc::new(EventEmitter::new()),
Arc::clone(&informer) as Arc<dyn crate::interviewer::Interviewer>,
@ -3271,8 +3271,8 @@ mod tests {
assert!(
messages
.iter()
.any(|(msg, stage)| msg.contains("Pipeline started") && stage == "pipeline"),
"expected 'Pipeline started' inform call, got: {messages:?}"
.any(|(msg, stage)| msg.contains("Run started") && stage == "run"),
"expected 'Run started' inform call, got: {messages:?}"
);
}
@ -3281,7 +3281,7 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let g = simple_graph();
let informer = Arc::new(RecordingInformer::new());
let engine = PipelineEngine::with_interviewer(
let engine = WorkflowRunEngine::with_interviewer(
make_registry(),
Arc::new(EventEmitter::new()),
Arc::clone(&informer) as Arc<dyn crate::interviewer::Interviewer>,
@ -3322,7 +3322,7 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let g = simple_graph();
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -3345,7 +3345,7 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let g = simple_graph();
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let cancel_token = Arc::new(AtomicBool::new(true));
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
@ -3368,7 +3368,7 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let g = simple_graph();
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let cancel_token = Arc::new(AtomicBool::new(false));
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
@ -3386,7 +3386,7 @@ mod tests {
}
#[tokio::test]
async fn engine_cancelled_mid_pipeline() {
async fn engine_cancelled_mid_run() {
let dir = tempfile::tempdir().unwrap();
let mut g = simple_graph();
// Insert a work node between start and exit
@ -3405,7 +3405,7 @@ mod tests {
let mut registry = make_registry();
registry.register("slow", Box::new(SlowHandler { sleep_ms: 200 }));
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: Some(cancel_token),
@ -3478,7 +3478,7 @@ mod tests {
g.attrs
.insert("max_node_visits".to_string(), AttrValue::Integer(3));
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -3504,7 +3504,7 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let g = cyclic_graph();
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -3532,7 +3532,7 @@ mod tests {
g.attrs
.insert("max_node_visits".to_string(), AttrValue::Integer(2));
let engine =
PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
WorkflowRunEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -3621,7 +3621,7 @@ mod tests {
let mut registry = make_registry();
registry.register("panicker", Box::new(PanickingHandler));
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -3846,7 +3846,7 @@ mod tests {
let mut registry = make_registry();
registry.register("always_fail", Box::new(AlwaysFailHandler));
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -3877,7 +3877,7 @@ mod tests {
let mut registry = make_registry();
registry.register("always_fail", Box::new(TransientFailHandler));
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -3915,7 +3915,7 @@ mod tests {
counter: std::sync::atomic::AtomicUsize::new(0),
}),
);
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -3938,7 +3938,7 @@ mod tests {
#[tokio::test]
async fn restart_circuit_breaker_aborts_on_repeated_failure() {
// In a pipeline with loop_restart edges, a repeating deterministic failure
// In a workflow with loop_restart edges, a repeating deterministic failure
// triggers a circuit breaker (either loop or restart, depending on topology).
let dir = tempfile::tempdir().unwrap();
let mut g = Graph::new("restart_test");
@ -3993,7 +3993,7 @@ mod tests {
let mut registry = make_registry();
registry.register("always_fail", Box::new(AlwaysFailHandler));
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -4036,7 +4036,7 @@ mod tests {
let start = Instant::now();
while start.elapsed() < Duration::from_millis(self.total_ms) {
tokio::time::sleep(Duration::from_millis(self.interval_ms)).await;
services.emitter.emit(&PipelineEvent::Prompt {
services.emitter.emit(&WorkflowRunEvent::Prompt {
stage: node.id.clone(),
text: "keepalive".to_string(),
});
@ -4084,7 +4084,7 @@ mod tests {
let mut registry = make_registry();
registry.register("slow", Box::new(SlowHandler { sleep_ms: 60_000 }));
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -4150,7 +4150,7 @@ mod tests {
total_ms: 500,
}),
);
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -4205,7 +4205,7 @@ mod tests {
let mut registry = make_registry();
registry.register("slow", Box::new(SlowHandler { sleep_ms: 200 }));
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -4224,7 +4224,7 @@ mod tests {
#[tokio::test]
async fn failure_signature_stored_in_context() {
let dir = tempfile::tempdir().unwrap();
// Simple pipeline: start -> work (fails) -> exit (via fail edge)
// Simple workflow: start -> work (fails) -> exit (via fail edge)
let mut g = Graph::new("sig_context_test");
g.attrs
.insert("goal".to_string(), AttrValue::String("test".to_string()));
@ -4259,7 +4259,7 @@ mod tests {
let mut registry = make_registry();
registry.register("always_fail", Box::new(AlwaysFailHandler));
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), local_env());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,

174
crates/arc-workflows/src/event.rs
View file

@ -5,10 +5,10 @@ use serde::{Deserialize, Serialize};
use crate::outcome::StageUsage;
use arc_agent::{AgentEvent, ExecutionEnvEvent};
/// Events emitted during pipeline execution for observability.
/// Events emitted during workflow run execution for observability.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum PipelineEvent {
PipelineStarted {
pub enum WorkflowRunEvent {
WorkflowRunStarted {
name: String,
run_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
@ -18,7 +18,7 @@ pub enum PipelineEvent {
#[serde(default, skip_serializing_if = "Option::is_none")]
worktree_dir: Option<String>,
},
PipelineCompleted {
WorkflowRunCompleted {
duration_ms: u64,
artifact_count: usize,
#[serde(default, skip_serializing_if = "Option::is_none")]
@ -26,7 +26,7 @@ pub enum PipelineEvent {
#[serde(default, skip_serializing_if = "Option::is_none")]
final_git_commit_sha: Option<String>,
},
PipelineFailed {
WorkflowRunFailed {
error: String,
duration_ms: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
@ -127,7 +127,7 @@ pub enum PipelineEvent {
stage: String,
text: String,
},
/// Forwarded from an agent session, tagged with the pipeline stage.
/// Forwarded from an agent session, tagged with the workflow stage.
Agent {
stage: String,
event: AgentEvent,
@ -179,24 +179,24 @@ pub enum PipelineEvent {
},
}
impl PipelineEvent {
impl WorkflowRunEvent {
pub fn trace(&self) {
use tracing::{debug, error, info, warn};
match self {
Self::PipelineStarted { name, run_id, .. } => {
info!(pipeline = name.as_str(), run_id, "Pipeline started");
Self::WorkflowRunStarted { name, run_id, .. } => {
info!(workflow = name.as_str(), run_id, "Workflow run started");
}
Self::PipelineCompleted {
Self::WorkflowRunCompleted {
duration_ms,
artifact_count,
..
} => {
info!(duration_ms, artifact_count, "Pipeline completed");
info!(duration_ms, artifact_count, "Workflow run completed");
}
Self::PipelineFailed {
Self::WorkflowRunFailed {
error, duration_ms, ..
} => {
error!(error, duration_ms, "Pipeline failed");
error!(error, duration_ms, "Workflow run failed");
}
Self::StageStarted {
name,
@ -446,10 +446,10 @@ fn epoch_millis() -> i64 {
.as_millis() as i64
}
/// Listener callback type for pipeline events.
type EventListener = Box<dyn Fn(&PipelineEvent) + Send + Sync>;
/// Listener callback type for workflow run events.
type EventListener = Box<dyn Fn(&WorkflowRunEvent) + Send + Sync>;
/// Callback-based event emitter for pipeline events.
/// Callback-based event emitter for workflow run events.
pub struct EventEmitter {
listeners: Vec<EventListener>,
/// Epoch milliseconds of the last `emit()` or `touch()` call. 0 until first event.
@ -480,11 +480,11 @@ impl EventEmitter {
}
}
pub fn on_event(&mut self, listener: impl Fn(&PipelineEvent) + Send + Sync + 'static) {
pub fn on_event(&mut self, listener: impl Fn(&WorkflowRunEvent) + Send + Sync + 'static) {
self.listeners.push(Box::new(listener));
}
pub fn emit(&self, event: &PipelineEvent) {
pub fn emit(&self, event: &WorkflowRunEvent) {
self.last_event_at.store(epoch_millis(), Ordering::Relaxed);
event.trace();
for listener in &self.listeners {
@ -498,7 +498,7 @@ impl EventEmitter {
self.last_event_at.load(Ordering::Relaxed)
}
/// Manually update the last-event timestamp (e.g. to seed the watchdog at pipeline start).
/// Manually update the last-event timestamp (e.g. to seed the watchdog at workflow run start).
pub fn touch(&self) {
self.last_event_at.store(epoch_millis(), Ordering::Relaxed);
}
@ -523,12 +523,12 @@ mod tests {
let received_clone = Arc::clone(&received);
emitter.on_event(move |event| {
let name = match event {
PipelineEvent::PipelineStarted { name, .. } => name.clone(),
WorkflowRunEvent::WorkflowRunStarted { name, .. } => name.clone(),
_ => "other".to_string(),
};
received_clone.lock().unwrap().push(name);
});
emitter.emit(&PipelineEvent::PipelineStarted {
emitter.emit(&WorkflowRunEvent::WorkflowRunStarted {
name: "test".to_string(),
run_id: "1".to_string(),
base_sha: None,
@ -541,8 +541,8 @@ mod tests {
}
#[test]
fn pipeline_event_serialization() {
let event = PipelineEvent::StageStarted {
fn workflow_run_event_serialization() {
let event = WorkflowRunEvent::StageStarted {
name: "plan".to_string(),
index: 0,
handler_type: Some("codergen".to_string()),
@ -557,7 +557,7 @@ mod tests {
assert!(json.contains("\"max_attempts\":3"));
// None handler_type serializes as null
let event_none = PipelineEvent::StageStarted {
let event_none = WorkflowRunEvent::StageStarted {
name: "plan".to_string(),
index: 0,
handler_type: None,
@ -576,7 +576,7 @@ mod tests {
#[test]
fn agent_event_wrapper_serialization() {
let event = PipelineEvent::Agent {
let event = WorkflowRunEvent::Agent {
stage: "plan".to_string(),
event: AgentEvent::ToolCallStarted {
tool_name: "read_file".to_string(),
@ -591,13 +591,13 @@ mod tests {
assert!(json.contains("plan"));
// Verify round-trip
let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(deserialized, PipelineEvent::Agent { stage, .. } if stage == "plan"));
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(deserialized, WorkflowRunEvent::Agent { stage, .. } if stage == "plan"));
}
#[test]
fn agent_assistant_message_serialization() {
let event = PipelineEvent::Agent {
let event = WorkflowRunEvent::Agent {
stage: "code".to_string(),
event: AgentEvent::AssistantMessage {
text: "Here is the implementation".to_string(),
@ -621,9 +621,9 @@ mod tests {
assert!(json.contains("\"reasoning_tokens\":100"));
// Round-trip
let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
match deserialized {
PipelineEvent::Agent {
WorkflowRunEvent::Agent {
event: AgentEvent::AssistantMessage { usage, .. },
..
} => {
@ -636,7 +636,7 @@ mod tests {
#[test]
fn agent_assistant_message_without_cache_tokens_omits_them() {
let event = PipelineEvent::Agent {
let event = WorkflowRunEvent::Agent {
stage: "code".to_string(),
event: AgentEvent::AssistantMessage {
text: "response".to_string(),
@ -657,7 +657,7 @@ mod tests {
#[test]
fn stage_completed_event_serialization_with_new_fields() {
let event = PipelineEvent::StageCompleted {
let event = WorkflowRunEvent::StageCompleted {
name: "plan".to_string(),
index: 0,
duration_ms: 1500,
@ -680,7 +680,7 @@ mod tests {
assert!(json.contains("\"max_attempts\":3"));
assert!(json.contains("\"failure_class\":null"));
let event_none = PipelineEvent::StageCompleted {
let event_none = WorkflowRunEvent::StageCompleted {
name: "plan".to_string(),
index: 0,
duration_ms: 1500,
@ -702,7 +702,7 @@ mod tests {
#[test]
fn stage_failed_event_serialization() {
let event = PipelineEvent::StageFailed {
let event = WorkflowRunEvent::StageFailed {
name: "plan".to_string(),
index: 0,
error: "timeout".to_string(),
@ -714,12 +714,12 @@ mod tests {
assert!(json.contains("\"failure_reason\":\"LLM request timed out\""));
assert!(json.contains("\"failure_class\":\"transient\""));
let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(
matches!(deserialized, PipelineEvent::StageFailed { failure_class: Some(fc), .. } if fc == "transient")
matches!(deserialized, WorkflowRunEvent::StageFailed { failure_class: Some(fc), .. } if fc == "transient")
);
let event_none = PipelineEvent::StageFailed {
let event_none = WorkflowRunEvent::StageFailed {
name: "plan".to_string(),
index: 0,
error: "timeout".to_string(),
@ -734,7 +734,7 @@ mod tests {
#[test]
fn parallel_branch_completed_event_serialization() {
let event = PipelineEvent::ParallelBranchCompleted {
let event = WorkflowRunEvent::ParallelBranchCompleted {
branch: "branch_a".to_string(),
index: 0,
duration_ms: 1500,
@ -744,15 +744,15 @@ mod tests {
assert!(json.contains("\"status\":\"success\""));
assert!(!json.contains("\"success\":"));
let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(
matches!(deserialized, PipelineEvent::ParallelBranchCompleted { status, .. } if status == "success")
matches!(deserialized, WorkflowRunEvent::ParallelBranchCompleted { status, .. } if status == "success")
);
}
#[test]
fn parallel_started_event_serialization() {
let event = PipelineEvent::ParallelStarted {
let event = WorkflowRunEvent::ParallelStarted {
branch_count: 3,
join_policy: "wait_all".to_string(),
error_policy: "continue".to_string(),
@ -761,15 +761,15 @@ mod tests {
assert!(json.contains("\"join_policy\":\"wait_all\""));
assert!(json.contains("\"error_policy\":\"continue\""));
let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(
matches!(deserialized, PipelineEvent::ParallelStarted { join_policy, error_policy, .. } if join_policy == "wait_all" && error_policy == "continue")
matches!(deserialized, WorkflowRunEvent::ParallelStarted { join_policy, error_policy, .. } if join_policy == "wait_all" && error_policy == "continue")
);
}
#[test]
fn interview_started_event_serialization() {
let event = PipelineEvent::InterviewStarted {
let event = WorkflowRunEvent::InterviewStarted {
question: "Review changes?".to_string(),
stage: "gate".to_string(),
question_type: "multiple_choice".to_string(),
@ -777,15 +777,15 @@ mod tests {
let json = serde_json::to_string(&event).unwrap();
assert!(json.contains("\"question_type\":\"multiple_choice\""));
let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(
matches!(deserialized, PipelineEvent::InterviewStarted { question_type, .. } if question_type == "multiple_choice")
matches!(deserialized, WorkflowRunEvent::InterviewStarted { question_type, .. } if question_type == "multiple_choice")
);
}
#[test]
fn agent_compaction_event_serialization() {
let started = PipelineEvent::Agent {
let started = WorkflowRunEvent::Agent {
stage: "code".to_string(),
event: AgentEvent::CompactionStarted {
estimated_tokens: 5000,
@ -794,10 +794,10 @@ mod tests {
};
let json = serde_json::to_string(&started).unwrap();
assert!(json.contains("CompactionStarted"));
let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(deserialized, PipelineEvent::Agent { stage, .. } if stage == "code"));
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(deserialized, WorkflowRunEvent::Agent { stage, .. } if stage == "code"));
let completed = PipelineEvent::Agent {
let completed = WorkflowRunEvent::Agent {
stage: "code".to_string(),
event: AgentEvent::CompactionCompleted {
original_turn_count: 20,
@ -808,13 +808,13 @@ mod tests {
};
let json = serde_json::to_string(&completed).unwrap();
assert!(json.contains("CompactionCompleted"));
let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(deserialized, PipelineEvent::Agent { stage, .. } if stage == "code"));
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(deserialized, WorkflowRunEvent::Agent { stage, .. } if stage == "code"));
}
#[test]
fn edge_selected_event_serialization() {
let event = PipelineEvent::EdgeSelected {
let event = WorkflowRunEvent::EdgeSelected {
from_node: "plan".to_string(),
to_node: "code".to_string(),
label: Some("success".to_string()),
@ -827,13 +827,13 @@ mod tests {
assert!(json.contains("\"label\":\"success\""));
assert!(json.contains("\"condition\":\"outcome == 'success'\""));
let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(
matches!(deserialized, PipelineEvent::EdgeSelected { from_node, to_node, .. } if from_node == "plan" && to_node == "code")
matches!(deserialized, WorkflowRunEvent::EdgeSelected { from_node, to_node, .. } if from_node == "plan" && to_node == "code")
);
// None label/condition
let event_none = PipelineEvent::EdgeSelected {
let event_none = WorkflowRunEvent::EdgeSelected {
from_node: "a".to_string(),
to_node: "b".to_string(),
label: None,
@ -846,7 +846,7 @@ mod tests {
#[test]
fn loop_restart_event_serialization() {
let event = PipelineEvent::LoopRestart {
let event = WorkflowRunEvent::LoopRestart {
from_node: "review".to_string(),
to_node: "code".to_string(),
};
@ -855,15 +855,15 @@ mod tests {
assert!(json.contains("\"from_node\":\"review\""));
assert!(json.contains("\"to_node\":\"code\""));
let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(
matches!(deserialized, PipelineEvent::LoopRestart { from_node, to_node } if from_node == "review" && to_node == "code")
matches!(deserialized, WorkflowRunEvent::LoopRestart { from_node, to_node } if from_node == "review" && to_node == "code")
);
}
#[test]
fn stage_retrying_event_serialization() {
let event = PipelineEvent::StageRetrying {
let event = WorkflowRunEvent::StageRetrying {
name: "lint".to_string(),
index: 2,
attempt: 3,
@ -876,10 +876,10 @@ mod tests {
assert!(json.contains("\"max_attempts\":5"));
assert!(json.contains("\"delay_ms\":400"));
let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(
deserialized,
PipelineEvent::StageRetrying {
WorkflowRunEvent::StageRetrying {
max_attempts: 5,
..
}
@ -888,7 +888,7 @@ mod tests {
#[test]
fn agent_llm_retry_event_serialization() {
let event = PipelineEvent::Agent {
let event = WorkflowRunEvent::Agent {
stage: "code".to_string(),
event: AgentEvent::LlmRetry {
provider: "anthropic".to_string(),
@ -903,13 +903,13 @@ mod tests {
assert!(json.contains("\"provider\":\"anthropic\""));
assert!(json.contains("\"delay_secs\":1.5"));
let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(deserialized, PipelineEvent::Agent { stage, .. } if stage == "code"));
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(deserialized, WorkflowRunEvent::Agent { stage, .. } if stage == "code"));
}
#[test]
fn parallel_early_termination_event_serialization() {
let event = PipelineEvent::ParallelEarlyTermination {
let event = WorkflowRunEvent::ParallelEarlyTermination {
reason: "fail_fast_branch_failed".to_string(),
completed_count: 2,
pending_count: 3,
@ -919,10 +919,10 @@ mod tests {
assert!(json.contains("\"completed_count\":2"));
assert!(json.contains("\"pending_count\":3"));
let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(
deserialized,
PipelineEvent::ParallelEarlyTermination {
WorkflowRunEvent::ParallelEarlyTermination {
completed_count: 2,
..
}
@ -931,7 +931,7 @@ mod tests {
#[test]
fn subgraph_started_event_serialization() {
let event = PipelineEvent::SubgraphStarted {
let event = WorkflowRunEvent::SubgraphStarted {
node_id: "sub_1".to_string(),
start_node: "start".to_string(),
};
@ -940,15 +940,15 @@ mod tests {
assert!(json.contains("\"node_id\":\"sub_1\""));
assert!(json.contains("\"start_node\":\"start\""));
let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(
matches!(deserialized, PipelineEvent::SubgraphStarted { node_id, .. } if node_id == "sub_1")
matches!(deserialized, WorkflowRunEvent::SubgraphStarted { node_id, .. } if node_id == "sub_1")
);
}
#[test]
fn subgraph_completed_event_serialization() {
let event = PipelineEvent::SubgraphCompleted {
let event = WorkflowRunEvent::SubgraphCompleted {
node_id: "sub_1".to_string(),
steps_executed: 5,
status: "success".to_string(),
@ -959,10 +959,10 @@ mod tests {
assert!(json.contains("\"steps_executed\":5"));
assert!(json.contains("\"duration_ms\":3200"));
let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(
deserialized,
PipelineEvent::SubgraphCompleted {
WorkflowRunEvent::SubgraphCompleted {
steps_executed: 5,
..
}
@ -973,7 +973,7 @@ mod tests {
fn execution_env_event_wrapper_serialization() {
use arc_agent::ExecutionEnvEvent;
let event = PipelineEvent::ExecutionEnv {
let event = WorkflowRunEvent::ExecutionEnv {
event: ExecutionEnvEvent::Initializing {
env_type: "docker".into(),
},
@ -983,8 +983,8 @@ mod tests {
assert!(json.contains("Initializing"));
assert!(json.contains("docker"));
let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(deserialized, PipelineEvent::ExecutionEnv { .. }));
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(deserialized, WorkflowRunEvent::ExecutionEnv { .. }));
}
#[test]
@ -997,7 +997,7 @@ mod tests {
fn emitter_last_event_at_updates_after_emit() {
let emitter = EventEmitter::new();
assert_eq!(emitter.last_event_at(), 0);
emitter.emit(&PipelineEvent::PipelineStarted {
emitter.emit(&WorkflowRunEvent::WorkflowRunStarted {
name: "test".to_string(),
run_id: "1".to_string(),
base_sha: None,
@ -1017,7 +1017,7 @@ mod tests {
#[test]
fn stall_watchdog_timeout_serialization() {
let event = PipelineEvent::StallWatchdogTimeout {
let event = WorkflowRunEvent::StallWatchdogTimeout {
node: "work".to_string(),
idle_seconds: 600,
};
@ -1026,28 +1026,28 @@ mod tests {
assert!(json.contains("\"node\":\"work\""));
assert!(json.contains("\"idle_seconds\":600"));
let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(
matches!(deserialized, PipelineEvent::StallWatchdogTimeout { node, idle_seconds } if node == "work" && idle_seconds == 600)
matches!(deserialized, WorkflowRunEvent::StallWatchdogTimeout { node, idle_seconds } if node == "work" && idle_seconds == 600)
);
}
#[test]
fn setup_events_serialization() {
let events = vec![
PipelineEvent::SetupStarted { command_count: 3 },
PipelineEvent::SetupCommandStarted {
WorkflowRunEvent::SetupStarted { command_count: 3 },
WorkflowRunEvent::SetupCommandStarted {
command: "npm install".into(),
index: 0,
},
PipelineEvent::SetupCommandCompleted {
WorkflowRunEvent::SetupCommandCompleted {
command: "npm install".into(),
index: 0,
exit_code: 0,
duration_ms: 5000,
},
PipelineEvent::SetupCompleted { duration_ms: 8000 },
PipelineEvent::SetupFailed {
WorkflowRunEvent::SetupCompleted { duration_ms: 8000 },
WorkflowRunEvent::SetupFailed {
command: "npm test".into(),
index: 1,
exit_code: 1,
@ -1057,7 +1057,7 @@ mod tests {
for event in &events {
let json = serde_json::to_string(event).unwrap();
let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
let json2 = serde_json::to_string(&deserialized).unwrap();
assert_eq!(json, json2);
}

10
crates/arc-workflows/src/handler/manager_loop.rs
View file

@ -8,15 +8,15 @@ use async_trait::async_trait;
use crate::condition::evaluate_condition;
use crate::context::Context;
use crate::engine::{PipelineEngine, RunConfig};
use crate::engine::{WorkflowRunEngine, RunConfig};
use crate::error::ArcError;
use crate::graph::{Graph, Node};
use crate::outcome::{Outcome, StageStatus};
use crate::pipeline::prepare_pipeline;
use crate::workflow::prepare_workflow;
use super::{EngineServices, Handler};
/// Orchestrates a child pipeline engine, polling for completion or stop conditions.
/// Orchestrates a child workflow engine, polling for completion or stop conditions.
pub struct SubWorkflowHandler;
/// Parse a duration string like "45s", "200ms", "5m" into a Duration.
@ -110,7 +110,7 @@ impl Handler for SubWorkflowHandler {
Err(e) => return Ok(Outcome::fail(e.to_string())),
};
let child_graph = match prepare_pipeline(&dot_source) {
let child_graph = match prepare_workflow(&dot_source) {
Ok(g) => g,
Err(e) => {
return Ok(Outcome::fail(format!(
@ -151,7 +151,7 @@ impl Handler for SubWorkflowHandler {
let before_snapshot = context.snapshot();
// Spawn child engine
let engine = PipelineEngine::from_services(services);
let engine = WorkflowRunEngine::from_services(services);
let mut child_handle =
tokio::spawn(
async move { engine.run_with_context(&child_graph, &child_config, child_context).await },

18
crates/arc-workflows/src/handler/parallel.rs
View file

@ -9,7 +9,7 @@ use tokio::sync::Semaphore;
use crate::context::Context;
use crate::engine::GitCheckpointMode;
use crate::error::ArcError;
use crate::event::PipelineEvent;
use crate::event::WorkflowRunEvent;
use crate::graph::{Graph, Node};
use crate::outcome::{Outcome, StageStatus};
@ -208,7 +208,7 @@ impl Handler for ParallelHandler {
.unwrap_or("continue"),
);
services.emitter.emit(&PipelineEvent::ParallelStarted {
services.emitter.emit(&WorkflowRunEvent::ParallelStarted {
branch_count: branches.len(),
join_policy: join_policy.to_string(),
error_policy: error_policy.to_string(),
@ -391,7 +391,7 @@ impl Handler for ParallelHandler {
.await
.map_err(|e| ArcError::Handler(format!("semaphore error: {e}")))?;
emitter.emit(&PipelineEvent::ParallelBranchStarted {
emitter.emit(&WorkflowRunEvent::ParallelBranchStarted {
branch: setup.target_id.clone(),
index: setup.branch_index,
});
@ -400,7 +400,7 @@ impl Handler for ParallelHandler {
let Some(target_node) = graph.nodes.get(&setup.target_id) else {
let outcome =
Outcome::fail(format!("branch target node not found: {}", setup.target_id));
emitter.emit(&PipelineEvent::ParallelBranchCompleted {
emitter.emit(&WorkflowRunEvent::ParallelBranchCompleted {
branch: setup.target_id.clone(),
index: setup.branch_index,
duration_ms: millis_u64(branch_start.elapsed()),
@ -466,7 +466,7 @@ impl Handler for ParallelHandler {
None
};
emitter.emit(&PipelineEvent::ParallelBranchCompleted {
emitter.emit(&WorkflowRunEvent::ParallelBranchCompleted {
branch: setup.target_id.clone(),
index: setup.branch_index,
duration_ms: millis_u64(branch_start.elapsed()),
@ -495,7 +495,7 @@ impl Handler for ParallelHandler {
results.push(result);
services
.emitter
.emit(&PipelineEvent::ParallelEarlyTermination {
.emit(&WorkflowRunEvent::ParallelEarlyTermination {
reason: "fail_fast_branch_failed".to_string(),
completed_count: results.len(),
pending_count: total_branches - handle_index - 1,
@ -515,7 +515,7 @@ impl Handler for ParallelHandler {
results.push(result);
services
.emitter
.emit(&PipelineEvent::ParallelEarlyTermination {
.emit(&WorkflowRunEvent::ParallelEarlyTermination {
reason: "fail_fast_handler_error".to_string(),
completed_count: results.len(),
pending_count: total_branches - handle_index - 1,
@ -535,7 +535,7 @@ impl Handler for ParallelHandler {
results.push(result);
services
.emitter
.emit(&PipelineEvent::ParallelEarlyTermination {
.emit(&WorkflowRunEvent::ParallelEarlyTermination {
reason: "fail_fast_join_error".to_string(),
completed_count: results.len(),
pending_count: total_branches - handle_index - 1,
@ -634,7 +634,7 @@ impl Handler for ParallelHandler {
let _ = tokio::fs::write(node_dir.join("parallel_results.json"), json).await;
}
services.emitter.emit(&PipelineEvent::ParallelCompleted {
services.emitter.emit(&WorkflowRunEvent::ParallelCompleted {
duration_ms: millis_u64(parallel_start.elapsed()),
success_count,
failure_count: fail_count,

10
crates/arc-workflows/src/handler/wait_human.rs
View file

@ -6,7 +6,7 @@ use async_trait::async_trait;
use crate::context::Context;
use crate::error::ArcError;
use crate::event::{EventEmitter, PipelineEvent};
use crate::event::{EventEmitter, WorkflowRunEvent};
use crate::graph::{Graph, Node};
use crate::interviewer::{
Answer, AnswerValue, Interviewer, Question, QuestionOption, QuestionType,
@ -90,7 +90,7 @@ impl WaitHumanHandler {
self
}
fn emit(&self, event: &PipelineEvent) {
fn emit(&self, event: &WorkflowRunEvent) {
if let Some(emitter) = &self.emitter {
emitter.emit(event);
}
@ -146,7 +146,7 @@ impl Handler for WaitHumanHandler {
// 3. Present to interviewer
let question_text = node.label().to_string();
self.emit(&PipelineEvent::InterviewStarted {
self.emit(&WorkflowRunEvent::InterviewStarted {
question: question_text.clone(),
stage: node.id.clone(),
question_type: question.question_type.to_string(),
@ -156,7 +156,7 @@ impl Handler for WaitHumanHandler {
// 4. Handle timeout
if answer.value == AnswerValue::Timeout {
self.emit(&PipelineEvent::InterviewTimeout {
self.emit(&WorkflowRunEvent::InterviewTimeout {
question: question_text,
stage: node.id.clone(),
duration_ms: millis_u64(interview_start.elapsed()),
@ -181,7 +181,7 @@ impl Handler for WaitHumanHandler {
}
// Emit interview completed for successful interactions
self.emit(&PipelineEvent::InterviewCompleted {
self.emit(&WorkflowRunEvent::InterviewCompleted {
question: question_text,
answer: answer_text(&answer),
duration_ms: millis_u64(interview_start.elapsed()),

2
crates/arc-workflows/src/lib.rs
View file

@ -13,7 +13,7 @@ pub mod handler;
pub mod interviewer;
pub mod outcome;
pub mod parser;
pub mod pipeline;
pub mod workflow;
pub mod preamble;
pub mod retro;
pub mod retro_agent;

24
crates/arc-workflows/src/retro.rs
View file

@ -6,7 +6,7 @@ use serde::{Deserialize, Serialize};
use crate::checkpoint::Checkpoint;
use crate::error::{ArcError, Result};
use crate::event::PipelineEvent;
use crate::event::WorkflowRunEvent;
use crate::outcome::StageStatus;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@ -113,7 +113,7 @@ pub struct RetroNarrative {
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Retro {
pub run_id: String,
pub pipeline_name: String,
pub workflow_name: String,
pub goal: String,
pub timestamp: DateTime<Utc>,
#[serde(default, skip_serializing_if = "Option::is_none")]
@ -186,10 +186,10 @@ pub fn extract_stage_durations(logs_root: &Path) -> HashMap<String, u64> {
let Some(event_value) = envelope.get("event") else {
continue;
};
let Ok(event) = serde_json::from_value::<PipelineEvent>(event_value.clone()) else {
let Ok(event) = serde_json::from_value::<WorkflowRunEvent>(event_value.clone()) else {
continue;
};
if let PipelineEvent::StageCompleted {
if let WorkflowRunEvent::StageCompleted {
name, duration_ms, ..
} = event
{
@ -205,11 +205,11 @@ pub fn extract_stage_durations(logs_root: &Path) -> HashMap<String, u64> {
#[allow(clippy::too_many_arguments)]
pub fn derive_retro(
run_id: &str,
pipeline_name: &str,
workflow_name: &str,
goal: &str,
checkpoint: &Checkpoint,
pipeline_failed: bool,
_pipeline_error: Option<&str>,
run_failed: bool,
_run_error: Option<&str>,
duration_ms: u64,
stage_durations: &HashMap<String, u64>,
) -> Retro {
@ -262,8 +262,8 @@ pub fn derive_retro(
});
}
// If pipeline failed with an error not captured in stages, record it
if pipeline_failed && stages_failed == 0 {
// If run failed with an error not captured in stages, record it
if run_failed && stages_failed == 0 {
stages_failed = 1;
}
@ -281,7 +281,7 @@ pub fn derive_retro(
Retro {
run_id: run_id.to_string(),
pipeline_name: pipeline_name.to_string(),
workflow_name: workflow_name.to_string(),
goal: goal.to_string(),
timestamp: Utc::now(),
smoothness: None,
@ -368,7 +368,7 @@ mod tests {
);
assert_eq!(retro.run_id, "run-1");
assert_eq!(retro.pipeline_name, "my_pipeline");
assert_eq!(retro.workflow_name, "my_pipeline");
assert_eq!(retro.goal, "Fix the bug");
assert_eq!(retro.stages.len(), 2);
assert_eq!(retro.stages[0].stage_id, "plan");
@ -388,7 +388,7 @@ mod tests {
}
#[test]
fn derive_retro_handles_failed_pipeline() {
fn derive_retro_handles_failed_run() {
let cp = Checkpoint {
timestamp: Utc::now(),
current_node: "start".to_string(),

22
crates/arc-workflows/src/retro_agent.rs
View file

@ -12,9 +12,9 @@ use arc_llm::types::ToolDefinition;
use crate::retro::RetroNarrative;
const RETRO_SYSTEM_PROMPT: &str = r#"You are a pipeline retrospective analyst. Your job is to analyze a completed pipeline run and generate a structured retrospective.
const RETRO_SYSTEM_PROMPT: &str = r#"You are a workflow run retrospective analyst. Your job is to analyze a completed workflow run and generate a structured retrospective.
You have access to the pipeline's data files:
You have access to the run's data files:
- `progress.ndjson` — the full event stream (stage starts/completions, agent tool calls, errors, retries)
- `checkpoint.json` — final execution state with node outcomes
- `manifest.json` — run metadata (if available)
@ -33,17 +33,17 @@ You have access to the pipeline's data files:
Grade the run on a 5-point scale:
- **effortless** — Pipeline achieved its goal on the first try with no retries, no wrong approaches. Agent moved efficiently from start to finish.
- **effortless** — Run achieved its goal on the first try with no retries, no wrong approaches. Agent moved efficiently from start to finish.
- **smooth** — Goal achieved with minor hiccups (1-2 retries or a brief wrong approach quickly corrected). No human intervention needed. Overall clean execution.
- **bumpy** — Goal achieved but with notable friction: multiple retries, at least one significant wrong approach, or substantial time spent on dead ends.
- **struggled** — Goal achieved only with difficulty: many retries, major approach changes, human intervention, or partial failures requiring recovery.
- **failed** — Pipeline did not achieve its stated goal. May have completed some stages but the overall intent was not fulfilled.
- **failed** — Run did not achieve its stated goal. May have completed some stages but the overall intent was not fulfilled.
Consider the full context: not just stage pass/fail, but the quality of the journey visible in the agent events (tool call patterns, error recovery, approach pivots).
## Guidelines for qualitative fields
- **intent**: What was the pipeline trying to accomplish? Summarize the goal in a sentence.
- **intent**: What was the workflow run trying to accomplish? Summarize the goal in a sentence.
- **outcome**: What actually happened? Did it succeed? What was produced?
- **learnings**: What was discovered about the repo, code, workflow, or tools?
- **friction_points**: Where did things get stuck? What caused slowdowns?
@ -57,11 +57,11 @@ const SUBMIT_RETRO_SCHEMA: &str = r#"{
"smoothness": {
"type": "string",
"enum": ["effortless", "smooth", "bumpy", "struggled", "failed"],
"description": "Overall smoothness rating for the pipeline run"
"description": "Overall smoothness rating for the workflow run"
},
"intent": {
"type": "string",
"description": "What was the pipeline trying to accomplish?"
"description": "What was the workflow run trying to accomplish?"
},
"outcome": {
"type": "string",
@ -108,7 +108,7 @@ const SUBMIT_RETRO_SCHEMA: &str = r#"{
"required": ["smoothness", "intent", "outcome"]
}"#;
/// Run a retro agent session that analyzes pipeline run data and produces
/// Run a retro agent session that analyzes workflow run data and produces
/// a structured narrative. The agent explores `progress.ndjson` and other
/// files via tool access, then calls `submit_retro` with its analysis.
pub async fn run_retro_agent(
@ -133,7 +133,7 @@ pub async fn run_retro_agent(
let submit_tool = arc_agent::tool_registry::RegisteredTool {
definition: ToolDefinition {
name: "submit_retro".to_string(),
description: "Submit the structured retrospective analysis. Call this once you have analyzed the pipeline run data.".to_string(),
description: "Submit the structured retrospective analysis. Call this once you have analyzed the workflow run data.".to_string(),
parameters: serde_json::from_str(SUBMIT_RETRO_SCHEMA)
.expect("submit_retro schema should be valid JSON"),
},
@ -171,7 +171,7 @@ pub async fn run_retro_agent(
session.initialize().await;
let prompt = format!(
"Analyze the pipeline run data at `{retro_data_dir}/` and generate a retrospective. \
"Analyze the workflow run data at `{retro_data_dir}/` and generate a retrospective. \
The key file is `{retro_data_dir}/progress.ndjson` which contains the full event stream. \
Also check `{retro_data_dir}/checkpoint.json` for stage outcomes. \
Use grep to search for interesting signals (failures, retries, errors, approach changes) \
@ -199,7 +199,7 @@ pub fn dry_run_narrative() -> RetroNarrative {
RetroNarrative {
smoothness: crate::retro::SmoothnessRating::Smooth,
intent: "[dry-run] No LLM analysis performed".to_string(),
outcome: "[dry-run] Pipeline completed in simulated mode".to_string(),
outcome: "[dry-run] Run completed in simulated mode".to_string(),
learnings: vec![],
friction_points: vec![],
open_items: vec![],

38
crates/arc-workflows/src/pipeline.rs → crates/arc-workflows/src/workflow.rs
View file

@ -3,13 +3,13 @@ use crate::graph::Graph;
use crate::transform::{StylesheetApplicationTransform, Transform, VariableExpansionTransform};
use crate::validation::{self, Diagnostic};
/// Builder for configuring and executing a pipeline preparation.
/// Builder for configuring and executing a workflow preparation.
/// Collects custom transforms that run after the built-in ones.
pub struct PipelineBuilder {
pub struct WorkflowBuilder {
transforms: Vec<Box<dyn Transform>>,
}
impl PipelineBuilder {
impl WorkflowBuilder {
#[must_use]
pub fn new() -> Self {
Self {
@ -23,7 +23,7 @@ impl PipelineBuilder {
self.transforms.push(transform);
}
/// Prepare a pipeline: parse DOT, apply built-in and custom transforms, validate.
/// Prepare a workflow: parse DOT, apply built-in and custom transforms, validate.
///
/// # Errors
///
@ -45,7 +45,7 @@ impl PipelineBuilder {
}
}
impl Default for PipelineBuilder {
impl Default for WorkflowBuilder {
fn default() -> Self {
Self::new()
}
@ -56,8 +56,8 @@ impl Default for PipelineBuilder {
/// # Errors
///
/// Returns an error if parsing fails or if validation produces Error-severity diagnostics.
pub fn prepare_pipeline(dot_source: &str) -> Result<Graph, ArcError> {
let builder = PipelineBuilder::new();
pub fn prepare_workflow(dot_source: &str) -> Result<Graph, ArcError> {
let builder = WorkflowBuilder::new();
let (graph, diagnostics) = builder.prepare(dot_source)?;
let errors: Vec<&Diagnostic> = diagnostics
@ -85,15 +85,15 @@ mod tests {
}"#;
#[test]
fn prepare_pipeline_minimal() {
let graph = prepare_pipeline(MINIMAL_DOT).unwrap();
fn prepare_workflow_minimal() {
let graph = prepare_workflow(MINIMAL_DOT).unwrap();
assert_eq!(graph.name, "Test");
assert!(graph.find_start_node().is_some());
assert!(graph.find_exit_node().is_some());
}
#[test]
fn prepare_pipeline_applies_variable_expansion() {
fn prepare_workflow_applies_variable_expansion() {
let dot = r#"digraph Test {
graph [goal="Fix bugs"]
start [shape=Mdiamond]
@ -101,7 +101,7 @@ mod tests {
exit [shape=Msquare]
start -> work -> exit
}"#;
let graph = prepare_pipeline(dot).unwrap();
let graph = prepare_workflow(dot).unwrap();
let prompt = graph.nodes["work"]
.attrs
.get("prompt")
@ -111,7 +111,7 @@ mod tests {
}
#[test]
fn prepare_pipeline_applies_stylesheet() {
fn prepare_workflow_applies_stylesheet() {
let dot = r#"digraph Test {
graph [goal="Test", model_stylesheet="* { llm_model: sonnet; }"]
start [shape=Mdiamond]
@ -119,7 +119,7 @@ mod tests {
exit [shape=Msquare]
start -> work -> exit
}"#;
let graph = prepare_pipeline(dot).unwrap();
let graph = prepare_workflow(dot).unwrap();
assert_eq!(
graph.nodes["work"].attrs.get("llm_model"),
Some(&AttrValue::String("sonnet".into()))
@ -127,18 +127,18 @@ mod tests {
}
#[test]
fn prepare_pipeline_returns_error_on_invalid_dot() {
let result = prepare_pipeline("not a graph");
fn prepare_workflow_returns_error_on_invalid_dot() {
let result = prepare_workflow("not a graph");
assert!(result.is_err());
}
#[test]
fn prepare_pipeline_returns_error_on_validation_failure() {
fn prepare_workflow_returns_error_on_validation_failure() {
let dot = r#"digraph Test {
graph [goal="Test"]
work [label="Work"]
}"#;
let result = prepare_pipeline(dot);
let result = prepare_workflow(dot);
assert!(result.is_err());
}
@ -154,7 +154,7 @@ mod tests {
}
}
let mut builder = PipelineBuilder::new();
let mut builder = WorkflowBuilder::new();
builder.register_transform(Box::new(TagTransform));
let (graph, _) = builder.prepare(MINIMAL_DOT).unwrap();
assert_eq!(
@ -165,7 +165,7 @@ mod tests {
#[test]
fn pipeline_builder_default() {
let builder = PipelineBuilder::default();
let builder = WorkflowBuilder::default();
let (graph, _) = builder.prepare(MINIMAL_DOT).unwrap();
assert_eq!(graph.name, "Test");
}

16
crates/arc-workflows/tests/daytona_integration.rs
View file

@ -13,7 +13,7 @@ use arc_workflows::artifact::sync_artifacts_to_env;
use arc_workflows::checkpoint::Checkpoint;
use arc_workflows::context::Context;
use arc_workflows::daytona_env::{DaytonaConfig, DaytonaExecutionEnvironment};
use arc_workflows::engine::{PipelineEngine, RunConfig};
use arc_workflows::engine::{WorkflowRunEngine, RunConfig};
use arc_workflows::error::ArcError;
use arc_workflows::event::EventEmitter;
use arc_workflows::graph::{AttrValue, Edge, Graph, Node};
@ -274,7 +274,7 @@ async fn daytona_pipeline_artifact_offload_and_sync() {
registry.register("start", Box::new(StartHandler));
registry.register("exit", Box::new(ExitHandler));
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), env.clone());
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), env.clone());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -465,7 +465,7 @@ async fn daytona_git_checkpoint_remote_emits_events() {
registry.register("start", Box::new(StartHandler));
registry.register("exit", Box::new(ExitHandler));
let engine = PipelineEngine::new(registry, Arc::new(emitter), env.clone());
let engine = WorkflowRunEngine::new(registry, Arc::new(emitter), env.clone());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,
@ -490,7 +490,7 @@ async fn daytona_git_checkpoint_remote_emits_events() {
let git_events: Vec<_> = events
.iter()
.filter_map(|e| {
if let arc_workflows::event::PipelineEvent::GitCheckpoint {
if let arc_workflows::event::WorkflowRunEvent::GitCheckpoint {
node_id,
git_commit_sha,
..
@ -644,7 +644,7 @@ async fn daytona_parallel_git_branching_e2e() {
registry.register("parallel", Box::new(ParallelHandler));
registry.register("parallel.fan_in", Box::new(FanInHandler::new(None)));
let engine = PipelineEngine::new(registry, Arc::new(emitter), Arc::clone(&env));
let engine = WorkflowRunEngine::new(registry, Arc::new(emitter), Arc::clone(&env));
let config = RunConfig {
logs_root: logs_dir.path().to_path_buf(),
@ -740,7 +740,7 @@ async fn daytona_parallel_git_branching_e2e() {
.filter(|e| {
matches!(
e,
arc_workflows::event::PipelineEvent::ParallelStarted { .. }
arc_workflows::event::WorkflowRunEvent::ParallelStarted { .. }
)
})
.collect();
@ -754,7 +754,7 @@ async fn daytona_parallel_git_branching_e2e() {
.filter(|e| {
matches!(
e,
arc_workflows::event::PipelineEvent::ParallelCompleted { .. }
arc_workflows::event::WorkflowRunEvent::ParallelCompleted { .. }
)
})
.collect();
@ -970,7 +970,7 @@ async fn daytona_git_checkpoint_with_shadow_branch() {
registry.register("exit", Box::new(ExitHandler));
let meta_branch = MetadataStore::branch_name(&run_id);
let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), env.clone());
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), env.clone());
let config = RunConfig {
logs_root: dir.path().to_path_buf(),
cancel_token: None,

288
crates/arc-workflows/tests/integration.rs
View file

File diff suppressed because it is too large Load diff