Fix undefined task outputs

And also fix the SSE connection getting reset on a infinite loop when runs are complete

apps/webapp/app/components/run/TaskDetail.tsx

@@ -33,7 +33,19 @@ import { Spinner } from "../primitives/Spinner";
 import type { DetailedTask } from "~/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam.tasks.$taskParam/route";
 
 export function TaskDetail({ task }: { task: DetailedTask }) {
-  const { name, description, icon, status, params, properties, output, style, attempts } = task;
+  const {
+    name,
+    description,
+    icon,
+    status,
+    params,
+    properties,
+    output,
+    outputIsUndefined,
+    style,
+    attempts,
+    noop,
+  } = task;
 
   const startedAt = task.startedAt ? new Date(task.startedAt) : undefined;
   const completedAt = task.completedAt ? new Date(task.completedAt) : undefined;
@@ -140,16 +152,18 @@ export function TaskDetail({ task }: { task: DetailedTask }) {
             <Paragraph variant="small">No input</Paragraph>
           )}
         </div>
-        <div className="mt-4 flex flex-col gap-2">
-          <Header3>Output</Header3>
-          {output ? (
-            <ClientOnly fallback={<Spinner />}>
-              {() => <CodeBlock code={output} maxLines={35} />}
-            </ClientOnly>
-          ) : (
-            <Paragraph variant="small">No output</Paragraph>
-          )}
-        </div>
+        {!noop && (
+          <div className="mt-4 flex flex-col gap-2">
+            <Header3>Output</Header3>
+            {output && !outputIsUndefined ? (
+              <ClientOnly fallback={<Spinner />}>
+                {() => <CodeBlock code={output} maxLines={35} />}
+              </ClientOnly>
+            ) : (
+              <Paragraph variant="small">No output</Paragraph>
+            )}
+          </div>
+        )}
       </RunPanelBody>
     </RunPanel>
   );

apps/webapp/app/hooks/useEventSource.tsx

@@ -3,6 +3,7 @@ import { useEffect, useState } from "react";
 type EventSourceOptions = {
   init?: EventSourceInit;
   event?: string;
+  disabled?: boolean;
 };
 
 /**
@@ -13,11 +14,15 @@ type EventSourceOptions = {
  */
 export function useEventSource(
   url: string | URL,
-  { event = "message", init }: EventSourceOptions = {}
+  { event = "message", init, disabled }: EventSourceOptions = {}
 ) {
   const [data, setData] = useState<string | null>(null);
 
   useEffect(() => {
+    if (disabled) {
+      return;
+    }
+
     const eventSource = new EventSource(url, init);
     eventSource.addEventListener(event ?? "message", handler);
 
@@ -32,7 +37,7 @@ export function useEventSource(
       eventSource.removeEventListener(event ?? "message", handler);
       eventSource.close();
     };
-  }, [url, event, init]);
+  }, [url, event, init, disabled]);
 
   return data;
 }

apps/webapp/app/models/task.server.ts

@@ -15,7 +15,7 @@ export function taskWithAttemptsToServerTask(task: TaskWithAttempts): ServerTask
     status: task.status,
     description: task.description,
     params: task.params as any,
-    output: task.output as any,
+    output: task.outputIsUndefined ? undefined : (task.output as any),
     context: task.context as any,
     properties: task.properties as any,
     style: task.style as any,
@@ -32,7 +32,7 @@ export function taskWithAttemptsToServerTask(task: TaskWithAttempts): ServerTask
 
 export type TaskForCaching = Pick<
   Task,
-  "id" | "status" | "idempotencyKey" | "noop" | "output" | "parentId"
+  "id" | "status" | "idempotencyKey" | "noop" | "output" | "parentId" | "outputIsUndefined"
 >;
 
 export function prepareTasksForCaching(
@@ -105,7 +105,7 @@ function prepareTaskForCaching(task: TaskForCaching): CachedTask {
     status: task.status,
     idempotencyKey: task.idempotencyKey,
     noop: task.noop,
-    output: task.output as any,
+    output: task.outputIsUndefined ? undefined : (task.output as any),
     parentId: task.parentId,
   };
 }

apps/webapp/app/presenters/TaskDetailsPresenter.server.ts

@@ -1,5 +1,4 @@
-import { RedactSchema } from "@trigger.dev/core";
-import { StyleSchema } from "@trigger.dev/core";
+import { RedactSchema, StyleSchema } from "@trigger.dev/core";
 import { PrismaClient, prisma } from "~/db.server";
 import { mergeProperties } from "~/utils/mergeProperties.server";
 import { Redactor } from "~/utils/redactor";
@@ -58,6 +57,7 @@ export class TaskDetailsPresenter {
         outputProperties: true,
         params: true,
         output: true,
+        outputIsUndefined: true,
         error: true,
         startedAt: true,
         completedAt: true,
@@ -89,9 +89,11 @@ export class TaskDetailsPresenter {
     return {
       ...task,
       redact: undefined,
-      output: task.output
-        ? JSON.stringify(this.#stringifyOutputWithRedactions(task.output, task.redact), null, 2)
-        : undefined,
+      output: JSON.stringify(
+        this.#stringifyOutputWithRedactions(task.output, task.redact),
+        null,
+        2
+      ),
       connection: task.runConnection,
       params: task.params as Record<string, any>,
       properties: mergeProperties(task.properties, task.outputProperties),
@@ -101,7 +103,7 @@ export class TaskDetailsPresenter {
 
   #stringifyOutputWithRedactions(output: any, redact: unknown): any {
     if (!output) {
-      return;
+      return output;
     }
 
     const parsedRedact = RedactSchema.safeParse(redact);

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam.tasks.$taskParam/route.tsx

@@ -1,6 +1,5 @@
-import { Await, useLoaderData } from "@remix-run/react";
-import { LoaderFunctionArgs, SerializeFrom, defer } from "@remix-run/server-runtime";
-import { Suspense } from "react";
+import { LoaderFunctionArgs } from "@remix-run/server-runtime";
+import { UseDataFunctionReturn, typedjson, useTypedLoaderData } from "remix-typedjson";
 import { Spinner } from "~/components/primitives/Spinner";
 import { TaskDetail } from "~/components/run/TaskDetail";
 import { TaskDetailsPresenter } from "~/presenters/TaskDetailsPresenter.server";
@@ -12,26 +11,28 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   const { taskParam } = TaskParamsSchema.parse(params);
 
   const presenter = new TaskDetailsPresenter();
-  const taskPromise = presenter.call({
+  const task = await presenter.call({
     userId,
     id: taskParam,
   });
 
-  return defer({
-    taskPromise,
+  return typedjson({
+    task,
   });
 };
 
-export type DetailedTask = NonNullable<Awaited<SerializeFrom<typeof loader>["taskPromise"]>>;
+export type DetailedTask = NonNullable<UseDataFunctionReturn<typeof loader>["task"]>;
 
 export default function Page() {
-  const { taskPromise } = useLoaderData<typeof loader>();
+  const { task } = useTypedLoaderData<typeof loader>();
 
-  return (
-    <Suspense fallback={<Spinner />}>
-      <Await resolve={taskPromise} errorElement={<p>Error loading task!</p>}>
-        {(resolvedTask) => resolvedTask && <TaskDetail task={resolvedTask as any} />}
-      </Await>
-    </Suspense>
-  );
+  if (!task) {
+    return (
+      <div className="flex h-full items-center justify-center">
+        <Spinner />
+      </div>
+    );
+  }
+
+  return <TaskDetail task={task} />;
 }

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam/route.tsx

@@ -67,6 +67,7 @@ export default function Page() {
   const revalidator = useRevalidator();
   const events = useEventSource(runStreamingPath(organization, project, job, run), {
     event: "message",
+    disabled: !!run.completedAt,
   });
   useEffect(() => {
     if (events !== null) {

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.external.$triggerParam_.runs.$runParam.tasks.$taskParam/route.tsx

@@ -1,6 +1,5 @@
-import { Await, useLoaderData } from "@remix-run/react";
-import { LoaderFunctionArgs, defer } from "@remix-run/server-runtime";
-import { Suspense } from "react";
+import { LoaderFunctionArgs } from "@remix-run/server-runtime";
+import { typedjson, useTypedLoaderData } from "remix-typedjson";
 import { Spinner } from "~/components/primitives/Spinner";
 import { TaskDetail } from "~/components/run/TaskDetail";
 import { TaskDetailsPresenter } from "~/presenters/TaskDetailsPresenter.server";
@@ -12,24 +11,26 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   const { taskParam } = TriggerSourceRunTaskParamsSchema.parse(params);
 
   const presenter = new TaskDetailsPresenter();
-  const taskPromise = presenter.call({
+  const task = await presenter.call({
     userId,
     id: taskParam,
   });
 
-  return defer({
-    taskPromise,
+  return typedjson({
+    task,
   });
 };
 
 export default function Page() {
-  const { taskPromise } = useLoaderData<typeof loader>();
+  const { task } = useTypedLoaderData<typeof loader>();
 
-  return (
-    <Suspense fallback={<Spinner />}>
-      <Await resolve={taskPromise} errorElement={<p>Error loading task!</p>}>
-        {(resolvedTask) => resolvedTask && <TaskDetail task={resolvedTask as any} />}
-      </Await>
-    </Suspense>
-  );
+  if (!task) {
+    return (
+      <div className="flex h-full items-center justify-center">
+        <Spinner />
+      </div>
+    );
+  }
+
+  return <TaskDetail task={task} />;
 }

apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete.ts

@@ -191,7 +191,8 @@ export class CompleteRunTaskService {
       },
       data: {
         status: "COMPLETED",
-        output: taskBody.output ?? undefined,
+        output: taskBody.output as any,
+        outputIsUndefined: typeof taskBody.output === "undefined",
         completedAt: new Date(),
         outputProperties: taskBody.properties,
       },

apps/webapp/app/services/runs/performRunExecutionV3.server.ts

@@ -1220,6 +1220,7 @@ async function findRun(prisma: PrismaClientOrTransaction, id: string) {
           status: true,
           noop: true,
           output: true,
+          outputIsUndefined: true,
           parentId: true,
         },
         orderBy: {

apps/webapp/app/utils/sse.server.ts

@@ -22,6 +22,16 @@ export function sse({ request, pingInterval = 1000, updateInterval = 348, run }:
     return new Response("SSE disabled", { status: 200 });
   }
 
+  let pinger: NodeJS.Timer | undefined = undefined;
+  let updater: NodeJS.Timer | undefined = undefined;
+  let timeout: NodeJS.Timeout | undefined = undefined;
+
+  const abort = () => {
+    clearInterval(pinger);
+    clearInterval(updater);
+    clearTimeout(timeout);
+  };
+
   return eventStream(request.signal, (send, close) => {
     const safeSend = (args: { event?: string; data: string }) => {
       try {
@@ -49,30 +59,26 @@ export function sse({ request, pingInterval = 1000, updateInterval = 348, run }:
       }
     };
 
-    const pinger = setInterval(() => {
+    pinger = setInterval(() => {
       if (request.signal.aborted) {
-        return close();
+        return abort();
       }
 
       safeSend({ event: "ping", data: new Date().toISOString() });
     }, pingInterval);
 
-    const updater = setInterval(() => {
+    updater = setInterval(() => {
       if (request.signal.aborted) {
-        return close();
+        return abort();
       }
 
-      run(safeSend, close);
+      run(safeSend, abort);
     }, updateInterval);
 
-    const timeout = setTimeout(() => {
-      close();
+    timeout = setTimeout(() => {
+      close(); // close the connection after 1 minute of inactivity, which will refresh the connection (that's why we aren't using abort)
     }, 60 * 1000); // 1 minute
 
-    return () => {
-      clearInterval(updater);
-      clearInterval(pinger);
-      clearTimeout(timeout);
-    };
+    return abort;
   });
 }

packages/database/prisma/migrations/20231113151412_add_output_is_undefined/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "Task" ADD COLUMN     "outputIsUndefined" BOOLEAN NOT NULL DEFAULT false;

packages/database/prisma/schema.prisma

@@ -882,17 +882,18 @@ model Task {
   delayUntil DateTime?
   noop       Boolean    @default(false)
 
-  description      String?
-  properties       Json?
-  outputProperties Json?
-  params           Json?
-  output           Json?
-  context          Json?
-  error            String?
-  redact           Json?
-  style            Json?
-  operation        String?
-  callbackUrl      String?
+  description       String?
+  properties        Json?
+  outputProperties  Json?
+  params            Json?
+  output            Json?
+  outputIsUndefined Boolean @default(false)
+  context           Json?
+  error             String?
+  redact            Json?
+  style             Json?
+  operation         String?
+  callbackUrl       String?
 
   startedAt   DateTime?
   completedAt DateTime?

references/job-catalog/package.json

@@ -31,6 +31,7 @@
     "cli-example": "nodemon --watch src/cli-example.ts -r tsconfig-paths/register -r dotenv/config src/cli-example.ts",
     "invoke": "nodemon --watch src/invoke.ts -r tsconfig-paths/register -r dotenv/config src/invoke.ts",
     "built-ins": "nodemon --watch src/built-ins.ts -r tsconfig-paths/register -r dotenv/config src/built-ins.ts",
+    "edge-cases": "nodemon --watch src/edge-cases.ts -r tsconfig-paths/register -r dotenv/config src/edge-cases.ts",
     "dev:trigger": "trigger-cli dev --port 8080"
   },
   "dependencies": {

references/job-catalog/src/edge-cases.ts

@@ -0,0 +1,99 @@
+import { createExpressServer } from "@trigger.dev/express";
+import { TriggerClient, invokeTrigger } from "@trigger.dev/sdk";
+import fs from "node:fs";
+import fsPromises from "node:fs/promises";
+
+export const client = new TriggerClient({
+  id: "job-catalog",
+  apiKey: process.env["TRIGGER_API_KEY"],
+  apiUrl: process.env["TRIGGER_API_URL"],
+  verbose: true,
+  ioLogLocalEnabled: true,
+});
+
+client.defineJob({
+  id: "task-output-edge-cases",
+  name: "Task Output Edge Cases",
+  version: "1.0.0",
+  trigger: invokeTrigger(),
+  run: async (payload, io, ctx) => {
+    const result1 = await io.runTask("undefined", async (task) => {
+      return undefined;
+    });
+
+    if (typeof result1 !== "undefined") {
+      throw new Error(`Expected undefined, got ${typeof result1}: ${JSON.stringify(result1)}`);
+    }
+
+    const result2 = await io.runTask("null", async (task) => {
+      return null;
+    });
+
+    if (result2 !== null) {
+      throw new Error(`Expected null, got ${typeof result2}: ${JSON.stringify(result2)}`);
+    }
+
+    const result3 = await io.runTask("false", async (task) => {
+      return false;
+    });
+
+    if (typeof result3 !== "boolean" && result3 !== false) {
+      throw new Error(`Expected false, got ${typeof result3}: ${JSON.stringify(result3)}`);
+    }
+
+    const result4 = await io.runTask("true", async (task) => {
+      return true;
+    });
+
+    if (typeof result4 !== "boolean" && result4 !== true) {
+      throw new Error(`Expected true, got ${typeof result4}: ${JSON.stringify(result4)}`);
+    }
+
+    const result5 = await io.runTask("date", async (task) => {
+      return new Date();
+    });
+
+    if (typeof result5 !== "string" || new Date(result5).toString() === "Invalid Date") {
+      throw new Error(`Expected string, got ${typeof result5}: ${JSON.stringify(result5)}`);
+    }
+
+    const result6 = await io.runTask("object", async (task) => {
+      return {
+        a: 1,
+        b: "2",
+        c: true,
+        d: new Date(),
+        e: null,
+        f: undefined,
+      };
+    });
+
+    if (typeof result6 !== "object" || result6 === null) {
+      throw new Error(`Expected object, got ${typeof result6}: ${JSON.stringify(result6)}`);
+    }
+
+    const result7 = await io.runTask("array", async (task) => {
+      return [1, "2", true, new Date(), null, undefined];
+    });
+
+    if (!Array.isArray(result7)) {
+      throw new Error(`Expected array, got ${typeof result7}: ${JSON.stringify(result7)}`);
+    }
+
+    const result8 = await io.runTask("file", async (task) => {
+      return fs.createReadStream(__filename) as any;
+    });
+
+    const result9 = await io.runTask("read-file", async (task) => {
+      return fsPromises.readFile(__filename, "utf-8");
+    });
+
+    if (typeof result9 !== "string") {
+      throw new Error(`Expected string, got ${typeof result9}: ${JSON.stringify(result9)}`);
+    }
+
+    await io.wait("wait-1", 1);
+  },
+});
+
+createExpressServer(client);