Linear pipeline of deterministic + LLM steps

State: open/active · Genre: analysis

Use case: linear-typed-pipeline RFC mapping: UC2 Source: cross-repo survey of sksizer/* (2026-07-05), companion to the workflow-constellation RFC (PR #682).

The need

The most common workflow shape: a fixed sequence of steps with hand-numbered banners (‘Step 1/4’, ‘[1/8]’), halt-on-failure semantics, and ad-hoc per-step logging — rebuilt per script, per app, and in three cases as entire bespoke engines (one of them restarted three times). The plumbing (sequencing, status, dry-run threading, placeholder substitution) always dwarfs the domain steps.

Evidence

DarkFactory

A complete hand-rolled workflow DSL (BuiltIn/Agent/Shell/Interactive task types, Workflow dataclass, RunContext templating) with a bespoke sequential dispatch loop producing per-step TaskStep records; canonical consumer is the PRD pipeline worktree→implement→gates→PR.

Where: python/darkfactory/workflow/_core.py · python/darkfactory/runner.py · python/darkfactory/workflow/definitions/prd/default/workflow.py

Pain: The entire linear typed-pipe engine is bespoke Python — and the repo is now rebuilding it a second time in TypeScript (PRD-637).

rust-worker-prototype

Three independent restarts of the same sequential step runner (enum-dispatch v1, StepRegistry+ExecutionState v2, bare-types v3) for a coding-agent pipeline; step IO untyped throughout, verify/commit/PR steps left as comments never reached.

Where: src/runner.rs · src/proto_two/executor.rs · crates/workflow/src/types.rs

Pain: The step-sequencing substrate was rebuilt from scratch three times and never got past linear bash steps — the strongest possible signal the plumbing is the blocker.

harness

ProvisionOrchestrator: a from-scratch step runner with per-step idempotency checks, streaming output batching, cancellation, fail-fast; plus a four-phase boot pipeline whose failure handling is copy-pasted per phase (the repo’s own audit names ‘Monolithic start_instance’, ‘Channel Parameter Sprawl’).

Where: prototypes/zoo/code2/src-tauri/src/vm/provision.rs · prototypes/zoo/code2/src-tauri/src/vm/manager.rs · prototypes/zoo/code2/.planning/phases/15.1-cleanup-state-machine/15.1-RESEARCH.md

Pain: Two full planning phases spent designing and then cleaning up hand-rolled pipeline machinery.

monosync-private

Hand-numbered ‘Step 1/4’ sync pipeline (manifest → diff → apply → commit/push) duplicated wholesale as ‘Step 1/3’ for reverse sync because there is no pipeline abstraction to share.

Where: src/sync_cmd.rs · src/reverse.rs · docs/how-it-works.md

Pain: Docs draw it as a dataflow graph; code encodes it as duplicated straight-line functions with drifting step labels.

trackr

ETL orchestrator sequencing steps with print banners and duplicated —fresh reset logic; a 1000-line derived-table build whose ~12 stages exist only as print() boundaries, with a hand-tallied parse_stats dict as its only report.

Where: src-python/main.py · src-python/trackr/modules/food/data_import/derived.py

Pain: No per-stage status, timing, or restartability; a mid-build failure means rerunning everything.

world-quiz

~30 copy-pasted puppeteer flow scripts, each re-implementing browser launch, error collectors, sleep-based step sequencing, and ad-hoc JSON reporting.

Where: scripts/smoke.mjs · scripts/features-flow.mjs · scripts/phase1-smoke.mjs

Pain: 30 files of duplicated skeleton around a handful of distinct flows.

Flowline fit

@flow/pipe (typed async composition) + @flow/workflow (linear typed tasks) are the direct target; structured per-step reporting and dry-run come along as the cross-cutting features every example hand-rolled.

Proposed API

Builds on: 01–02 Introduces: pipe(), workflow().add().build(), run()

Two tiers. pipe is the five-line script floor; workflow is what today’s hand-numbered "[1/8] fetch..." banner scripts become:

import { pipe } from "flowline/pipe";
import { agent } from "flowline/llm";
import { workflow, run } from "flowline/workflow";

// Script tier: fn -> LLM -> fn, no engine.
const toEntity = pipe(
  (raw: string) => raw.trim(),
  (text) => agent({ prompt: `Extract:\n${text}`, output: Entity }),
  (r) => writeEntity(r.value!),
);

// Named-step tier: the monosync/trackr shape.
const sync = workflow("sync", "forward sync")
  .seed(Config.of(cfg))
  .add(fetchRemote)     // typed reads/writes; compile error on a missing input
  .add(diffStates)
  .add(applyChanges)
  .add(writeReport)
  .build();

const result = await run(sync);
// result.steps: [{ name, ok, durationMs, failureReason? }] — step numbering,
// halt-on-failure, and per-step timing come from the runner, not the script.

---

← Back to Notes