Implement attractor CLI binary with run and validate subcommands

Add a [[bin]] target to the attractor crate with two subcommands:
- `attractor validate <pipeline.dot>` -- parse and validate only
- `attractor run <pipeline.dot>` -- full pipeline execution with LLM backend

The CLI supports --dry-run, --auto-approve, --resume, --model, --provider,
and two-level verbosity (-v one-line summaries, -vv full event details).

Extracts a shared `default_registry()` function in handler/mod.rs so both
the CLI and server can build a fully-wired HandlerRegistry without
duplicating handler registration boilerplate.

The AgentBackend in cli/backend.rs implements CodergenBackend by creating
a coding-agent-loop Session per node invocation, giving LLM nodes access
to file and shell tools.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-02-23 09:11:42 -05:00
parent 54d0c887dc
commit 433a7c6247
9 changed files with 661 additions and 0 deletions

2
Cargo.lock generated
View file

@ -132,9 +132,11 @@ checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0"
name = "attractor"
version = "0.1.0"
dependencies = [
"anyhow",
"async-trait",
"axum",
"chrono",
"clap",
"coding-agent-loop",
"dotenvy",
"futures",

View file

@ -9,11 +9,18 @@ keywords = ["llm", "ai", "pipeline", "workflow", "dot"]
categories = ["development-tools"]
readme = "README.md"
[[bin]]
name = "attractor"
path = "src/main.rs"
[features]
default = []
server = ["axum", "tower", "tokio-stream"]
[dependencies]
clap.workspace = true
anyhow.workspace = true
dotenvy.workspace = true
coding-agent-loop = { path = "../coding-agent-loop" }
unified-llm = { path = "../unified-llm" }
thiserror.workspace = true

View file

@ -0,0 +1,85 @@
use std::path::PathBuf;
use std::sync::Arc;
use async_trait::async_trait;
use coding_agent_loop::{
AnthropicProfile, GeminiProfile, LocalExecutionEnvironment, OpenAiProfile, ProviderProfile,
Session, SessionConfig, Turn,
};
use unified_llm::client::Client;
use crate::context::Context;
use crate::error::AttractorError;
use crate::graph::Node;
use crate::handler::codergen::{CodergenBackend, CodergenResult};
/// LLM backend that delegates to a `coding-agent-loop` Session per invocation.
pub struct AgentBackend {
model: String,
provider: Option<String>,
}
impl AgentBackend {
#[must_use]
pub const fn new(model: String, provider: Option<String>) -> Self {
Self { model, provider }
}
fn build_profile(&self) -> Arc<dyn ProviderProfile> {
let provider = self.provider.as_deref().unwrap_or("anthropic");
match provider {
"openai" => Arc::new(OpenAiProfile::new(&self.model)),
"gemini" => Arc::new(GeminiProfile::new(&self.model)),
_ => Arc::new(AnthropicProfile::new(&self.model)),
}
}
}
#[async_trait]
impl CodergenBackend for AgentBackend {
async fn run(
&self,
node: &Node,
prompt: &str,
_context: &Context,
_thread_id: Option<&str>,
) -> Result<CodergenResult, AttractorError> {
let client = Client::from_env()
.await
.map_err(|e| AttractorError::Handler(format!("Failed to create LLM client: {e}")))?;
let profile = self.build_profile();
let cwd = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
let exec_env = Arc::new(LocalExecutionEnvironment::new(cwd));
let config = SessionConfig {
reasoning_effort: Some(node.reasoning_effort().to_string()),
..SessionConfig::default()
};
let mut session = Session::new(client, profile, exec_env, config);
session.initialize().await;
session.process_input(prompt).await.map_err(|e| {
AttractorError::Handler(format!("Agent session failed: {e}"))
})?;
// Extract last assistant response from the session history.
let response = session
.history()
.turns()
.iter()
.rev()
.find_map(|turn| {
if let Turn::Assistant { content, .. } = turn {
if !content.is_empty() {
return Some(content.clone());
}
}
None
})
.unwrap_or_default();
Ok(CodergenResult::Text(response))
}
}

View file

@ -0,0 +1,275 @@
pub mod backend;
pub mod run;
pub mod validate;
use std::path::Path;
use clap::{Args, Parser, Subcommand};
use std::path::PathBuf;
use crate::event::PipelineEvent;
use crate::validation::{Diagnostic, Severity};
#[derive(Parser)]
#[command(name = "attractor", version, about = "DOT-based pipeline runner for AI workflows")]
pub struct Cli {
#[command(subcommand)]
pub command: Command,
}
#[derive(Subcommand)]
pub enum Command {
/// Launch a pipeline from a .dot file
Run(RunArgs),
/// Parse and validate a pipeline without executing
Validate(ValidateArgs),
}
#[derive(Args)]
pub struct RunArgs {
/// Path to the .dot pipeline file
pub pipeline: PathBuf,
/// Log/artifact directory
#[arg(long)]
pub logs_dir: Option<PathBuf>,
/// Execute with simulated LLM backend
#[arg(long)]
pub dry_run: bool,
/// Auto-approve all human gates
#[arg(long)]
pub auto_approve: bool,
/// Resume from a checkpoint file
#[arg(long)]
pub resume: Option<PathBuf>,
/// Override default LLM model
#[arg(long)]
pub model: Option<String>,
/// Override default LLM provider
#[arg(long)]
pub provider: Option<String>,
/// Verbosity level (-v summary, -vv full details)
#[arg(short, long, action = clap::ArgAction::Count)]
pub verbose: u8,
}
#[derive(Args)]
pub struct ValidateArgs {
/// Path to the .dot pipeline file
pub pipeline: PathBuf,
}
/// Read a .dot file from disk.
///
/// # Errors
///
/// Returns an error if the file cannot be read.
pub fn read_dot_file(path: &Path) -> anyhow::Result<String> {
std::fs::read_to_string(path)
.map_err(|e| anyhow::anyhow!("Failed to read {}: {e}", path.display()))
}
/// Print diagnostics to stderr, grouped by severity.
pub fn print_diagnostics(diagnostics: &[Diagnostic]) {
for d in diagnostics {
let prefix = match d.severity {
Severity::Error => "error",
Severity::Warning => "warning",
Severity::Info => "info",
};
let location = match (&d.node_id, &d.edge) {
(Some(node), _) => format!(" [node: {node}]"),
(_, Some((from, to))) => format!(" [edge: {from} -> {to}]"),
_ => String::new(),
};
eprintln!("{prefix}{location}: {} ({})", d.message, d.rule);
}
}
/// One-line summary of a pipeline event for `-v` output.
#[must_use]
pub fn format_event_summary(event: &PipelineEvent) -> String {
match event {
PipelineEvent::PipelineStarted { name, id } => {
format!("[PIPELINE_STARTED] name={name} id={id}")
}
PipelineEvent::PipelineCompleted {
duration_ms,
artifact_count,
} => {
format!("[PIPELINE_COMPLETED] duration={duration_ms}ms artifacts={artifact_count}")
}
PipelineEvent::PipelineFailed { error, duration_ms } => {
format!("[PIPELINE_FAILED] error=\"{error}\" duration={duration_ms}ms")
}
PipelineEvent::StageStarted { name, index } => {
format!("[STAGE_STARTED] name={name} index={index}")
}
PipelineEvent::StageCompleted {
name,
index,
duration_ms,
} => {
format!("[STAGE_COMPLETED] name={name} index={index} duration={duration_ms}ms")
}
PipelineEvent::StageFailed {
name,
index,
error,
will_retry,
} => {
format!(
"[STAGE_FAILED] name={name} index={index} error=\"{error}\" will_retry={will_retry}"
)
}
PipelineEvent::StageRetrying {
name,
index,
attempt,
delay_ms,
} => {
format!(
"[STAGE_RETRYING] name={name} index={index} attempt={attempt} delay={delay_ms}ms"
)
}
PipelineEvent::ParallelStarted { branch_count } => {
format!("[PARALLEL_STARTED] branches={branch_count}")
}
PipelineEvent::ParallelBranchStarted { branch, index } => {
format!("[PARALLEL_BRANCH_STARTED] branch={branch} index={index}")
}
PipelineEvent::ParallelBranchCompleted {
branch,
index,
duration_ms,
success,
} => {
format!("[PARALLEL_BRANCH_COMPLETED] branch={branch} index={index} duration={duration_ms}ms success={success}")
}
PipelineEvent::ParallelCompleted {
duration_ms,
success_count,
failure_count,
} => {
format!("[PARALLEL_COMPLETED] duration={duration_ms}ms succeeded={success_count} failed={failure_count}")
}
PipelineEvent::InterviewStarted { question, stage } => {
format!("[INTERVIEW_STARTED] stage={stage} question=\"{question}\"")
}
PipelineEvent::InterviewCompleted {
question,
answer,
duration_ms,
} => {
format!(
"[INTERVIEW_COMPLETED] question=\"{question}\" answer=\"{answer}\" duration={duration_ms}ms"
)
}
PipelineEvent::InterviewTimeout {
stage, duration_ms, ..
} => {
format!("[INTERVIEW_TIMEOUT] stage={stage} duration={duration_ms}ms")
}
PipelineEvent::CheckpointSaved { node_id } => {
format!("[CHECKPOINT_SAVED] node={node_id}")
}
}
}
/// Multi-line detail view of a pipeline event for `-vv` output.
#[must_use]
pub fn format_event_detail(event: &PipelineEvent) -> String {
match event {
PipelineEvent::PipelineStarted { name, id } => {
format!(
"── PIPELINE_STARTED ─────────────────────────\n name: {name}\n id: {id}\n"
)
}
PipelineEvent::PipelineCompleted {
duration_ms,
artifact_count,
} => {
format!("── PIPELINE_COMPLETED ───────────────────────\n duration_ms: {duration_ms}\n artifact_count: {artifact_count}\n")
}
PipelineEvent::PipelineFailed { error, duration_ms } => {
format!("── PIPELINE_FAILED ──────────────────────────\n error: {error}\n duration_ms: {duration_ms}\n")
}
PipelineEvent::StageStarted { name, index } => {
format!(
"── STAGE_STARTED ────────────────────────────\n name: {name}\n index: {index}\n"
)
}
PipelineEvent::StageCompleted {
name,
index,
duration_ms,
} => {
format!("── STAGE_COMPLETED ──────────────────────────\n name: {name}\n index: {index}\n duration_ms: {duration_ms}\n")
}
PipelineEvent::StageFailed {
name,
index,
error,
will_retry,
} => {
format!("── STAGE_FAILED ─────────────────────────────\n name: {name}\n index: {index}\n error: {error}\n will_retry: {will_retry}\n")
}
PipelineEvent::StageRetrying {
name,
index,
attempt,
delay_ms,
} => {
format!("── STAGE_RETRYING ───────────────────────────\n name: {name}\n index: {index}\n attempt: {attempt}\n delay_ms: {delay_ms}\n")
}
PipelineEvent::ParallelStarted { branch_count } => {
format!("── PARALLEL_STARTED ─────────────────────────\n branch_count: {branch_count}\n")
}
PipelineEvent::ParallelBranchStarted { branch, index } => {
format!("── PARALLEL_BRANCH_STARTED ──────────────────\n branch: {branch}\n index: {index}\n")
}
PipelineEvent::ParallelBranchCompleted {
branch,
index,
duration_ms,
success,
} => {
format!("── PARALLEL_BRANCH_COMPLETED ────────────────\n branch: {branch}\n index: {index}\n duration_ms: {duration_ms}\n success: {success}\n")
}
PipelineEvent::ParallelCompleted {
duration_ms,
success_count,
failure_count,
} => {
format!("── PARALLEL_COMPLETED ───────────────────────\n duration_ms: {duration_ms}\n success_count: {success_count}\n failure_count: {failure_count}\n")
}
PipelineEvent::InterviewStarted { question, stage } => {
format!("── INTERVIEW_STARTED ────────────────────────\n stage: {stage}\n question: {question}\n")
}
PipelineEvent::InterviewCompleted {
question,
answer,
duration_ms,
} => {
format!("── INTERVIEW_COMPLETED ──────────────────────\n question: {question}\n answer: {answer}\n duration_ms: {duration_ms}\n")
}
PipelineEvent::InterviewTimeout {
question,
stage,
duration_ms,
} => {
format!("── INTERVIEW_TIMEOUT ────────────────────────\n question: {question}\n stage: {stage}\n duration_ms: {duration_ms}\n")
}
PipelineEvent::CheckpointSaved { node_id } => {
format!(
"── CHECKPOINT_SAVED ─────────────────────────\n node_id: {node_id}\n"
)
}
}
}

View file

@ -0,0 +1,164 @@
use std::path::PathBuf;
use std::sync::Arc;
use anyhow::bail;
use chrono::Local;
use crate::checkpoint::Checkpoint;
use crate::engine::{PipelineEngine, 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::validation::Severity;
use super::backend::AgentBackend;
use super::{format_event_detail, format_event_summary, print_diagnostics, read_dot_file, RunArgs};
/// Execute a full pipeline run.
///
/// # Errors
///
/// Returns an error if the pipeline cannot be read, parsed, validated, or executed.
#[allow(clippy::too_many_lines)]
pub async fn run_command(args: RunArgs) -> anyhow::Result<()> {
// 1. Parse and validate pipeline
let source = read_dot_file(&args.pipeline)?;
let (graph, diagnostics) = PipelineBuilder::new().prepare(&source)?;
println!(
"Parsed pipeline: {} ({} nodes, {} edges)",
graph.name,
graph.nodes.len(),
graph.edges.len(),
);
let goal = graph.goal();
if !goal.is_empty() {
println!("Goal: {goal}");
}
print_diagnostics(&diagnostics);
if diagnostics.iter().any(|d| d.severity == Severity::Error) {
bail!("Validation failed");
}
// 2. Create logs directory
let logs_dir = args.logs_dir.unwrap_or_else(|| {
PathBuf::from(format!(
"attractor-run-{}",
Local::now().format("%Y%m%d-%H%M%S")
))
});
tokio::fs::create_dir_all(&logs_dir).await?;
// 3. Build event emitter
let mut emitter = EventEmitter::new();
if args.verbose >= 2 {
emitter.on_event(|event| {
eprint!("{}", format_event_detail(event));
});
} else if args.verbose >= 1 {
emitter.on_event(|event| {
eprintln!("{}", format_event_summary(event));
});
}
// 4. Build interviewer
let interviewer: Arc<dyn Interviewer> = if args.auto_approve {
Arc::new(AutoApproveInterviewer)
} else {
Arc::new(ConsoleInterviewer)
};
// 5. Resolve backend, model, and provider
let dry_run_mode = if args.dry_run {
true
} else {
match unified_llm::client::Client::from_env().await {
Ok(c) if c.provider_names().is_empty() => {
eprintln!("Warning: No LLM providers configured. Running in dry-run mode.");
true
}
Ok(_) => false,
Err(e) => {
eprintln!("Warning: Failed to initialize LLM client: {e}. Running in dry-run mode.");
true
}
}
};
let provider = args.provider.or_else(|| {
graph
.attrs
.get("default_provider")
.and_then(|v| v.as_str())
.map(String::from)
});
let model = args
.model
.or_else(|| {
graph
.attrs
.get("default_model")
.and_then(|v| v.as_str())
.map(String::from)
})
.unwrap_or_else(|| match provider.as_deref() {
Some("openai") => "gpt-5.2".to_string(),
Some("gemini") => "gemini-3-pro-preview".to_string(),
_ => "claude-sonnet-4-5".to_string(),
});
// 6. Build engine
let registry = default_registry(interviewer.clone(), || {
if dry_run_mode {
None
} else {
Some(Box::new(AgentBackend::new(
model.clone(),
provider.clone(),
)))
}
});
let engine = PipelineEngine::with_interviewer(registry, emitter, interviewer);
// 7. Execute
let config = RunConfig {
logs_root: logs_dir.clone(),
cancel_token: None,
};
let outcome = if let Some(ref checkpoint_path) = args.resume {
let checkpoint = Checkpoint::load(checkpoint_path)?;
engine
.run_from_checkpoint(&graph, &config, &checkpoint)
.await?
} else {
engine.run(&graph, &config).await?
};
// 8. Print result
println!("\n=== Pipeline Result ===");
println!("Status: {}", outcome.status.to_string().to_uppercase());
if let Some(notes) = &outcome.notes {
println!("Notes: {notes}");
}
if let Some(failure) = &outcome.failure_reason {
println!("Failure: {failure}");
}
println!("Logs: {}", logs_dir.display());
// 9. Exit code
match outcome.status {
StageStatus::Success | StageStatus::PartialSuccess => Ok(()),
_ => {
std::process::exit(1);
}
}
}

View file

@ -0,0 +1,32 @@
use anyhow::bail;
use crate::pipeline::PipelineBuilder;
use crate::validation::Severity;
use super::{print_diagnostics, read_dot_file, ValidateArgs};
/// Parse and validate a pipeline 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) -> anyhow::Result<()> {
let source = read_dot_file(&args.pipeline)?;
let (graph, diagnostics) = PipelineBuilder::new().prepare(&source)?;
println!(
"Parsed pipeline: {} ({} nodes, {} edges)",
graph.name,
graph.nodes.len(),
graph.edges.len(),
);
print_diagnostics(&diagnostics);
if diagnostics.iter().any(|d| d.severity == Severity::Error) {
bail!("Validation failed");
}
println!("Validation: OK");
Ok(())
}

View file

@ -11,12 +11,14 @@ pub mod wait_human;
use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use async_trait::async_trait;
use crate::context::Context;
use crate::error::AttractorError;
use crate::graph::{shape_to_handler_type, Graph, Node};
use crate::interviewer::Interviewer;
use crate::outcome::Outcome;
/// The handler interface for node execution.
@ -29,6 +31,12 @@ pub trait Handler: Send + Sync {
graph: &Graph,
logs_root: &Path,
) -> Result<Outcome, AttractorError>;
/// Determines whether an error should be retried.
/// Default implementation retries transient errors only.
fn should_retry(&self, err: &AttractorError) -> bool {
err.is_retryable()
}
}
/// Maps handler type strings to handler implementations.
@ -74,6 +82,40 @@ impl HandlerRegistry {
}
}
/// Build a [`HandlerRegistry`] with all built-in handler types registered.
///
/// The `make_backend` closure is called once per handler that needs a backend
/// (`CodergenHandler` default, explicit `"codergen"`, and `"parallel.fan_in"`).
#[must_use]
pub fn default_registry(
interviewer: Arc<dyn Interviewer>,
make_backend: impl Fn() -> Option<Box<dyn codergen::CodergenBackend>>,
) -> HandlerRegistry {
let mut registry =
HandlerRegistry::new(Box::new(codergen::CodergenHandler::new(make_backend())));
registry.register("start", Box::new(start::StartHandler));
registry.register("exit", Box::new(exit::ExitHandler));
registry.register(
"codergen",
Box::new(codergen::CodergenHandler::new(make_backend())),
);
registry.register("conditional", Box::new(conditional::ConditionalHandler));
registry.register(
"wait.human",
Box::new(wait_human::WaitHumanHandler::new(interviewer)),
);
registry.register("tool", Box::new(tool::ToolHandler));
registry.register(
"parallel.fan_in",
Box::new(fan_in::FanInHandler::new(make_backend())),
);
registry.register(
"stack.manager_loop",
Box::new(manager_loop::ManagerLoopHandler::new(None)),
);
registry
}
#[cfg(test)]
mod tests {
use super::*;
@ -150,6 +192,41 @@ mod tests {
let _ = handler;
}
#[test]
fn default_should_retry_uses_is_retryable() {
let handler = TestHandler {
_name: "test".to_string(),
};
assert!(handler.should_retry(&AttractorError::Handler("timeout".to_string())));
assert!(!handler.should_retry(&AttractorError::Parse("bad".to_string())));
}
struct NeverRetryHandler;
#[async_trait]
impl Handler for NeverRetryHandler {
async fn execute(
&self,
_node: &Node,
_context: &Context,
_graph: &Graph,
_logs_root: &Path,
) -> Result<Outcome, AttractorError> {
Ok(Outcome::success())
}
fn should_retry(&self, _err: &AttractorError) -> bool {
false
}
}
#[test]
fn custom_should_retry_override() {
let handler = NeverRetryHandler;
assert!(!handler.should_retry(&AttractorError::Handler("timeout".to_string())));
assert!(!handler.should_retry(&AttractorError::Io("connection reset".to_string())));
}
#[test]
fn register_replaces_existing() {
let mut registry = HandlerRegistry::new(Box::new(TestHandler {

View file

@ -1,5 +1,6 @@
pub mod artifact;
pub mod checkpoint;
pub mod cli;
pub mod condition;
pub mod context;
pub mod engine;

View file

@ -0,0 +1,18 @@
use clap::Parser;
#[tokio::main]
async fn main() {
dotenvy::dotenv().ok();
let cli = attractor::cli::Cli::parse();
let result = match cli.command {
attractor::cli::Command::Run(args) => attractor::cli::run::run_command(args).await,
attractor::cli::Command::Validate(args) => attractor::cli::validate::validate_command(&args),
};
if let Err(e) = result {
eprintln!("Error: {e:#}");
std::process::exit(1);
}
}