From cc0acc786ce642bd633acee9b81f6c9ddd7fabf9 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Fri, 21 Aug 2026 18:21:54 +0100 Subject: [PATCH 1/9] feat(run-store): freeze the completed-waitpoints pointer, record and resolver types Reserves completedWaitpoints on the snapshot entry and rejects a set value. The pointer's physical form stays the #c sidecar field, because the append script mints both halves after the entry is serialized. --- .../run-store/src/redisSnapshotStore.test.ts | 38 ++++++++ .../run-store/src/redisSnapshotStore.ts | 97 ++++++++++++++++++- 2 files changed, 134 insertions(+), 1 deletion(-) diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts index 369d79e5338..25546194e13 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.test.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts @@ -10,6 +10,7 @@ import { isValidFor, RedisSnapshotStore, type SnapshotEntryInput, + type CompletedWaitpointsPointer, } from "./redisSnapshotStore.js"; describe("snapshotKeys", () => { @@ -1224,3 +1225,40 @@ describe("observability", () => { } ); }); + +describe("the reserved completedWaitpoints field", () => { + redisTest("append rejects an entry that sets it", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 }); + try { + const runId = "run_reserved_throw"; + const pointer: CompletedWaitpointsPointer = { cycleSeq: 1, count: 0 }; + await expect( + store.append({ + entry: { ...entry({ id: "snap_1", runId }), completedWaitpoints: pointer }, + kind: "birth", + isTerminal: false, + }) + ).rejects.toThrow(/reserved/i); + } finally { + await store.quit(); + } + }); + + redisTest("a stored entry never holds the key", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 }); + try { + const runId = "run_reserved_absent"; + await store.append({ + entry: entry({ id: "snap_1", runId }), + kind: "birth", + isTerminal: false, + }); + const read = await store.getLatest(runId); + expect(read).not.toBeNull(); + expect(read!.raw).not.toContain("completedWaitpoints"); + expect(read!.entry).not.toHaveProperty("completedWaitpoints"); + } finally { + await store.quit(); + } + }); +}); diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts index 7b60843e2b4..75e8f3088de 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.ts @@ -6,6 +6,7 @@ import { type Result, } from "@internal/redis"; import { Logger } from "@trigger.dev/core/logger"; +import type { CompletedWaitpoint } from "@trigger.dev/core/v3/schemas"; export type SnapshotKeys = { e: string; idx: string; cur: string; seq: string }; @@ -33,6 +34,83 @@ export function isValidFor(entry: { error?: unknown }): boolean { return !entry.error; } +// --------------------------------------------------------------------------- +// The completed-waitpoints freeze. Frozen jointly with the waitpoint lane. +// Do not change a field here without re-agreeing the contract with that lane. +// --------------------------------------------------------------------------- + +/** + * The once-per-wait-cycle pointer. `cycleSeq` names the snap:{runId}:wp: + * key. `count` is order.length -- NOT the record count -- so it is zero for any + * wait that carries no batch index. + */ +export type CompletedWaitpointsPointer = { + cycleSeq: number; + count: number; +}; + +/** + * A record's output. + * - `inline` holds the literal value, bounded by the pre-existing offload thresholds. + * - `ref` holds an application/store reference that was already offloaded. + * - `deriveFromRun` means the resolver reads TaskRun.output for completedByTaskRunId. + * Only a RUN record with outputIsError false uses it: TaskRun.output is a String + * column holding the same string verbatim, so the re-read is byte-identical. A RUN + * error cannot use it, because TaskRun.error is jsonb and never round-trips. + */ +export type CompletedWaitpointRecordOutput = + | { inline: string } + | { ref: string } + | { deriveFromRun: true } + | null; + +/** + * One completed waitpoint, one per DISTINCT id in a wait cycle. The resolver expands + * this into one CompletedWaitpoint per position of the id in the cycle's order list. + */ +export type CompletedWaitpointRecord = { + id: string; + friendlyId: string; + type: "RUN" | "BATCH" | "DATETIME" | "MANUAL"; + /** ISO. The writer pins it, applying the null fallback once. */ + completedAt: string; + /** Defaults to "application/json" at source. */ + outputType: string; + outputIsError: boolean; + output: CompletedWaitpointRecordOutput; + /** RUN. The resolver derives friendlyId, and batch{} from the READING entry's batchId. */ + completedByTaskRunId?: string; + /** BATCH. The resolver derives friendlyId. */ + completedByBatchId?: string; + /** ISO. Any type may set it: a MANUAL waitpoint with a timeout does. */ + completedAfter?: string; + /** Already resolved: userProvidedIdempotencyKey && !inactiveIdempotencyKey. */ + idempotencyKey?: string; +}; + +/** + * What the store hands the resolver. The store owns the keyspace, so the store reads + * and parses the cycle hash. The resolver never touches Redis and never derives a key. + */ +export type ResolveCompletedWaitpointsArgs = { + runId: string; + /** The batchId of the entry being READ, never the entry that minted the cycle. */ + batchId?: string; + pointer: CompletedWaitpointsPointer; + /** Index oracle only. A SUBSET of the record ids. Repeats preserved. */ + order: string[]; + /** The authoritative, complete set. Iterate this, never `order`. */ + records: CompletedWaitpointRecord[]; +}; + +/** + * This lane owns the signature. The waitpoint lane owns the implementation, which + * lives in run-engine because a deriveFromRun record needs a Postgres read. + */ +export type CompletedWaitpointResolver = ( + args: ResolveCompletedWaitpointsArgs +) => Promise; + export type SnapshotEntryInput = { id: string; engine: "V2"; @@ -53,6 +131,16 @@ export type SnapshotEntryInput = { runnerId?: string; metadata?: unknown; error?: string; + /** + * RESERVED. Always unset. `append` rejects a set value. + * + * The pointer's physical form is the `#c` sidecar field on the `e` hash, + * because the append Lua mints both halves after the client serializes the entry. + * The entry JSON must stay byte-identical to the caller's document, and the Postgres + * snapshot row has no pointer column, so a pointer inside the JSON would stop the two + * documents from being comparable for the dual-write comparator. + */ + completedWaitpoints?: CompletedWaitpointsPointer; }; export type WaitpointIds = { present: boolean; distinctIds: string[]; order: string[] }; @@ -67,7 +155,7 @@ export type SnapshotRead = { isValid: boolean; entry: Record; raw: string; - cycle?: { cycleSeq: number; count: number }; + cycle?: CompletedWaitpointsPointer; completedWaitpointIds?: WaitpointIds; }; @@ -158,6 +246,13 @@ export class RedisSnapshotStore { | { kind: "new"; completedWaitpoints: CompletedWaitpointRef[]; records?: string } | { kind: "carryForward"; cycleSeq: number }; }): Promise { + if (args.entry.completedWaitpoints !== undefined) { + throw new Error( + "completedWaitpoints is a reserved entry field and must stay unset. The pointer's " + + "physical form is the `#c` sidecar field, which the append script mints. " + + "Writing it into the entry JSON breaks byte-comparability with the Postgres row." + ); + } return this.#timed("append", async () => { const k = snapshotKeys(args.entry.runId); const raw = JSON.stringify(args.entry); From 86050e1a81675447513836e2c8ac3f87a19b28c5 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Fri, 21 Aug 2026 18:30:33 +0100 Subject: [PATCH 2/9] test(run-engine): prove the frozen waitpoint record matches the enhance oracle Runs the real enhanceExecutionSnapshotWithWaitpoints against a reference resolver over equivalent records and asserts field-for-field parity, so the frozen shape is machine-checked instead of asserted. --- .../systems/completedWaitpointFreeze.test.ts | 487 ++++++++++++++++++ .../engine/systems/executionSnapshotSystem.ts | 2 +- 2 files changed, 488 insertions(+), 1 deletion(-) create mode 100644 internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts diff --git a/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts b/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts new file mode 100644 index 00000000000..9253e6215b8 --- /dev/null +++ b/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts @@ -0,0 +1,487 @@ +// The freeze's executable definition. It runs the real oracle, +// enhanceExecutionSnapshotWithWaitpoints, against a reference resolver over equivalent +// records, and asserts the two agree field for field. The waitpoint lane owns the +// production resolver; this reference exists so the frozen shapes are checked rather +// than asserted. +import { describe, expect, it } from "vitest"; +import type { Waitpoint } from "@trigger.dev/database"; +import { BatchId, RunId } from "@trigger.dev/core/v3/isomorphic"; +import type { + CompletedWaitpoint, +} from "@trigger.dev/core/v3"; +import type { + CompletedWaitpointRecord, + CompletedWaitpointResolver, + ResolveCompletedWaitpointsArgs, +} from "@internal/run-store"; +import { enhanceExecutionSnapshotWithWaitpoints } from "./executionSnapshotSystem.js"; + +function makeWaitpoint(overrides: Partial): Waitpoint { + return { + id: "wp_default", + friendlyId: "waitpoint_default", + type: "MANUAL", + status: "COMPLETED", + completedAt: new Date("2026-01-01T00:00:00.000Z"), + idempotencyKey: "idem_generated", + userProvidedIdempotencyKey: false, + inactiveIdempotencyKey: null, + idempotencyKeyExpiresAt: null, + completedByTaskRunId: null, + completedByBatchId: null, + completedAfter: null, + output: null, + outputType: "application/json", + outputIsError: false, + projectId: "proj_1", + environmentId: "env_1", + tags: [], + createdAt: new Date("2026-01-01T00:00:00.000Z"), + updatedAt: new Date("2026-01-01T00:00:00.000Z"), + ...overrides, + } as Waitpoint; +} + +// The WRITE side of the freeze: one Waitpoint row becomes one record. +function toRecord(w: Waitpoint): CompletedWaitpointRecord { + return { + id: w.id, + friendlyId: w.friendlyId, + type: w.type, + completedAt: (w.completedAt ?? new Date()).toISOString(), + outputType: w.outputType, + outputIsError: w.outputIsError, + output: recordOutputFor(w), + completedByTaskRunId: w.completedByTaskRunId ?? undefined, + completedByBatchId: w.completedByBatchId ?? undefined, + completedAfter: w.completedAfter?.toISOString(), + idempotencyKey: + w.userProvidedIdempotencyKey && !w.inactiveIdempotencyKey ? w.idempotencyKey : undefined, + }; +} + +function recordOutputFor(w: Waitpoint): CompletedWaitpointRecord["output"] { + if (w.output === null) return null; + // A RUN success re-derives byte-identically from TaskRun.output. A RUN error cannot, + // because TaskRun.error is jsonb, so it carries inline. + // + // This branch is deliberately BEFORE the application/store branch. An offloaded RUN + // success is still deriveFromRun, and that is correct: completeAttemptSuccess receives + // the same `output` and `outputType` the waitpoint got, so TaskRun.output holds the + // same ref string. The re-read stays byte-identical either way. + if (w.type === "RUN" && !w.outputIsError) return { deriveFromRun: true }; + if (w.outputType === "application/store") return { ref: w.output }; + return { inline: w.output }; +} + +// The READ side of the freeze. Iterates `records`, never `order`. +async function referenceResolver( + args: ResolveCompletedWaitpointsArgs, + lookupRunOutput: (runId: string) => Promise +): Promise { + const out: CompletedWaitpoint[] = []; + for (const record of args.records) { + const indexes: (number | undefined)[] = []; + for (let i = 0; i < args.order.length; i++) { + if (args.order[i] === record.id) indexes.push(i); + } + if (indexes.length === 0) indexes.push(undefined); + + let output: string | undefined; + if (record.output === null) { + output = undefined; + } else if ("inline" in record.output) { + output = record.output.inline; + } else if ("ref" in record.output) { + output = record.output.ref; + } else { + output = record.completedByTaskRunId + ? await lookupRunOutput(record.completedByTaskRunId) + : undefined; + } + + for (const index of indexes) { + out.push({ + id: record.id, + // Unreachable: the oracle's own loop pushes a non-negative integer or undefined. + // Reproduced because the frozen index-expansion rule names it. + index: index === -1 ? undefined : index, + friendlyId: record.friendlyId, + type: record.type, + completedAt: new Date(record.completedAt), + idempotencyKey: record.idempotencyKey, + completedByTaskRun: record.completedByTaskRunId + ? { + id: record.completedByTaskRunId, + friendlyId: RunId.toFriendlyId(record.completedByTaskRunId), + batch: args.batchId + ? { id: args.batchId, friendlyId: BatchId.toFriendlyId(args.batchId) } + : undefined, + } + : undefined, + completedAfter: record.completedAfter ? new Date(record.completedAfter) : undefined, + completedByBatch: record.completedByBatchId + ? { + id: record.completedByBatchId, + friendlyId: BatchId.toFriendlyId(record.completedByBatchId), + } + : undefined, + output, + outputType: record.outputType, + outputIsError: record.outputIsError, + }); + } + } + return out; +} + +// Proves the frozen hook signature is implementable exactly as declared. The reference +// resolver takes its TaskRun lookup as a second parameter, so the production shape is +// the curried form -- which is what the waitpoint lane will bind to a Prisma client. +// If this assignment stops compiling, the frozen signature has drifted. +const resolverUnderTest: CompletedWaitpointResolver = (args) => + referenceResolver(args, async () => undefined); + +// The oracle spreads the snapshot, so it needs the two fields the mapping reads. +function makeSnapshot(batchId: string | null) { + return { id: "snap_1", runId: "run_1", batchId, checkpoint: null } as never; +} + +async function assertParity( + waitpoints: Waitpoint[], + order: string[], + batchId: string | null, + 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] + ); + expect(resolved).toEqual(enhanced.completedWaitpoints); + return { enhanced, resolved }; +} + +describe("the frozen record shape", () => { + // Literals, not a mirror of the writer. A field rename or an encoding change must + // fail HERE, because the parity suite cannot see it. + it("pins the RUN record", () => { + expect( + toRecord( + makeWaitpoint({ + id: "wp_run", + friendlyId: "waitpoint_run", + type: "RUN", + completedByTaskRunId: "run_child", + output: '{"value":42}', + }) + ) + ).toEqual({ + id: "wp_run", + friendlyId: "waitpoint_run", + type: "RUN", + completedAt: "2026-01-01T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: { deriveFromRun: true }, + completedByTaskRunId: "run_child", + completedByBatchId: undefined, + completedAfter: undefined, + idempotencyKey: undefined, + }); + }); + + it("pins the BATCH record", () => { + expect( + toRecord( + makeWaitpoint({ + id: "wp_batch", + friendlyId: "waitpoint_batch", + type: "BATCH", + completedByBatchId: "batch_child", + output: "Batch waitpoint completed", + }) + ) + ).toEqual({ + id: "wp_batch", + friendlyId: "waitpoint_batch", + type: "BATCH", + completedAt: "2026-01-01T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: { inline: "Batch waitpoint completed" }, + completedByTaskRunId: undefined, + completedByBatchId: "batch_child", + completedAfter: undefined, + idempotencyKey: undefined, + }); + }); + + it("pins the DATETIME record", () => { + expect( + toRecord( + makeWaitpoint({ + id: "wp_dt", + friendlyId: "waitpoint_dt", + type: "DATETIME", + completedAfter: new Date("2026-02-02T00:00:00.000Z"), + }) + ) + ).toEqual({ + id: "wp_dt", + friendlyId: "waitpoint_dt", + type: "DATETIME", + completedAt: "2026-01-01T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: null, + completedByTaskRunId: undefined, + completedByBatchId: undefined, + completedAfter: "2026-02-02T00:00:00.000Z", + idempotencyKey: undefined, + }); + }); + + it("pins the MANUAL record, with a user idempotency key and an offloaded output", () => { + expect( + toRecord( + makeWaitpoint({ + id: "wp_manual", + friendlyId: "waitpoint_manual", + type: "MANUAL", + idempotencyKey: "idem_user", + userProvidedIdempotencyKey: true, + output: "s3://bucket/key", + outputType: "application/store", + }) + ) + ).toEqual({ + id: "wp_manual", + friendlyId: "waitpoint_manual", + type: "MANUAL", + completedAt: "2026-01-01T00:00:00.000Z", + outputType: "application/store", + outputIsError: false, + output: { ref: "s3://bucket/key" }, + completedByTaskRunId: undefined, + completedByBatchId: undefined, + completedAfter: undefined, + idempotencyKey: "idem_user", + }); + }); + + it("keeps an offloaded RUN success on deriveFromRun, not ref", () => { + // Branch precedence. TaskRun.output holds the same ref string, so the re-read is + // still byte-identical. A later edit that reorders the branches must fail here. + const record = toRecord( + makeWaitpoint({ + id: "wp_run_offloaded", + type: "RUN", + completedByTaskRunId: "run_child", + output: "s3://bucket/key", + outputType: "application/store", + }) + ); + expect(record.output).toEqual({ deriveFromRun: true }); + }); + + it("carries a RUN error inline, never deriveFromRun", () => { + const record = toRecord( + makeWaitpoint({ + id: "wp_run_err", + type: "RUN", + completedByTaskRunId: "run_child", + output: '{"type":"BUILT_IN_ERROR"}', + outputIsError: true, + }) + ); + expect(record.output).toEqual({ inline: '{"type":"BUILT_IN_ERROR"}' }); + }); +}); + +describe("the completed-waitpoints freeze", () => { + it("expands a repeated id at each of its positions", async () => { + const w = makeWaitpoint({ id: "wp_a", type: "RUN", completedByTaskRunId: "run_child" }); + const { resolved } = await assertParity([w], ["wp_a", "wp_other", "wp_a"], "batch_1"); + expect(resolved.map((r) => r.index)).toEqual([0, 2]); + }); + + it("yields one entry with an undefined index for a record absent from order", async () => { + const w = makeWaitpoint({ id: "wp_absent", type: "MANUAL" }); + const { resolved } = await assertParity([w], ["wp_other"], null); + expect(resolved).toHaveLength(1); + expect(resolved[0]!.index).toBeUndefined(); + }); + + it("resolves a non-batch wait, where order is empty and one record exists", async () => { + // The commonest resume. Postgres's join holds the id while order does not, which is + // why `records` is authoritative and the mint comparison never reads `order`. + // batchId is null here on purpose: a triggerAndWait outside a batch is the shape + // this case is named for, and it exercises the oracle's `batchId ? ... : undefined` + // false branch, which no other case reaches. + // output stays null, so no TaskRun lookup is involved: both halves yield undefined. + const w = makeWaitpoint({ id: "wp_single", type: "RUN", completedByTaskRunId: "run_child" }); + const { resolved } = await assertParity([w], [], null); + expect(resolved).toHaveLength(1); + expect(resolved[0]!.index).toBeUndefined(); + expect(resolved[0]!.completedByTaskRun?.id).toBe("run_child"); + expect(resolved[0]!.completedByTaskRun?.batch).toBeUndefined(); + }); + + it("keys completedByBatch on the id alone, not on the type", async () => { + // The oracle checks completedByBatchId without looking at `type`, at + // executionSnapshotSystem.ts:107-113. A resolver keyed on type would pass every + // other case here and diverge in production. + const w = makeWaitpoint({ + id: "wp_manual_with_batch", + type: "MANUAL", + completedByBatchId: "batch_child", + }); + const { resolved } = await assertParity([w], ["wp_manual_with_batch"], null); + expect(resolved[0]!.completedByBatch?.id).toBe("batch_child"); + }); + + it("carries outputIsError on a non-RUN type", async () => { + const w = makeWaitpoint({ + id: "wp_manual_err", + type: "MANUAL", + output: '{"type":"STRING_ERROR"}', + outputIsError: true, + }); + const { resolved } = await assertParity([w], ["wp_manual_err"], null); + expect(resolved[0]!.outputIsError).toBe(true); + expect(resolved[0]!.output).toBe('{"type":"STRING_ERROR"}'); + }); + + it("returns an empty list for no waitpoints", async () => { + const { resolved } = await assertParity([], [], "batch_1"); + expect(resolved).toEqual([]); + }); + + it("resolves through the frozen hook signature", async () => { + // Exercises resolverUnderTest, so the declared CompletedWaitpointResolver type is + // proved implementable at runtime and not only at compile time. + const w = makeWaitpoint({ id: "wp_hook", type: "MANUAL" }); + const resolved = await resolverUnderTest({ + runId: "run_1", + batchId: undefined, + pointer: { cycleSeq: 1, count: 1 }, + order: ["wp_hook"], + records: [toRecord(w)], + }); + expect(resolved).toHaveLength(1); + expect(resolved[0]!.id).toBe("wp_hook"); + expect(resolved[0]!.index).toBe(0); + }); + + it("round-trips all four waitpoint types", async () => { + const waitpoints = [ + makeWaitpoint({ id: "wp_run", type: "RUN", completedByTaskRunId: "run_child" }), + makeWaitpoint({ id: "wp_batch", type: "BATCH", completedByBatchId: "batch_child" }), + makeWaitpoint({ + id: "wp_dt", + type: "DATETIME", + completedAfter: new Date("2026-02-02T00:00:00.000Z"), + }), + makeWaitpoint({ id: "wp_manual", type: "MANUAL" }), + ]; + const { resolved } = await assertParity(waitpoints, ["wp_run", "wp_batch"], "batch_1"); + expect(resolved.map((r) => r.type)).toEqual(["RUN", "BATCH", "DATETIME", "MANUAL"]); + }); + + it("applies the idempotency-key rule in all four combinations", async () => { + const combos: Array<[boolean, string | null, string | undefined]> = [ + [true, null, "idem_user"], + [true, "cleared", undefined], + [false, null, undefined], + [false, "cleared", undefined], + ]; + for (const [userProvided, inactive, expected] of combos) { + const w = makeWaitpoint({ + id: "wp_idem", + idempotencyKey: "idem_user", + userProvidedIdempotencyKey: userProvided, + inactiveIdempotencyKey: inactive, + }); + const { resolved } = await assertParity([w], ["wp_idem"], null); + expect(resolved[0]!.idempotencyKey).toBe(expected); + } + }); + + it("forwards completedAfter on a MANUAL waitpoint with a timeout", async () => { + // The plan comment scopes completedAfter to DATETIME. The oracle forwards it for any + // type, so the resolver must too. + const w = makeWaitpoint({ + id: "wp_timeout", + type: "MANUAL", + completedAfter: new Date("2026-03-03T00:00:00.000Z"), + }); + const { resolved } = await assertParity([w], ["wp_timeout"], null); + expect(resolved[0]!.completedAfter).toEqual(new Date("2026-03-03T00:00:00.000Z")); + }); + + it("maps every output variant", async () => { + const runSuccess = makeWaitpoint({ + id: "wp_run_ok", + type: "RUN", + completedByTaskRunId: "run_ok", + output: '{"value":42}', + }); + const runError = makeWaitpoint({ + id: "wp_run_err", + type: "RUN", + completedByTaskRunId: "run_err", + output: '{"type":"BUILT_IN_ERROR"}', + outputIsError: true, + }); + const offloaded = makeWaitpoint({ + id: "wp_ref", + type: "MANUAL", + output: "s3://bucket/key", + outputType: "application/store", + }); + const empty = makeWaitpoint({ id: "wp_none", type: "MANUAL", output: null }); + + const { resolved } = await assertParity( + [runSuccess, runError, offloaded, empty], + ["wp_run_ok", "wp_run_err", "wp_ref", "wp_none"], + "batch_1", + // deriveFromRun: the same string TaskRun.output holds verbatim. + { run_ok: '{"value":42}' } + ); + expect(resolved.map((r) => r.output)).toEqual([ + '{"value":42}', + '{"type":"BUILT_IN_ERROR"}', + "s3://bucket/key", + undefined, + ]); + }); +}); + +describe("the freeze's two deliberate divergences", () => { + it("pins completedAt at write time, where the oracle samples the clock", async () => { + // The oracle applies `w.completedAt ?? new Date()`, so a null value changes on every + // read. No deterministic record can match that. The record pins it once instead. + const w = makeWaitpoint({ id: "wp_null_at", completedAt: null }); + const record = toRecord(w); + expect(record.completedAt).toMatch(/^\d{4}-\d{2}-\d{2}T/); + expect(new Date(record.completedAt).getTime()).not.toBeNaN(); + }); + + it("takes batch{} from the reading entry, not the completing run's own batch", async () => { + // A known, deliberate conflation in the oracle. Byte-compatibility requires it. + const w = makeWaitpoint({ id: "wp_run", type: "RUN", completedByTaskRunId: "run_child" }); + const { resolved } = await assertParity([w], ["wp_run"], "batch_reading_entry"); + expect(resolved[0]!.completedByTaskRun?.batch?.id).toBe("batch_reading_entry"); + }); +}); diff --git a/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts b/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts index e79383a8bb6..81c41d2c2ae 100644 --- a/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts @@ -58,7 +58,7 @@ function enhanceExecutionSnapshot( * Transforms a snapshot (with checkpoint but without waitpoints) into an EnhancedExecutionSnapshot * by combining it with pre-fetched waitpoints. */ -function enhanceExecutionSnapshotWithWaitpoints( +export function enhanceExecutionSnapshotWithWaitpoints( snapshot: ExecutionSnapshotWithCheckpoint, waitpoints: Waitpoint[], completedWaitpointOrder: string[] From 86b5e630468d8eda1dd78dd48c890f4b84fb406b Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Fri, 21 Aug 2026 18:43:49 +0100 Subject: [PATCH 3/9] fix(run-engine): close review findings on the waitpoint-freeze conformance test - Fix a real parity gap: an orphaned RUN waitpoint (completing run row deleted, completedByTaskRunId null) lost its output under deriveFromRun. recordOutputFor now requires a non-null completedByTaskRunId; documents the precondition on the frozen CompletedWaitpointRecordOutput type. - Wire a tsconfig.freeze-test.json into this package's typecheck script so the conformance test file is actually typechecked, closing the compile-time-proof gap the two related comments were claiming. - Add a literal assertion pinning the pointer's count-is-order.length rule, including the empty-order case. - Tighten the completedAt divergence assertion to check it was set within the last minute, and drop an unused async. - Format the file with oxfmt. --- internal-packages/run-engine/package.json | 2 +- .../systems/completedWaitpointFreeze.test.ts | 51 ++++++++++++++----- .../run-engine/tsconfig.freeze-test.json | 9 ++++ .../run-store/src/redisSnapshotStore.ts | 9 ++-- 4 files changed, 53 insertions(+), 18 deletions(-) create mode 100644 internal-packages/run-engine/tsconfig.freeze-test.json diff --git a/internal-packages/run-engine/package.json b/internal-packages/run-engine/package.json index 96ace3a0e43..f4cfcd629d1 100644 --- a/internal-packages/run-engine/package.json +++ b/internal-packages/run-engine/package.json @@ -43,7 +43,7 @@ }, "scripts": { "clean": "rimraf dist", - "typecheck": "tsc --noEmit -p tsconfig.build.json", + "typecheck": "tsc --noEmit -p tsconfig.build.json && tsc --noEmit -p tsconfig.freeze-test.json", "test": "vitest --sequence.concurrent=false --no-file-parallelism", "test:coverage": "vitest --sequence.concurrent=false --no-file-parallelism --coverage.enabled", "build": "pnpm run clean && tsc -p tsconfig.build.json", 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 9253e6215b8..4da62e8da90 100644 --- a/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts +++ b/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts @@ -6,12 +6,11 @@ import { describe, expect, it } from "vitest"; import type { Waitpoint } from "@trigger.dev/database"; import { BatchId, RunId } from "@trigger.dev/core/v3/isomorphic"; -import type { - CompletedWaitpoint, -} from "@trigger.dev/core/v3"; +import type { CompletedWaitpoint } from "@trigger.dev/core/v3"; import type { CompletedWaitpointRecord, CompletedWaitpointResolver, + CompletedWaitpointsPointer, ResolveCompletedWaitpointsArgs, } from "@internal/run-store"; import { enhanceExecutionSnapshotWithWaitpoints } from "./executionSnapshotSystem.js"; @@ -69,7 +68,8 @@ function recordOutputFor(w: Waitpoint): CompletedWaitpointRecord["output"] { // success is still deriveFromRun, and that is correct: completeAttemptSuccess receives // the same `output` and `outputType` the waitpoint got, so TaskRun.output holds the // same ref string. The re-read stays byte-identical either way. - if (w.type === "RUN" && !w.outputIsError) return { deriveFromRun: true }; + if (w.type === "RUN" && !w.outputIsError && w.completedByTaskRunId) + return { deriveFromRun: true }; if (w.outputType === "application/store") return { ref: w.output }; return { inline: w.output }; } @@ -138,7 +138,9 @@ async function referenceResolver( // Proves the frozen hook signature is implementable exactly as declared. The reference // resolver takes its TaskRun lookup as a second parameter, so the production shape is // the curried form -- which is what the waitpoint lane will bind to a Prisma client. -// If this assignment stops compiling, the frozen signature has drifted. +// If this assignment stops compiling, the frozen signature has drifted. This file is +// typechecked by tsconfig.freeze-test.json, wired into this package's `typecheck` +// script, so that drift is caught -- vitest's esbuild transform alone would not catch it. const resolverUnderTest: CompletedWaitpointResolver = (args) => referenceResolver(args, async () => undefined); @@ -153,11 +155,7 @@ async function assertParity( batchId: string | null, runOutputs: Record = {} ) { - const enhanced = enhanceExecutionSnapshotWithWaitpoints( - makeSnapshot(batchId), - waitpoints, - order - ); + const enhanced = enhanceExecutionSnapshotWithWaitpoints(makeSnapshot(batchId), waitpoints, order); const resolved = await referenceResolver( { runId: "run_1", @@ -309,6 +307,20 @@ describe("the frozen record shape", () => { }); }); +describe("the frozen pointer shape", () => { + it("pins count to order.length, not the record count", () => { + const order = ["wp_a", "wp_b", "wp_a"]; + const pointer: CompletedWaitpointsPointer = { cycleSeq: 7, count: order.length }; + expect(pointer).toEqual({ cycleSeq: 7, count: 3 }); + }); + + it("pins count at 0 when order is empty, even if records exist", () => { + const order: string[] = []; + const pointer: CompletedWaitpointsPointer = { cycleSeq: 7, count: order.length }; + expect(pointer).toEqual({ cycleSeq: 7, count: 0 }); + }); +}); + describe("the completed-waitpoints freeze", () => { it("expands a repeated id at each of its positions", async () => { const w = makeWaitpoint({ id: "wp_a", type: "RUN", completedByTaskRunId: "run_child" }); @@ -370,7 +382,8 @@ describe("the completed-waitpoints freeze", () => { it("resolves through the frozen hook signature", async () => { // Exercises resolverUnderTest, so the declared CompletedWaitpointResolver type is - // proved implementable at runtime and not only at compile time. + // proved implementable at runtime, on top of the compile-time proof at its + // declaration above (checked by tsconfig.freeze-test.json). const w = makeWaitpoint({ id: "wp_hook", type: "MANUAL" }); const resolved = await resolverUnderTest({ runId: "run_1", @@ -430,6 +443,17 @@ describe("the completed-waitpoints freeze", () => { expect(resolved[0]!.completedAfter).toEqual(new Date("2026-03-03T00:00:00.000Z")); }); + it("falls back off deriveFromRun when the completing run was deleted", async () => { + const w = makeWaitpoint({ + id: "wp_orphan", + type: "RUN", + completedByTaskRunId: null, + output: '{"value":42}', + }); + const { resolved } = await assertParity([w], ["wp_orphan"], null); + expect(resolved[0]!.output).toBe('{"value":42}'); + }); + it("maps every output variant", async () => { const runSuccess = makeWaitpoint({ id: "wp_run_ok", @@ -469,13 +493,12 @@ describe("the completed-waitpoints freeze", () => { }); describe("the freeze's two deliberate divergences", () => { - it("pins completedAt at write time, where the oracle samples the clock", async () => { + it("pins completedAt at write time, where the oracle samples the clock", () => { // The oracle applies `w.completedAt ?? new Date()`, so a null value changes on every // read. No deterministic record can match that. The record pins it once instead. const w = makeWaitpoint({ id: "wp_null_at", completedAt: null }); const record = toRecord(w); - expect(record.completedAt).toMatch(/^\d{4}-\d{2}-\d{2}T/); - expect(new Date(record.completedAt).getTime()).not.toBeNaN(); + expect(Math.abs(Date.now() - new Date(record.completedAt).getTime())).toBeLessThan(60_000); }); it("takes batch{} from the reading entry, not the completing run's own batch", async () => { diff --git a/internal-packages/run-engine/tsconfig.freeze-test.json b/internal-packages/run-engine/tsconfig.freeze-test.json new file mode 100644 index 00000000000..04de1a6e252 --- /dev/null +++ b/internal-packages/run-engine/tsconfig.freeze-test.json @@ -0,0 +1,9 @@ +{ + "extends": "./tsconfig.build.json", + "include": ["src/engine/systems/completedWaitpointFreeze.test.ts"], + "exclude": [], + "compilerOptions": { + "composite": false, + "declaration": false + } +} diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts index 75e8f3088de..cc66fba0842 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.ts @@ -54,9 +54,12 @@ export type CompletedWaitpointsPointer = { * - `inline` holds the literal value, bounded by the pre-existing offload thresholds. * - `ref` holds an application/store reference that was already offloaded. * - `deriveFromRun` means the resolver reads TaskRun.output for completedByTaskRunId. - * Only a RUN record with outputIsError false uses it: TaskRun.output is a String - * column holding the same string verbatim, so the re-read is byte-identical. A RUN - * error cannot use it, because TaskRun.error is jsonb and never round-trips. + * Only a RUN record with outputIsError false AND a non-null completedByTaskRunId uses + * it: TaskRun.output is a String column holding the same string verbatim, so the + * re-read is byte-identical. A RUN error cannot use it, because TaskRun.error is + * jsonb and never round-trips. Waitpoint.completedByTaskRun is onDelete: SetNull, so + * an orphaned RUN waitpoint (the completing run row was deleted) has no run left to + * derive from -- its output carries inline instead. */ export type CompletedWaitpointRecordOutput = | { inline: string } From 594027d0d4712958f8b9e37a098895e655ad1de9 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Fri, 21 Aug 2026 19:15:18 +0100 Subject: [PATCH 4/9] fix(run-store,run-engine): attach the frozen record type to Redis, close the output union - RedisSnapshotStore.append's cycle.records now takes a typed CompletedWaitpointRecord[] and serializes it, instead of accepting an opaque pre-serialized string with no compile-time link to the frozen type. Adds a round-trip test reading the cycle hash's records field back with a raw client. - The test file's referenceResolver output discrimination is now exhaustive: an explicit deriveFromRun branch plus a `never` fallback, so a future output variant fails to compile here instead of silently resolving through a TaskRun re-read. - tsconfig.freeze-test.json documents that it resolves @internal/run-store from dist, so the gate must run through turbo rather than directly inside run-engine against a stale build. - assertParity now asserts pointer.count === order.length on every parity case, binding the frozen count-is-order.length rule instead of leaving it decorative. --- .../systems/completedWaitpointFreeze.test.ts | 26 +++++++----- .../run-engine/tsconfig.freeze-test.json | 10 +++++ .../run-store/src/redisSnapshotStore.test.ts | 41 +++++++++++++++++++ .../run-store/src/redisSnapshotStore.ts | 8 +++- 4 files changed, 72 insertions(+), 13 deletions(-) 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 4da62e8da90..adbd6dc2a3d 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 04de1a6e252..284a132e293 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 25546194e13..0a95f7feb62 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 cc66fba0842..90e6dbf8198 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"; From c1b9630ef247bc2425e5c5cffb7081f8bde797bd Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Mon, 24 Aug 2026 10:07:48 +0100 Subject: [PATCH 5/9] fix(run-store): clear a stale records field when a new wait cycle reuses a key The seq counter can vanish under maxmemory eviction while a wp: key survives. A birth does not check seq, unlike a transition, so the counter restarts and re-mints a cycleSeq whose key still holds another cycle's records. order and count are both overwritten, so they stay consistent with each other and the mismatch check cannot see the drift. A resolver would then read a previous cycle's records paired with a new order. --- .../run-store/src/redisSnapshotStore.test.ts | 48 +++++++++++++++++++ .../run-store/src/redisSnapshotStore.ts | 5 ++ 2 files changed, 53 insertions(+) diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts index 0a95f7feb62..e6deddc69b1 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.test.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts @@ -278,6 +278,54 @@ describe("append", () => { } ); + redisTest( + "a recordless new cycle clears another cycle's records off a reused key", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + const raw = createRedisClient(redisOptions); + try { + await store.append({ + entry: entry({ id: "snap_1" }), + kind: "birth", + isTerminal: false, + cycle: { + kind: "new", + completedWaitpoints: [{ id: "w_a", index: 0 }], + records: [ + { + id: "w_a", + friendlyId: "waitpoint_a", + type: "MANUAL", + completedAt: "2026-01-01T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: { inline: "stale" }, + }, + ], + }, + }); + expect(await raw.hget("snap:{run_1}:wp:1", "records")).not.toBeNull(); + + // Only the counter is lost, as under maxmemory eviction. A birth does not check seq, so + // the next new cycle re-mints cycleSeq 1 onto the surviving key. + await raw.del("snap:{run_1}:seq"); + + await store.append({ + entry: entry({ id: "snap_2" }), + kind: "birth", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_b", index: 0 }] }, + }); + + expect(await raw.hget("snap:{run_1}:wp:1", "order")).toBe(JSON.stringify(["w_b"])); + expect(await raw.hget("snap:{run_1}:wp:1", "records")).toBeNull(); + } 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 90e6dbf8198..09a791a1b4e 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.ts @@ -571,6 +571,11 @@ export class RedisSnapshotStore { redis.call('HSET', wpKey(cycleSeq), 'order', orderJson, 'count', orderCount) if records ~= '' then redis.call('HSET', wpKey(cycleSeq), 'records', records) + else + -- A new cycle owns the whole key: a lost seq counter can re-mint a cycleSeq whose key + -- still holds another cycle's records, and order/count stay mutually consistent so the + -- mismatch check cannot see it. No-op on a fresh key. + redis.call('HDEL', wpKey(cycleSeq), 'records') end elseif cycleMode == 'carry' then cycleSeq = cycleSeqIn From 1548018f465e8a8c6f1b1cc429d958d19e7887a2 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Mon, 24 Aug 2026 10:49:22 +0100 Subject: [PATCH 6/9] test(run-store,run-engine): enumerate the parity input space, refuse an unminted cycle Replaces hand-picked parity sampling with an exhaustive grid over every Waitpoint column the enhance oracle reads: 6144 combinations in ~35ms, compared with isDeepStrictEqual so key presence is checked too. The combination count is pinned, so a new column the oracle reads fails the assertion instead of silently shrinking coverage. Reverting the orphaned-RUN guard makes the grid report 192 divergences. A carryForward now attaches a pointer only if this incarnation actually minted the cycle. seq can be evicted while a wp: key survives, and a bare key-exists check adopted a dead incarnation's order and records under a count that agreed with them, reporting no mismatch. Also drops three tests that could not fail: two asserted a literal equalled its own construction, and one compared two structurally empty arrays. --- .../systems/completedWaitpointFreeze.test.ts | 165 +++++++++++++++--- .../run-store/src/redisSnapshotStore.test.ts | 51 ++++++ .../run-store/src/redisSnapshotStore.ts | 10 +- 3 files changed, 201 insertions(+), 25 deletions(-) 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 adbd6dc2a3d..f3eab876a19 100644 --- a/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts +++ b/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts @@ -3,6 +3,7 @@ // records, and asserts the two agree field for field. The waitpoint lane owns the // production resolver; this reference exists so the frozen shapes are checked rather // than asserted. +import { isDeepStrictEqual } from "node:util"; import { describe, expect, it } from "vitest"; import type { Waitpoint } from "@trigger.dev/database"; import { BatchId, RunId } from "@trigger.dev/core/v3/isomorphic"; @@ -10,7 +11,6 @@ import type { CompletedWaitpoint } from "@trigger.dev/core/v3"; import type { CompletedWaitpointRecord, CompletedWaitpointResolver, - CompletedWaitpointsPointer, ResolveCompletedWaitpointsArgs, } from "@internal/run-store"; import { enhanceExecutionSnapshotWithWaitpoints } from "./executionSnapshotSystem.js"; @@ -166,9 +166,8 @@ async function assertParity( 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); + // count-carried-forward behaviour (order.length, not the record count) is covered by + // the run-store Redis suite, not here -- this line only constructs `args`, not asserts. const resolved = await referenceResolver(args, async (id) => runOutputs[id]); expect(resolved).toEqual(enhanced.completedWaitpoints); return { enhanced, resolved }; @@ -311,19 +310,8 @@ describe("the frozen record shape", () => { }); }); -describe("the frozen pointer shape", () => { - it("pins count to order.length, not the record count", () => { - const order = ["wp_a", "wp_b", "wp_a"]; - const pointer: CompletedWaitpointsPointer = { cycleSeq: 7, count: order.length }; - expect(pointer).toEqual({ cycleSeq: 7, count: 3 }); - }); - - it("pins count at 0 when order is empty, even if records exist", () => { - const order: string[] = []; - const pointer: CompletedWaitpointsPointer = { cycleSeq: 7, count: order.length }; - expect(pointer).toEqual({ cycleSeq: 7, count: 0 }); - }); -}); +// The pointer's shape is pinned by CompletedWaitpointsPointer and tsconfig.freeze-test.json, +// not by a runtime assertion here -- a value that only echoes its own construction can't fail. describe("the completed-waitpoints freeze", () => { it("expands a repeated id at each of its positions", async () => { @@ -379,11 +367,6 @@ describe("the completed-waitpoints freeze", () => { expect(resolved[0]!.output).toBe('{"type":"STRING_ERROR"}'); }); - it("returns an empty list for no waitpoints", async () => { - const { resolved } = await assertParity([], [], "batch_1"); - expect(resolved).toEqual([]); - }); - it("resolves through the frozen hook signature", async () => { // Exercises resolverUnderTest, so the declared CompletedWaitpointResolver type is // proved implementable at runtime, on top of the compile-time proof at its @@ -496,6 +479,144 @@ describe("the completed-waitpoints freeze", () => { }); }); +describe("the exhaustive parity grid", () => { + // Dimensions mirror every Waitpoint column the oracle reads (type, output, outputType, + // outputIsError, completedByTaskRunId, completedByBatchId, completedAfter, + // userProvidedIdempotencyKey, inactiveIdempotencyKey), plus order-membership and the + // reading entry's batchId. A new column the oracle reads must widen a dimension here, + // so the pinned combination count below fails instead of coverage silently shrinking. + const TYPES: Waitpoint["type"][] = ["RUN", "BATCH", "DATETIME", "MANUAL"]; + const OUTPUTS: (string | null)[] = [null, '{"value":42}']; + const OUTPUT_TYPES = ["application/json", "application/store"]; + const OUTPUT_IS_ERRORS = [false, true]; + const TASK_RUN_IDS: (string | null)[] = [null, "run_child"]; + const BATCH_IDS: (string | null)[] = [null, "batch_child"]; + const COMPLETED_AFTERS: (Date | null)[] = [null, new Date("2026-02-02T00:00:00.000Z")]; + const IDEMPOTENCY_COMBOS: Array<[boolean, string | null]> = [ + [false, null], + [false, "cleared"], + [true, null], + [true, "cleared"], + ]; + const ORDER_MEMBERSHIPS = ["absent", "once", "twice"] as const; + const READING_BATCH_IDS: (string | null)[] = [null, "batch_reading_entry"]; + + // Only reached when type is RUN, output is set, outputIsError is false, and + // completedByTaskRunId is "run_child": the deriveFromRun branch. The value matches + // OUTPUTS' non-null entry so a correct resolver is byte-identical to the oracle. + const RUN_OUTPUT_LOOKUP: Record = { run_child: '{"value":42}' }; + + it("agrees with the oracle across every combination", async () => { + type Combo = { + type: Waitpoint["type"]; + output: string | null; + outputType: string; + outputIsError: boolean; + completedByTaskRunId: string | null; + completedByBatchId: string | null; + completedAfter: Date | null; + userProvidedIdempotencyKey: boolean; + inactiveIdempotencyKey: string | null; + orderMembership: (typeof ORDER_MEMBERSHIPS)[number]; + readingBatchId: string | null; + }; + const failures: Array<{ combo: Combo; oracle: unknown; resolver: unknown }> = []; + let cases = 0; + + for (const type of TYPES) { + for (const output of OUTPUTS) { + for (const outputType of OUTPUT_TYPES) { + for (const outputIsError of OUTPUT_IS_ERRORS) { + for (const completedByTaskRunId of TASK_RUN_IDS) { + for (const completedByBatchId of BATCH_IDS) { + for (const completedAfter of COMPLETED_AFTERS) { + for (const [ + userProvidedIdempotencyKey, + inactiveIdempotencyKey, + ] of IDEMPOTENCY_COMBOS) { + for (const orderMembership of ORDER_MEMBERSHIPS) { + for (const readingBatchId of READING_BATCH_IDS) { + cases++; + const combo: Combo = { + type, + output, + outputType, + outputIsError, + completedByTaskRunId, + completedByBatchId, + completedAfter, + userProvidedIdempotencyKey, + inactiveIdempotencyKey, + orderMembership, + readingBatchId, + }; + + const id = "wp_grid"; + const w = makeWaitpoint({ + id, + type, + output, + outputType, + outputIsError, + completedByTaskRunId, + completedByBatchId, + completedAfter, + idempotencyKey: "idem_user", + userProvidedIdempotencyKey, + inactiveIdempotencyKey, + }); + const order = + orderMembership === "absent" + ? ["wp_other"] + : orderMembership === "once" + ? [id] + : [id, id]; + + const enhanced = enhanceExecutionSnapshotWithWaitpoints( + makeSnapshot(readingBatchId), + [w], + order + ); + const args: ResolveCompletedWaitpointsArgs = { + runId: "run_1", + batchId: readingBatchId ?? undefined, + pointer: { cycleSeq: 1, count: order.length }, + order, + records: [toRecord(w)], + }; + const resolved = await referenceResolver( + args, + async (runId) => RUN_OUTPUT_LOOKUP[runId] + ); + + if (!isDeepStrictEqual(resolved, enhanced.completedWaitpoints)) { + failures.push({ + combo, + oracle: enhanced.completedWaitpoints, + resolver: resolved, + }); + } + } + } + } + } + } + } + } + } + } + } + + expect(cases).toBe(6144); + expect( + failures.length, + failures.length > 0 + ? `${failures.length}/${cases} combinations diverged. First: ${JSON.stringify(failures[0], null, 2)}` + : undefined + ).toBe(0); + }); +}); + describe("the freeze's two deliberate divergences", () => { it("pins completedAt at write time, where the oracle samples the clock", () => { // The oracle applies `w.completedAt ?? new Date()`, so a null value changes on every diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts index e6deddc69b1..8b738e6b106 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.test.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts @@ -326,6 +326,57 @@ describe("append", () => { } ); + redisTest( + "a carry-forward refuses a cycle this incarnation never minted", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 }); + const raw = createRedisClient(redisOptions); + try { + await store.append({ + entry: entry({ id: "snap_1" }), + kind: "birth", + isTerminal: false, + cycle: { + kind: "new", + completedWaitpoints: [{ id: "w_old", index: 0 }], + records: [ + { + id: "w_old", + friendlyId: "waitpoint_old", + type: "MANUAL", + completedAt: "2026-01-01T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: { inline: "stale" }, + }, + ], + }, + }); + + // Lose the whole keyspace except the cycle key, as under maxmemory eviction. + await raw.del("snap:{run_1}:e", "snap:{run_1}:idx", "snap:{run_1}:cur", "snap:{run_1}:seq"); + + const carried = await store.append({ + entry: entry({ id: "snap_2" }), + kind: "birth", + isTerminal: false, + cycle: { kind: "carryForward", cycleSeq: 1 }, + }); + + // Written, flagged, and carrying NO pointer: the dead incarnation's waitpoints must not + // be served to a fresh run under a count that agrees with them. + expect(carried).toMatchObject({ outcome: "written", cycleMismatch: true }); + expect(carried).not.toHaveProperty("cycleSeq"); + const read = await store.getLatest("run_1"); + expect(read?.cycle).toBeUndefined(); + expect(read?.completedWaitpointIds).toBeUndefined(); + } 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 09a791a1b4e..f82355a3216 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.ts @@ -578,11 +578,15 @@ export class RedisSnapshotStore { redis.call('HDEL', wpKey(cycleSeq), 'records') end elseif cycleMode == 'carry' then - cycleSeq = cycleSeqIn - local c = redis.call('HGET', wpKey(cycleSeq), 'count') - if not c then + -- Attach a pointer only if this incarnation actually minted the cycle. seq can be + -- evicted while a wp: key survives, so a bare key-exists check would adopt a dead + -- incarnation's order and records under a consistent count, invisibly. + local minted = tonumber(redis.call('HGET', seqKey, 'c') or '0') + local c = redis.call('HGET', wpKey(cycleSeqIn), 'count') + if not c or minted < cycleSeqIn then mismatch = 1 else + cycleSeq = cycleSeqIn orderCount = c end end From 39b0ef887e4501b9f62e24ce113064ba0367c5a3 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Mon, 24 Aug 2026 10:58:39 +0100 Subject: [PATCH 7/9] fix(run-store): count records in the cycle-key size metric The metric is on the wp: key, but it measured only the order field. records dominates that key once populated, so a 20KB cycle key was reported as 7 bytes and the high-water log could never fire on the field that actually grows. --- .../run-store/src/redisSnapshotStore.test.ts | 47 +++++++++++++++++++ .../run-store/src/redisSnapshotStore.ts | 23 +++++++-- 2 files changed, 65 insertions(+), 5 deletions(-) diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts index 8b738e6b106..39004b81a56 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.test.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts @@ -1254,6 +1254,53 @@ describe("hash tag and keyPrefix", () => { }); describe("observability", () => { + redisTest("cycle-key bytes cover records, not just order", async ({ redisOptions }) => { + // The plan puts a metric on the wp: KEY size. records dominates that key once + // populated, so measuring order alone understates it by orders of magnitude. + const calls: Array<[string, number]> = []; + const metrics = { + recordAppend: () => {}, + recordEntryBytes: () => {}, + recordCycleKeyBytes: (b: number) => calls.push(["cycleBytes", b]), + recordCycleCount: () => {}, + recordSkippedNoKeyspace: () => {}, + recordCycleMismatch: () => {}, + recordLatency: () => {}, + }; + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000, metrics }); + try { + const records: CompletedWaitpointRecord[] = [ + { + id: "w_a", + friendlyId: "waitpoint_a", + type: "MANUAL", + completedAt: "2026-01-01T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: { inline: "y".repeat(20_000) }, + }, + ]; + const orderJson = JSON.stringify(["w_a"]); + const recordsJson = JSON.stringify(records); + + await store.append({ + entry: entry({ id: "snap_1" }), + kind: "birth", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }], records }, + }); + + expect(calls).toEqual([ + [ + "cycleBytes", + Buffer.byteLength(orderJson, "utf8") + Buffer.byteLength(recordsJson, "utf8"), + ], + ]); + } finally { + await store.quit(); + } + }); + redisTest("records sizes and outcomes without ever rejecting", async ({ redisOptions }) => { const calls: unknown[][] = []; const metrics = { diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts index f82355a3216..10175988fed 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.ts @@ -301,11 +301,17 @@ export class RedisSnapshotStore { args.expectedCur !== undefined ? "1" : "0" )) as string[]; - return this.#interpretAppend(reply, raw, orderJson, args.entry.runId); + return this.#interpretAppend(reply, raw, orderJson, records, args.entry.runId); }); } - #interpretAppend(reply: string[], raw: string, orderJson: string, runId: string): AppendResult { + #interpretAppend( + reply: string[], + raw: string, + orderJson: string, + records: string, + runId: string + ): AppendResult { if (reply[0] === SKIPPED) { this.metrics?.recordSkippedNoKeyspace(); this.metrics?.recordAppend("skippedNoKeyspace", "none"); @@ -326,7 +332,7 @@ export class RedisSnapshotStore { if (cycleMismatch) { this.metrics?.recordCycleMismatch(); } - this.#observeSizes(raw, orderJson, cycleSeq, runId); + this.#observeSizes(raw, orderJson, records, cycleSeq, runId); this.metrics?.recordAppend("written", ttl); return { outcome: "written", @@ -337,14 +343,21 @@ export class RedisSnapshotStore { }; } - #observeSizes(raw: string, orderJson: string, cycleSeq: number, runId: string): void { + #observeSizes( + raw: string, + orderJson: string, + records: string, + cycleSeq: number, + runId: string + ): void { const entryBytes = Buffer.byteLength(raw, "utf8"); this.metrics?.recordEntryBytes(entryBytes); if (this.highWater.entryBytes !== undefined && entryBytes > this.highWater.entryBytes) { this.logger.warn("RedisSnapshotStore entry above high-water mark", { runId, entryBytes }); } if (orderJson !== "") { - const cycleBytes = Buffer.byteLength(orderJson, "utf8"); + // The whole wp: key, not just its order field: records dominates it once populated. + const cycleBytes = Buffer.byteLength(orderJson, "utf8") + Buffer.byteLength(records, "utf8"); this.metrics?.recordCycleKeyBytes(cycleBytes); if (this.highWater.cycleKeyBytes !== undefined && cycleBytes > this.highWater.cycleKeyBytes) { this.logger.warn("RedisSnapshotStore cycle key above high-water mark", { From 31d504c02b5ef8268c4b19fc2668b3f06a7e7d98 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Mon, 24 Aug 2026 11:16:00 +0100 Subject: [PATCH 8/9] test(run-store,run-engine): sweep keyspace damage, pin the frozen key sets Two bugs where a surviving cycle key no longer belonged to the cycle its pointer named were both found by hand. Replaces hand-picked damage cases with a sweep: the powerset of deletions over the five key shapes, across two injection points, 128 replays, asserting that records read back for a pointer never mention an id outside that pointer's own order. Reverting the HDEL in the new-cycle branch makes it report 56 violations. Also pins the frozen key sets exactly and bidirectionally. Renames, removals, widenings and required-to-optional already broke compilation through the usage sites; an added optional field did not, and on a jointly-owned frozen type that is the change neither lane may make alone. --- .../systems/completedWaitpointFreeze.test.ts | 29 +++++ .../run-store/src/redisSnapshotStore.test.ts | 123 ++++++++++++++++++ 2 files changed, 152 insertions(+) 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 f3eab876a19..a1af90582b6 100644 --- a/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts +++ b/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts @@ -11,8 +11,37 @@ import type { CompletedWaitpoint } from "@trigger.dev/core/v3"; import type { CompletedWaitpointRecord, CompletedWaitpointResolver, + CompletedWaitpointsPointer, ResolveCompletedWaitpointsArgs, } from "@internal/run-store"; + +// The frozen key sets, pinned exactly and bidirectionally. Renames, removals, widenings and +// required-to-optional all already break compilation through the usage sites below; an ADDED +// OPTIONAL field does not, and on a jointly-owned frozen type that is the change neither lane +// may make unilaterally. These fail on it. +type Exact = [A] extends [B] ? ([B] extends [A] ? true : never) : never; + +const _recordKeys: Exact< + keyof CompletedWaitpointRecord, + | "id" + | "friendlyId" + | "type" + | "completedAt" + | "outputType" + | "outputIsError" + | "output" + | "completedByTaskRunId" + | "completedByBatchId" + | "completedAfter" + | "idempotencyKey" +> = true; + +const _pointerKeys: Exact = true; + +const _argsKeys: Exact< + keyof ResolveCompletedWaitpointsArgs, + "runId" | "batchId" | "pointer" | "order" | "records" +> = true; import { enhanceExecutionSnapshotWithWaitpoints } from "./executionSnapshotSystem.js"; function makeWaitpoint(overrides: Partial): Waitpoint { diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts index 39004b81a56..0c7b4c0720a 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.test.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts @@ -525,6 +525,129 @@ describe("cycle keys", () => { }); }); +describe("cycle key deletion sweep", () => { + const KEY_SUFFIXES = ["e", "idx", "cur", "seq", "wp:1", "wp:2"] as const; + const INJECTION_POINTS = ["beforeCarryForward", "beforeSecondNewCycle"] as const; + + function powerset(items: readonly T[]): T[][] { + let out: T[][] = [[]]; + for (const item of items) { + out = out.concat(out.map((s) => [...s, item])); + } + return out; + } + + // Mechanized replacement for the two hand-picked eviction regressions above: replays the same + // birth/carryForward/new-cycle/terminal shape once per element of the powerset of key deletions, + // at two points in the sequence, and checks that a cycle key's records never leak into a pointer + // naming a different cycle than the one the key's own order field currently describes. + redisTest( + "records read back for a pointer never mention an id outside that pointer's own order", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 }); + const raw = createRedisClient(redisOptions); + type Violation = { + subset: string[]; + injectionPoint: string; + runId: string; + pointer: CompletedWaitpointsPointer; + recordsFound: unknown; + }; + const violations: Violation[] = []; + let replays = 0; + + try { + for (const injectionPoint of INJECTION_POINTS) { + for (const subset of powerset(KEY_SUFFIXES)) { + replays++; + const runId = `run_sweep_${injectionPoint}_${replays}`; + const base = `snap:{${runId}}`; + const damage = async () => { + if (subset.length > 0) { + await raw.del(...subset.map((s) => `${base}:${s}`)); + } + }; + + await store.append({ + entry: entry({ id: "snap_1", runId }), + kind: "birth", + isTerminal: false, + cycle: { + kind: "new", + completedWaitpoints: [{ id: "w_c1", index: 0 }], + records: [ + { + id: "w_c1", + friendlyId: "waitpoint_c1", + type: "MANUAL", + completedAt: "2026-01-01T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: { inline: "payload_c1" }, + }, + ], + }, + }); + + if (injectionPoint === "beforeCarryForward") await damage(); + await store.append({ + entry: entry({ id: "snap_2", runId }), + kind: "transition", + isTerminal: false, + cycle: { kind: "carryForward", cycleSeq: 1 }, + }); + + if (injectionPoint === "beforeSecondNewCycle") await damage(); + // birth, not transition: both eviction regressions above only reproduce past a birth's + // liveness bypass -- a transition here would just report skippedNoKeyspace once e or seq + // is gone, exempting the replay before the cycle-mint branch ever ran. + await store.append({ + entry: entry({ id: "snap_3", runId }), + kind: "birth", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_c2", index: 0 }] }, + }); + + await store.append({ + entry: entry({ id: "snap_4", runId, executionStatus: "FINISHED" }), + kind: "transition", + isTerminal: true, + }); + + for (const id of ["snap_1", "snap_2", "snap_3", "snap_4"]) { + const read = await store.getById(runId, id); + if (!read?.cycle) continue; + const orderIds = new Set(read.completedWaitpointIds?.order ?? []); + const recordsRaw = await raw.hget(`${base}:wp:${read.cycle.cycleSeq}`, "records"); + if (recordsRaw === null) continue; + const recordIds = (JSON.parse(recordsRaw) as { id: string }[]).map((r) => r.id); + if (recordIds.some((rid) => !orderIds.has(rid))) { + violations.push({ + subset, + injectionPoint, + runId, + pointer: read.cycle, + recordsFound: recordIds, + }); + } + } + } + } + } finally { + raw.disconnect(); + await store.quit(); + } + + if (violations.length > 0) { + throw new Error( + `${violations.length} violation(s) across ${replays} replays. First: ` + + JSON.stringify(violations[0]) + ); + } + } + ); +}); + describe("read-side cycle mismatch", () => { redisTest( "warns and records a metric when a cycle's count disagrees with its order", From 400145f797ab473bda4915ca5eb4b3c763a44797 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Mon, 24 Aug 2026 12:44:08 +0100 Subject: [PATCH 9/9] docs(run-store): correct the boundedness claim on inline record output Only BUILT_IN_ERROR is truncated; a STRING_ERROR, a CUSTOM_ERROR and a cancel-reason error pass through, so the bound for an error payload is the completion body limit and not the offload thresholds. Postgres stores the identical strings, so this is a copy of an existing payload class. --- internal-packages/run-store/src/redisSnapshotStore.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts index 10175988fed..2964959c4fe 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.ts @@ -51,7 +51,9 @@ export type CompletedWaitpointsPointer = { /** * A record's output. - * - `inline` holds the literal value, bounded by the pre-existing offload thresholds. + * - `inline` holds the literal value. MANUAL and DATETIME are bounded by the offload + * thresholds; error outputs are not (only BUILT_IN_ERROR truncates), so the bound is + * the completion body limit. Postgres holds the same strings, so this is a copy. * - `ref` holds an application/store reference that was already offloaded. * - `deriveFromRun` means the resolver reads TaskRun.output for completedByTaskRunId. * Only a RUN record with outputIsError false AND a non-null completedByTaskRunId uses