fabro/lib/components/fabro-workflow
Bryan Helmkamp 6659ae768a
feat(events): make inference in-flight state observable
During a long LLM turn the durable event stream was silent: between
`agent.tool.completed` and the next `agent.message` nothing was emitted,
so "the model is generating" and "the worker is wedged" were
indistinguishable from the run store, SSE, or the UI.

The signal already existed. `AssistantTextStart` fired at exactly the
right point — after `build_request()`, after compaction, immediately
before the stream opens — then was classified as streaming noise and
thrown away. This promotes it rather than inventing a new one.

Two events, each asserting only what is provable when it is emitted:

- `agent.llm.started` carries the *requested* provider/model. No usage,
  no cost, no context window: none of it exists yet, and failover can
  re-target, so `agent.message` stays authoritative for what answered.
- `agent.llm.first_output` is edge-triggered on the first output of an
  attempt and names what arrived. `ToolCall` is required, not optional:
  a turn that opens with a tool call produces no text or reasoning
  delta, so a latch keyed on those two would stay silent for exactly
  the tool-heavy rounds where liveness matters most.

`agent.llm.retry` now also fires on the one previously invisible
mid-turn path — a stream that ends without a finish event, which
replays the turn and discards its output with nothing to show for it.
Its `attempt` field was already fed by two independent counters, so an
optional `phase` (open | consume) names which loop it counts.

`StageProjection.inference` projects the open bracket. `Some` means
"the event log contains an unclosed inference bracket", not "the model
is computing now" — a SIGKILLed worker leaves it open, which is the
truthful statement of what we know, and `watchdog.timeout` remains the
authority on actually-stuck.

The close is the subtle part. Terminal cancel and wall-clock timeout
tear the session down through `discard_session` without emitting a
message, error, or interrupt, so a session-lifecycle backstop is
required. It has to be `agent.session.ended`, not
`agent.session.deactivated`: deactivation is emitted by `lease.release()`
*before* the forwarder drains queued agent events, so a queued
`agent.llm.started` can arrive after it and re-open the bracket. But
`agent.session.ended` carries no stage identity, so the close takes
ordering from the event and identity from the projection, scanning for
brackets the ending session opened. A normal stage lookup there finds
no target and silently no-ops.

Presentation states what the log proves and nothing more: no progress
bar or ETA (no completion estimate exists), "reasoning" only when the
provider sent reasoning output, elapsed counted since the request
opened, and no live animation once the run is terminal.

Scope is session-backed agent stages. One-shot completions call
`client.complete` directly and never build a session; covering them
means moving the emit point into `fabro-llm`, filed as a follow-up.

`agent.output.start` was never persisted — it existed in a name map,
an `unreachable!` arm, and docs — so the rename carries no migration
risk. Corrects `events.md`, which documented it as a real emitted
event, and the v2 proposal, which mapped it to `message.part.started`
despite it firing before the request opens.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-24 21:47:51 -04:00
..
src feat(events): make inference in-flight state observable 2026-07-24 21:47:51 -04:00
tests Merge pull request #607 from fabro-sh/feat/shared-checkout-parallel 2026-07-24 09:33:03 -04:00
Cargo.toml refactor: organize crates into three layers 2026-07-23 17:59:34 -04:00
README.md Merge branch 'main' into feat/shared-checkout-parallel 2026-07-24 06:29:57 -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 -> 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)