fix(core): Harden RunEvent schema and add backward-compatibility transform

bharathkumar39293 · bharathkumar39293 · commit 399373073710 · 2026-03-15T08:55:17.000+06:00
diff --git a/.changeset/define-runevent-schema.md b/.changeset/define-runevent-schema.md
@@ -1,5 +1,5 @@
 ---
-"@trigger.dev/core": patch
+"@trigger.dev/core": minor
 ---
 
 Define RunEvent schema and update ApiClient to use it
diff --git a/packages/cli-v3/src/commands/deploy.ts b/packages/cli-v3/src/commands/deploy.ts
@@ -501,9 +501,8 @@ async function _deployCommand(dir: string, options: DeployCommandOptions) {
   const version = deployment.version;
 
   const rawDeploymentLink = `${authorization.dashboardUrl}/projects/v3/${resolvedConfig.project}/deployments/${deployment.shortCode}`;
-  const rawTestLink = `${authorization.dashboardUrl}/projects/v3/${
-    resolvedConfig.project
-  }/test?environment=${options.env === "prod" ? "prod" : "stg"}`;
+  const rawTestLink = `${authorization.dashboardUrl}/projects/v3/${resolvedConfig.project
+    }/test?environment=${options.env === "prod" ? "prod" : "stg"}`;
 
   const deploymentLink = cliLink("View deployment", rawDeploymentLink);
   const testLink = cliLink("Test tasks", rawTestLink);
@@ -720,8 +719,7 @@ async function _deployCommand(dir: string, options: DeployCommandOptions) {
     }
   } else {
     outro(
-      `Version ${version} deployed with ${taskCount} detected task${taskCount === 1 ? "" : "s"} ${
-        isLinksSupported ? `| ${deploymentLink} | ${testLink}` : ""
+      `Version ${version} deployed with ${taskCount} detected task${taskCount === 1 ? "" : "s"} ${isLinksSupported ? `| ${deploymentLink} | ${testLink}` : ""
       }`
     );
 
@@ -745,18 +743,16 @@ async function _deployCommand(dir: string, options: DeployCommandOptions) {
       TRIGGER_VERSION: version,
       TRIGGER_DEPLOYMENT_SHORT_CODE: deployment.shortCode,
       TRIGGER_DEPLOYMENT_URL: `${authorization.dashboardUrl}/projects/v3/${resolvedConfig.project}/deployments/${deployment.shortCode}`,
-      TRIGGER_TEST_URL: `${authorization.dashboardUrl}/projects/v3/${
-        resolvedConfig.project
-      }/test?environment=${options.env === "prod" ? "prod" : "stg"}`,
+      TRIGGER_TEST_URL: `${authorization.dashboardUrl}/projects/v3/${resolvedConfig.project
+        }/test?environment=${options.env === "prod" ? "prod" : "stg"}`,
     },
     outputs: {
       deploymentVersion: version,
       workerVersion: version,
       deploymentShortCode: deployment.shortCode,
       deploymentUrl: `${authorization.dashboardUrl}/projects/v3/${resolvedConfig.project}/deployments/${deployment.shortCode}`,
-      testUrl: `${authorization.dashboardUrl}/projects/v3/${
-        resolvedConfig.project
-      }/test?environment=${options.env === "prod" ? "prod" : "stg"}`,
+      testUrl: `${authorization.dashboardUrl}/projects/v3/${resolvedConfig.project
+        }/test?environment=${options.env === "prod" ? "prod" : "stg"}`,
       needsPromotion: options.skipPromotion ? "true" : "false",
     },
   });
@@ -799,8 +795,7 @@ async function failDeploy(
       checkLogsForErrors(logs);
 
       outro(
-        `${chalkError(`${prefix}:`)} ${
-          error.message
+        `${chalkError(`${prefix}:`)} ${error.message
         }. Full build logs have been saved to ${logPath}`
       );
 
@@ -1100,9 +1095,8 @@ async function handleNativeBuildServerDeploy({
   const deployment = initializeDeploymentResult.data;
 
   const rawDeploymentLink = `${dashboardUrl}/projects/v3/${config.project}/deployments/${deployment.shortCode}`;
-  const rawTestLink = `${dashboardUrl}/projects/v3/${config.project}/test?environment=${
-    options.env === "prod" ? "prod" : "stg"
-  }`;
+  const rawTestLink = `${dashboardUrl}/projects/v3/${config.project}/test?environment=${options.env === "prod" ? "prod" : "stg"
+    }`;
 
   const exposedDeploymentLink = isLinksSupported
     ? cliLink(chalk.bold(rawDeploymentLink), rawDeploymentLink)
@@ -1155,8 +1149,9 @@ async function handleNativeBuildServerDeploy({
   const [readSessionError, readSession] = await tryCatch(
     stream.readSession(
       {
-        start: { from: { seqNum: 0 }, clamp: true },
-        stop: { waitSecs: 60 * 20 }, // 20 minutes
+        seq_num: 0,
+        clamp: true,
+        wait: 60 * 20, // 20 minutes
       },
       { signal: abortController.signal }
     )
@@ -1167,8 +1162,7 @@ async function handleNativeBuildServerDeploy({
     log.warn(`Failed streaming build logs, open the deployment in the dashboard to view the logs`);
 
     outro(
-      `Version ${deployment.version} is being deployed ${
-        isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
+      `Version ${deployment.version} is being deployed ${isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
       }`
     );
 
@@ -1193,7 +1187,7 @@ async function handleNativeBuildServerDeploy({
 
     switch (event.type) {
       case "log": {
-        if (record.seqNum === 0) {
+        if (record.seq_num === 0) {
           $queuedSpinner.stop("Build started");
           console.log("│");
           queuedSpinnerStopped = true;
@@ -1214,10 +1208,10 @@ async function handleNativeBuildServerDeploy({
           level === "error"
             ? chalk.bold(chalkError(message))
             : level === "warn"
-            ? chalkWarning(message)
-            : level === "debug"
-            ? chalkGrey(message)
-            : message;
+              ? chalkWarning(message)
+              : level === "debug"
+                ? chalkGrey(message)
+                : message;
 
         // We use console.log here instead of clack's logger as the current version does not support changing the line spacing.
         // And the logs look verbose with the default spacing.
@@ -1250,8 +1244,7 @@ async function handleNativeBuildServerDeploy({
     log.error("Failed dequeueing build, please try again shortly");
 
     throw new OutroCommandError(
-      `Version ${deployment.version} ${
-        isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
+      `Version ${deployment.version} ${isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
       }`
     );
   }
@@ -1266,8 +1259,7 @@ async function handleNativeBuildServerDeploy({
     }
 
     throw new OutroCommandError(
-      `Version ${deployment.version} ${
-        isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
+      `Version ${deployment.version} ${isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
       }`
     );
   }
@@ -1293,13 +1285,12 @@ async function handleNativeBuildServerDeploy({
       }
 
       outro(
-        `Version ${deployment.version} was deployed ${
-          isLinksSupported
-            ? `| ${cliLink("Test tasks", rawTestLink)} | ${cliLink(
-                "View deployment",
-                rawDeploymentLink
-              )}`
-            : ""
+        `Version ${deployment.version} was deployed ${isLinksSupported
+          ? `| ${cliLink("Test tasks", rawTestLink)} | ${cliLink(
+            "View deployment",
+            rawDeploymentLink
+          )}`
+          : ""
         }`
       );
       return process.exit(0);
@@ -1313,14 +1304,13 @@ async function handleNativeBuildServerDeploy({
         chalk.bold(
           chalkError(
             "Deployment failed" +
-              (finalDeploymentEvent.message ? `: ${finalDeploymentEvent.message}` : "")
+            (finalDeploymentEvent.message ? `: ${finalDeploymentEvent.message}` : "")
           )
         )
       );
 
       throw new OutroCommandError(
-        `Version ${deployment.version} deployment failed ${
-          isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
+        `Version ${deployment.version} deployment failed ${isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
         }`
       );
     }
@@ -1333,14 +1323,13 @@ async function handleNativeBuildServerDeploy({
         chalk.bold(
           chalkError(
             "Deployment timed out" +
-              (finalDeploymentEvent.message ? `: ${finalDeploymentEvent.message}` : "")
+            (finalDeploymentEvent.message ? `: ${finalDeploymentEvent.message}` : "")
           )
         )
       );
 
       throw new OutroCommandError(
-        `Version ${deployment.version} deployment timed out ${
-          isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
+        `Version ${deployment.version} deployment timed out ${isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
         }`
       );
     }
@@ -1353,14 +1342,13 @@ async function handleNativeBuildServerDeploy({
         chalk.bold(
           chalkError(
             "Deployment was canceled" +
-              (finalDeploymentEvent.message ? `: ${finalDeploymentEvent.message}` : "")
+            (finalDeploymentEvent.message ? `: ${finalDeploymentEvent.message}` : "")
           )
         )
       );
 
       throw new OutroCommandError(
-        `Version ${deployment.version} deployment canceled ${
-          isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
+        `Version ${deployment.version} deployment canceled ${isLinksSupported ? `| ${cliLink("View deployment", rawDeploymentLink)}` : ""
         }`
       );
     }
@@ -1379,13 +1367,12 @@ async function handleNativeBuildServerDeploy({
       }
 
       outro(
-        `Version ${deployment.version} ${
-          isLinksSupported
-            ? `| ${cliLink("Test tasks", rawTestLink)} | ${cliLink(
-                "View deployment",
-                rawDeploymentLink
-              )}`
-            : ""
+        `Version ${deployment.version} ${isLinksSupported
+          ? `| ${cliLink("Test tasks", rawTestLink)} | ${cliLink(
+            "View deployment",
+            rawDeploymentLink
+          )}`
+          : ""
         }`
       );
       return process.exit(0);
diff --git a/packages/cli-v3/tsc_output.txt b/packages/cli-v3/tsc_output.txt
diff --git a/packages/core/src/v3/realtimeStreams/streamsWriterV2.ts b/packages/core/src/v3/realtimeStreams/streamsWriterV2.ts
@@ -61,11 +61,11 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
       accessToken: options.accessToken,
       ...(options.endpoint
         ? {
-            endpoints: {
-              account: options.endpoint,
-              basin: options.endpoint,
-            },
-          }
+          endpoints: {
+            account: options.endpoint,
+            basin: options.endpoint,
+          },
+        }
         : {}),
     });
     this.flushIntervalMs = options.flushIntervalMs ?? 200;
@@ -135,7 +135,7 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
       const stream = basin.stream(this.options.stream);
 
       const session = await stream.appendSession({
-        maxInflightBytes: this.maxInflightBytes,
+        maxQueuedBytes: this.maxInflightBytes,
       });
 
       this.sessionWritable = session.writable;
@@ -152,7 +152,7 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
                 return;
               }
               // Convert each chunk to JSON string and wrap in AppendRecord
-              controller.enqueue(AppendRecord.string({ body: JSON.stringify({ data: chunk, id: nanoid(7) }) }));
+              controller.enqueue(AppendRecord.make(JSON.stringify({ data: chunk, id: nanoid(7) })));
             },
           })
         )
@@ -169,9 +169,9 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
       const lastAcked = session.lastAckedPosition();
 
       if (lastAcked?.end) {
-        const recordsWritten = lastAcked.end.seqNum;
+        const recordsWritten = lastAcked.end.seq_num;
         this.log(
-          `[S2MetadataStream] Written ${recordsWritten} records, ending at seqNum=${lastAcked.end.seqNum}`
+          `[S2MetadataStream] Written ${recordsWritten} records, ending at seq_num=${lastAcked.end.seq_num}`
         );
       }
     } catch (error) {
@@ -223,5 +223,5 @@ async function* streamToAsyncIterator<T>(stream: ReadableStream<T>): AsyncIterab
 function safeReleaseLock(reader: ReadableStreamDefaultReader<any>) {
   try {
     reader.releaseLock();
-  } catch (error) {}
+  } catch (error) { }
 }
diff --git a/packages/core/src/v3/schemas/api-type.test.ts b/packages/core/src/v3/schemas/api-type.test.ts
@@ -1,5 +1,5 @@
 import { describe, it, expect } from "vitest";
-import { InitializeDeploymentRequestBody, RunEvent, ListRunEventsResponse } from "./api.js";
+import { InitializeDeploymentRequestBody, RunEvent, ListRunEventsResponse, ListRunEventsResponseWithStringDates } from "./api.js";
 import type { InitializeDeploymentRequestBody as InitializeDeploymentRequestBodyType } from "./api.js";
 
 describe("InitializeDeploymentRequestBody", () => {
@@ -197,6 +197,18 @@ describe("RunEvent Schema", () => {
     expect(result.startTime.toISOString()).toBe("2024-03-14T00:00:00.000Z");
   });
 
+  it("should handle Date object", () => {
+    const now = new Date();
+    const result = RunEvent.safeParse({
+      ...validEvent,
+      startTime: now,
+    });
+    expect(result.success).toBe(true);
+    if (result.success) {
+      expect(result.data.startTime.toISOString()).toBe(now.toISOString());
+    }
+  });
+
   it("handles bigint nanosecond startTime", () => {
     const event = { ...validEvent, startTime: 1710374400000000000n };
     const result = RunEvent.parse(event as any);
@@ -238,6 +250,19 @@ describe("RunEvent Schema", () => {
     const result = RunEvent.parse(eventWithSlug);
     expect(result.taskSlug).toBe("my-task");
   });
+
+  it("ListRunEventsResponseWithStringDates correctly transforms Dates to strings", () => {
+    const rawResponse = {
+      events: [validEvent],
+    };
+
+    const parsed = ListRunEventsResponse.parse(rawResponse);
+    expect(parsed.events[0]!.startTime).toBeInstanceOf(Date);
+
+    const legacy = ListRunEventsResponseWithStringDates.parse(rawResponse);
+    expect(typeof legacy.events[0]!.startTime).toBe("string");
+    expect(legacy.events[0]!.startTime).toBe(parsed.events[0]!.startTime.toISOString());
+  });
 });
 
 describe("ListRunEventsResponse Schema", () => {
diff --git a/packages/core/src/v3/schemas/api.ts b/packages/core/src/v3/schemas/api.ts
