fix(run-engine): resume a child that returned no output, and read its output on the writer

Two defects in the deferred-output path, both found in review.

A task that returns nothing completes its waitpoint with no output at all. The
record build still marked it derivable, so the resolver read a null
TaskRun.output and refused the resume as a lost output -- for every
triggerAndWait on a void task. The hydration this replaces resumes cleanly with
no output. A record now defers only when the waitpoint actually carried an
output; one that did, whose run row has since gone, still refuses, which is
what the refusal is for. An empty string stays derivable, because empty is a
value.

A test asserted the broken behaviour was correct, which is why nothing caught
it. It now pins the case it meant to cover: a record that defers to a run whose
output is gone.

The run-output read also needs read-your-writes, and an earlier revision
removed its client to keep the read off the writer. The ordering argument for
that was wrong: the child does commit its output before completing the
waitpoint, but the reader can observe the completion by another route while the
run-output replica still trails, and the output then reads null and the resume
is refused. A hard refusal is worse than a read on the writer, and this read is
bounded and projected to one column. The client is required rather than
optional, because the router turns a missing client into a replica read and an
optional parameter would reintroduce the window whenever a caller omitted it.

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

Author: d-cs

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

@@ -89,7 +89,7 @@ async function bothPaths(
   const runStore = new PostgresRunStore({ prisma, readOnlyPrisma: prisma });
 
   const actual = await createCompletedWaitpointResolver({
-    readRunOutputs: createRunOutputsReader(runStore),
+    readRunOutputs: createRunOutputsReader(runStore, prisma),
   })({
     runId: RUN_ID,
     ...(batchId ? { batchId } : {}),
@@ -232,6 +232,29 @@ describe("the resolver reproduces the existing hydration", () => {
     expect(actual.find((w) => w.id === "wp_run_b")?.output).toBe('{"child":"b"}');
   });
 
+  // A child that returned nothing. The oracle emits `output: w.output ?? undefined`, i.e. resumes
+  // with no output; an earlier revision marked this derivable, read a null TaskRun.output and
+  // refused the resume outright. Comparing against the oracle is what makes that a failure rather
+  // than a design choice, so the case belongs here and not only in the unit suite.
+  postgresTest("for a RUN waitpoint whose child returned no output", async ({ prisma }) => {
+    const childRunId = await seedChildRunWithOutput(prisma, null);
+    const { expected, actual } = await bothPaths(
+      prisma,
+      [
+        pair({
+          id: "wp_run_void",
+          type: "RUN",
+          output: null,
+          completedByTaskRunId: childRunId,
+        }),
+      ],
+      ["wp_run_void"]
+    );
+
+    expect(actual).toEqual(expected);
+    expect(actual[0]?.output).toBeUndefined();
+  });
+
   postgresTest("for a RUN waitpoint read under a batch", async ({ prisma }) => {
     const childRunId = await seedChildRunWithOutput(prisma, '{"ok":true}');
     const { expected, actual } = await bothPaths(

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

@@ -107,6 +107,27 @@ describe("buildCompletedWaitpointRecords", () => {
       expect(record?.output).toEqual({ deriveFromRun: true });
     });
 
+    // A task that returns nothing completes its waitpoint with NO output, so there is nothing to
+    // derive. Marking it derivable made the resolver read a null TaskRun.output and refuse the
+    // resume as a lost output -- for every triggerAndWait on a void task. The hydration this
+    // replaces resumes cleanly with no output, so the record must carry none, not a marker.
+    it("does not defer a RUN whose output is absent", () => {
+      const [record] = buildCompletedWaitpointRecords([
+        source({ type: "RUN", completedByTaskRunId: "run_1", output: undefined }),
+      ]);
+
+      expect(record?.output).toBeNull();
+    });
+
+    // An empty string is a value, so it IS derivable: TaskRun.output holds it verbatim.
+    it("defers a RUN whose output is an empty string", () => {
+      const [record] = buildCompletedWaitpointRecords([
+        source({ type: "RUN", completedByTaskRunId: "run_1", output: "" }),
+      ]);
+
+      expect(record?.output).toEqual({ deriveFromRun: true });
+    });
+
     // TaskRun.error is jsonb and does not round-trip to the same string, so a RUN error can
     // never be re-read from the run row.
     it("keeps a RUN error inline", () => {

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

@@ -49,11 +49,21 @@ function chooseOutput(source: CompletionEnvelopeSource): CompletedWaitpointRecor
     return { ref: source.outputRef };
   }
 
-  // A plain RUN output is re-readable from TaskRun.output verbatim. Two RUN cases are not,
-  // and both must stay inline: an ERROR, because TaskRun.error is jsonb and does not
-  // round-trip to the same string, and an ORPHAN, because the back-reference is
-  // onDelete: SetNull so the completing row may be gone.
-  if (source.type === "RUN" && !source.outputIsError && source.completedByTaskRunId) {
+  // A plain RUN output is re-readable from TaskRun.output verbatim. Three RUN cases are not.
+  // An ERROR, because TaskRun.error is jsonb and does not round-trip to the same string. An
+  // ORPHAN, because the back-reference is onDelete: SetNull so the completing row may be gone.
+  // And an ABSENT output, which is the case that has to be checked here rather than left to the
+  // read: a task that returns nothing completes its waitpoint with no output at all, and there
+  // is then nothing to derive. Marking it derivable makes the resolver read a null TaskRun.output
+  // and refuse the resume as a lost output, where the hydration this replaces resumes cleanly
+  // with no output (`output: w.output ?? undefined`). A waitpoint that DID carry an output whose
+  // run row has since gone still refuses, which is what the refusal is for.
+  if (
+    source.type === "RUN" &&
+    !source.outputIsError &&
+    source.output !== undefined &&
+    source.completedByTaskRunId
+  ) {
     return { deriveFromRun: true };
   }
 

internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointResolver.runOutput.test.ts

@@ -30,7 +30,9 @@ function deriveRecord(completedByTaskRunId: string): CompletedWaitpointRecord {
 
 function resolverFor(prisma: PrismaClient) {
   const runStore = new PostgresRunStore({ prisma, readOnlyPrisma: prisma });
-  return createCompletedWaitpointResolver({ readRunOutputs: createRunOutputsReader(runStore) });
+  return createCompletedWaitpointResolver({
+    readRunOutputs: createRunOutputsReader(runStore, prisma),
+  });
 }
 
 /**
@@ -42,7 +44,7 @@ function resolverFor(prisma: PrismaClient) {
  */
 function countingResolver(prisma: PrismaClient) {
   const runStore = new PostgresRunStore({ prisma, readOnlyPrisma: prisma });
-  const read = createRunOutputsReader(runStore);
+  const read = createRunOutputsReader(runStore, prisma);
   const batches: string[][] = [];
 
   return {
@@ -106,7 +108,14 @@ describe("the deriveFromRun branch", () => {
     expect(failure.reason).toBe("lost-run-output");
   });
 
-  postgresTest("refuses when the run exists with no output", async ({ prisma }) => {
+  // A record that DEFERS to a run whose output is gone still refuses -- that is the case the
+  // refusal exists for, and the record only defers when the waitpoint carried an output.
+  //
+  // This is deliberately reached by hand-building a derive record for an output-less run, a shape
+  // chooseOutput no longer produces. An earlier revision produced it for every non-error RUN
+  // waitpoint, which refused the resume of any task that returns nothing; the test below pins
+  // that case, and this one keeps the refusal itself honest.
+  postgresTest("refuses when a derive record's run output is gone", async ({ prisma }) => {
     const runId = await seedChildRunWithOutput(prisma, null);
 
     const failure = await resolverFor(prisma)({

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

@@ -1,5 +1,6 @@
 import type {
   CompletedWaitpointRecord,
+  ReadClient,
   ResolveCompletedWaitpointsArgs,
   RunStore,
 } from "@internal/run-store";
@@ -64,22 +65,29 @@ const RUN_OUTPUT_CHUNK_SIZE = 100;
  * and it forces `id` into the projection so the map keys correctly even though this select
  * names only `output`.
  *
- * Takes no read client on purpose. The router reads the owning store's REPLICA when no client is
- * passed, and forces its PRIMARY for any client that is not replica-branded -- so accepting one
- * would let a caller put a wide `TaskRun.output` read on the writer by reflex. This read does not
- * need read-your-writes: the child run committed its output before it completed the waitpoint that
- * unblocked this parent, so replica lag cannot hide it. That is the opposite of the envelope read
- * in the legacy arm, which reads a waitpoint completed moments earlier and must use the writer.
+ * The client is REQUIRED, and must be the writer, because this read needs read-your-writes.
+ *
+ * It is tempting to argue the replica is safe here: the child commits its output before it
+ * completes the waitpoint that unblocks this parent, so the write happens first. That ordering is
+ * real but it does not help, because the READER can observe the completion by another route --
+ * from Redis, or from the primary -- while the run-output replica still trails. The output then
+ * reads null and the resolver refuses the resume as a lost output. A hard refusal is a far worse
+ * outcome than a read on the writer, and this read is bounded and projected to one column.
+ *
+ * The router turns any client that is not replica-branded into the owning store's primary, and
+ * turns NO client into its replica -- so an optional parameter would silently reintroduce the lag
+ * window every time a caller omitted it. Hence required.
  */
 export function createRunOutputsReader(
-  runStore: Pick<RunStore, "findRunsByIds">
+  runStore: Pick<RunStore, "findRunsByIds">,
+  client: ReadClient
 ): (taskRunIds: string[]) => Promise<Map<string, string>> {
   return async (taskRunIds) => {
     const outputs = new Map<string, string>();
 
     for (let i = 0; i < taskRunIds.length; i += RUN_OUTPUT_CHUNK_SIZE) {
       const chunk = taskRunIds.slice(i, i + RUN_OUTPUT_CHUNK_SIZE);
-      const rows = await runStore.findRunsByIds(chunk, { select: { output: true } });
+      const rows = await runStore.findRunsByIds(chunk, { select: { output: true } }, client);
 
       for (const [id, row] of rows) {
         // A row present with a null output is the same absence as a missing row: either way the

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

@@ -126,7 +126,7 @@ function harness(prisma: PrismaClient, redisOptions: never): Harness {
       logger: new Logger("roundtrip", "error"),
     }),
     resolve: createCompletedWaitpointResolver({
-      readRunOutputs: createRunOutputsReader(runStore),
+      readRunOutputs: createRunOutputsReader(runStore, prisma),
     }),
     runStore,
   };