flowing

name	flowing
description	DAG workflow runner that encodes control flow in code, not prose. Use when a procedure has 3+ steps with branching, retries, or validation that must be enforced — gates as `when=`, edge contracts as `validate=`, predicate loops as `retry_until=`. The runner owns the graph; the LLM provides leaves. Also covers parallel execution, checkpoint resume, detached side-effects.
metadata	{"version":"1.4.0"}

NOT SUPERSEDED BY DYNAMIC WORKFLOWS — read first

Claude Code's dynamic workflows orchestrate subagents (separate contexts, fan-out to 16-concurrent / 1000-agent). This skill is a different primitive: single-context control flow over YOUR OWN tool calls, with durable side-effects and checkpoint resume. The workflows runtime explicitly cannot touch the filesystem or shell directly — its agents do the work and the script only coordinates them. Flowing is the inverse: the script does the work.

Use flowing for an in-context pipeline (3+ steps, branches, retries, validation, detached side-effects). Use a workflow when you need many subagents. They compose; they do not compete. Do not abandon flowing for a workflow — you would lose the durable side-effects and the cross-session checkpoint that hub-spoke depends on.

Flowing — Control Flow in Code, Not Prose

When a procedure needs 3+ steps with branches, retries, or contracts, encode it as a DAG of Python tasks instead of prose imperatives. Prose like "first X, then Y, then if Z retry 3×" is read and generated past. A @task graph is structural: a step physically cannot run until its inputs are bound, and gates that fire on bad inputs can't be skipped.

The runner owns control flow — branching, retrying, validating, propagating failures, parallelizing. You provide judgment at the leaves. Runner: scripts/flowing.py.

Quick Start

from flowing import task, Flow

@task
def fetch_data():
    return {"items": [1, 2, 3]}

@task(depends_on=[fetch_data])
def process(fetch_data):          # param name must match the dep's name
    return sum(fetch_data["items"])

@task(depends_on=[process])
def store(process):
    print(f"Result: {process}")

Flow(store).run()                 # topo-sorts, runs each layer, parallel within a layer

Each task receives its dependencies as kwargs named after them. Independent tasks in the same layer run in parallel.

Control-Flow Primitives

Encode branches and contracts as graph structure, not if statements inside task bodies.

when= — conditional gate

Run the task only if the predicate (over gathered dep values) is truthy. Falsy → SKIPPED, and the skip propagates to dependents.

@task(depends_on=[fetch], when=lambda fetch: fetch["needs_processing"])
def process(fetch):
    return transform(fetch["payload"])

validate= — edge contract

Check gathered dep values before the body runs. Raise → FAILED with no retry (bad inputs don't fix themselves). Pass → proceed.

def must_have_items(fetch):
    if not fetch.get("items"):
        raise ValueError("fetch returned empty payload")

@task(depends_on=[fetch], validate=must_have_items)
def process(fetch):
    return sum(fetch["items"])

retry_until= — predicate-driven loop

Run the body, then call retry_until(value). True → done. False → retry, consuming the retry= budget. Use for self-correcting LLM steps: generate, check, regenerate.

@task(retry=4, retry_until=lambda r: r["valid"])
def generate_until_valid():
    candidate = llm_call(...)
    return {"valid": passes_schema(candidate), "candidate": candidate}

Distinct from retry= alone, which only retries on a raised exception.

Other capabilities

• Parallel execution — independent tasks in a layer run on a thread pool (max_workers=).
• detached=True — side-effect tasks (memory writes, notifications) that run after the main DAG and never block it on failure.
• In-process resume — flow.run() → fix → flow.resume() re-runs from the failure point, keeping succeeded tasks cached in memory (same process only). flow.override(task, value) injects a corrected result.
• Durable journal (journal_path=) — opt-in content-addressed replay that survives container death. Flow(term, journal_path="/path/run.jsonl").run() appends each succeeded task's result to an append-only JSONL keyed by a step_key = SHA-256 over the task's bytecode + its when/validate/retry_until bodies + its dependencies' keys (chained, so an upstream change propagates downstream). A later run() — even in a fresh container — replays the unchanged prefix from the journal and only executes tasks whose key is absent; editing a task body busts its key and re-runs it and its dependents, while cosmetic knobs (retry=, timeout_s=, name) do not. This is the cross-session checkpoint hub-spoke work relies on. Caveat: results are pickled, so non-picklable return values simply re-run; closure-captured values are not part of the key (only the task body's own code is).
• timeout_s=, retry= with exponential backoff, fail_fast=.

Read references/reference.md before using anything beyond the quick start and the three primitives above — it covers every @task parameter, the Flow methods, resume/override, detached auto-discovery, and the validate=/when= signature-matching gotcha.

When to use

• A procedure has branches that matter → when= makes them structural.
• Steps have input contracts → validate= makes them enforceable.
• An LLM step needs to converge → retry_until= puts the check in the loop.
• 3+ independent operations that can parallelize.
• Multi-step pipelines where late failures shouldn't waste early work.
• Side-effects that shouldn't block the critical path → detached=True.

When NOT to use

• A single sequential operation — just call the function.
• The next step needs reasoning about the prior result that can't be a predicate — use a think loop.
• Async or distributed workflows — this is single-container, thread-pool based.

Authoring discipline

If you find yourself writing prose like "first call X, validate Y, then if Z retry up to 3 times" — that is a flowing graph. Refactor before shipping. Prose imperatives don't enforce; @task graphs do.