Every run executes on Petri, so the in-process legacy executor goes: `fabro-core` and, in `fabro-workflow`, the handlers, lifecycle, pipeline execution, routing, retry, conditions, node handlers, steering, agent memory, artifacts, checkpoints, command log, and the `start`, `resume`, `retry`, `fork`, `rewind` and `timeline` operations. The two are deleted together because the engine half of `fabro-workflow` was the only user of `fabro-core` and `fabro-core` the only runtime of that half; neither compiles without the other. Kept in `fabro-workflow`, narrowed: the parse/transform/validate/persist pipeline and `create`, `archive`, `validate` (workflow definitions still come from DOT and settings); the run tools (`run_tools`, moved from `handler/llm/fabro_tools.rs`) for Ask Fabro, `fabro exec` and Petri's host tools; the pull request pipeline (`pull_request`, moved from `pipeline/`, for the step 0 port); Run Files' diff helpers in `sandbox_git`; `git_identity`, `usage_rollup`, `run_status`, `run_materialization`, `web_search` and `workflow_bundle`. Server: `RegistryFactoryOverride` becomes `execute_in_process`; `RunAnswerTransport::InProcess` carries only the interviewer; the interrupt endpoint answers 501 `interrupt_unsupported` and every pair endpoint 501 `pair_unsupported` (status lists none); rewind, fork, retry and timeline handlers and routes are removed; the command log is served from the stage output blob; usage rollups accumulate from the settled projection after an in-process run as after a worker exit. Ported while here: - `materialize_admitted_run` materializes the goal and drops a disabled pull request block, as the legacy materializer did. - A run whose admitted graph has an agent or prompt node is refused at create when no LLM provider is ready (`fabro.model.no_ready_provider`); a workflow of commands and gates needs no model and is admitted. - The projection's question type falls back on the options, as the interview adapter does, so a gate with edge-label options answers as multiple choice. Tests: the server scenarios (lifecycle, run completion, SSE, helpers) run in process on Petri and assert Petri's stage labels and stream names; the reconcile tests assert Petri's relaunch semantics; legacy unit tests of the deleted executor are removed; three server unit tests the removal took with it are restored; the pair fixtures go with the pair feature. Petri test fixtures no longer name `[workflow] engine`. Still red after this commit, all legacy consumers the next steps delete or port: fabro-store's Slate/reducer fixtures and fabro-types legacy JSON tests (step 4); server unit tests over legacy run events (retry endpoints, list_run_events, artifacts, per-event pause/unpause, run history activation, legacy sandbox fixtures) (steps 3-4); CLI tests that parse legacy event envelopes, the legacy `events`/`attach`/`diff`/ `dump`/`inspect` snapshots, `run rewind`/`run fork`, the ACP and git-identity workflow tests, and the runner tests that drive the legacy worker by hand (steps 3-4); the web app's Petri fixtures still carry `engine` (regenerate with `FABRO_CAPTURE_PETRI_FIXTURES` in step 4). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> |
||
|---|---|---|
| .. | ||
| src | ||
| tests | ||
| Cargo.toml | ||
| README.md | ||
fabro-workflow
A DOT-based pipeline runner for multi-stage AI workflows. Define workflows as Graphviz digraph files and execute them with pluggable handlers, conditional routing, human-in-the-loop gates, parallel branching, retry policies, and checkpoint-based recovery.
Key Concepts
- Graph -- A directed graph parsed from DOT syntax containing nodes, edges, and attributes. The graph carries a
goaldescribing the pipeline's purpose. - Node -- A workflow step. Graphviz shapes map to handler types (e.g.,
Mdiamond= start,Msquare= exit,box= agent,tab= prompt,diamond= conditional,hexagon= human gate,component= parallel). - Edge -- A connection between nodes with optional
condition,label,weight, andfidelityattributes that control routing. - Handler -- An async trait implementation that executes a node and returns an
Outcome. Built-in handlers includeStartHandler,ExitHandler,AgentHandler,PromptHandler,ConditionalHandler,HumanHandler,ParallelHandler,FanInHandler,CommandHandler, andSubWorkflowHandler. - Outcome -- The result of executing a handler, carrying a
StageOutcome(Success, Fail, PartialSuccess, Retry, Skipped), optional routing hints (preferred_label,suggested_next_ids), and context updates. - Context -- A thread-safe key-value store shared across pipeline stages, supporting snapshots and isolated cloning for parallel branches.
- Interviewer -- A trait for human-in-the-loop interactions. Implementations include
AutoApproveInterviewer,QueueInterviewer,CallbackInterviewer,ConsoleInterviewer, andRecordingInterviewer. - Checkpoint -- A serializable snapshot of execution state (completed nodes, context values) for crash recovery and resume.
Pipeline Definition
Pipelines are defined using Graphviz DOT syntax:
digraph MyPipeline {
graph [goal="Implement and validate a feature"]
rankdir=LR
node [shape=box, timeout="900s"]
start [shape=Mdiamond, label="Start"]
exit [shape=Msquare, label="Exit"]
plan [label="Plan", prompt="Plan the implementation"]
implement [label="Implement", prompt="Implement the plan"]
validate [label="Validate", prompt="Run tests"]
gate [shape=diamond, label="Tests passing?"]
start -> plan -> implement -> validate -> gate
gate -> exit [label="Yes", condition="outcome=succeeded"]
gate -> implement [label="No", condition="outcome!=succeeded"]
}
Usage
Parsing and Validating a Pipeline
use fabro_workflow::operations::{create, CreateOptions};
let dot_source = r#"digraph Simple {
graph [goal="Run tests"]
start [shape=Mdiamond]
exit [shape=Msquare]
work [shape=box, prompt="Run the test suite"]
start -> work -> exit
}"#;
let validated = create(dot_source, CreateOptions::default())
.expect("pipeline should parse");
validated.raise_on_errors().expect("pipeline should validate");
let (graph, _, _) = validated.into_parts();
assert_eq!(graph.name, "Simple");
assert_eq!(graph.goal(), "Run tests");
operations::create parses the DOT source, applies built-in transforms (variable expansion, stylesheet application, preamble injection), and returns diagnostics through Validated.
Running a Pipeline
use fabro_workflow::operations::start;
use fabro_workflow::pipeline;
// Use `operations::start(...)` for the full
// initialize -> execute -> conclude -> publish -> finalize flow.
// Use `pipeline::initialize(...)` + `pipeline::execute(...)` when you need partial lifecycle control.
Custom Handlers
Implement the Handler trait to add custom node behavior:
use arc_workflows::handler::Handler;
use arc_workflows::context::Context;
use arc_workflows::graph::{Graph, Node};
use arc_workflows::outcome::Outcome;
use arc_workflows::error::ArcError;
use async_trait::async_trait;
use std::path::Path;
struct MyHandler;
#[async_trait]
impl Handler for MyHandler {
async fn execute(
&self,
node: &Node,
context: &Context,
graph: &Graph,
run_dir: &Path,
) -> Result<Outcome, ArcError> {
// Custom logic here
Ok(Outcome::success())
}
}
Model Stylesheets
CSS-like stylesheets control LLM model assignment with specificity-based cascading:
digraph Styled {
graph [
goal="Build feature",
model_stylesheet="
* { model: claude-sonnet-4-5;}
.code { model: claude-opus-4-6; }
#critical_review { model: gpt-5.2;}
"
]
// ...
}
Selectors by specificity: * (universal, 0) < shape (1) < .class (2) < #id (3). Explicit node attributes are never overridden.
Condition Expressions
Edge conditions use a simple expression syntax for routing:
outcome=succeeded
outcome!=failed
outcome=succeeded && context.tests_passed=true
my_flag
Clauses support =, !=, and bare key truthiness checks, joined with &&.
Human-in-the-Loop Gates
Nodes with shape=hexagon or type="human" pause execution for human input. Outgoing edge labels become selectable options, with accelerator key parsing for patterns like [A] Approve and F) Fix.
Parallel Execution
Nodes with shape=component fan out to branches concurrently. Branches receive isolated context forks, share the same sandbox checkout, and always finish before the workflow continues. Use max_parallel to limit concurrency; concurrent workspace writes are user-managed.
Checkpoints and Resume
The engine saves a checkpoint after each node. Resume from a checkpoint with engine.run_from_checkpoint(&graph, &config, &checkpoint).
Architecture
parser (DOT -> AST -> Graph)
-> transform (variable expansion, stylesheet, preamble)
-> validation (14 lint rules)
-> engine (execution loop with retry, edge selection, goal gates)
-> handler (pluggable node executors)
-> interviewer (human-in-the-loop I/O)