workflow

Tier 2 · domain. Depends on store-sqlite, agent, hitl, and authority. zod v4 is a peer.

npm install @titan-design/workflow zod

The problem it solves

A long-running agentic process — plan, review, revise, get approval, ship — has to survive the process dying halfway through. The usual answers are a state machine you hand-maintain, or a durable-execution engine you now have to operate.

Here you write an ordinary async function. Every dispatch, seed, and assisted call is memoized in SQLite, so after a crash or restart the function re-runs from the top and reconciles unfinished agent work before replay. Uncertain work stays marked for recovery instead of being submitted again. Human pauses are durable gates, resolvable from any process.

When to reach for it

Multi-step agent work with branches, loops, and approval points that must not lose progress across a restart. For a single agent call with no memoization, use agent directly.

Example

Verified against 0.1.0 with inlineRunner.

import { WorkflowRuntime, inlineRunner, workflowMigration } from "@titan-design/workflow";
import { SqliteGateStore, gateMigration, gateResolverMigration } from "@titan-design/hitl/sqlite";
import { openDatabase, runMigrations } from "@titan-design/store-sqlite";

const db = openDatabase("state.sqlite3");
runMigrations(db, [gateMigration(1), workflowMigration(2), gateResolverMigration(3)]);
const owner = { class: "owner-terminal", id: "alice", channel: "cli" } as const;

const runtime = new WorkflowRuntime({
  db,
  gates: new SqliteGateStore(db, { migrate: false }),
  runner: inlineRunner(({ prompt }) => `handled: ${prompt}`),   // returns a string
  onEvent: (event) => {
    // In a real product the answer comes from a human, from anywhere.
    if (event.type === "gate_opened") runtime.signal(event.runId, event.stepId, { signal: "approved" }, owner);
  },
});

runtime.register("review", async (ctx) => {
  const plan = await ctx.dispatch("plan", "Plan {{brief}}.");
  // plan.output === 'handled: Plan the docs site.'
  const approval = await ctx.assisted("approve", "Ship it?");
  // approval.signal === 'approved'
});

const runId = runtime.start("review", { brief: "the docs site" });
await runtime.wait(runId);
runtime.status(runId)?.status;   // 'completed'

In production, swap the runner for agentRunner, which runs each step as one headless Claude Code session:

import { agentRunner } from "@titan-design/workflow";

runner: agentRunner({ cwd: repo, maxTurns: 40, maxBudgetUsd: 5 })

After a restart, await runtime.hydrate() claims eligible runs and reconciles their active steps. Legacy agent runs without confirmed attachment remain recovery_required.

The three step kinds

• dispatch(stepId, template, { model?, vars? }) renders the template — over the run's params, STEP_OUTPUT_<STEP>, PREVIOUS_STEP_OUTPUTS, WORKFLOW_NAME, RUN_ID — hands the prompt to the StepRunner, and parses a signal from the output. Results are keyed stepId:iteration, so calling the same step again in a loop is a new iteration and ctx.iteration(stepId) is the loop guard. A retryable runner failure is re-dispatched once (maxRetries); anything else fails the run. StepResult.usage sums every attempt, retries included, and the active step persists earlier attempts' cost as priorUsage so a resumed run counts each attempt once.
• dispatch(stepId, template, { schema }) takes a zod schema for a typed step. The runner receives it as StepRunInput.outputSchema and may pass it to the model; the workflow's own parse is authoritative either way. After the runner succeeds, the output is parsed as JSON, validated, and stored as StepResult.data, typed as the schema's output, beside the raw output. A payload that is not JSON, fails the schema, or serialises to more than maxStepDataBytes (default 64 KiB) throws StepOutputInvalidError, a non-retryable StepFailedError with kind (not_json, schema, too_large), issues and the attempt's usage. It costs one runner call and records no result. z.object strips undeclared keys, but any top-level titan.trace.* key in the output that the schema does not declare is carried into data unchanged, before the size bound applies; a key the schema declares is the schema's. Schemas should produce JSON values, because the stored data is serialised. On replay, data is re-derived from the recorded output under the current schema, so a run stored before typed output replays with typed data and a memoised step is never re-run. A recorded output that no longer fits the schema fails the run with WorkflowSchemaDriftError. A replayed call without a schema returns the recorded result unchanged. signal still comes from the signal parser, not from data.
• seed(stepId, fn) runs a deterministic function once per call; its data merges into the params for later prompts.
• assisted(stepId, prompt) opens a gate and waits. The row survives restarts; runtime.signal(runId, stepId, payload, resolvedBy) resolves it from anywhere and records resolvedBy on the gate. A signal field in the payload becomes the step's signal. Like dispatch it advances ctx.iteration(stepId), so each call in a loop opens a new gate: the first is keyed stepId with gate <runId>/<stepId>, iteration n is keyed stepId:n with gate <runId>/<stepId>:n. runtime.signal resolves the call that is waiting, and replay returns recorded answers without reopening their gates. A gate cancelled while the call waits fails the run, unless the call passes { recordCancel: true }. Then the cancel is recorded as the answer, with signal GATE_CANCELLED_SIGNAL and data { reason }, and the next call opens a fresh gate.

All three share one call counter per stepId, so seed("x") then assisted("x") gates on <runId>/x:1, and every result records the method that wrote it as StepResult.operation. Runs stored before 0.5 carry no operation and keep that release's keys until they finish.

Authority steps

ctx.authorize(stepId, request, { prompt?, expiresAt? }) asks the authority table before a governed action. Give the runtime authority: { actor: { class, id }, table? }; the table defaults to DEFAULT_TABLE, and a run without the option fails at its first authorize. allow returns { verdict: "allow", ruleId }. deny records the decision and throws AuthorityDeniedError without opening a gate. gate opens a hitl gate at <runId>/<stepId> bound to the rule (rule: { table: "F5", version, ruleId, resolvers }), so the store refuses a resolver class the rule does not name. The answer is { decision: "approve" | "refuse", subject }, where subject repeats the request's subject exactly. An approval returns { verdict: "approved", ruleId, gateId, resolvedBy }; a refusal, or a gate resolved with no resolver, throws AuthorityRefusedError.

const runtime = new WorkflowRuntime({ db, gates, runner, authority: { actor: { class: "automation", id: "factory" } } });
runtime.register("land", async (ctx) => {
  await ctx.authorize("merge-authorize", { action: "merge", subject: { repo, pr, headSha } });
  // merge exactly headSha
});

A restarted run resumes onto the open gate and judges the answer by the rule recorded on it, so a table edit during the pause does not flip the decision. Replay never consults the table.

Fan-out

mapItems(ctx, stepId, items, fn, { key, concurrency?, budgetUsd?, maxFailures? }) runs fn over a list with a concurrency cap (default 1). Each item runs as step ${stepId}/${key}, and fn receives that id to pass to ctx.dispatch, so each item memoizes on its own. After a restart, replay reuses finished items and runs only the rest. Matching is by key, not position, so the input may be reordered between runs. Duplicate keys throw before anything launches.

import { mapItems } from "@titan-design/workflow";

const review = await mapItems(ctx, "review", files, (file, itemStepId, c) =>
  c.dispatch(itemStepId, `Review ${file}.`), { key: (file) => file, concurrency: 4, budgetUsd: 2 });
// review.results in input order; review.stoppedBy is "budget", "failure", "cost_unknown" or null

budgetUsd is checked before each launch against the cost items reported in StepResult.usage (a StepUsage of costUsd plus optional token counts) or on their failure. Items still in flight are not counted, so a run can overshoot by up to concurrency - 1 items. An item whose usage is marked costUnknown (an unpriced measurement, such as a Codex subscription run) stops launches with stoppedBy: "cost_unknown" when budgetUsd is set, rather than counting the unpriced spend as zero.

A failed item lands in failed with its error, its retryable flag and the usage of its failed attempts, and the other items keep launching. A non-retryable failure stops new launches at once; retryable failures stop them once there are more than maxFailures (default 3). spentUsd includes failed-call cost. The result carries results, failed, skipped, spentUsd, and stoppedBy. Any other error is rethrown once the items in flight settle.

Runners

agentRunner classifies agent failures: rate limits, runtime errors, and inactivity are retryable; budget, auth, refusal, and schema failures are not. It reports usage (costUsd, inputTokens, outputTokens) on every successful step and on a failed step when the agent run reported it, and a run with an outputSchema stores its output as JSON text. A step's own schema is forwarded as that outputSchema and wins over defaults.outputSchema. A schema_invalid failure from the agent ends as the same non-retryable StepOutputInvalidError (kind: "schema") as a workflow-side parse failure, through idempotentRunner and routedRunner too; its message carries zod issue paths, never the payload.

durableHarnessRunner reports usage on a successful step from the harness's measurements, selected with agent-protocol's foldUsage. Response deltas count once each and supersede snapshots; without deltas it keeps the highest-sequence snapshot per scope, and a conversation snapshot supersedes turn snapshots in its epoch. An unpriced measurement adds its tokens at zero cost, and a null token count adds nothing. Its failed steps report none, because the durable failure record carries no usage.

idempotentRunner(live) wraps a live runner whose steps are safe to repeat, such as read-only judgements. A step that was in flight at a crash is dispatched again on hydrate instead of parking the run as recovery_required. With the wrapper, a step's agentId is its request key, not the agent session id.

routedRunner(routes) sends each dispatch step to the runner its route names, so one workflow can mix code steps and model steps. A route's match covers a step id and its <id>:<suffix> family, and the longest match wins. Its onRestart rule decides what a crash does to a step in flight: "repeat" dispatches it again on hydrate, and "park" leaves the run recovery_required, as does a step with no route. Call runner.assertRoutes(workflowName, dispatchStepIds) when you register a workflow; it throws naming every uncovered step id. Two routes with the same match throw at construction.

inlineRunner(fn) is for tests and for steps that are not agents. Its function returns a plain string, not a StepRunOutcome — the wrapper turns a thrown error into a non-retryable failure.

Implement RecoverableStepRunner for a queue or a subprocess. Its early acknowledgment and terminal completion must already be durable. The optional legacy attach(step) API is deprecated; missing or failed attachment requires recovery.

Execution recovery

durableHarnessRunner(dispatcher, { request }) connects the durable dispatcher from agent to the workflow. The request builder receives the rendered prompt, step model, and execution intent; it supplies harness configuration and bounds. Workflow owns the cancellation signal.

The runtime persists execution identity and request key before dispatch, then stores the acknowledged runner reference before awaiting completion. On restart, a stored terminal result or a still-owned live handle can resume replay. Only explicit safe absence permits a replacement attempt. Unknown state, ownership loss, timeout, or failed persistence keeps the active step intact. A later hydrate() may reconcile new evidence, but never blindly resubmits an uncertain execution.

runtime.completeStep(runId, stepId, output) records output as the answer of an active dispatch step of a run no runtime holds, so the next hydrate() replays that answer instead of dispatching the step again. It returns false for a run a live runtime drives.

runtime.annotate(runId, stepId, data) records data as step stepId of a run that already completed, failed or was cancelled, for a fact learnt after it ended. It never rewrites a step the run has, and returns false for that or for an unfinished run. WorkflowRunStore.annotate is the same write by key.

A replay runs the workflow function again, so code that changed since the run was recorded can take a different path through old results. ctx.historyNext() names the step recorded right after the call just answered from the record, and is undefined once the run is live; a workflow reads it to follow the path the run already took.

Existing databases must apply workflowOwnershipMigration(version) with a new migration version. Fresh workflow DDL already includes ownership columns; the additive migration also tolerates that case. Saves and renewals require the captured runtime generation, unexpired lease, and expected row revision.

Signals

parseSignals(output) returns every signal an output carries, highest precedence first; parseSignal(output) returns the first of them or null, and is what dispatch stores.

• Empty or whitespace-only output yields only empty_output (EMPTY_OUTPUT_SIGNAL). A reviewer that said nothing has not approved; escalate to a human.
• <!-- signal: needs_revision --> in the output is authoritative. Markers that name a known signal replace the prose patterns; unknown names fall through to them.
• Otherwise the default patterns apply in this order: high_risk (a risk score of 4 or more), needs_revision, has_open_questions, approved, needs_fixes, changes_requested, needs_clarification, needs_changes. A high risk score beats Verdict: PASS.

Pass parseSignal: createSignalParser(patterns) to use your own; the key order of patterns is the precedence. unfilledVariables(template, vars) lists the placeholders a template needs that vars does not supply, read from the template before substitution.

Lifecycle

start persists the run and launches it. wait resolves when it finishes. status and list read state. cancel persists cancelling before aborting live work and pending gates. Confirmed cancellation becomes cancelled; unconfirmed cancellation retains the active step as recovery_required. A success racing cancellation is memoized before the workflow finishes cancelled. Events — step_started, step_complete, step_retry, step_failed, gate_opened, workflow_* — go to onEvent.

Trace projection

projectRunTrace(run, { gates, executions?, digest }) projects a stored run into the strict trace records of titan.trace/v1. It reads the hitl gate store's get, optionally an execution ledger's findByRequestKey, and an injected sha256 digest, and never the clock. Re-projecting the same database, after a restart or from a rebuilt runtime, gives JSON-byte-identical records, so a consumer can upsert them by id.

Record	Source	One per
run	the WorkflowRun	run, first
attempt	the ledger entry at workflowStepRequestKey(runId, stepId, iteration, attempt), else the step result	dispatch attempt, retries included, each with its own usage
gate (policy)	StepResult.data["titan.trace.gates"] (TRACE_GATES_KEY), written by ctx.authorize	policy decision
gate (human)	the hitl gate an assisted step or an owner-gated authorize waited on	gate, plus a paused run's pending gate
artifact	StepResult.data["titan.trace.artifacts"] (TRACE_ARTIFACTS_KEY)	entry

Records follow the run's recording order: each step's attempts, then its gates, then its artifacts. In-flight steps and a pending gate come last. Without a ledger, a step projects one attempt at StepResult.attempt, the attempt that succeeded. A human gate's decidedBy is human:owner for an owner-* resolver and agent:<id> otherwise.

Free text stays out of the trace. The params (canonical JSON), a step's output, and a gate's prompt and payload are carried only as digests, and recovery evidence is dropped. Entries under the two data keys are parsed strictly; a malformed one throws an error naming the step rather than being dropped. A dispatch stores data only when it has a schema, so an untyped dispatch cannot carry artifacts.

Gotchas

You cannot signal a gate before it opens. runtime.start returns immediately, and the workflow reaches assisted some steps later. Calling runtime.signal(runId, stepId, …) before then throws GateNotFound, and the run then waits forever for an answer that was already thrown away. Drive it from the gate_opened event, or from a real human surface listing store.listPending().

Your workflow function re-runs from the top on resume. That is the mechanism, not a bug: memoized steps return their stored result instead of re-executing. Keep side effects inside seed or dispatch, never in the function body between them.

Editing a workflow under a live run can fail it. Replay matches calls by step id and call index. If the edited function reaches a recorded call through a different method, the run fails with WorkflowNonDeterminismError rather than hand back an answer recorded for another question. Add new steps under new step ids.

An agent cannot answer a gate. The gate store checks the resolver before it writes. A coordinator, worker, headless or automation class, or a refusal from the store's authorize, makes runtime.signal throw GateResolverRefused; the gate stays pending and the run stays paused. Once gateResolverMigration has run, a signal with no resolver is refused too.

Gate and workflow migrations share a database. Pass migrate: false to SqliteGateStore and put gateMigration(n) in your own migration list, or the two will fight over the schema.

Pushing events to a client is the product's job. brain sent these over an MCP channel; that adapter is deliberately not in the package.

Where it came from

brain's bespoke SQLite runtime, ported with its PM, template, and process-polling couplings turned into injectable seams. What stayed in brain — PM task claiming, prompt rendering from notes, model routing by turn complexity, .plans/ output files, the agent-process reconciler — each maps onto one of those seams: onEvent, render, DispatchOptions.model, and RecoverableStepRunner.reconcile.