fabro/lib/components/fabro-workflow
Bryan Helmkamp 8d14b54994
Bound for_each fan-out memory
Addresses a Copilot review comment on #653.

The source array is runtime data, usually produced by a model, so its
length is not something a workflow author reviewed. Two changes, so an
over-long array degrades into a clear error rather than memory pressure.

Cap the item count at 1000. Above that the stage fails deterministically
before `parallel.started`, alongside the other for_each contract
violations, and the message says how to reduce the array.

Fork the parent context inside the branch task, after it acquires a
`max_parallel` slot, instead of at dispatch time. Live context copies now
track `max_parallel` rather than item count. Only the branch's own
preamble entry is moved into the task, so the shared stash is not cloned
per branch either.

The reviewer also suggested replacing spawn-all with `max_parallel`
workers pulling from a queue. Not done here: with the fork deferred, a
pending task holds little beyond its item, and reshaping the dispatch
loop would change cancellation and scope-reservation ordering, which
deserves its own review.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-28 20:16:20 -04:00
..
src Bound for_each fan-out memory 2026-07-28 20:16:20 -04:00
tests Merge remote-tracking branch 'origin/main' into feat/for-each-item-injection 2026-07-28 20:03:25 -04:00
Cargo.toml Merge remote-tracking branch 'origin/main' into refactor/remove-env-interpolation 2026-07-28 17:44:03 -04:00
README.md fix(workflow): make publish failures terminal 2026-07-27 11:25:18 -04:00

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 goal describing 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, and fidelity attributes that control routing.
  • Handler -- An async trait implementation that executes a node and returns an Outcome. Built-in handlers include StartHandler, ExitHandler, AgentHandler, PromptHandler, ConditionalHandler, HumanHandler, ParallelHandler, FanInHandler, CommandHandler, and SubWorkflowHandler.
  • 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, and RecordingInterviewer.
  • 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)