Stop threading the extraction result through the pipeline

The intake step put extracted/matches into its output, so every downstream step's
input schema carried fields none of them read (matching runs on the trusted
record). That phantom data suggested a link that doesn't exist.

Intake now writes the document + extraction result to the trace via the stream
writer (intake-document / intake-result chunks) and outputs the run input
unchanged, so matching receives exactly what it consumes. The trace adapter owns
the intake node from those chunks; intake's bare step lifecycle events are dropped.

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

Author: DylanMerigaud

lib/trace.ts

@@ -173,6 +173,10 @@ export function toTraceEvent(
 
       case "workflow-step-start":
         if (isMappingStep(stepId)) return null; // hide the .map() plumbing step
+        // Intake owns its node via the intake-document/intake-result chunks it
+        // writes (they carry the document + extraction result); its bare
+        // lifecycle events would be dataless duplicates, so drop them.
+        if (stage === "intake") return null;
         return {
           kind: "step",
           stage,
@@ -236,6 +240,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;
+          return {
+            kind: "step",
+            stage: "intake",
+            status: extracted ? "ok" : "warn",
+            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 },
+          };
+        }
+
         return null;
       }
 
@@ -258,29 +282,16 @@ export function toTraceEvent(
       case "workflow-step-result":
       case "workflow-step-finish": {
         if (isMappingStep(stepId)) return null; // hide the .map() plumbing step
+        // Intake's node is owned by its intake-document/intake-result chunks; its
+        // step-result is a bare passthrough (the run input) with nothing to show.
+        if (stage === "intake") return null;
         const rawOut = asRecord(payload?.["output"]);
         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
+        // 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.

src/mastra/workflows/p2p.ts

@@ -73,44 +73,38 @@ const EXTRACTION_TIMEOUT_MS = 20_000;
    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(),
-  }),
-);
+   model returns), and the final extraction result, are written to the TRACE via
+   the stream writer (an `intake-document` then an `intake-result` chunk) — NOT
+   threaded into the step output, because the downstream steps don't consume them
+   (they run on the trusted record). So intake's output is the run input passed
+   through unchanged: matching receives exactly what it uses, no phantom fields. */
 const intakeStep = createStep({
   id: "intake",
   inputSchema: RunInput,
-  outputSchema: IntakeOut,
+  outputSchema: RunInput,
   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: "" };
-    }
+    if (inputData.skipExtraction) return inputData;
+
+    const emit = async (chunk: unknown) => {
+      try {
+        await writer?.write(chunk);
+      } catch {
+        /* ignore writer errors — never let the trace affect the result */
+      }
+    };
 
     // 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 */
-    }
+    await emit({ type: "intake-document", payload: { document: seed } });
 
-    let extracted: Invoice | null = null;
-    let matches = false;
-    let narration =
-      "Couldn't extract the document just now; proceeding on the seeded record.";
+    let result: { extracted: Invoice; matches: boolean } | { extracted: null } =
+      { extracted: null };
     try {
       const pdfBase64 = await renderInvoicePdfBase64(seed);
-      const result = await Promise.race([
+      const out = await Promise.race([
         extractInvoice(pdfBase64),
         new Promise<{ ok: false; kind: "timeout" }>((resolve) =>
           setTimeout(
@@ -119,25 +113,22 @@ const intakeStep = createStep({
           ),
         ),
       ]);
-      if (result.ok) {
-        extracted = result.invoice;
+      if (out.ok) {
         // 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).";
+        const matches =
+          out.invoice.invoiceNumber === seed.invoiceNumber &&
+          (out.invoice.poNumber ?? null) === (seed.poNumber ?? null) &&
+          Math.abs(out.invoice.total - seed.total) < 0.01;
+        result = { extracted: out.invoice, matches };
       }
     } catch {
-      /* fail open — narration already set to the fallback */
+      /* fail open — result stays { extracted: null } */
     }
 
-    // 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 };
+    await emit({ type: "intake-result", payload: result });
+    // Matching runs on the trusted record — pass the input through unchanged.
+    return inputData;
   },
 });
 
@@ -152,7 +143,7 @@ const MatchStepOut = MatchResult.merge(Narrated)
   .merge(HumanApproval);
 const matchingStep = createStep({
   id: "matching",
-  inputSchema: IntakeOut,
+  inputSchema: RunInput,
   outputSchema: MatchStepOut,
   execute: async ({ inputData }) => {
     const match = runMatch({