fabro/lib/crates/fabro-workflow
fabro-sh-0530[bot] 475b4ab650
Replace stdin JSONL control pipe with WebSocket worker control bus (#440)
## Summary

Workers no longer receive control messages over stdin JSONL. A new
`WorkerControlBus` abstraction (backed by `LocalWorkerControlBus` for
local/single-node deployments) publishes `WorkerControlEnvelope`
messages server-side; a worker-initiated WebSocket at `GET
/runs/{id}/worker/control-stream` delivers them with ordered, replayable
delivery frames. The bus API is designed so a Redis Streams backend can
slot in later without touching API handlers or worker message handling.

### Plan Summary

- **Task 1 – Bus contract:** `WorkerControlBus` trait,
`WorkerControlDelivery`, `WorkerControlCursor` (`Start` / `After(id)`),
bus errors.
- **Task 2 – Local backend:** `LocalWorkerControlBus` — in-memory
per-run stream, replay from `Start`, reconnect via `After(id)`, 1
024-message trim bound, cleanup on terminal runs.
- **Task 3 – Server state:** `Arc<dyn WorkerControlBus>` added to
`AppState`; `LocalWorkerControlBus` constructed at startup.
- **Task 4 – Protocol extension:** `WorkerControlMessage::RunPause` /
`RunUnpause`, `WorkerControlDeliveryFrame`, WebSocket liveness constants
(`WORKER_CONTROL_WS_PING_INTERVAL = 15s`,
`WORKER_CONTROL_WS_LIVENESS_TIMEOUT = 45s`), close-reason strings.
- **Task 5 – Worker message handler:** `apply_worker_control_message`
split out; pause/unpause routing; delivery-id dedupe
(`AppliedWorkerControlDeliveryIds`, capacity 2 048).
- **Task 6 – Worker WebSocket client:** `spawn_worker_control_manager` —
HTTP→ws/wss and Unix-socket connection, backoff 100ms→5s,
first-connection gate before `operations::start/resume`, ping/pong
watchdog, fatal loss wired back to `execute`.
- **Task 7 – Server route:** `GET /runs/{id}/worker/control-stream`,
worker-only auth via new `RequireWorkerRunScoped` extractor,
`Start`/`After` cursor dispatch, 410 on invalid cursor, server-side
ping/pong.
- **Task 8 – Stdin removal:** `RunAnswerTransport::Subprocess` renamed
to `Worker { run_id, bus }`; `pump_worker_control_jsonl` deleted; worker
launched with `stdin(Stdio::null())`; pause/unpause transport methods
added.
- **Tasks 9–10 – E2E & verification:** reconnect, invalid-cursor,
cancel-over-WebSocket, and human-interview regression tests; no Redis
dependency added.

### Key design decisions

**`RunAnswerTransport::Subprocess` → `Worker { run_id, bus }`** — all
existing transport methods (`submit`, `cancel_run`, `steer`,
`interrupt`, `pair_*`) now call `bus.publish(run_id, envelope)` instead
of writing to a channel that fed stdin. The match arms are symmetric, so
the diff is mechanical but large.

**First-connection gate** — `execute()` calls
`control_manager.wait_for_first_connection().await?` before
`operations::start` or `operations::resume`. Temporary failures spin
with backoff; a fatal invalid-cursor or request-build failure propagates
as an error before the workflow starts.

**Fatal vs. reconnectable** — HTTP 410 or a WebSocket close with reason
`"invalid_cursor"` is fatal (infrastructure failure, not user
cancellation). Any other close/error triggers the reconnect loop while
the run is non-terminal.

**`AutomationStore::load` made synchronous** — startup load now uses
`std::fs` under a `clippy::disallowed_methods` exception; async
`tokio::fs` is no longer needed for the one-shot directory scan. Invalid
automation files now fail loudly instead of being silently skipped.

**`canRetry` extended to succeeded runs** — `status.kind ===
"succeeded"` is now retryable (non-archived). Tests and API docs updated
to match.

**Default model bumps** — OpenAI default: `gpt-5.4` → `gpt-5.5`; Gemini
default: `gemini-3.1-pro-preview` → `gemini-3.5-flash`.


### Fabro Details

<details>
<summary>Ran 9 stages in 129m 19s for $58.27</summary>

| Stage | Duration | Cost | Retries |
|---|---|---|---|
| start | 0s | – | 0 |
| toolchain | 1s | – | 0 |
| preflight_compile | 2m 10s | – | 0 |
| preflight_lint | 2m 23s | – | 0 |
| implement | 73m 48s | $41.53 | 0 |
| simplify_opus | 22m 55s | $11.75 | 0 |
| simplify_gpt | 7m 19s | $2.74 | 0 |
| verify | 8m 51s | – | 0 |
| fixup | 10m 59s | $2.24 | 0 |
| **Total** | **129m 19s** | **$58.27** | **0** |

</details>

<details>
<summary>Ran <code>ImplementPlan.fabro</code> (11 nodes and 14
edges)</summary>

```dot
digraph ImplementPlan {
    graph [
        goal="Implement and simplify",
        model_stylesheet="
            * { model: claude-opus-4-7; }
        "
    ]
    rankdir=LR

    start [shape=Mdiamond, label="Start"]
    exit  [shape=Msquare, label="Exit"]

    toolchain         [label="Toolchain", shape=parallelogram, script="command -v cargo >/dev/null || { curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y && sudo ln -sf $HOME/.cargo/bin/* /usr/local/bin/; }; cargo --version 2>&1", max_retries=0]
    preflight_compile [label="Preflight Compile", shape=parallelogram, script="cargo check -q --workspace 2>&1", max_retries=0]
    preflight_lint    [label="Preflight Lint", shape=parallelogram, script="cargo +nightly-2026-04-14 clippy -q --workspace --all-targets -- -D warnings 2>&1", max_retries=0]
    fix_lints         [label="Fix Lints", prompt="The preflight lint step failed. Read the build output from context and fix all clippy lint warnings.", max_visits=3]
    implement         [label="Implement", prompt="Read the plan file referenced in the goal and implement every step. Make all the code changes described in the plan. Use red/green TDD.", model="gpt-55", reasoning_effort="xhigh"]
    simplify_opus     [label="Simplify (Opus)", prompt="@prompts/simplify.md"]
    simplify_gpt      [label="Simplify (GPT-55)", prompt="@prompts/simplify.md", model="gpt-55"]
    verify            [label="Verify", shape=parallelogram, script="git fetch origin main 2>&1 && git merge --no-edit --no-stat origin/main 2>&1 && cargo +nightly-2026-04-14 fmt --all 2>&1 && cargo dev docs refresh 2>&1 && cargo +nightly-2026-04-14 fmt --check --all 2>&1 && { command -v rg >/dev/null 2>&1 || { echo 'rg is required for verify'; exit 127; }; } && ! rg -n 'AuthMode::Disabled|RunAuthMethod|RunSubjectProvenance|\bActorRef\b|\bActorKind\b|AuthenticatedSubject|AuthenticatedService|AuthorizeRunScoped|AuthorizeRunBlob|AuthorizeStageArtifact|AuthorizeCommandLog|auth_method\s*==\s*\"disabled\"' lib/crates apps lib/packages docs/public/api-reference/fabro-api.yaml 2>&1 && cargo +nightly-2026-04-14 clippy --workspace --all-targets -- -D warnings 2>&1 && cargo nextest run --workspace --status-level slow --profile ci 2>&1 && cargo dev docs check 2>&1 && bun install --frozen-lockfile 2>&1 && (cd apps/fabro-web && bun run typecheck) 2>&1 && (cd apps/fabro-web && bun run test) 2>&1 && (cd lib/packages/fabro-api-client && bun run typecheck) 2>&1 && cargo dev build -- -p fabro-cli --release 2>&1", goal_gate=true, retry_target="fixup"]
    fixup             [label="Fixup", prompt="The verify step failed. Read the build output from context and fix all format, clippy, Rust test, docs, TypeScript typecheck/test, and build failures.", max_visits=3]

    start -> toolchain
    toolchain -> preflight_compile [condition="outcome=succeeded"]
    toolchain -> exit
    preflight_compile -> preflight_lint [condition="outcome=succeeded"]
    preflight_compile -> exit
    preflight_lint -> implement [condition="outcome=succeeded"]
    preflight_lint -> fix_lints
    fix_lints -> preflight_lint
    implement -> simplify_opus -> simplify_gpt -> verify
    verify -> exit  [condition="outcome=succeeded"]
    verify -> fixup
    fixup -> verify
}

```

</details>

⚒️ Generated with [Fabro](https://fabro.sh)

---------

Co-authored-by: Fabro <noreply@fabro.sh>
Co-authored-by: Bryan Helmkamp <bryan@brynary.com>
2026-05-27 20:24:25 -04:00
..
src Replace stdin JSONL control pipe with WebSocket worker control bus (#440) 2026-05-27 20:24:25 -04:00
tests Model run sandbox lifecycle explicitly (#431) 2026-05-27 12:48:56 -04:00
Cargo.toml refactor: remove devcontainer support (#433) 2026-05-27 12:48:40 -04:00
README.md refactor(workflow): remove retro stage (#230) 2026-05-09 10:18:20 -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. Configurable join policies: wait_all (default), first_success.

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)