Run the pipeline on the extracted invoice, not the seed

The extraction was theatre: the vision model read the PDF but its output was
discarded — matching ran on the seeded record. Now the extracted invoice is what
the pipeline runs on (matching joins the extracted lines against the PO/receipt),
so the data genuinely comes from the read, like production.

- The demo PDFs now print the SKU (lib/invoice-pdf.ts) — matching joins on it, so
  the model has to read the item code off the page; without it every line missed
  its PO line and clean invoices turned into exceptions. Verified: all 10 seeded
  cases produce the same verdict from extracted data as from the record.
- Intake fails closed: a failed/timed-out/invalid read stops the run with an
  error, no silent fallback to a stored record (lib/intake.ts).
- lib/intake.ts is the injectable core; lib/intake.test.ts is the "mock vision"
  coverage (success, header-mismatch, validation/refusal/timeout/throw).
- The offline pipeline test runs with skipExtraction so it stays key-free.
- README/comments corrected — nothing claims the pipeline runs on the seed.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: DylanMerigaud

README.md

@@ -38,7 +38,7 @@ flowchart LR
     class BLK,REJ stop;
 ```
 
-- **Extraction (AI)** — the vendor's invoice PDF is read by a vision model into a schema-validated `Invoice`. You watch the document get scanned and the fields fill in. (The seeded pipeline then runs on the trusted record, so a model misread degrades the reveal, never the downstream verdicts — see below.)
+- **Extraction (AI)** — the vendor's invoice PDF is read by a vision model into a schema-validated `Invoice`. You watch the document get scanned and the fields fill in. That extracted invoice is what the rest of the pipeline runs on — matching joins the extracted lines against the PO, like production.
 - **Matching** — a 2-way (invoice ↔ PO) or 3-way (invoice ↔ PO ↔ goods receipt) match, returning a verdict: `clean`, `exception`, or `duplicate`.
 - **Investigation** — runs only on an exception, and the one step that's an agent. A number ("9% over the PO") doesn't tell a reviewer whether it's a legitimate price increase or an overcharge; that lives in unstructured records, and which records matter depends on what you find. The agent **chooses** which tools to call, reads the records, and writes a recommendation. It decides nothing about the money.
 - **Approval routing** — tiers an exception (manager / director by the money and variance at stake) and **pauses for a human**; a duplicate is blocked so it's never paid twice; a clean match skips straight through.
@@ -63,7 +63,7 @@ The split-view dashboard shows the **invoice queue** (color-coded by outcome) an
 
 **AI at the edges, deterministic code in the core.** The two ends of the pipeline are language/perception problems — reading a messy PDF, judging a fuzzy exception — so they use a model. The middle (is this a 9%-over variance? a duplicate? which approval tier?) is arithmetic and policy, so it's pure, unit-tested functions ([`lib/matching.ts`](lib/matching.ts), [`lib/policy.ts`](lib/policy.ts), [`lib/erp.ts`](lib/erp.ts)): exact, auditable, identical on every run. An LLM never decides a payment amount.
 
-**Extraction reads the document; the record decides.** Extraction is the **first step of the Mastra workflow** (`intake`): the vendor's PDF is rendered ([`lib/invoice-pdf.ts`](lib/invoice-pdf.ts)) and read by a vision model into a schema-validated `Invoice` ([`lib/extract.ts`](lib/extract.ts), `Invoice.safeParse`) — so it's one pipeline end to end, from PDF to ERP, not a step bolted on outside. Here's the deliberate part: the extracted invoice is *shown* (the reveal proves the read happened), but the downstream steps run on the **trusted seeded record**, not the model's output. So a misread degrades the reveal, never the verdicts — the demo's edge cases stay reliable. It also fails open: if the model is slow or down, the step notes it and the pipeline proceeds on the record. In production you'd reconcile extraction against the PO and surface the diff; the seam is the same.
+**Extraction feeds the pipeline; the document is the source of truth.** Extraction is the **first step of the Mastra workflow** (`intake`): the vendor's PDF is rendered ([`lib/invoice-pdf.ts`](lib/invoice-pdf.ts)) and read by a vision model into a schema-validated `Invoice` ([`lib/extract.ts`](lib/extract.ts), `Invoice.safeParse`). The **extracted invoice is what the rest of the pipeline runs on** — matching joins the extracted lines against the PO and goods receipt (which come from the DB, as they'd come from the ERP). One pipeline end to end, PDF to ERP, and the data genuinely comes from the read. Like production: if the read fails (timeout, refusal, invalid output), the run stops with an error rather than fabricating data — there's no silent fallback to a stored record. (The seeded invoice is only what we render the PDF from — the stand-in for "a vendor PDF arrived" — and it's why the demo's PDFs carry SKUs: matching joins on them, so the model has to read the item code off the page.)
 
 **The investigator agent.** [`src/mastra/agents/investigator.ts`](src/mastra/agents/investigator.ts) is a Mastra [`Agent`](https://mastra.ai) with three tools — `get-vendor-price-history`, `get-po-notes`, `get-receipt-notes` — returning deliberately unstructured, free-text records ([`lib/vendor-context.ts`](lib/vendor-context.ts)). It runs an open-ended loop: it picks which tools to call and in what order (a quantity problem pulls the receipt notes first; a price problem pulls the price history), reads them, and writes a recommendation (`likely_legitimate` / `likely_overcharge` / `unclear`). It only *recommends* — the deterministic routing and the human gate own the outcome, so a wrong call is caught by the reviewer. Each tool call and the recommendation appear live on the trace. Tools read the trusted vendor from `requestContext`, not model-generated args, so the agent can't pull the wrong vendor's file. Model id lives in one place ([`src/mastra/model.ts`](src/mastra/model.ts)), resolved by Mastra's router (`"anthropic/claude-haiku-4-5"`).
 
@@ -153,7 +153,7 @@ This is a stateless demo with a fake ERP; the decision logic (matching, routing,
 - swap the fake ERP adapter for a real one (same `ErpAdapter` interface),
 - add persistence and an audit trail (the `agent_runs` table is already shaped for it),
 - wire real approver identity to the human gate,
-- make extraction authoritative — reconcile the extracted invoice against the PO and surface the diff for review, rather than running on the seeded record,
+- accept real uploaded PDFs at intake (today the demo renders its own from seed data),
 - batch processing across the whole queue.
 
 ---

components/extraction-reveal.tsx

@@ -13,17 +13,18 @@ import type { Invoice } from "@/lib/schema";
  *
  * Two states, driven by the intake trace node:
  *   • running  — scanning: the sweep animates, fields are still "reading…"
- *   • done     — the extracted fields are shown; matching ones are checked.
+ *   • done     — the extracted fields are shown; a badge says whether the header
+ *                reconciled with the PO record.
  *
- * The downstream pipeline runs on the trusted seeded record, so a model misread
- * degrades this reveal, never the verdicts.
+ * The extracted invoice is what the rest of the pipeline runs on (matching joins
+ * it against the PO) — the data on screen is the data that drives the verdicts.
  */
 
 export interface ExtractionState {
   status: "running" | "done";
   /** The invoice the model extracted (present once done). */
   extracted: Invoice | null;
-  /** Did the extracted key fields reconcile with the seeded record? */
+  /** Did the extracted header reconcile with the PO record? */
   matches: boolean;
 }
 

lib/extract.ts

@@ -16,9 +16,8 @@ import {
  * model reads the PDF directly (vision); nothing is guessed off filenames.
  *
  * Result is a tagged union so callers map each failure to a state instead of a
- * thrown error. Note: in the demo, the extracted invoice is shown to prove the
- * read happened, but the downstream pipeline runs on the trusted seeded record —
- * so a model misread degrades the reveal, never the deterministic verdicts.
+ * thrown error. The extracted invoice is what the pipeline then runs on (see
+ * `lib/intake.ts`); a failure stops the run rather than fabricating data.
  */
 
 /** Sonnet reads the PDF directly via vision; extraction is transcription, not reasoning. */
@@ -40,7 +39,7 @@ Rules:
 - Numbers are plain JSON numbers — no currency symbols, no thousands separators, a period as decimal.
 - "lineItems" is one entry per billed line; "amount" is that line's total.
 - "subtotal" is the pre-tax sum; "total" is the final amount due. Transcribe the numbers as printed, even if they don't add up — a downstream checker flags inconsistencies. Do NOT silently fix the math.
-- Each line item needs a "sku": use the printed code if present, else a short slug of the description.`;
+- Each line item needs a "sku": copy the printed item/SKU code for that line EXACTLY as shown (e.g. the "Item" column). Only if no code is printed, fall back to a short slug of the description.`;
 
 class MissingApiKeyError extends Error {
   constructor() {

lib/intake.test.ts

@@ -0,0 +1,114 @@
+import { test } from "node:test";
+import assert from "node:assert/strict";
+import { runIntake, type Extractor } from "./intake";
+import type { Invoice } from "./schema";
+
+/**
+ * Intake tests — the "mock vision" coverage. The real extraction calls the
+ * Anthropic vision API; here we inject a mock extractor so the whole intake path
+ * (render → extract → reconcile-with-record / fail) is exercised offline.
+ */
+
+const SOURCE: Invoice = {
+  invoiceNumber: "INV-9001",
+  poNumber: "PO-9001",
+  vendor: "Acme Corp",
+  issueDate: "2026-05-01",
+  currency: "USD",
+  lineItems: [
+    { sku: "A-1", description: "Widget", qty: 2, unitPrice: 10, amount: 20 },
+  ],
+  subtotal: 20,
+  tax: null,
+  total: 20,
+};
+
+// A render stub so no PDF is actually generated.
+const render = async () => "ZmFrZS1wZGY=";
+
+/** An extractor that returns a given invoice (or failure). */
+function mockExtractor(invoice: Invoice): Extractor {
+  return async () => ({ ok: true, invoice });
+}
+
+test("successful extraction returns the extracted invoice", async () => {
+  const extracted: Invoice = { ...SOURCE, vendor: "Acme Corporation Ltd" };
+  const res = await runIntake(SOURCE, {
+    extract: mockExtractor(extracted),
+    render,
+  });
+  assert.equal(res.ok, true);
+  if (res.ok) {
+    assert.equal(res.invoice.vendor, "Acme Corporation Ltd");
+    // Header reconciles (invoiceNumber + poNumber + total all match the record).
+    assert.equal(res.matchesRecord, true);
+  }
+});
+
+test("extracted header differing from the record → matchesRecord false", async () => {
+  const extracted: Invoice = { ...SOURCE, total: 999 };
+  const res = await runIntake(SOURCE, {
+    extract: mockExtractor(extracted),
+    render,
+  });
+  assert.equal(res.ok, true);
+  if (res.ok) assert.equal(res.matchesRecord, false);
+});
+
+test("the pipeline runs on the EXTRACTED data, not the source", async () => {
+  // The model reads a different unit price than the record — runIntake must
+  // return the model's number (that's the whole point: data comes from the read).
+  const extracted: Invoice = {
+    ...SOURCE,
+    lineItems: [{ ...SOURCE.lineItems[0]!, unitPrice: 11, amount: 22 }],
+    subtotal: 22,
+    total: 22,
+  };
+  const res = await runIntake(SOURCE, {
+    extract: mockExtractor(extracted),
+    render,
+  });
+  assert.equal(res.ok, true);
+  if (res.ok) assert.equal(res.invoice.lineItems[0]?.unitPrice, 11);
+});
+
+test("a validation failure surfaces as a failure (no fabricated data)", async () => {
+  const res = await runIntake(SOURCE, {
+    extract: async () => ({
+      ok: false,
+      kind: "validation",
+      issues: ["total: expected a number"],
+    }),
+    render,
+  });
+  assert.equal(res.ok, false);
+  if (!res.ok) assert.match(res.reason, /validation/i);
+});
+
+test("a refusal surfaces as a failure", async () => {
+  const res = await runIntake(SOURCE, {
+    extract: async () => ({ ok: false, kind: "refusal" }),
+    render,
+  });
+  assert.equal(res.ok, false);
+});
+
+test("a timeout surfaces as a failure (does not hang)", async () => {
+  const res = await runIntake(SOURCE, {
+    // Never resolves → the internal timeout must win.
+    extract: () => new Promise(() => {}),
+    render,
+    timeoutMs: 20,
+  });
+  assert.equal(res.ok, false);
+});
+
+test("a thrown extractor is caught, not propagated", async () => {
+  const res = await runIntake(SOURCE, {
+    extract: async () => {
+      throw new Error("network down");
+    },
+    render,
+  });
+  assert.equal(res.ok, false);
+});

lib/intake.ts

@@ -0,0 +1,97 @@
+import type { Invoice } from "./schema";
+import { extractInvoice, type ExtractionResult } from "./extract";
+import { renderInvoicePdfBase64 } from "./invoice-pdf";
+
+/**
+ * The intake core — render the source document to a PDF, read it back with the
+ * vision model, and return the structured invoice the pipeline will run on.
+ *
+ * This is what makes the extraction REAL: the downstream matching runs on what
+ * the model extracted, not on the seeded record. Like production — the document
+ * is the source of truth; if the read fails, you don't invent data, the run
+ * can't proceed (the caller surfaces an error). The seeded record is only the
+ * thing we render the PDF from (our stand-in for "a vendor PDF arrived").
+ *
+ * The extractor is injectable so tests can run the whole pipeline offline with a
+ * mock instead of a live vision call.
+ */
+
+/** Render `source` to a PDF and extract it; same tagged result as `extractInvoice`. */
+export type Extractor = (pdfBase64: string) => Promise<ExtractionResult>;
+
+interface IntakeOk {
+  ok: true;
+  /** The invoice the pipeline runs on (the model's output). */
+  invoice: Invoice;
+  /** True when the extracted header reconciles with the source record. */
+  matchesRecord: boolean;
+}
+interface IntakeFail {
+  ok: false;
+  reason: string;
+}
+export type IntakeResult = IntakeOk | IntakeFail;
+
+const DEFAULT_TIMEOUT_MS = 20_000;
+
+/**
+ * Read `source` (the seeded record we render to a PDF) into a structured invoice.
+ * Returns the extracted invoice on success, or a failure the caller turns into an
+ * error trace event. `extract` + `render` are injectable for offline tests.
+ */
+export async function runIntake(
+  source: Invoice,
+  opts: {
+    extract?: Extractor;
+    render?: (inv: Invoice) => Promise<string>;
+    timeoutMs?: number;
+  } = {},
+): Promise<IntakeResult> {
+  const extract = opts.extract ?? extractInvoice;
+  const render = opts.render ?? renderInvoicePdfBase64;
+  const timeoutMs = opts.timeoutMs ?? DEFAULT_TIMEOUT_MS;
+
+  let result: ExtractionResult;
+  try {
+    const pdfBase64 = await render(source);
+    result = await Promise.race([
+      extract(pdfBase64),
+      new Promise<ExtractionResult>((resolve) =>
+        setTimeout(
+          () =>
+            resolve({
+              ok: false,
+              kind: "api_error",
+              message: "extraction timed out",
+            }),
+          timeoutMs,
+        ),
+      ),
+    ]);
+  } catch {
+    return { ok: false, reason: "Could not read the document." };
+  }
+
+  if (!result.ok) {
+    return {
+      ok: false,
+      reason:
+        result.kind === "validation"
+          ? "Extracted data failed validation."
+          : result.kind === "refusal"
+            ? "The model declined to read the document."
+            : result.kind === "no_json"
+              ? "The model returned no structured data."
+              : "Could not read the document.",
+    };
+  }
+
+  // Did the extracted header reconcile with the source record on the key fields?
+  // (A clean signal for the reveal; the line-level match is the matching step.)
+  const matchesRecord =
+    result.invoice.invoiceNumber === source.invoiceNumber &&
+    (result.invoice.poNumber ?? null) === (source.poNumber ?? null) &&
+    Math.abs(result.invoice.total - source.total) < 0.01;
+
+  return { ok: true, invoice: result.invoice, matchesRecord };
+}

lib/invoice-pdf.ts

@@ -77,11 +77,16 @@ export async function renderInvoicePdf(invoice: Invoice): Promise<Uint8Array> {
   }
 
   // ── Line-item table ───────────────────────────────────────────────────────
+  // The SKU column is printed because matching joins invoice ↔ PO ↔ receipt by
+  // SKU — the extraction has to be able to read the item code off the document
+  // (a real invoice carries one), otherwise the downstream match has no key.
   y += 24;
-  const colDesc = MARGIN;
-  const colQty = 330;
-  const colUnit = 400;
+  const colSku = MARGIN;
+  const colDesc = MARGIN + 110;
+  const colQty = 360;
+  const colUnit = 420;
   const colAmt = PAGE_W - MARGIN;
+  draw("Item", colSku, y, bold, 9, muted);
   draw("Description", colDesc, y, bold, 9, muted);
   draw("Qty", colQty, y, bold, 9, muted);
   draw("Unit price", colUnit, y, bold, 9, muted);
@@ -96,6 +101,7 @@ export async function renderInvoicePdf(invoice: Invoice): Promise<Uint8Array> {
   y += 18;
 
   for (const li of invoice.lineItems) {
+    draw(li.sku, colSku, y, font, 9);
     draw(li.description, colDesc, y, font, 10);
     draw(String(li.qty), colQty, y, font, 10);
     draw(money(li.unitPrice, invoice.currency), colUnit, y, font, 10);

lib/trace.ts

@@ -241,22 +241,26 @@ export function toTraceEvent(
         }
 
         if (innerType === "intake-result") {
-          // The extraction result (the extracted invoice + whether it reconciled
-          // with the record), written when intake finishes. Upserts the same
-          // intake node. A failed extraction (extracted = null) is a soft warn.
-          const extracted = innerPayload?.["extracted"] ?? null;
+          // The intake result (`runIntake`): on success the extracted invoice +
+          // whether its header reconciled with the record; on failure a reason.
+          // Upserts the same intake node.
+          const ok = innerPayload?.["ok"] === true;
+          const extracted = ok ? (innerPayload?.["invoice"] ?? null) : null;
+          const matches = innerPayload?.["matchesRecord"] === true;
           return {
             kind: "step",
             stage: "intake",
-            status: extracted ? "ok" : "warn",
+            status: ok ? "ok" : "error",
             stepId: "intake",
-            label: extracted ? "Intake — extracted" : "Intake — using record",
-            detail: extracted
-              ? innerPayload?.["matches"] === true
-                ? "Extracted and reconciled with the PO record."
-                : "Extracted; key fields differ from the record (pipeline uses the record)."
-              : "Couldn't extract the document; proceeding on the seeded record.",
-            data: { extracted, matches: innerPayload?.["matches"] === true },
+            label: ok ? "Intake — extracted" : "Intake — failed",
+            detail: ok
+              ? matches
+                ? "Read the document and reconciled it with the PO record."
+                : "Read the document; header differs from the PO record."
+              : typeof innerPayload?.["reason"] === "string"
+                ? (innerPayload["reason"] as string)
+                : "Could not read the document.",
+            data: { extracted, matches },
           };
         }
 

src/mastra/testing/pipeline-trace.test.ts

@@ -58,6 +58,10 @@ async function runTrace(mastra: Mastra, b: SeedBundle) {
       purchaseOrder: b.purchaseOrder ?? null,
       goodsReceipt: b.goodsReceipt ?? null,
       priorInvoiceNumbers,
+      // Skip the intake vision call — this test runs offline (no API key) and is
+      // about the matching → investigation → routing wiring, not extraction.
+      // The intake step + runIntake have their own test (lib/intake.test.ts).
+      skipExtraction: true,
     },
   });
   const raw: unknown[] = [];