Processing Workflow

Processing Workflow

How an uploaded file becomes a reviewable document, and what keeps that process alive through timeouts, cold starts, provider outages and crashes.

Canonical source: docs/workflow.md · code: lib/workflows/process-document.ts. Failure codes: Failure Codes.

---

1. Why not a request

The obvious implementation is a long POST handler. It fails in three ways that matter here:

• A timeout loses the work. Extraction, OCR and a model pass over a 40-page PDF can exceed any sane request budget, and retrying means starting over.
• A background promise loses it silently. Kick the work off and return, and a cold start or a recycled instance ends it with nothing written down.
• Partial failure is unrecoverable. If analysis fails after extraction succeeded, a request-scoped implementation has no way to resume.

A durable run (the Workflow SDK) fixes all three. Each step's result is persisted, a failure resumes from the last completed step, and a step that throws is retried on its own. FatalError marks the failures retrying cannot fix: an unsupported file type does not become supported on the third attempt.

Where the runs live

Deployment	World	Worker
Vercel	supplied by the platform	none needed; leave WORKFLOW_TARGET_WORLD / WORKFLOW_POSTGRES_URL unset
Docker / self-hosted	@workflow/world-postgres, in the same Postgres	started by instrumentation.ts

Two things are easy to get wrong, and both fail quietly:

• The worker has to be started. Without it, a self-hosted install accepts uploads and never processes them.
• The world reads WORKFLOW_POSTGRES_URL, not DATABASE_URL, and falls back to postgres://world:world@localhost:5432/world rather than failing. instrumentation.ts defaults it from DATABASE_URL.

---

2. Getting in line: admission

POST /api/documents/:id/process does not start a run directly. It:

1. applies the processing rate limit (by network key) and checks ownership;
2. returns the existing run if the document already has a workflowRunId ({ runId, resumed: true }), so a retried click never starts a rival run;
3. marks the document queued and asks admission (lib/documents/admission.ts) whether the owner has a free slot.

A document waits in queued until one of four things admits it:

When	Why
The process route	the bytes have just landed
The retry route	pressing Retry is not a way around the limit
The end of every run, success or failure (admitNext)	this is what makes it a queue rather than a throttle; forgetting the failure path is how a queue drains only when everything goes right
The cleanup endpoint, /api/cron/cleanup (admitStalled)	a backstop for a run killed mid-flight, when the event that would free the slot never arrives. It runs from Vercel cron, the sweep.yml Action or the Compose scheduler; pnpm cleanup does not run it

Slots are per owner: ANONIFY_BATCH_PROCESSING, 3 on the demo profile and 6 self-hosted.

---

3. The orchestrator

The "use workflow" function only sequences. It runs in a sandbox without Node built-ins, so it holds no logic beyond ordering and error handling:

export async function processDocument(documentId: string) {
  "use workflow"
  try {
    await publishStatus(documentId, "queued",      "document.queued",      { progress: 10 })
    await publishStatus(documentId, "extracting",  "document.extracting",  { progress: 30 })
    await ingestUpload(documentId)
    const expansion = await expandContainerStep(documentId)

    if (expansion.container) {                    // a mailbox stops here: it IS the batch
      await finishContainer(documentId, expansion.children)
      await admitNext(documentId)
      return { documentId, status: "expanded" }
    }

    await publishStatus(documentId, "normalizing", "document.normalizing", { progress: 55 })
    const { pageCount } = await extractAndNormalize(documentId)

    await publishStatus(documentId, "analyzing",   "document.ai.started",  { progress: 70 })
    const { suggestions } = await analyze(documentId)

    await carryDecisions(documentId)              // batch + global rules reach this document
    await finish(documentId, pageCount, suggestions)
    await admitNext(documentId)                   // this slot is free; start the next
    return { documentId, status: "ready" }
  } catch (error) {
    await fail(documentId, message(error))
    await admitNext(documentId)                   // a failure frees the slot too
    return { documentId, status: "failed" }
  }
}

Everything real is a "use step" with full Node access: parsing, crypto, database, model calls.

flowchart TD
    Q["queued"] --> ING["ingestUpload<br/>stream, size-check, sniff, SHA-256,<br/>seal in chunks, delete the plaintext"]
    ING --> EXP{"expandContainer<br/>message with attachments?<br/>mailbox?"}
    EXP -- "mailbox" --> FC["finishContainer<br/>status: expanded (terminal)"]
    EXP -- "message with attachments" --> KIDS["each attachment → a child document<br/>with its own run (started at most once)"]
    EXP -- "anything else" --> EX
    KIDS --> EX["extractAndNormalize<br/>re-verify checksum · ranged/streamed read<br/>OCR scanned pages · write model + page index<br/>charge the allowance once"]
    EX --> AN["analyze<br/>patterns → classify → contextual → verify → expand<br/>vision pass for pixels · column pass for grids<br/>every result: status suggested"]
    AN --> CD["carryDecisions<br/>batch + global rules applied here"]
    CD --> FIN["finish · status: ready"]
    FIN --> ADM["admitNext"]
    FC --> ADM
    ING -. throws .-> F["fail<br/>code + fixed sentence on the row"]
    EXP -. throws .-> F
    EX -. throws .-> F
    AN -. throws .-> F
    CD -. throws .-> F
    F --> ADM

Loading

---

4. Retries: paced, bounded, and honest

Every step has maxRetries = 4. The SDK would enqueue each retry immediately, which is close to no retry at all against the failures that are actually transient (a cold worker, a storage blip). So a throwing step is rethrown through paced, which delays the next attempt with exponential backoff:

const MAX_BACKOFF_MS = 30_000
function backoffMs(attempt: number) {
  return Math.min(MAX_BACKOFF_MS, 1_000 * 2 ** Math.max(0, attempt - 1))
}   // 1 s, 2 s, 4 s, 8 s … capped at 30 s

• FatalError passes through paced untouched. It exists to say retrying cannot help, and wrapping it would throw that away.
• A step retry is the wrong altitude for a provider rate limit. It re-runs the whole step (the decrypt, the rasterisation, every page already recognised). Calls to services this install does not own go through lib/services/throttle.ts instead, which paces them and retries per request, with jitter and honouring Retry-After. See AI Providers.
• Retrying is idempotent by construction. Ingest returns early when sourceBlobKey is set. Child runs are guarded on the workflowRunId column. Children are unique on (parentDocumentId, sourcePartPath) or the message index. The quota charge is recorded on the row. A rule carry skips a decision whose copy already exists.

---

5. The steps

ingestUpload: take ownership of the bytes

The first moment the server sees the file.

1. Read the stored size without reading the object, and refuse empty (empty-file) or oversized (too-large) uploads.
2. Stream the plaintext upload.
3. Sniff the content. The declared MIME type and the filename are hints; the magic bytes decide. A .pdf that is really a zip is extension-mismatch, an unknown kind is unsupported-type.
4. SHA-256 the bytes in the same pass.
5. Seal them under a fresh per-document data key in the chunked v1 envelope, stored as documents/:id/source.bin.
6. Delete the plaintext upload.

Idempotent: on replay it sees sourceBlobKey already set and returns early, rather than re-encrypting under a second key and orphaning the first.

expandContainer: when one file becomes many

Runs after ingest and before extraction, because reservation happens before there are any bytes to parse, and because running it first queues the children while the parent is still being read.

• A message (eml) with attachments in supported formats gets one first-class child document per attachment, each with its own key, run, extraction, quota, review and export. Provenance is the parent and the MIME part path. The message itself continues to extraction.
• A mailbox (mbox) becomes one document per message, each exactly like the same .eml uploaded on its own. The mailbox itself ends at the terminal status expanded. It is not ready (there is no model to open) and not failed.
• Anything else falls straight through.

A parser or expansion limit is a verdict about this file, not the weather, so it is refused whole as too-complex. A partly expanded message would look complete and would not be. A mailbox with bytes but no readable messages is empty-container. Each child's run starts at most once, guarded on the workflowRunId column rather than on the step replaying. Detail: Batches and Mailboxes.

extractAndNormalize: one vocabulary from every format

1. Re-verify the source checksum (corrupt-source on mismatch). Storage that returns different bytes than it was given is worth catching before those bytes are parsed.
2. Dispatch on the sniffed kind to the format pipeline (Supported Formats). Most formats are read by ranged or streamed reads of the sealed source, not decrypted whole (Streaming IO). An archive the ranged zip reader cannot be certain about falls back to whole-file extraction.
3. PDF pages with fewer than 16 characters of text are rasterised and OCR'd (Tesseract or Mistral). A misconfigured OCR provider fails at once as configuration.
4. The normalized model is written, sealed under the same key, as normalized.json.bin, together with normalizedIndex (byte offsets of every page) in the same database update. A retried extraction therefore cannot leave one model's key beside another model's offsets.
5. Charge the daily allowance. This is the first moment the real cost is known: pages, filled cells, slides, KiB of decoded email text. Going over is quota, and it never deletes the upload. The charge is idempotent: metadata.quotaCharged is set in the same transaction as the UsageRecord increment, and a retry reads it first.

The model is stored outside the database. It contains the document's text, and text in a column is text in every backup and every query log.

analyze: propose, never decide

Runs the detection pipeline (Redaction Engine), streaming document.ai.progress as chunks complete:

• deterministic detectors on every page's text (free);
• classify, then the contextual pass in ~6 KB chunks through the throttle, then a batched verify of shaky candidates, then local expansion of every value already judged (free);
• for images and scanned pages, a vision pass over the pixels, driven here so its skip reason is kept;
• for spreadsheets, one column-level judgement from headers plus five samples, written as one column suggestion rather than one per cell;
• the preset's categoryAllowed filter applied to model proposals too.

Every result is written with status: "suggested". When the model pass fell short (no key, an unsupported model, a rate limit, an empty balance, the spend cap), the step writes document.ai.degraded with the reason, and the usage panel says so out loud.

The pipeline never marks its own findings accepted. There is no code path in which detection produces an accepted redaction.

carryDecisions: rules agreed elsewhere reach this document

A batch is reviewed while its documents are still arriving, so a rule agreed on the first file has to reach the fourth, which was still being analysed when the reviewer agreed it. Two kinds of rule are carried here, after analysis so their redactions sit beside the suggestions:

• batch rules (BatchRule) of the document's batch;
• the owner's global rules (OwnerRule), which apply to everything uploaded after they were made.

Each is materialised as a per-document GlobalRule whose redactions are accepted, because a rule is a decision. A switched-off rule is skipped. A carry is not swallowed on failure: a decision that silently did not arrive is a value the reviewer believes is gone everywhere. A rule that runs past the pattern budget in this document fails it as rule-too-broad. That is the one retryable verdict, because narrowing or switching off the rule changes the outcome, and the document still opens for review because extraction succeeded.

finish / finishContainer / fail

These set the terminal status (ready, expanded, failed), emit the last event, and close the stream. fail classifies the thrown error into one of the Failure Codes and stores the fixed sentence for that code, never the thrown message, because a parser or driver error is not guaranteed to be free of document content. Then it frees the slot.

---

6. Retry

POST /api/documents/:id/retry re-runs a failed document. Ingest is idempotent and every later step recomputes from the sealed source, so it is safe. Suggestions from the failed attempt are discarded, and anything a person touched (accepted, rejected) is kept. It goes back through admission and is refused (409, with errorCode and retryable: false) when the failure is not retryable or the upload is gone.

---

7. Streaming progress

Steps write newline-delimited JSON to the run's durable stream:

const writer = getWritable<string>().getWriter()
try { await writer.write(encodeStreamEvent(event)) }
finally { writer.releaseLock() }   // an unreleased lock keeps the request alive

GET /api/documents/:id/stream converts each line into an SSE frame with its index, for up to 15 minutes, and is declared supportsCancellation, so a client that navigates away tears the invocation down. The client hook (use-processing-stream.ts) tracks the last index it saw and reconnects with startIndex = last + 1, so a dropped connection resumes rather than replaying the run or missing its middle.

Events carry ids, stages, counts and durations. They never carry document text. The stream is a progress channel, not a data channel.

stateDiagram-v2
    [*] --> queued: document.queued
    queued --> extracting: document.extracting
    extracting --> expanded: document.expanded (a mailbox is the batch)
    extracting --> normalizing: document.normalizing
    normalizing --> analyzing: document.ai.started
    analyzing --> analyzing: document.ai.progress ×N · document.ai.degraded
    analyzing --> ready: document.redaction.created, then document.ready
    extracting --> failed: document.failed {code, retryable}
    normalizing --> failed
    analyzing --> failed
    ready --> [*]
    expanded --> [*]
    failed --> [*]

Loading

An event looks like this. Only type, documentId and at are required:

{ "type": "document.ai.progress", "documentId": "doc_…", "at": "2026-09-05T12:00:00.000Z",
  "status": "analyzing", "progress": 70, "message": "Analyzing…",
  "payload": { "stage": "text", "completed": 3, "total": 8, "suggestions": 12 } }

decodeStreamEvent returns null on a blank or malformed line rather than throwing, so a bad frame drops quietly instead of tearing down a live stream. When the stream reports ready or failed, the workspace calls router.refresh() once per status and opens the editor.

---

8. Batch export: a durable run of its own

lib/workflows/export-batch.ts

POST   /api/batches/:id/export         create a BatchExport row, start the run → 202
GET    /api/batches/:id/export         where the run has got to (and a fresh archive link)
DELETE /api/batches/:id/export         stop it
GET    /api/batches/:id/export/stream  follow it as SSE

• planExport reads the wave width once and carries it through the run. A workflow function replays, and configuration read there would let a value changed mid-run reshape the waves.
• exportDocument is one step per document (maxRetries = 4, paced), run in waves of ANONIFY_BATCH_EXPORTING. It uses the same deliver function and the same verification gate as a single export, and charges the export allowance per document as a single export would.
• The row is the record. Each step writes its outcome into BatchExport.documents under SELECT … FOR UPDATE, and totals are recomputed from states, never incremented, because a retried step would otherwise count twice.
• Progress is pushed, not polled. After writing the row, each step writes a whole snapshot to the run's stream. /export/stream relays it with the same indexed resume. If the stream cannot be held open, the client falls back to reading the row every 5 seconds.
• Stopping sets cancelRequested and then cancels the run. The document being redacted right now finishes and is kept. Everything already exported stays downloadable, because it was verified before the reviewer changed their mind.
• settleRemaining names every document the run did not reach; finishExport / failExport write the terminal state.

The archive is not built by the run. GET /api/batches/:id/download assembles it later from the artifacts the run produced, in the shape the reviewer chose. See Batches and Mailboxes.

---

9. Expiry

lib/workflows/cleanup.ts, every 15 minutes: Vercel cron from vercel.json (daily on Hobby), the sweep.yml GitHub Action as a Hobby backstop, pnpm cleanup from cron, or the opt-in Compose scheduler. See Deployment.

1. Mark documents past expiresAt as expired, so the workspace stops serving them mid-window (410).
2. Purge each one through purgeDocument: every object under documents/:id/ (upload, source, normalized model, exports, reports, renders), then the row.
3. Prune empty batches, owner/global rules unused for 30 days, and stale rate-limit buckets. A prune failure is swallowed to 0, so it cannot abort the load-bearing expiry pass.
4. Admit anything left queued behind a slot nothing freed (admitStalled). This step belongs to the HTTP endpoint /api/cron/cleanup only. pnpm cleanup talks to the database directly and does the first three steps, but not this one.

A run is bounded to 50 documents, and a backlog is worked down over successive runs. Order matters: the row is the only thing that knows where the bytes are, so it is deleted last, and only if storage cleared. A blob already gone counts as deleted, and so does a document removed by someone else mid-sweep (fixed in 1.8.0, #140). One document that cannot be purged is counted, logged and retried next run instead of stopping the rest. A message and its attachments share an expiry, and purging the message cascades to its children.

Retention is 1 h, 6 h, 24 h or 3 days, chosen at upload. Extending it recomputes from creation, capped at 72 hours, so renewing repeatedly converges on the ceiling instead of walking it forward. Attachments share their parent's clock.

---

10. Failure, from the user's side

Failure	What the user sees	What is true
Model provider down or unconfigured	Sparse or absent suggestions, and a degraded notice in the usage panel	Deterministic detection ran; manual redaction, search and rules work
Extractor throws	"We couldn't analyze this document" + Retry (if retryable)	Source untouched
Failed after extraction (e.g. rule-too-broad)	The editor opens with a failure banner	Manual review is real; an empty list is not "nothing to redact"
Quota exceeded	The limit, and when it resets	Nothing was deleted
Stream drops	Reconnects and resumes	The run continues regardless
Export verification fails	"did not pass verification and was not saved"	No file written, no link issued

The recurring sentence in the error copy, your original file is safe and was not modified, is not reassurance. It is the architecture: the source is never mutated, so it is simply true.

---

See also

• Failure Codes · Architecture · Streaming IO · Batches and Mailboxes · Troubleshooting