diff --git a/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts b/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts index 4da62e8da..adbd6dc2a 100644 --- a/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts +++ b/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts @@ -94,10 +94,13 @@ async function referenceResolver( output = record.output.inline; } else if ("ref" in record.output) { output = record.output.ref; - } else { + } else if ("deriveFromRun" in record.output) { output = record.completedByTaskRunId ? await lookupRunOutput(record.completedByTaskRunId) : undefined; + } else { + const _never: never = record.output; + throw new Error(`unknown record output variant: ${JSON.stringify(_never)}`); } for (const index of indexes) { @@ -156,16 +159,17 @@ async function assertParity( runOutputs: Record = {} ) { const enhanced = enhanceExecutionSnapshotWithWaitpoints(makeSnapshot(batchId), waitpoints, order); - const resolved = await referenceResolver( - { - runId: "run_1", - batchId: batchId ?? undefined, - pointer: { cycleSeq: 1, count: order.length }, - order, - records: waitpoints.map(toRecord), - }, - async (id) => runOutputs[id] - ); + const args: ResolveCompletedWaitpointsArgs = { + runId: "run_1", + batchId: batchId ?? undefined, + pointer: { cycleSeq: 1, count: order.length }, + order, + records: waitpoints.map(toRecord), + }; + // The frozen rule: count is order.length, NOT the record count. Binding it here means every + // parity case enforces it, not only the dedicated "the frozen pointer shape" cases. + expect(args.pointer.count).toBe(order.length); + const resolved = await referenceResolver(args, async (id) => runOutputs[id]); expect(resolved).toEqual(enhanced.completedWaitpoints); return { enhanced, resolved }; } diff --git a/internal-packages/run-engine/tsconfig.freeze-test.json b/internal-packages/run-engine/tsconfig.freeze-test.json index 04de1a6e2..284a132e2 100644 --- a/internal-packages/run-engine/tsconfig.freeze-test.json +++ b/internal-packages/run-engine/tsconfig.freeze-test.json @@ -1,3 +1,13 @@ +// Typechecks completedWaitpointFreeze.test.ts, which tsconfig.build.json otherwise excludes +// (src/**/*.test.ts) and vitest's esbuild transform never checks. This config has no +// "@triggerdotdev/source" customCondition, so it resolves @internal/run-store from its built +// `dist`, not from source -- same as tsconfig.build.json. That means this gate only sees a +// source change in run-store once run-store has been rebuilt, so it MUST be run through turbo +// (`pnpm run typecheck --filter @internal/run-engine`), whose `typecheck` task declares +// `dependsOn: ["^build"]`. Running `tsc -p tsconfig.freeze-test.json` (or `pnpm run typecheck`) +// directly inside this package, against a stale dist/, passes green while the frozen type has +// already drifted in source. Do not "fix" this with customConditions: that pulls +// @trigger.dev/core's source in too, which fails to typecheck here on `lib: ES2020`. { "extends": "./tsconfig.build.json", "include": ["src/engine/systems/completedWaitpointFreeze.test.ts"], diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts index 25546194e..0a95f7feb 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.test.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts @@ -11,6 +11,7 @@ import { RedisSnapshotStore, type SnapshotEntryInput, type CompletedWaitpointsPointer, + type CompletedWaitpointRecord, } from "./redisSnapshotStore.js"; describe("snapshotKeys", () => { @@ -237,6 +238,46 @@ describe("append", () => { } ); + redisTest( + "round-trips a typed records array through the cycle hash's records field", + async ({ redisOptions }) => { + // The only place CompletedWaitpointRecord[] physically enters Redis. If the writer ever + // serializes a different envelope, this is where that would show up as a broken round trip. + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + const raw = createRedisClient(redisOptions); + const records: CompletedWaitpointRecord[] = [ + { + id: "w_a", + friendlyId: "waitpoint_a", + type: "RUN", + completedAt: "2026-01-01T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: { deriveFromRun: true }, + completedByTaskRunId: "run_child", + }, + ]; + try { + await store.append({ + entry: entry({ id: "snap_1" }), + kind: "birth", + isTerminal: false, + cycle: { + kind: "new", + completedWaitpoints: [{ id: "w_a", index: 0 }], + records, + }, + }); + + const storedRaw = await raw.hget("snap:{run_1}:wp:1", "records"); + expect(JSON.parse(storedRaw!)).toEqual(records); + } finally { + raw.disconnect(); + await store.quit(); + } + } + ); + redisTest( "reports a duplicate id without overwriting the original entry", async ({ redisOptions }) => { diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts index cc66fba08..90e6dbf81 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.ts @@ -246,7 +246,11 @@ export class RedisSnapshotStore { isTerminal: boolean; expectedCur?: string; cycle?: - | { kind: "new"; completedWaitpoints: CompletedWaitpointRef[]; records?: string } + | { + kind: "new"; + completedWaitpoints: CompletedWaitpointRef[]; + records?: CompletedWaitpointRecord[]; + } | { kind: "carryForward"; cycleSeq: number }; }): Promise { if (args.entry.completedWaitpoints !== undefined) { @@ -270,7 +274,7 @@ export class RedisSnapshotStore { const order = deriveOrder(args.cycle.completedWaitpoints); cycleMode = "new"; orderJson = JSON.stringify(order); - records = args.cycle.records ?? ""; + records = args.cycle.records ? JSON.stringify(args.cycle.records) : ""; orderCount = String(order.length); } else if (args.cycle?.kind === "carryForward") { cycleMode = "carry";