fix(run-engine): make both envelope arms honour one omission contract

The store arm cannot return a pending waitpoint, because a pending one has no completion to
read. The legacy arm read rows by id with no status filter, so it could hand back an envelope
for a PENDING waitpoint with completedAt defaulted to now. The resolver's coverage check reads
an omission as "fail loud", so the arms disagreeing there would turn a pending waitpoint into
a resumable one. Filters to COMPLETED.

Also states why the ref branch precedes the RUN branch, which is the opposite order to the
reference implementation in the freeze test. Both are byte-identical at read time by that
reference's own reasoning, and this order needs no Postgres read to recover a string already
in hand — and keeps an offloaded RUN success resolvable when the completing run row is gone,
which now refuses rather than resolving empty. Adds the offloaded-RUN-success case that both
suites were missing.

Author: d-cs

internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointEquivalence.test.ts

@@ -239,6 +239,26 @@ describe("the resolver reproduces the existing hydration", () => {
     expect(actual).toEqual(expected);
   });
 
+  // The case the suite was blind to, and the one the frozen reference orders the other way. The
+  // oracle emits the ref string; so does this, by a different branch.
+  it("for an offloaded RUN success", async () => {
+    const { expected, actual } = await bothPaths(
+      [
+        pair({
+          id: "wp_run_ref",
+          type: "RUN",
+          output: "s3://bucket/key",
+          outputType: "application/store",
+          completedByTaskRunId: CHILD_RUN_ID,
+        }),
+      ],
+      []
+    );
+
+    expect(actual).toEqual(expected);
+    expect(actual[0]?.output).toBe("s3://bucket/key");
+  });
+
   it("for one run present at two batch indexes", async () => {
     const { expected, actual } = await bothPaths(
       [

internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointRecords.test.ts

@@ -83,6 +83,22 @@ describe("buildCompletedWaitpointRecords", () => {
       expect(record?.output).toEqual({ ref: "store-key-1" });
     });
 
+    // Deliberately a ref, not deriveFromRun, and the opposite of the reference implementation in
+    // completedWaitpointFreeze.test.ts. Byte-identical either way, and this route stays
+    // resolvable when the completing run row is gone.
+    it("routes an offloaded RUN success down the ref branch", () => {
+      const [record] = buildCompletedWaitpointRecords([
+        source({
+          type: "RUN",
+          outputRef: "s3://bucket/key",
+          outputType: "application/store",
+          completedByTaskRunId: "run_1",
+        }),
+      ]);
+
+      expect(record?.output).toEqual({ ref: "s3://bucket/key" });
+    });
+
     it("marks a plain RUN output as derivable from the run", () => {
       const [record] = buildCompletedWaitpointRecords([
         source({ type: "RUN", output: '{"ok":true}', completedByTaskRunId: "run_1" }),

internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointRecords.ts

@@ -38,6 +38,13 @@ export function buildCompletedWaitpointRecords(
 }
 
 function chooseOutput(source: CompletionEnvelopeSource): CompletedWaitpointRecordOutput {
+  // Ref BEFORE the RUN branch, which is the opposite order to the reference implementation in
+  // completedWaitpointFreeze.test.ts. Both are byte-identical at read time, by that reference's
+  // own reasoning: an offloaded RUN success has the same ref string in TaskRun.output. This
+  // order is preferred because it needs no Postgres read to recover a string already in hand,
+  // and because a deriveFromRun record whose run row is later deleted now refuses rather than
+  // resolving empty — so routing an offloaded RUN success down the ref branch keeps it
+  // resolvable when that row is gone.
   if (source.outputRef !== undefined) {
     return { ref: source.outputRef };
   }

internal-packages/run-engine/src/engine/waitpointCoordinator/completionEnvelopeSource.test.ts

@@ -98,6 +98,16 @@ describe("envelopeSourceFromWaitpointRow", () => {
     });
   });
 
+  // The mapper itself is status-blind by design: the legacy arm filters to COMPLETED before
+  // calling it, so both arms omit a pending waitpoint rather than describing one. This states
+  // that the mapper is not where that decision lives.
+  it("does not itself inspect status", () => {
+    const source = envelopeSourceFromWaitpointRow(row({ status: "PENDING", completedAt: null }));
+
+    expect(source.id).toBe("wp_1");
+    expect(source.completedAt).toBeInstanceOf(Date);
+  });
+
   it("carries the RUN and BATCH back-references", () => {
     expect(
       envelopeSourceFromWaitpointRow(row({ type: "RUN", completedByTaskRunId: "run_child" }))

internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts

@@ -107,7 +107,12 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator
 
     const rows = await fetchWaitpointsInChunks(this.prisma, waitpointIds, this.runStore, runId);
 
-    return rows.map(envelopeSourceFromWaitpointRow);
+    // COMPLETED only, so both arms honour one omission contract. The store arm cannot return a
+    // pending waitpoint because a pending one has no completion to read; this arm reads rows by
+    // id and would otherwise hand back an envelope with completedAt defaulted to now. The
+    // resolver's coverage check reads an omission as "fail loud", so the two arms disagreeing
+    // here would turn a pending waitpoint into a resumable one.
+    return rows.filter((row) => row.status === "COMPLETED").map(envelopeSourceFromWaitpointRow);
   }
 
   async registerBlocks({