wait.retrieveToken working

Author: matt-aitken

apps/webapp/app/components/runs/v3/WaitpointDetails.tsx

@@ -87,7 +87,7 @@ export function WaitpointDetailTable({
                   {waitpoint.status === "WAITING" && <ForceTimeout waitpoint={waitpoint} />}
                 </div>
                 <Paragraph variant="extra-small" className="text-text-dimmed/70">
-                  {waitpoint.isTimeout
+                  {waitpoint.status === "TIMED_OUT"
                     ? "The waitpoint timed out"
                     : waitpoint.status === "COMPLETED"
                     ? "The waitpoint completed before this timeout was reached"
@@ -112,7 +112,7 @@ export function WaitpointDetailTable({
           </Property.Item>
         </>
       )}
-      {waitpoint.status === "WAITING" ? null : waitpoint.isTimeout ? (
+      {waitpoint.status === "WAITING" ? null : waitpoint.status === "TIMED_OUT" ? (
         <></>
       ) : waitpoint.output ? (
         <PacketDisplay title="Output" data={waitpoint.output} dataType={waitpoint.outputType} />

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

@@ -0,0 +1,47 @@
+import { type RuntimeEnvironmentType } from "@trigger.dev/core/v3";
+import { type RunEngineVersion } from "@trigger.dev/database";
+import { ServiceValidationError } from "~/v3/services/baseService.server";
+import { BasePresenter } from "./basePresenter.server";
+import { WaitpointPresenter } from "./WaitpointPresenter.server";
+
+export class ApiWaitpointPresenter extends BasePresenter {
+  public async call(
+    environment: {
+      id: string;
+      type: RuntimeEnvironmentType;
+      project: {
+        id: string;
+        engine: RunEngineVersion;
+      };
+    },
+    waitpointId: string
+  ) {
+    return this.trace("call", async (span) => {
+      const presenter = new WaitpointPresenter();
+      const result = await presenter.call({
+        friendlyId: waitpointId,
+        environmentId: environment.id,
+        projectId: environment.project.id,
+      });
+
+      if (!result) {
+        throw new ServiceValidationError("Waitpoint not found");
+      }
+
+      return {
+        id: result.id,
+        status: result.status,
+        completedAt: result.completedAt ?? undefined,
+        timeoutAt: result.timeoutAt ?? undefined,
+        completedAfter: result.completedAfter ?? undefined,
+        idempotencyKey: result.userProvidedIdempotencyKey ? result.idempotencyKey : undefined,
+        idempotencyKeyExpiresAt: result.idempotencyKeyExpiresAt ?? undefined,
+        tags: result.tags ?? [],
+        createdAt: result.createdAt,
+        output: result.output,
+        outputType: result.outputType,
+        outputIsError: result.outputIsError,
+      };
+    });
+  }
+}

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

@@ -33,6 +33,8 @@ export class WaitpointPresenter extends BasePresenter {
         outputType: true,
         outputIsError: true,
         completedAfter: true,
+        completedAt: true,
+        createdAt: true,
         connectedRuns: {
           select: {
             friendlyId: true,
@@ -89,9 +91,11 @@ export class WaitpointPresenter extends BasePresenter {
       output: output,
       outputType: waitpoint.outputType,
       outputIsError: waitpoint.outputIsError,
+      timeoutAt: waitpoint.completedAfter,
       completedAfter: waitpoint.completedAfter,
+      completedAt: waitpoint.completedAt,
+      createdAt: waitpoint.createdAt,
       tags: waitpoint.tags,
-      isTimeout,
       connectedRuns,
     };
   }

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

@@ -248,14 +248,13 @@ export class WaitpointTokenListPresenter extends BasePresenter {
         id: token.friendlyId,
         status: waitpointStatusToApiStatus(token.status, token.outputIsError),
         completedAt: token.completedAt ?? undefined,
+        timeoutAt: token.completedAfter ?? undefined,
         completedAfter: token.completedAfter ?? undefined,
         idempotencyKey: token.userProvidedIdempotencyKey
           ? token.inactiveIdempotencyKey ?? token.idempotencyKey
           : undefined,
         idempotencyKeyExpiresAt: token.idempotencyKeyExpiresAt ?? undefined,
         tags: token.tags ? token.tags.sort((a, b) => a.localeCompare(b)) : [],
-        //we can assume that all errors for tokens are timeouts
-        isTimeout: token.outputIsError,
         createdAt: token.createdAt,
       })),
       pagination: {

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

@@ -0,0 +1,22 @@
+import { json } from "@remix-run/server-runtime";
+import { type WaitpointRetrieveTokenResponse } from "@trigger.dev/core/v3";
+import { z } from "zod";
+import { ApiWaitpointPresenter } from "~/presenters/v3/ApiWaitpointPresenter.server";
+import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+export const loader = createLoaderApiRoute(
+  {
+    params: z.object({
+      waitpointFriendlyId: z.string(),
+    }),
+    findResource: async () => 1, // This is a dummy function, we don't need to find a resource
+  },
+  async ({ params, authentication }) => {
+    const presenter = new ApiWaitpointPresenter();
+    const result: WaitpointRetrieveTokenResponse = await presenter.call(
+      authentication.environment,
+      params.waitpointFriendlyId
+    );
+    return json(result);
+  }
+);

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

@@ -38,6 +38,7 @@ import {
   WaitForDurationRequestBody,
   WaitForDurationResponseBody,
   WaitForWaitpointTokenResponseBody,
+  WaitpointRetrieveTokenResponse,
   WaitpointTokenItem,
 } from "../schemas/index.js";
 import { taskContext } from "../task-context-api.js";
@@ -697,7 +698,7 @@ export class ApiClient {
 
   retrieveWaitpointToken(friendlyId: string, requestOptions?: ZodFetchOptions) {
     return zodfetch(
-      WaitpointTokenItem,
+      WaitpointRetrieveTokenResponse,
       `${this.baseUrl}/api/v1/waitpoints/tokens/${friendlyId}`,
       {
         method: "GET",

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

@@ -957,14 +957,28 @@ export const WaitpointTokenItem = z.object({
   status: WaitpointTokenStatus,
   completedAt: z.coerce.date().optional(),
   completedAfter: z.coerce.date().optional(),
+  timeoutAt: z.coerce.date().optional(),
   idempotencyKey: z.string().optional(),
   idempotencyKeyExpiresAt: z.coerce.date().optional(),
   tags: z.array(z.string()),
-  isTimeout: z.boolean(),
   createdAt: z.coerce.date(),
 });
 export type WaitpointTokenItem = z.infer<typeof WaitpointTokenItem>;
 
+export const WaitpointListTokenItem = WaitpointTokenItem.omit({
+  completedAfter: true,
+});
+export type WaitpointListTokenItem = z.infer<typeof WaitpointListTokenItem>;
+
+export const WaitpointRetrieveTokenResponse = WaitpointListTokenItem.and(
+  z.object({
+    output: z.string().optional(),
+    outputType: z.string().optional(),
+    outputIsError: z.boolean().optional(),
+  })
+);
+export type WaitpointRetrieveTokenResponse = z.infer<typeof WaitpointRetrieveTokenResponse>;
+
 export const CompleteWaitpointTokenRequestBody = z.object({
   data: z.any().nullish(),
 });

packages/trigger-sdk/src/v3/wait.ts

@@ -16,6 +16,8 @@ import {
   CursorPagePromise,
   WaitpointTokenItem,
   flattenAttributes,
+  WaitpointListTokenItem,
+  WaitpointTokenStatus,
 } from "@trigger.dev/core/v3";
 import { tracer } from "./tracer.js";
 import { conditionallyImportAndParsePacket } from "@trigger.dev/core/v3/utils/ioSerialization";
@@ -120,7 +122,7 @@ function createToken(
 function listTokens(
   params?: ListWaitpointTokensQueryParams,
   requestOptions?: ApiRequestOptions
-): CursorPagePromise<typeof WaitpointTokenItem> {
+): CursorPagePromise<typeof WaitpointListTokenItem> {
   const apiClient = apiClientManager.clientOrThrow();
 
   const $requestOptions = mergeRequestOptions(
@@ -138,6 +140,26 @@ function listTokens(
   return apiClient.listWaitpointTokens(params, $requestOptions);
 }
 
+/**
+ * A waitpoint token that has been retrieved.
+ *
+ * If the status is `WAITING`, this means the waitpoint is still pending.
+ * For `COMPLETED` the `output` will be the data you passed in when completing the waitpoint.
+ * For `TIMED_OUT` there will be an `error`.
+ */
+export type WaitpointRetrievedToken<T> = {
+  id: string;
+  status: WaitpointTokenStatus;
+  completedAt?: Date;
+  timeoutAt?: Date;
+  idempotencyKey?: string;
+  idempotencyKeyExpiresAt?: Date;
+  tags: string[];
+  createdAt: Date;
+  output?: T;
+  error?: Error;
+};
+
 /**
  * Retrieves a waitpoint token by its ID.
  *
@@ -151,12 +173,12 @@ function listTokens(
  * @param token - The token to retrieve.
  * This can be a string token ID or an object with an `id` property.
  * @param requestOptions - Optional API request options.
- * @returns The waitpoint token details.
+ * @returns The waitpoint token details, including the output or error if the waitpoint is completed or timed out.
  */
-function retrieveToken(
+async function retrieveToken<T>(
   token: string | { id: string },
   requestOptions?: ApiRequestOptions
-): ApiPromise<WaitpointTokenItem> {
+): Promise<WaitpointRetrievedToken<T>> {
   const apiClient = apiClientManager.clientOrThrow();
 
   const $tokenId = typeof token === "string" ? token : token.id;
@@ -182,7 +204,36 @@ function retrieveToken(
     requestOptions
   );
 
-  return apiClient.retrieveWaitpointToken($tokenId, $requestOptions);
+  const result = await apiClient.retrieveWaitpointToken($tokenId, $requestOptions);
+
+  const data = result.output
+    ? await conditionallyImportAndParsePacket(
+        { data: result.output, dataType: result.outputType ?? "application/json" },
+        apiClient
+      )
+    : undefined;
+
+  let error: Error | undefined = undefined;
+  let output: T | undefined = undefined;
+
+  if (result.outputIsError) {
+    error = new WaitpointTimeoutError(data.message);
+  } else {
+    output = data as T;
+  }
+
+  return {
+    id: result.id,
+    status: result.status,
+    completedAt: result.completedAt,
+    timeoutAt: result.timeoutAt,
+    idempotencyKey: result.idempotencyKey,
+    idempotencyKeyExpiresAt: result.idempotencyKeyExpiresAt,
+    tags: result.tags,
+    createdAt: result.createdAt,
+    output,
+    error,
+  };
 }
 
 /**

references/hello-world/src/trigger/waits.ts

@@ -51,6 +51,9 @@ export const waitToken = task({
       logger.log("Token", token);
     }
 
+    const retrievedToken = await wait.retrieveToken(token.id);
+    logger.log("Retrieved token", retrievedToken);
+
     //wait for the token
     const result = await wait.forToken<{ foo: string }>(token, { releaseConcurrency: true });
     if (!result.ok) {
@@ -63,6 +66,9 @@ export const waitToken = task({
     for await (const token of tokens2) {
       logger.log("Token2", token);
     }
+
+    const retrievedToken2 = await wait.retrieveToken(token.id);
+    logger.log("Retrieved token2", retrievedToken2);
   },
 });