fix(core): define RunEvent schema and update ApiClient validation (PR #3220)

- Add RunEvent, TaskEventLevel schemas to api.ts
- Add ListRunEventsResponse schema
- Add RunEvent import to ApiClient
- Add changeset for define-runevent-schema

Closes #3220

Author: deepshekhardas

.changeset/define-runevent-schema.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": minor
+---
+
+Define RunEvent schema and update ApiClient to use it

packages/cli-v3/src/commands/deploy.ts

@@ -1149,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
+        start: {
+          from: { seqNum: 0 },
+        },
       },
       { signal: abortController.signal }
     )

packages/core/src/v3/apiClient/index.ts

@@ -41,6 +41,8 @@ import {
   EnvironmentVariableResponseBody,
   EnvironmentVariableWithSecret,
   ListQueueOptions,
+  RunEvent,
+  ListRunEventsResponse,
   ListRunResponseItem,
   ListScheduleOptions,
   QueueItem,
@@ -748,7 +750,7 @@ export class ApiClient {
 
   listRunEvents(runId: string, requestOptions?: ZodFetchOptions) {
     return zodfetch(
-      z.any(), // TODO: define a proper schema for this
+      ListRunEventsResponse,
       `${this.baseUrl}/api/v1/runs/${runId}/events`,
       {
         method: "GET",

packages/core/src/v3/realtimeStreams/streamsWriterV2.ts

@@ -71,11 +71,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;
@@ -238,7 +238,7 @@ async function* streamToAsyncIterator<T>(stream: ReadableStream<T>): AsyncIterab
 function safeReleaseLock(reader: ReadableStreamDefaultReader<any>) {
   try {
     reader.releaseLock();
-  } catch (error) {}
+  } catch (error) { }
 }
 
 // chat.agent emits two chunk shapes through this writer:

packages/core/src/v3/schemas/api-type.test.ts

@@ -1,5 +1,5 @@
 import { describe, it, expect } from "vitest";
-import { InitializeDeploymentRequestBody } from "./api.js";
+import { InitializeDeploymentRequestBody, RunEvent, ListRunEventsResponse, ListRunEventsResponseWithStringDates } from "./api.js";
 import type { InitializeDeploymentRequestBody as InitializeDeploymentRequestBodyType } from "./api.js";
 
 describe("InitializeDeploymentRequestBody", () => {
@@ -139,3 +139,202 @@ describe("InitializeDeploymentRequestBody", () => {
     });
   });
 });
+
+describe("RunEvent Schema", () => {
+  const validEvent = {
+    spanId: "span_123",
+    parentId: "span_root",
+    runId: "run_abc",
+    message: "Test event",
+    style: {
+      icon: "task",
+      variant: "primary",
+    },
+    startTime: "2024-03-14T00:00:00Z",
+    duration: 1234,
+    isError: false,
+    isPartial: false,
+    isCancelled: false,
+    level: "INFO",
+    kind: "TASK",
+    attemptNumber: 1,
+  };
+
+  it("parses a valid event correctly", () => {
+    const result = RunEvent.safeParse(validEvent);
+    expect(result.success).toBe(true);
+    if (result.success) {
+      expect(result.data.spanId).toBe("span_123");
+      expect(result.data.startTime).toBeInstanceOf(Date);
+      expect(result.data.level).toBe("INFO");
+    }
+  });
+
+  it("fails on missing required fields", () => {
+    const invalidEvent = { ...validEvent };
+    delete (invalidEvent as any).spanId;
+    const result = RunEvent.safeParse(invalidEvent);
+    expect(result.success).toBe(false);
+  });
+
+  it("fails on invalid level", () => {
+    const invalidEvent = { ...validEvent, level: "INVALID_LEVEL" };
+    const result = RunEvent.safeParse(invalidEvent);
+    expect(result.success).toBe(false);
+  });
+
+  it("coerces startTime to Date", () => {
+    const result = RunEvent.parse(validEvent);
+    expect(result.startTime).toBeInstanceOf(Date);
+    expect(result.startTime.toISOString()).toBe("2024-03-14T00:00:00.000Z");
+  });
+
+  it("handles 19-digit nanosecond startTime strings", () => {
+    const event = { ...validEvent, startTime: "1710374400000000000" };
+    const result = RunEvent.parse(event);
+    expect(result.startTime).toBeInstanceOf(Date);
+    // 1710374400000000000 ns = 1710374400000 ms = 2024-03-14T00:00:00Z
+    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);
+    expect(result.startTime).toBeInstanceOf(Date);
+    expect(result.startTime.toISOString()).toBe("2024-03-14T00:00:00.000Z");
+  });
+
+  it("fails on invalid startTime", () => {
+    const event = { ...validEvent, startTime: "not-a-date" };
+    const result = RunEvent.safeParse(event);
+    expect(result.success).toBe(false);
+  });
+
+  describe("startTime edge cases", () => {
+    it("should handle whitespace-padded strings", () => {
+      const result = RunEvent.safeParse({
+        ...validEvent,
+        startTime: "  2024-03-14T00:00:00Z  ",
+      });
+      expect(result.success).toBe(true);
+      if (result.success) {
+        expect(result.data.startTime.toISOString()).toBe("2024-03-14T00:00:00.000Z");
+      }
+    });
+
+    it("should handle whitespace-padded nanosecond strings", () => {
+      const result = RunEvent.safeParse({
+        ...validEvent,
+        startTime: "  1710374400000000000  ",
+      });
+      expect(result.success).toBe(true);
+      if (result.success) {
+        expect(result.data.startTime.toISOString()).toBe("2024-03-14T00:00:00.000Z");
+      }
+    });
+
+    it("should fail on empty string", () => {
+      const result = RunEvent.safeParse({
+        ...validEvent,
+        startTime: "",
+      });
+      expect(result.success).toBe(false);
+    });
+
+    it("should fail on whitespace-only string", () => {
+      const result = RunEvent.safeParse({
+        ...validEvent,
+        startTime: "   ",
+      });
+      expect(result.success).toBe(false);
+    });
+  });
+
+  it("allows optional/null parentId", () => {
+    const eventWithoutParent = { ...validEvent };
+    delete (eventWithoutParent as any).parentId;
+    expect(RunEvent.safeParse(eventWithoutParent).success).toBe(true);
+
+    const eventWithNullParent = { ...validEvent, parentId: null };
+    expect(RunEvent.safeParse(eventWithNullParent).success).toBe(true);
+  });
+
+  it("allows nullish attemptNumber", () => {
+    const eventWithNullAttempt = { ...validEvent, attemptNumber: null };
+    const result = RunEvent.safeParse(eventWithNullAttempt);
+    expect(result.success).toBe(true);
+    if (result.success) {
+      expect(result.data.attemptNumber).toBe(null);
+    }
+
+    const eventWithoutAttempt = { ...validEvent };
+    delete (eventWithoutAttempt as any).attemptNumber;
+    const result2 = RunEvent.safeParse(eventWithoutAttempt);
+    expect(result2.success).toBe(true);
+  });
+
+  it("supports taskSlug", () => {
+    const eventWithSlug = { ...validEvent, taskSlug: "my-task" };
+    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", () => {
+  it("parses a valid wrapped response", () => {
+    const response = {
+      events: [
+        {
+          spanId: "span_1",
+          runId: "run_1",
+          message: "Event 1",
+          style: {},
+          startTime: "2024-03-14T00:00:00Z",
+          duration: 100,
+          isError: false,
+          isPartial: false,
+          isCancelled: false,
+          level: "INFO",
+          kind: "TASK",
+        },
+      ],
+    };
+
+    const result = ListRunEventsResponse.safeParse(response);
+    expect(result.success).toBe(true);
+    if (result.success && result.data) {
+      expect(result.data.events[0]!.spanId).toBe("span_1");
+    }
+  });
+
+  it("fails on plain array", () => {
+    const response = [{ spanId: "span_1" }];
+    const result = ListRunEventsResponse.safeParse(response);
+    expect(result.success).toBe(false);
+  });
+});

packages/core/src/v3/schemas/api.ts

@@ -9,6 +9,8 @@ import {
 } from "./common.js";
 import { BackgroundWorkerMetadata } from "./resources.js";
 import { DequeuedMessage, MachineResources } from "./runEngine.js";
+import { TaskEventStyle } from "./style.js";
+import { SpanEvents } from "./openTelemetry.js";
 
 export const RunEngineVersion = z.union([z.literal("V1"), z.literal("V2")]);
 
@@ -1991,14 +1993,35 @@ export const SendInputStreamResponseBody = z.object({
 });
 export type SendInputStreamResponseBody = z.infer<typeof SendInputStreamResponseBody>;
 
-/**
- * Response body for `GET /realtime/v1/sessions/:id/:io/records`. A non-SSE,
- * `wait=0` drain of a session channel — used at run boot for snapshot
- * replay where the SSE long-poll tax (~1s on empty streams) was the
- * dominant cost. The shape mirrors the webapp's internal `StreamRecord`
- * type (`apps/webapp/app/services/realtime/types.ts`); each record's
- * `data` is a JSON-encoded chunk body that callers parse client-side.
- */
+export const TaskEventLevel = z.enum(["TRACE", "DEBUG", "INFO", "LOG", "WARN", "ERROR"]);
+export type TaskEventLevel = z.infer<typeof TaskEventLevel>;
+
+export const RunEvent = z.object({
+  spanId: z.string(),
+  parentId: z.string().nullish(),
+  runId: z.string(),
+  message: z.string(),
+  style: TaskEventStyle,
+  startTime: z.coerce.date(),
+  duration: z.number(),
+  isError: z.boolean(),
+  isPartial: z.boolean(),
+  isCancelled: z.boolean(),
+  level: TaskEventLevel,
+  events: SpanEvents.optional(),
+  kind: z.string(),
+  attemptNumber: z.number().nullish(),
+  taskSlug: z.string().optional(),
+});
+
+export type RunEvent = z.infer<typeof RunEvent>;
+
+export const ListRunEventsResponse = z.object({
+  events: z.array(RunEvent),
+});
+
+export type ListRunEventsResponse = z.infer<typeof ListRunEventsResponse>;
+
 export const ReadSessionStreamRecordsResponseBody = z.object({
   records: z.array(
     z.object({