fix(run-engine): enforce the writer-client rule, and close the deferred-output fail-open

An exhaustive walk of the deferred-output state space found the mechanism
sound, and asked for one change plus two cheap ones.

The required-writer rule could not be enforced by the type system. ReadClient
admits both a writer and a replica, and the two are structurally identical --
separable only by a runtime brand -- so a caller passing readOnlyPrisma would
type-check and quietly reinstate the replica-lag window that turns a committed
child output into a refused resume. The reader now asserts the brand. It is
built once at wiring time, so the check costs nothing per read.

A record marked as deferring its output while carrying no run id was the one
place this design failed OPEN rather than loud: it resolved the waitpoint with
no output and no error. Unreachable from the current record build, which
requires the run id before it marks anything derivable, but a reordering of
those conditions is all it would take. It now throws, like a record arriving
with no reader wired.

Also pins the orphan case against the oracle: a RUN waitpoint whose completing
run is gone keeps its own output inline and must omit the run sub-object, which
only the differential comparison catches.

Records the premise the deferred read rests on: the completing run's output
still holds what the waitpoint was completed with. Nothing overwrites it today,
and if that changes the divergence would be silent rather than loud.

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

Author: d-cs

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

@@ -236,6 +236,29 @@ describe("the resolver reproduces the existing hydration", () => {
   // 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.
+  // An orphan: the completing run is gone, so onDelete: SetNull cleared the back-reference while
+  // the waitpoint kept its output. The record must stay inline -- derivable would send the
+  // resolver to a run that no longer exists -- and the entry must omit completedByTaskRun, which
+  // only the oracle comparison pins.
+  postgresTest("for an orphaned RUN waitpoint that kept its output", async ({ prisma }) => {
+    const { expected, actual } = await bothPaths(
+      prisma,
+      [
+        pair({
+          id: "wp_run_orphan",
+          type: "RUN",
+          output: '{"orphan":true}',
+          completedByTaskRunId: null,
+        }),
+      ],
+      ["wp_run_orphan"]
+    );
+
+    expect(actual).toEqual(expected);
+    expect(actual[0]?.output).toBe('{"orphan":true}');
+    expect(actual[0]?.completedByTaskRun).toBeUndefined();
+  });
+
   postgresTest("for a RUN waitpoint whose child returned no output", async ({ prisma }) => {
     const childRunId = await seedChildRunWithOutput(prisma, null);
     const { expected, actual } = await bothPaths(

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

@@ -4,7 +4,7 @@
 // by a pure test: the claim is that TaskRun.output holds the same string the waitpoint carried,
 // and only a real row can settle that. The pure suite covers everything that does not read.
 import { postgresTest } from "@internal/testcontainers";
-import { PostgresRunStore } from "@internal/run-store";
+import { markReadReplicaClient, PostgresRunStore } from "@internal/run-store";
 import type { CompletedWaitpointRecord } from "@internal/run-store";
 import type { PrismaClient } from "@trigger.dev/database";
 import { describe, expect } from "vitest";
@@ -248,6 +248,31 @@ describe("the deriveFromRun branch", () => {
     expect(batches).toEqual([]);
   });
 
+  // The required-writer rule is a runtime check because it cannot be a type one: ReadClient admits
+  // a writer and a replica, and they are structurally identical apart from the brand.
+  postgresTest("refuses to be built with a replica client", async ({ prisma }) => {
+    const runStore = new PostgresRunStore({ prisma, readOnlyPrisma: prisma });
+    const replica = markReadReplicaClient({ ...prisma });
+
+    expect(() => createRunOutputsReader(runStore, replica as never)).toThrow(/needs a writer/);
+  });
+
+  // The one place the design could fail OPEN instead of loud: a derive marker with no run to
+  // derive from would otherwise resolve the waitpoint with no output and no error.
+  postgresTest("throws on a derive record carrying no run id", async ({ prisma }) => {
+    const { id: _drop, ...rest } = deriveRecord("run_unused");
+
+    await expect(
+      resolverFor(prisma)({
+        runId: "run_parent",
+        pointer: { cycleSeq: 1, count: 0 },
+        order: [],
+        distinctIds: ["wp_run"],
+        records: [{ ...rest, id: "wp_run", completedByTaskRunId: undefined }],
+      })
+    ).rejects.toThrow(/carries no run id/);
+  });
+
   postgresTest("throws when a derive record arrives with no reader wired", async ({ prisma }) => {
     const runId = await seedChildRunWithOutput(prisma, '{"value":42}');
 

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

@@ -1,3 +1,4 @@
+import { isReadReplicaClient } from "@internal/run-store";
 import type {
   CompletedWaitpointRecord,
   ReadClient,
@@ -76,12 +77,29 @@ const RUN_OUTPUT_CHUNK_SIZE = 100;
  *
  * 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.
+ * window every time a caller omitted it. Hence required, and hence checked: `ReadClient` admits
+ * both a writer and a replica, and the two are structurally identical, separable only by the
+ * runtime brand. A caller passing `readOnlyPrisma` would type-check and quietly reinstate the
+ * window, so the constraint has to be asserted rather than documented. Constructed once at wiring
+ * time, so the assertion costs nothing per read.
+ *
+ * The premise this read rests on: `TaskRun.output` still holds what the waitpoint was completed
+ * with. `completeAttemptSuccess` writes the run's output and the waitpoint's completion from the
+ * same values, and nothing overwrites a completed run's output afterwards. If that ever stops
+ * being true, a deferred record resolves the NEWER value while the legacy path would emit the
+ * frozen one -- and unlike a lost output, that divergence is silent.
  */
 export function createRunOutputsReader(
   runStore: Pick<RunStore, "findRunsByIds">,
   client: ReadClient
 ): (taskRunIds: string[]) => Promise<Map<string, string>> {
+  if (isReadReplicaClient(client)) {
+    throw new Error(
+      "createRunOutputsReader needs a writer: a replica can trail the child's committed output " +
+        "and the resolver would refuse the resume as a lost output."
+    );
+  }
+
   return async (taskRunIds) => {
     const outputs = new Map<string, string>();
 
@@ -295,6 +313,17 @@ function hydrateOutput(
     return record.output.ref;
   }
 
+  // A record marked deriveFromRun with no run to derive from. Unreachable from today's
+  // chooseOutput, which requires the id before it marks anything derivable -- but this is the one
+  // spot where the design could fail OPEN rather than loud, resolving the waitpoint with no output
+  // at all, and a reordering of those conjuncts is all it would take. Treated as the build fault
+  // it would be, like a record arriving with no reader wired.
+  if (record.output !== null && "deriveFromRun" in record.output && !record.completedByTaskRunId) {
+    throw new Error(
+      `Waitpoint ${record.id} defers its output to its completing run, but carries no run id.`
+    );
+  }
+
   const runId = deferredRunIdOf(record);
   if (runId === undefined) {
     return undefined;