gpumon · integration API

broker · writer · ingress · ocr-api

Envelope v1 — wire format

Why this exists / when to use what

Every queue in federation.work uses the same outer JSON shape. The envelope is the contract between producers and consumers across repos, languages, and machines. It is forever-compatible: consumers MUST accept version: 1 and ignore unknown top-level keys so the format can grow without coordinated upgrades.

If you're publishing or consuming a message — read this file. The per-domain payload schemas are the only thing that varies; the outer wrapper is constant.

Outer envelope

interface EnvelopeV1<P = unknown> {
  version: 1;          // schema version, locked at 1
  id:      string;     // uuidv7 — sortable by time, globally unique
  ts:      number;     // unix epoch seconds when the envelope was minted
  attempt: number;     // 1 on first publish, +1 on each transient retry
  payload: P;          // domain-specific (see below)
}

Wire encoding:

  • content-type: application/json
  • UTF-8 JSON, no comments
  • delivery_mode: 2 (persistent) — set by the producer
  • correlation_id: optional, mirrors id
  • message_id: optional, mirrors id

Rules

  1. version: 1 is permanent. A future schema change is additive: new optional top-level fields, new optional payload fields. Both sides ignore what they don't understand.
  2. id is a uuidv7. Sortable by mint time, good for log triage. The same id MUST be reused across retries — i.e. a republish with attempt+1 keeps the same id. Consumers use id for idempotency.
  3. attempt starts at 1. Worker republishes (transient retries) bump it. Consumer give-up happens at attempt > MAX_ATTEMPTS (default 5).
  4. ts is the first publish time. Republishes preserve the original ts so DLQ archaeology can compute end-to-end latency.
  5. payload is mandatory and non-empty. No "ping" messages on federation queues — those go on a separate health queue if ever needed.

Domain payloads

Each queue carries exactly one payload shape. Cross-domain envelopes (e.g. edgar publishing into ocr.jobs) use the destination payload shape, not the source — the producer translates.

OcrJobPayload (queue ocr.jobs)

interface OcrJobPayload {
  kind:        "ocr";
  job_id:      string;     // matches the writer's ocr_jobs.id
  ipfs_cid:    string;     // CIDv1, base32. Required.
  filename:    string;     // for display + result naming
  sha1?:       string;     // hex sha1 of the source bytes
  bytes?:      number;     // source byte length
  total_pages?: number;    // if pre-counted
  // Optional cross-domain follow-up annotation:
  origin?: {
    domain: "edgar" | "fraud" | "manual";
    ref:    string;        // e.g. "edgar/0000320193/0001628280-24-001234/exhibit-99.pdf"
  };
}

Example envelope:

{
  "version": 1,
  "id":      "0192f4e2-7c10-7c71-a4b1-0a3f1c2b8d44",
  "ts":      1747613000,
  "attempt": 1,
  "payload": {
    "kind":     "ocr",
    "job_id":   "ocr_2026-05-19_abc123",
    "ipfs_cid": "bafybeigdyrztpnv5q...",
    "filename": "MEP-v-Rudkins-exhibit-12.pdf",
    "sha1":     "9c8a2d1e5f4b3c2a1e8f9d7b6a5c4d3e2f1a0b9c",
    "bytes":    4_823_104
  }
}

FraudRowPayload (queue fraud.rows)

interface FraudRowPayload {
  kind:        "fraud.row";
  corpus_id:   string;     // fraud-corpus.db.corpora.id
  row_no:      number;     // fraud-corpus.db.rows.row_no within corpus
  chunk_text:  string;     // the chunk to enrich (extract triples / cleanup / verify)
  task?:       "triples" | "cleanup" | "verify";  // default: triples
}

EdgarCikPayload (queue edgar.ciks)

interface EdgarCikPayload {
  kind:       "edgar.cik";
  cik:        string;      // zero-padded 10-digit ("0000320193")
  reason?:    "new_filing" | "stale_resolve" | "manual";
  entity_name?: string;    // hint, resolver verifies via SEC
}

EdgarFilingPayload (queue edgar.filings)

interface EdgarFilingPayload {
  kind:       "edgar.filing";
  cik:        string;      // zero-padded 10-digit
  accession:  string;      // "0001628280-24-001234"
  form:       string;      // "10-K", "10-Q", "8-K", "DEF 14A", ...
  filed_at?:  string;      // ISO date
  // If the ingester decides this filing needs OCR (scanned PDF exhibit),
  // it publishes a follow-up envelope to ocr.jobs (see below).
  needs_ocr?: boolean;
}

Cross-domain follow-up pattern

A consumer of one queue is often a producer for another. The canonical example: edgar-filing-ingester downloads a 10-K, notices an exhibit is a scanned PDF (rather than text), uploads it to IPFS, and publishes to ocr.jobs so the OCR worker pool can fan out.

// Inside edgar-filing-ingester, after the filing is processed:
if (exhibit.isScanned) {
  const cid = await uploadToIpfs(exhibit.bytes, exhibit.filename);
  await producer.publish({
    version: 1,
    id:      uuidv7(),
    ts:      Math.floor(Date.now()/1000),
    attempt: 1,
    payload: {
      kind:     "ocr",
      job_id:   `ocr_edgar_${env.payload.accession}_${exhibit.index}`,
      ipfs_cid: cid,
      filename: exhibit.filename,
      sha1:     exhibit.sha1,
      bytes:    exhibit.bytes.length,
      origin: {
        domain: "edgar",
        ref:    `${env.payload.cik}/${env.payload.accession}/${exhibit.filename}`,
      },
    },
  }, "ocr.jobs");  // routing key = destination queue
}

The OCR worker doesn't care that the envelope came from edgar — it just sees an OcrJobPayload, runs the pipeline, calls /api/ocr/jobs/finish on the writer. The origin field is optional metadata for traceability; consumers MUST function correctly when it's absent.

Bun helper

import { randomUUID } from "node:crypto";

export interface EnvelopeV1<P = unknown> {
  version: 1; id: string; ts: number; attempt: number; payload: P;
}

export function newEnvelope<P>(payload: P, attempt = 1): EnvelopeV1<P> {
  return {
    version: 1,
    id:      randomUUID(),         // uuidv4 here; swap for uuidv7 if you have a lib
    ts:      Math.floor(Date.now() / 1000),
    attempt,
    payload,
  };
}

Cross-references

rendered from docs/api/envelope.md  ·  v0 reference: writer hand-coded catalog ↗ (the layout that inspired this style guide)