Completing and skipping waitpoints is working

Author: matt-aitken

apps/webapp/app/presenters/v3/SpanPresenter.server.ts

@@ -11,6 +11,7 @@ import { machinePresetFromName } from "~/v3/machinePresets.server";
 import { FINAL_ATTEMPT_STATUSES, isFailedRunStatus, isFinalRunStatus } from "~/v3/taskStatus";
 import { BasePresenter } from "./basePresenter.server";
 import { getMaxDuration } from "@trigger.dev/core/v3/apps";
+import { logger } from "~/services/logger.server";
 
 type Result = Awaited<ReturnType<SpanPresenter["call"]>>;
 export type Span = NonNullable<NonNullable<Result>["span"]>;
@@ -43,6 +44,7 @@ export class SpanPresenter extends BasePresenter {
     const parentRun = await this._prisma.taskRun.findFirst({
       select: {
         traceId: true,
+        runtimeEnvironmentId: true,
       },
       where: {
         friendlyId: runFriendlyId,
@@ -64,7 +66,7 @@ export class SpanPresenter extends BasePresenter {
     }
 
     //get the run
-    const span = await this.getSpan(traceId, spanId);
+    const span = await this.getSpan(traceId, spanId, parentRun.runtimeEnvironmentId);
 
     if (!span) {
       throw new Error("Span not found");
@@ -399,7 +401,7 @@ export class SpanPresenter extends BasePresenter {
     };
   }
 
-  async getSpan(traceId: string, spanId: string) {
+  async getSpan(traceId: string, spanId: string, environmentId: string) {
     const span = await eventRepository.getSpan(spanId, traceId);
     if (!span) {
       return;
@@ -451,14 +453,42 @@ export class SpanPresenter extends BasePresenter {
           },
         });
 
+        if (!waitpoint) {
+          logger.error(`SpanPresenter: Waitpoint not found`, {
+            spanId,
+            waitpointFriendlyId: span.entity.id,
+          });
+          return { ...data, entity: null };
+        }
+
+        const output =
+          waitpoint.outputType === "application/store"
+            ? `/resources/packets/${environmentId}/${waitpoint.output}`
+            : typeof waitpoint.output !== "undefined" && waitpoint.output !== null
+            ? await prettyPrintPacket(waitpoint.output, waitpoint.outputType ?? undefined)
+            : undefined;
+
         return {
           ...data,
-          entityType: "waitpoint" as const,
-          waitpoint,
+          entity: {
+            type: "waitpoint" as const,
+            object: {
+              friendlyId: waitpoint.friendlyId,
+              type: waitpoint.type,
+              status: waitpoint.status,
+              idempotencyKey: waitpoint.idempotencyKey,
+              userProvidedIdempotencyKey: waitpoint.userProvidedIdempotencyKey,
+              idempotencyKeyExpiresAt: waitpoint.idempotencyKeyExpiresAt,
+              output: output,
+              outputType: waitpoint.outputType,
+              outputIsError: waitpoint.outputIsError,
+              completedAfter: waitpoint.completedAfter,
+            },
+          },
         };
 
       default:
-        return { ...data, entityType: "span" as const };
+        return { ...data, entity: null };
     }
   }
 }

apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.complete.ts

@@ -8,6 +8,7 @@ import {
 import { WaitpointId } from "@trigger.dev/core/v3/apps";
 import { z } from "zod";
 import { $replica } from "~/db.server";
+import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { engine } from "~/v3/runEngine.server";
@@ -18,20 +19,14 @@ const { action } = createActionApiRoute(
       waitpointFriendlyId: z.string(),
     }),
     body: CompleteWaitpointTokenRequestBody,
-    maxContentLength: 1024 * 10, // 10KB
+    maxContentLength: env.TASK_PAYLOAD_MAXIMUM_SIZE,
     method: "POST",
   },
   async ({ authentication, body, params }) => {
     // Resume tokens are actually just waitpoints
     const waitpointId = WaitpointId.toId(params.waitpointFriendlyId);
 
     try {
-      const stringifiedData = await stringifyIO(body.data);
-      const finalData = await conditionallyExportPacket(
-        stringifiedData,
-        `${waitpointId}/waitpoint/token`
-      );
-
       //check permissions
       const waitpoint = await $replica.waitpoint.findFirst({
         where: {
@@ -44,6 +39,12 @@ const { action } = createActionApiRoute(
         throw json({ error: "Waitpoint not found" }, { status: 404 });
       }
 
+      const stringifiedData = await stringifyIO(body.data);
+      const finalData = await conditionallyExportPacket(
+        stringifiedData,
+        `${waitpointId}/waitpoint/token`
+      );
+
       const result = await engine.completeWaitpoint({
         id: waitpointId,
         output: finalData.data

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.waitpoints.$waitpointFriendlyId.complete/route.tsx

@@ -2,9 +2,9 @@ import { parse } from "@conform-to/zod";
 import { InformationCircleIcon } from "@heroicons/react/20/solid";
 import { Form, useLocation, useNavigation, useSubmit } from "@remix-run/react";
 import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
+import { conditionallyExportPacket, stringifyIO } from "@trigger.dev/core/v3";
 import { WaitpointId } from "@trigger.dev/core/v3/apps";
 import { Waitpoint } from "@trigger.dev/database";
-import { motion } from "framer-motion";
 import { useCallback, useRef } from "react";
 import { z } from "zod";
 import { AnimatedHourglassIcon } from "~/assets/icons/AnimatedHourglassIcon";
@@ -14,16 +14,14 @@ import { Button } from "~/components/primitives/Buttons";
 import { DateTime } from "~/components/primitives/DateTime";
 import { Paragraph } from "~/components/primitives/Paragraph";
 import { LiveCountdown } from "~/components/runs/v3/LiveTimer";
-import { $replica, prisma } from "~/db.server";
+import { $replica } from "~/db.server";
 import { useOrganization } from "~/hooks/useOrganizations";
 import { useProject } from "~/hooks/useProject";
 import { redirectWithErrorMessage, redirectWithSuccessMessage } from "~/models/message.server";
 import { logger } from "~/services/logger.server";
 import { requireUserId } from "~/services/session.server";
-import { ProjectParamSchema, v3RunsPath, v3SchedulesPath } from "~/utils/pathBuilder";
+import { ProjectParamSchema, v3RunsPath } from "~/utils/pathBuilder";
 import { engine } from "~/v3/runEngine.server";
-import { UpsertSchedule } from "~/v3/schedules";
-import { UpsertTaskScheduleService } from "~/v3/services/upsertTaskSchedule.server";
 
 const CompleteWaitpointFormData = z.discriminatedUnion("type", [
   z.object({
@@ -106,8 +104,34 @@ export const action = async ({ request, params }: ActionFunctionArgs) => {
         );
       }
       case "MANUAL": {
-        //todo packet
-        //todo completeWaitpoint
+        let data: any;
+        try {
+          data = JSON.parse(submission.value.payload);
+        } catch (e) {
+          return redirectWithErrorMessage(
+            submission.value.failureRedirect,
+            request,
+            "Invalid payload, must be valid JSON"
+          );
+        }
+        const stringifiedData = await stringifyIO(data);
+        const finalData = await conditionallyExportPacket(
+          stringifiedData,
+          `${waitpointId}/waitpoint/token`
+        );
+
+        const result = await engine.completeWaitpoint({
+          id: waitpointId,
+          output: finalData.data
+            ? { type: finalData.dataType, value: finalData.data, isError: false }
+            : undefined,
+        });
+
+        return redirectWithSuccessMessage(
+          submission.value.successRedirect,
+          request,
+          "Waitpoint completed"
+        );
       }
     }
   } catch (error: any) {
@@ -259,7 +283,7 @@ function CompleteManualWaitpointForm({ waitpoint }: { waitpoint: { friendlyId: s
         onSubmit={(e) => submitForm(e)}
         className="grid h-full max-h-full grid-rows-[2.5rem_1fr_2.5rem] overflow-hidden rounded-md border border-grid-bright"
       >
-        <input type="hidden" name="type" value={"DATETIME"} />
+        <input type="hidden" name="type" value={"MANUAL"} />
         <input
           type="hidden"
           name="successRedirect"
@@ -313,7 +337,7 @@ function CompleteManualWaitpointForm({ waitpoint }: { waitpoint: { friendlyId: s
         }
         code={`
 await wait.completeToken<YourType>(tokenId,
-output
+  output
 );`}
         showLineNumbers={false}
       />