Make extraction the first step of the Mastra workflow

Extraction was running in the API route before the workflow — an arbitrary split
that left "extraction" outside the orchestration. It's now the workflow's intake
step, so the pipeline is one Mastra flow end to end (PDF → extract → match →
investigate → approve → reconcile).

- intakeStep renders the PDF + runs vision extraction, writes an intake-document
  chunk early so the reveal shows the page immediately, and passes the input
  through unchanged (the pipeline still runs on the trusted seeded record).
- Fails open: a timeout/error proceeds on the record (now proven offline too,
  since the mock-model integration test runs intake with no API key).
- A phase-2 resume sets skipExtraction so approve/reject doesn't re-run the
  vision call (3s vs 15s).
- Route no longer does extraction; trace adapter surfaces the intake step's
  document + extracted result. README/comments updated — nothing says extraction
  is outside Mastra.

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

Author: DylanMerigaud

README.md

@@ -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.** 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`). Here's the deliberate part: the extracted invoice is *shown* (the reveal proves the read happened), but the downstream pipeline runs 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 run notes it and proceeds on the record. In production you'd reconcile extraction against the PO and surface the diff; the seam is the same.
+**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.
 
 **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"`).
 

app/api/run/route.ts

@@ -3,9 +3,6 @@ import { loadRunBundle } from "@/db/client";
 import { toTraceEvent, pipelineErrorEvent, type TraceEvent } from "@/lib/trace";
 import { ndjsonLine } from "@/lib/ndjson";
 import { checkRateLimit, clientIpFrom } from "@/lib/ratelimit";
-import { renderInvoicePdfBase64 } from "@/lib/invoice-pdf";
-import { extractInvoice } from "@/lib/extract";
-import type { Invoice } from "@/lib/schema";
 import {
   RunRequest,
   STREAM_CONTENT_TYPE,
@@ -43,86 +40,6 @@ function line(obj: unknown): Uint8Array {
   return new TextEncoder().encode(ndjsonLine(obj));
 }
 
-/** Hard cap on the extraction call so a slow model never stalls the demo. */
-const EXTRACTION_TIMEOUT_MS = 20_000;
-
-/**
- * Phase-0 intake extraction. Renders the seeded invoice to a PDF, has the model
- * read it back, and emits intake trace events carrying the extracted invoice (for
- * the reveal) — then leaves the pipeline to run on the trusted seed. Never throws:
- * any failure becomes a "could not extract, proceeding on record" note.
- */
-async function emitExtraction(
-  emit: (e: Omit<TraceEvent, "seq" | "atMs">) => void,
-  seed: Invoice,
-): Promise<void> {
-  emit({
-    kind: "step",
-    stage: "intake",
-    status: "running",
-    stepId: "intake",
-    label: "Intake — reading invoice PDF",
-    detail: "Extracting the document with the vision model…",
-    // Carry the document so the reveal can render + scan it immediately, before
-    // the model returns. (This is the on-screen twin of the PDF being read.)
-    data: { document: seed },
-  });
-
-  try {
-    const pdfBase64 = await renderInvoicePdfBase64(seed);
-    const result = await Promise.race([
-      extractInvoice(pdfBase64),
-      new Promise<{ ok: false; kind: "timeout" }>((resolve) =>
-        setTimeout(
-          () => resolve({ ok: false, kind: "timeout" }),
-          EXTRACTION_TIMEOUT_MS,
-        ),
-      ),
-    ]);
-
-    if (result.ok) {
-      // Did the model's read agree with the seeded record on the key fields the
-      // pipeline routes on? (It drives a "validated against PO record" note.)
-      const matches =
-        result.invoice.invoiceNumber === seed.invoiceNumber &&
-        (result.invoice.poNumber ?? null) === (seed.poNumber ?? null) &&
-        Math.abs(result.invoice.total - seed.total) < 0.01;
-      emit({
-        kind: "step",
-        stage: "intake",
-        status: "ok",
-        stepId: "intake",
-        label: "Intake — extracted",
-        detail: matches
-          ? "Extracted and reconciled with the PO record."
-          : "Extracted; key fields differ from the record (pipeline uses the record).",
-        data: { extracted: result.invoice, matches },
-      });
-      return;
-    }
-
-    // Extraction failed/timed out — proceed on the seed.
-    emit({
-      kind: "step",
-      stage: "intake",
-      status: "warn",
-      stepId: "intake",
-      label: "Intake — using record",
-      detail:
-        "Couldn't extract the document just now; proceeding on the seeded record.",
-    });
-  } catch {
-    emit({
-      kind: "step",
-      stage: "intake",
-      status: "warn",
-      stepId: "intake",
-      label: "Intake — using record",
-      detail: "Extraction unavailable; proceeding on the seeded record.",
-    });
-  }
-}
-
 export async function POST(request: Request): Promise<Response> {
   // 0. Rate-limit by IP first — before any work or model calls. The demo is
   //    public and each run spends Anthropic tokens, so this caps abuse. Fails
@@ -201,17 +118,6 @@ export async function POST(request: Request): Promise<Response> {
         controller.enqueue(line(stamp(e)));
 
       try {
-        // Phase 0: INTAKE EXTRACTION. Only on the first request (not the
-        // Approve/Reject resume — re-extracting then would be wasted work). The
-        // agent reads the rendered invoice PDF and we surface what it extracted.
-        // The extracted invoice is shown to prove the read happened; the pipeline
-        // below runs on the TRUSTED seeded record, so a model misread degrades the
-        // reveal, never the deterministic verdicts. Fails open: if extraction
-        // errors or times out, we note it and proceed on the seed.
-        if (decision === undefined) {
-          await emitExtraction(emit, bundle.invoice);
-        }
-
         const workflow = mastra.getWorkflow("p2p");
         const run = await workflow.createRun();
         const result = run.stream({
@@ -221,6 +127,10 @@ export async function POST(request: Request): Promise<Response> {
             goodsReceipt: bundle.goodsReceipt,
             priorInvoiceNumbers: bundle.priorInvoiceNumbers,
             humanApproval,
+            // On a phase-2 resume (approve/reject) the workflow re-runs from the
+            // top, but the document was already read — skip the (costly) vision
+            // call the second time. The reveal from phase 1 still stands.
+            skipExtraction: decision !== undefined,
           },
         });
 

lib/trace.ts

@@ -222,6 +222,20 @@ export function toTraceEvent(
           };
         }
 
+        if (innerType === "intake-document") {
+          // The intake step writes this the instant it starts so the reveal can
+          // show the document + scan before the vision model returns.
+          return {
+            kind: "step",
+            stage: "intake",
+            status: "running",
+            stepId: "intake",
+            label: "Intake — reading invoice PDF",
+            detail: "Extracting the document with the vision model…",
+            data: { document: asRecord(innerPayload?.["document"]) },
+          };
+        }
+
         return null;
       }
 
@@ -245,14 +259,31 @@ export function toTraceEvent(
       case "workflow-step-finish": {
         if (isMappingStep(stepId)) return null; // hide the .map() plumbing step
         const rawOut = asRecord(payload?.["output"]);
-        // Steps emit a domain object plus a `narration` string, and some wrap the
-        // domain object (the approval step outputs `{ decision, match, vendor }`).
-        // `domain` is the object the UI should colour + render; `narration` is the
-        // agent's prose. Unwrapping here keeps the UI decoupled from step shapes.
         const narration =
           typeof rawOut?.["narration"] === "string"
             ? (rawOut["narration"] as string)
             : undefined;
+
+        // Intake carries the extraction result for the reveal (extracted invoice
+        // + whether it reconciled with the record). Surface just that as the
+        // node's data; a failed extraction (extracted = null) is a soft warn.
+        if (stage === "intake") {
+          const extracted = rawOut?.["extracted"];
+          return {
+            kind: "step",
+            stage: "intake",
+            status: extracted ? "ok" : "warn",
+            stepId: "intake",
+            label: extracted ? "Intake — extracted" : "Intake — using record",
+            detail: narration,
+            data: { extracted, matches: rawOut?.["matches"] === true },
+          };
+        }
+
+        // Other steps emit a domain object plus a `narration` string, and some
+        // wrap the domain object (approval outputs `{ decision, match, vendor }`).
+        // `domain` is the object the UI colours + renders. Unwrapping here keeps
+        // the UI decoupled from step shapes.
         const domain = unwrapStageData(rawOut);
         return {
           kind: "step",

src/mastra/workflows/p2p.ts

@@ -14,6 +14,8 @@ import { runMatch } from "@/lib/matching";
 import { routeApproval } from "@/lib/policy";
 import { reconcile } from "@/lib/erp";
 import { runInvestigation, type InvestigatorAgent } from "@/lib/investigation";
+import { renderInvoicePdfBase64 } from "@/lib/invoice-pdf";
+import { extractInvoice } from "@/lib/extract";
 import { CTX } from "../tools/context";
 
 /**
@@ -47,15 +49,100 @@ const RunInput = z.object({
   goodsReceipt: GoodsReceipt.nullable(),
   priorInvoiceNumbers: z.array(z.string()),
   humanApproval: z.enum(["pending", "approve", "reject"]).default("pending"),
+  /* On a phase-2 resume the document was already read in phase 1 — skip the
+     vision call so an approve/reject doesn't re-extract (wasted cost + latency). */
+  skipExtraction: z.boolean().default(false),
 });
 
 /* A narration field every stage adds for the trace. */
 const Narrated = z.object({ narration: z.string() });
 
-/* ── Step 1: Matching ───────────────────────────────────────────────────────
-   The first workflow step. (Intake — the document extraction — runs in the route
-   before the workflow and is surfaced on the trace there; see app/api/run.) The
-   authoritative verdict comes from the pure matcher; its `verdict` drives the
+/* Hard cap on the extraction call so a slow model never stalls a run. */
+const EXTRACTION_TIMEOUT_MS = 20_000;
+
+/* ── Step 1: Intake (document extraction) ───────────────────────────────────
+   The first workflow step: render the invoice to a PDF and have the vision model
+   read it back into structured data. This is one of the two AI touch-points (the
+   other is the exception investigator); the rest of the pipeline is deterministic.
+
+   Deliberate split: the extraction is SHOWN (the reveal proves the read happened),
+   but the pipeline downstream runs on the TRUSTED seeded record, so a model
+   misread degrades the reveal, never the verdicts — the seeded edge cases stay
+   reliable. So intake passes the input through UNCHANGED to matching, and only
+   surfaces what it extracted for the trace. Fails open: a timeout/error just
+   notes it and proceeds on the record.
+
+   The reveal needs the document on screen the instant the run starts (before the
+   model returns), so we write an `intake-document` chunk early via the stream
+   writer; the step's final output then carries `{ extracted, matches }`. */
+const IntakeOut = RunInput.merge(Narrated).merge(
+  z.object({
+    extracted: Invoice.nullable(),
+    matches: z.boolean(),
+  }),
+);
+const intakeStep = createStep({
+  id: "intake",
+  inputSchema: RunInput,
+  outputSchema: IntakeOut,
+  execute: async ({ inputData, writer }) => {
+    const seed = inputData.invoice;
+
+    // Phase-2 resume: the document was already read in phase 1, so skip the vision
+    // call and pass straight through (the reveal from phase 1 still stands).
+    if (inputData.skipExtraction) {
+      return { ...inputData, extracted: null, matches: false, narration: "" };
+    }
+
+    // Show the document immediately (the on-screen twin of the PDF being read).
+    try {
+      await writer?.write({
+        type: "intake-document",
+        payload: { document: seed },
+      });
+    } catch {
+      /* ignore writer errors — never let the trace affect the result */
+    }
+
+    let extracted: Invoice | null = null;
+    let matches = false;
+    let narration =
+      "Couldn't extract the document just now; proceeding on the seeded record.";
+    try {
+      const pdfBase64 = await renderInvoicePdfBase64(seed);
+      const result = await Promise.race([
+        extractInvoice(pdfBase64),
+        new Promise<{ ok: false; kind: "timeout" }>((resolve) =>
+          setTimeout(
+            () => resolve({ ok: false, kind: "timeout" }),
+            EXTRACTION_TIMEOUT_MS,
+          ),
+        ),
+      ]);
+      if (result.ok) {
+        extracted = result.invoice;
+        // Did the model's read agree with the seeded record on the fields the
+        // pipeline routes on? (Drives the "reconciled with PO record" note.)
+        matches =
+          result.invoice.invoiceNumber === seed.invoiceNumber &&
+          (result.invoice.poNumber ?? null) === (seed.poNumber ?? null) &&
+          Math.abs(result.invoice.total - seed.total) < 0.01;
+        narration = matches
+          ? "Extracted and reconciled with the PO record."
+          : "Extracted; key fields differ from the record (pipeline uses the record).";
+      }
+    } catch {
+      /* fail open — narration already set to the fallback */
+    }
+
+    // Pass the input through unchanged (matching runs on the trusted record);
+    // `extracted`/`matches` ride along only for the trace/reveal.
+    return { ...inputData, extracted, matches, narration };
+  },
+});
+
+/* ── Step 2: Matching ───────────────────────────────────────────────────────
+   The authoritative verdict comes from the pure matcher; its `verdict` drives the
    branch below. We carry the vendor + the reviewer's decision forward. */
 const HumanApproval = z.object({
   humanApproval: z.enum(["pending", "approve", "reject"]),
@@ -65,7 +152,7 @@ const MatchStepOut = MatchResult.merge(Narrated)
   .merge(HumanApproval);
 const matchingStep = createStep({
   id: "matching",
-  inputSchema: RunInput,
+  inputSchema: IntakeOut,
   outputSchema: MatchStepOut,
   execute: async ({ inputData }) => {
     const match = runMatch({
@@ -252,6 +339,7 @@ export const p2pWorkflow = createWorkflow({
   inputSchema: RunInput,
   outputSchema: ReconResult.merge(Narrated),
 })
+  .then(intakeStep)
   .then(matchingStep)
   .branch([
     // Exception → investigate (agent) then route to Approval.