A Workflow Engine: Turning the Pipeline Into a Resumable Task Graph

Idempotency comes down to the key

Idempotent sounds abstract; on this pipeline it means one thing: the same input must not spend money a second time.

You implement it by computing a fingerprint per node and using it to ask whether that work's artifacts already exist. All the difficulty lies in what goes into the fingerprint.

Four things must be in it, none optional:

The node id. Different nodes must not collide. Obvious.

The implementation version. You changed the node's prompt template, encoding parameters, algorithm — the input did not change, but the output should. If the version is not in the key, you read old artifacts with new code, undetectably. This is the easiest of the four to omit and the most insidious.

This node's inputs. Model id, prompt, duration, resolution — everything that affects the output.

Every dependency's fingerprint. This is the only mechanism by which an upstream change invalidates downstream work. Hash only your own inputs and, when the script changes upstream, you keep using the old shot files: no error, just a wrong cut.

And what must never be in it: the run id, timestamps, random numbers, absolute paths. Include any of them and every key is new, the cache never hits, and you will assume the cache is broken. "Cache never hits" and "cache serves the wrong take" both come from the key, in opposite directions.

import { createHash } from 'node:crypto'
 
// All four parts are required. Without version, a changed implementation reads stale artifacts;
// without deps, upstream changes never propagate; with runId, the cache never hits.
export function fingerprint(node, depFingerprints) {
  const payload = JSON.stringify({
    id: node.id,
    version: node.version,
    inputs: node.inputs,
    deps: depFingerprints,
  })
  return createHash('sha256').update(payload).digest('hex').slice(0, 16)
}

import hashlib
import json
 
 
# All four parts are required. Without version, a changed implementation reads stale artifacts;
# without deps, upstream changes never propagate; with run_id, the cache never hits.
def fingerprint(node: NodeSpec, dep_fingerprints: list[str]) -> str:
    payload = json.dumps(
        {"id": node.id, "version": node.version, "inputs": node.inputs, "deps": dep_fingerprints},
        sort_keys=True,
        ensure_ascii=False,
    )
    return hashlib.sha256(payload.encode()).hexdigest()[:16]

One boundary in passing: D3's asset deduplication — "the same description does not regenerate" — is this mechanism applied to the image stage; how much the cache actually saves and how to degrade when hit rate is low is D12's topic. Today, just get the key right.

Where artifacts land: the directory name is the fingerprint

With a fingerprint, artifacts should land under it. That is content addressing — the directory name is decided by content, not by which run produced it.

So the disk splits cleanly into two layers:

work/
├── cache/<nodeId>/<fingerprint>/     artifacts; the same input always writes the same directory
│   ├── clips/a276b932372bc7c6/s01.mp4
│   └── voice/182a684769cb79e8/s01.mp3
└── runs/<runId>/
    ├── state.json                    per-node status, elapsed time, spend and fingerprint
    ├── run.json                      the cost ledger
    └── output/                       cut, subtitles and cover taken from compose's cache dir

The cache is global; state is per run. That division is worth remembering, because it maps onto two completely different capabilities: idempotency comes from the cache — re-run under a new run id and you still spend nothing; resumption comes from state — knowing where the last run got to, and which node failed rather than never being reached.

Separating them has another benefit: run directories become very light — one state file, one ledger and a few final artifacts — so dozens of experiments' run directories add up to nothing, while the heavy material lives in the cache deduplicated by content, naturally single-copy per input.

Resumption: a re-run becomes a set difference

With those two sections laid down, resuming holds no mystery. The engine's main loop walks the topological order and asks three questions per node: did anything upstream fail? are my artifacts complete? did I persist the result after running?

export async function runWorkflow(specs, opts) {
  const state = await loadState(opts.statePath, opts.runId)
  const fingerprints = new Map()
  const dirs = new Map()
  let failedAt
 
  for (const node of topoSort(specs)) {
    const fp = fingerprint(node, node.deps.map((d) => fingerprints.get(d) ?? ''))
    fingerprints.set(node.id, fp)
    const outDir = `${opts.cacheDir}/${node.id}/${fp}`
    dirs.set(node.id, outDir)
 
    // Upstream unfinished means incomplete input downstream; running only wastes money again.
    if (failedAt) {
      state.nodes[node.id] = { fingerprint: fp, status: 'blocked', ms: 0, calls: 0, costCny: 0 }
      continue
    }
 
    // Idempotency check: look only at whether the artifacts exist, never at the state file.
    if (await allExist(outDir, node.outputs)) {
      state.nodes[node.id] = { fingerprint: fp, status: 'done', source: 'cache', ms: 0, calls: 0, costCny: 0 }
      await saveState(opts.statePath, state) // persist per step, so a kill loses only the running node
      continue
    }
 
    const started = Date.now()
    const callsBefore = opts.callCount()
    try {
      const { note } = await node.run({ outDir, depDir: (id) => dirs.get(id), record: opts.onCost })
      state.nodes[node.id] = { fingerprint: fp, status: 'done', source: 'executed', ms: Date.now() - started, calls: opts.callCount() - callsBefore, note }
    } catch (err) {
      // Persist failed nodes too: the money they spent is real, and partial artifacts stay.
      state.nodes[node.id] = { fingerprint: fp, status: 'failed', ms: Date.now() - started, calls: opts.callCount() - callsBefore, error: String(err) }
      failedAt = node.id
    }
    await saveState(opts.statePath, state)
  }
  return { state, dirs, failedAt }
}

def run_workflow(specs: list[NodeSpec], opts: EngineOptions) -> WorkflowResult:
    state = load_state(opts.state_path, opts.run_id)
    fingerprints: dict[str, str] = {}
    dirs: dict[str, str] = {}
    failed_at: str | None = None
 
    for node in topo_sort(specs):
        fp = fingerprint(node, [fingerprints.get(d, "") for d in node.deps])
        fingerprints[node.id] = fp
        out_dir = f"{opts.cache_dir}/{node.id}/{fp}"
        dirs[node.id] = out_dir
 
        # Upstream unfinished means incomplete input downstream; running only wastes money.
        if failed_at:
            state["nodes"][node.id] = {"fingerprint": fp, "status": "blocked", "ms": 0, "calls": 0, "costCny": 0}
            continue
 
        # Idempotency check: look only at whether the artifacts exist, never at the state file.
        if all_exist(out_dir, node.outputs):
            state["nodes"][node.id] = {"fingerprint": fp, "status": "done", "source": "cache", "ms": 0, "calls": 0}
            save_state(opts.state_path, state)  # persist per step, so a kill loses only one node
            continue
 
        started, calls_before = time.time(), opts.call_count()
        try:
            note = node.run(Ctx(out_dir=out_dir, dep_dir=dirs.get, record=opts.on_cost)).get("note")
            state["nodes"][node.id] = {"fingerprint": fp, "status": "done", "source": "executed", "note": note,
                                       "ms": int((time.time() - started) * 1000), "calls": opts.call_count() - calls_before}
        except Exception as err:
            # Persist failed nodes too: the money they spent is real, and partial artifacts stay.
            state["nodes"][node.id] = {"fingerprint": fp, "status": "failed", "error": str(err),
                                       "ms": int((time.time() - started) * 1000), "calls": opts.call_count() - calls_before}
            failed_at = node.id
        save_state(opts.state_path, state)
 
    return WorkflowResult(state=state, dirs=dirs, failed_at=failed_at)

In practice: the first run executes all six nodes; the second hits the cache on all six with zero paid calls:

  skip script    cache hit e5eba27d2715034e
  skip assets    cache hit 3d4e77e25b7af76f
  skip frames    cache hit 6a17ff8efdb8e2b8
  skip clips     cache hit a276b932372bc7c6
  skip voice     cache hit 182a684769cb79e8
  skip compose   cache hit be81c7a6875a8931

Inject a failure at the fourth node and then resume, and you get the direct answer to yesterday's pain: the first three nodes hit cache, 0 calls, CNY 0, and real execution begins at the fourth. Yesterday's wasted CNY 3.1250 is 0 today.

Note one thing in the main loop that is easy to omit: state must be persisted immediately after each node, not written once at the end. Written at the end, a hard kill, a power loss or an evicted container loses all of it — and none of those are rare during a video task that runs for a quarter of an hour.

And one boundary to state honestly: this course's idempotency granularity is the node, not the shot. Four shots live in one node, so a failure on the third redoes all four and the first two's money is wasted. Splitting to one node per shot is thriftier, at the price of a much larger task graph and more complex scheduling. That is the classic trade-off between parallel granularity and failure radius, and D9 tackles it head on.

The ledger: which node spent whose money

With node state in hand, the cost ledger comes almost free — each node records which provider it called, how many times, how much it cost and what it produced. Print a table at the end:

Ledger (by node)
  node       status  source  elapsed  paid calls   estimated   fingerprint
  script     ok      cache       0ms        0   CNY  0.0000   e5eba27d2715034e
  assets     ok      cache       0ms        0   CNY  0.0000   dd68d1cf67dfdb0b
  frames     ok      cache       0ms        0   CNY  0.0000   6fccd700817dfc5c
  clips      ok      exec     2666ms        3   CNY 10.4950   14741e30ae0d75c0
  voice      ok      exec      258ms        5   CNY  0.0193   9fa591ead4503587
  compose    ok      exec     2730ms        0   CNY  0.0000   e4a2658619e63c9e
  total                      5654ms        8   CNY 10.5143
  real spend this run: video CNY 10.4950, tts CNY 0.0193

This table has two columns yesterday's did not, and those two columns are today's entire result: source tells you whether the node really ran or hit the cache, and fingerprint lets any artifact be traced back to which run, which version and which input produced it.

One rule for reading it: some columns are reproducible and some are not, so do not copy the two kinds together. Elapsed time depends on your machine and gives ten different numbers in ten runs; fingerprint, source and paid call count are deterministic — the same input necessarily computes the same fingerprint, and the second run's call count is necessarily 0. So acceptance watches those columns and treats elapsed time as a reference. Not knowing which numbers should reproduce is the most common way to misread someone else's performance report.

The paid-calls column deserves emphasis: it must be counted, never inferred. "It hit cache so it should be 0" is an inference, and an inference becomes a lie after one careless code change. The right way is a counter wrapped around the four providers, incremented on every real call. A metric that cannot prove anything is worse than no metric.

Offline, the ledger's amounts are estimates — usage converted at published unit prices, with the speech tier approximated from a resource package. Estimating is fine, but it must be labeled as an estimate. How to use this ledger to save money — draft tier versus final tier, model routing, budget circuit breakers — is D12; today we only keep the books straight.

When to switch to something off the shelf

One contrarian note to close: the engine you wrote today is not meant to be a general orchestration system.

It is under three hundred lines and does four things: topological sort, fingerprinting, cache hits, state persistence. No distributed scheduling, no retry policy, no cron triggers, no visualization, no multi-tenancy, no permissions. Every one of those absences is deliberate, because those four things are exactly what you need right now, and writing them yourself buys you a thorough understanding of why an idempotency key must include dependency fingerprints and why state must be persisted per step. That understanding transfers to any off-the-shelf engine, whereas someone who only clicks buttons is helpless the moment the framework changes.

So when should you switch? Three signals; any one of them warrants a serious evaluation.

One: you need cross-machine scheduling. In-process scheduling on one machine no longer holds, and nodes must be dispatched to several machines. Implementing distributed scheduling yourself grows exponentially in complexity, and not switching here is reinventing the wheel.

Two: you need human-in-the-loop nodes. Something like "generate, then wait for a human review before continuing" requires a flow that suspends for hours or days, with state externalized to a database rather than a JSON file. D10's review desk is where you feel this need for the first time.

Three: non-engineers need to see it. Operations needs to know which step an episode is stuck at and to click retry themselves. At that point what you need is not an engine but a product with an interface.

With none of the three present, keep your own three hundred lines. Adopting a heavyweight orchestration framework too early costs you a detour around its abstractions on every business change, while the payoff arrives only at scale.

Source Reading

Hands-On Lab

Before you start, dig out yesterday's pain report and keep the "money already spent" figure to hand — today's acceptance is watching it become 0. The lab persists state to a JSON file; no database and no API key needed.

1. Run the same command twice in a row, compare the two ledgers, and confirm the second run's paid call count is 0 rather than merely "no errors."
2. Delete one artifact file from one node's cache by hand and run again; confirm that node redoes its work while the others still hit.
3. Inject a failure at the fourth node, read the spend on its ledger row and the status of the two downstream rows, then resume and confirm execution starts at the fourth node.
4. Bump one node's implementation version and run again; confirm it and everything downstream invalidate while its upstream is unaffected.
5. Change the sentence and the run id, run once, put the two fingerprint columns side by side, and say why all six changed.

Interview Questions

Today's 3 questions are in the question bank below, focused on separating scheduling from business logic, idempotency key design, and what resumption must persist. Read the analysis before the key points — practicing the derivation beats memorizing them. The cn / global labels let you pick by target market.

Checklist and Tomorrow

• I can implement a minimal task-graph executor that schedules nodes by dependency and records each node's state
• I can make every node idempotent: the same input doesn't spend money twice, and a rerun only completes the unfinished parts
• I can log each run's outputs and cost into a ledger, traceable by run id
• I can name the four things an idempotency key must include and the four it must never include
• I can explain what the cache layer and the state layer each provide, and why they are separate
• All 5 lab acceptance criteria pass
• I can answer at least 2 of the 3 interview questions without looking at the key points

Tomorrow (D9) we start several episodes at once. Today solved "do not do the same thing twice"; tomorrow solves "how to do different things simultaneously without blowing through the vendor's quota" — the speech API allows twenty requests per minute for topped-up accounts, and one episode with forty storyboard entries means forty syntheses, so without a gate you will hit the limit. Why idempotency before concurrency? Because concurrency only burns money faster, and turning it on while every failure still restarts from scratch multiplies yesterday's pain by the number of episodes.