feat(run-engine): resolve completed waitpoints from the cycle record set

Rebuilds CompletedWaitpoint[] from a wait cycle's ordered id list and records, field-for-
field equivalent to the existing snapshot hydration, which is what the executor consumes.

It iterates the records, never the order. The order holds only batch-indexed ids, so
iterating it would drop every index-less wait: each wait.for, each single triggerAndWait
and each token. The equivalence suite pins that, and fails on 10 of 12 cases if the
iteration is inverted.

The coverage check is the fail-loud rule. The id classifier is total and never throws, so
an unrecognised shape would otherwise classify as legacy, find no row, and vanish from the
resumed run's completed set. An id that no half resolves throws, and so does an id that
both halves claim.

Author: d-cs

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

@@ -0,0 +1,330 @@
+// The resolver must produce what the executor already consumes, so the oracle is the
+// existing hydration and not a hand-written literal. A literal cannot catch a drift in
+// enhanceExecutionSnapshotWithWaitpoints itself; this can.
+import type { Waitpoint } from "@trigger.dev/database";
+import { describe, expect, it } from "vitest";
+import { enhanceExecutionSnapshotWithWaitpoints } from "../systems/executionSnapshotSystem.js";
+import { buildCompletedWaitpointRecords } from "./completedWaitpointRecords.js";
+import { createCompletedWaitpointResolver } from "./completedWaitpointResolver.js";
+import type { CompletionEnvelopeSource } from "./types.js";
+
+const COMPLETED_AT = new Date("2026-08-25T00:00:00.000Z");
+const RUN_ID = "run_0123456789abcdefghijklm";
+const CHILD_RUN_ID = "run_zyxwvutsrqponmlkjihgfe";
+const BATCH_ID = "batch_0123456789abcdefghijk";
+
+/**
+ * One waitpoint, in both shapes, from one description. Keeping them in one factory is what
+ * makes the comparison meaningful: a field added to only one shape shows up as a diff.
+ */
+function pair(overrides: {
+  id: string;
+  type: Waitpoint["type"];
+  output?: string | null;
+  outputType?: string;
+  outputIsError?: boolean;
+  completedByTaskRunId?: string | null;
+  completedByBatchId?: string | null;
+  completedAfter?: Date | null;
+  idempotencyKey?: string;
+  userProvidedIdempotencyKey?: boolean;
+  inactiveIdempotencyKey?: string | null;
+}): { row: Waitpoint; source: CompletionEnvelopeSource } {
+  const outputType = overrides.outputType ?? "application/json";
+  const outputIsError = overrides.outputIsError ?? false;
+  const output = overrides.output ?? null;
+  const isRef = outputType === "application/store";
+
+  const row = {
+    id: overrides.id,
+    friendlyId: `waitpoint_${overrides.id}`,
+    type: overrides.type,
+    status: "COMPLETED",
+    completedAt: COMPLETED_AT,
+    output,
+    outputType,
+    outputIsError,
+    completedByTaskRunId: overrides.completedByTaskRunId ?? null,
+    completedByBatchId: overrides.completedByBatchId ?? null,
+    completedAfter: overrides.completedAfter ?? null,
+    idempotencyKey: overrides.idempotencyKey ?? "internal",
+    userProvidedIdempotencyKey: overrides.userProvidedIdempotencyKey ?? false,
+    inactiveIdempotencyKey: overrides.inactiveIdempotencyKey ?? null,
+  } as unknown as Waitpoint;
+
+  const source: CompletionEnvelopeSource = {
+    id: overrides.id,
+    friendlyId: `waitpoint_${overrides.id}`,
+    type: overrides.type,
+    completedAt: COMPLETED_AT,
+    outputType,
+    outputIsError,
+    ...(output !== null ? (isRef ? { outputRef: output } : { output }) : {}),
+    ...(overrides.completedByTaskRunId && {
+      completedByTaskRunId: overrides.completedByTaskRunId,
+    }),
+    ...(overrides.completedByBatchId && { completedByBatchId: overrides.completedByBatchId }),
+    ...(overrides.completedAfter && { completedAfter: overrides.completedAfter }),
+    ...(overrides.userProvidedIdempotencyKey &&
+    !overrides.inactiveIdempotencyKey &&
+    overrides.idempotencyKey
+      ? { idempotencyKey: overrides.idempotencyKey }
+      : {}),
+  };
+
+  return { row, source };
+}
+
+function snapshot(batchId: string | null) {
+  return { id: "snap_1", runId: RUN_ID, batchId } as never;
+}
+
+function sortEntries<T extends { id: string; index?: number }>(entries: T[]): T[] {
+  return [...entries].sort((a, b) => a.id.localeCompare(b.id) || (a.index ?? -1) - (b.index ?? -1));
+}
+
+/**
+ * Run one description through both paths and assert the results match.
+ *
+ * `deriveFromRun` is the one case where the two paths cannot be identical by construction:
+ * the row carries the value and the record carries a marker. Feeding the row's own output
+ * back as the run's output is what makes them comparable, which is exactly the claim the
+ * variant makes — that TaskRun.output holds the same string.
+ */
+async function bothPaths(
+  pairs: ReturnType<typeof pair>[],
+  order: string[],
+  batchId: string | null = null
+) {
+  const outputsByRunId = new Map<string, string>();
+  for (const { row } of pairs) {
+    if (row.completedByTaskRunId && row.output !== null) {
+      outputsByRunId.set(row.completedByTaskRunId, row.output);
+    }
+  }
+
+  const expected = enhanceExecutionSnapshotWithWaitpoints(
+    snapshot(batchId),
+    pairs.map((p) => p.row),
+    order
+  ).completedWaitpoints;
+
+  const actual = await createCompletedWaitpointResolver({
+    readRunOutput: async (taskRunId) => outputsByRunId.get(taskRunId),
+  })({
+    runId: RUN_ID,
+    ...(batchId ? { batchId } : {}),
+    pointer: { cycleSeq: 1, count: order.length },
+    order,
+    records: buildCompletedWaitpointRecords(pairs.map((p) => p.source)),
+  });
+
+  return { expected: sortEntries(expected), actual: sortEntries(actual) };
+}
+
+describe("the resolver reproduces the existing hydration", () => {
+  it("for a single MANUAL waitpoint with an inline output", async () => {
+    const { expected, actual } = await bothPaths(
+      [pair({ id: "wp_manual", type: "MANUAL", output: '{"token":1}' })],
+      []
+    );
+
+    expect(actual).toEqual(expected);
+  });
+
+  it("for a MANUAL waitpoint with a user-provided idempotency key", async () => {
+    const { expected, actual } = await bothPaths(
+      [
+        pair({
+          id: "wp_manual",
+          type: "MANUAL",
+          output: '{"token":1}',
+          idempotencyKey: "user-key",
+          userProvidedIdempotencyKey: true,
+        }),
+      ],
+      []
+    );
+
+    expect(actual).toEqual(expected);
+    expect(actual[0]?.idempotencyKey).toBe("user-key");
+  });
+
+  it("for an idempotency key the user provided but that went inactive", async () => {
+    const { expected, actual } = await bothPaths(
+      [
+        pair({
+          id: "wp_manual",
+          type: "MANUAL",
+          output: '{"token":1}',
+          idempotencyKey: "user-key",
+          userProvidedIdempotencyKey: true,
+          inactiveIdempotencyKey: "old",
+        }),
+      ],
+      []
+    );
+
+    expect(actual).toEqual(expected);
+    expect(actual[0]?.idempotencyKey).toBeUndefined();
+  });
+
+  it("for a DATETIME waitpoint", async () => {
+    const { expected, actual } = await bothPaths(
+      [
+        pair({
+          id: "wp_datetime",
+          type: "DATETIME",
+          completedAfter: new Date("2026-08-26T00:00:00.000Z"),
+        }),
+      ],
+      []
+    );
+
+    expect(actual).toEqual(expected);
+  });
+
+  it("for a RUN waitpoint outside a batch", async () => {
+    const { expected, actual } = await bothPaths(
+      [
+        pair({
+          id: "wp_run",
+          type: "RUN",
+          output: '{"ok":true}',
+          completedByTaskRunId: CHILD_RUN_ID,
+        }),
+      ],
+      []
+    );
+
+    expect(actual).toEqual(expected);
+  });
+
+  it("for a RUN waitpoint read under a batch", async () => {
+    const { expected, actual } = await bothPaths(
+      [
+        pair({
+          id: "wp_run",
+          type: "RUN",
+          output: '{"ok":true}',
+          completedByTaskRunId: CHILD_RUN_ID,
+        }),
+      ],
+      ["wp_run"],
+      BATCH_ID
+    );
+
+    expect(actual).toEqual(expected);
+    expect(actual[0]?.completedByTaskRun?.batch?.id).toBe(BATCH_ID);
+  });
+
+  it("for a RUN waitpoint whose output is an error", async () => {
+    const { expected, actual } = await bothPaths(
+      [
+        pair({
+          id: "wp_run",
+          type: "RUN",
+          output: '{"message":"boom"}',
+          outputIsError: true,
+          completedByTaskRunId: CHILD_RUN_ID,
+        }),
+      ],
+      []
+    );
+
+    expect(actual).toEqual(expected);
+  });
+
+  it("for a BATCH waitpoint", async () => {
+    const { expected, actual } = await bothPaths(
+      [pair({ id: "wp_batch", type: "BATCH", completedByBatchId: BATCH_ID })],
+      []
+    );
+
+    expect(actual).toEqual(expected);
+  });
+
+  it("for an already-offloaded output", async () => {
+    const { expected, actual } = await bothPaths(
+      [
+        pair({
+          id: "wp_manual",
+          type: "MANUAL",
+          output: "store-key-1",
+          outputType: "application/store",
+        }),
+      ],
+      []
+    );
+
+    expect(actual).toEqual(expected);
+  });
+
+  it("for one run present at two batch indexes", async () => {
+    const { expected, actual } = await bothPaths(
+      [
+        pair({
+          id: "wp_run",
+          type: "RUN",
+          output: '{"ok":true}',
+          completedByTaskRunId: CHILD_RUN_ID,
+        }),
+      ],
+      ["wp_run", "wp_run"],
+      BATCH_ID
+    );
+
+    expect(actual).toEqual(expected);
+    expect(actual.map((w) => w.index)).toEqual([0, 1]);
+  });
+
+  it("for an index-less waitpoint sitting beside indexed ones", async () => {
+    const { expected, actual } = await bothPaths(
+      [
+        pair({ id: "wp_indexless", type: "MANUAL", output: '{"token":1}' }),
+        pair({
+          id: "wp_run",
+          type: "RUN",
+          output: '{"ok":true}',
+          completedByTaskRunId: CHILD_RUN_ID,
+        }),
+      ],
+      ["wp_run"],
+      BATCH_ID
+    );
+
+    expect(actual).toEqual(expected);
+    expect(actual.find((w) => w.id === "wp_indexless")?.index).toBeUndefined();
+  });
+
+  it("for every type at once, under a batch", async () => {
+    const { expected, actual } = await bothPaths(
+      [
+        pair({
+          id: "wp_run",
+          type: "RUN",
+          output: '{"ok":true}',
+          completedByTaskRunId: CHILD_RUN_ID,
+        }),
+        pair({ id: "wp_batch", type: "BATCH", completedByBatchId: BATCH_ID }),
+        pair({
+          id: "wp_datetime",
+          type: "DATETIME",
+          completedAfter: new Date("2026-08-26T00:00:00.000Z"),
+        }),
+        pair({
+          id: "wp_manual",
+          type: "MANUAL",
+          output: '{"token":1}',
+          idempotencyKey: "user-key",
+          userProvidedIdempotencyKey: true,
+        }),
+      ],
+      ["wp_run", "wp_batch", "wp_datetime"],
+      BATCH_ID
+    );
+
+    expect(actual).toEqual(expected);
+    expect(actual).toHaveLength(4);
+  });
+});