Waitpoint token callback URLs (#2025)

* Initial commit with a plan for what we’re going to do * Some initial types and improved plan * Add Waitpoint resolver * Add resolver + status index * Remove type + status index * Only drop if exists * Remove type index * Update waitpoint list presenter to use resolver * Added resolver to the engine * Made the existing waitpoint list presenter more flexible * Initial implentation ofr wait.forHttpCallback() * Added the callback endpoint (no API rate limit) * schema version * Added jsdocs, removed schema version because of errors * Show callback URL if it’s set * Dashboard pages and panels * Remove todos * Added temporary icon * Added a blank state * Some tweaks and added a Replicate example * Implement unwrap() for httpCallback * Added unwrap to wait.forToken() as well * Improved jsdocs * Added docs * Added unwrap to the token docs * Show a dash if there are no tags Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> * Make the timeout error safer * Fixed migrations… should use id desc not createdAt desc * Fixed page title * Fixed migration so it only adds them if they don’t exist. This allows us to manuall run in cloud first * Respect the max content length by getting the length of the body * Added more docs details about the callback format * Remove code comment * Improved the error * Added a hash to the HTTP callback URLs * Add the apiKey to the API input type to fix TS error * Return the error responses. They were being caught and not preserved * The content-length header is required. Deal with an empty body * Removed unused types * Added some new span icons * Reworked http callback to be a create call then just use wait.forToken() * Added a changeset * Updated the docs * Updated the wait overview docs * Simplify to just a call * WIP stripping right back to waitpoints just having a URL associated with them… * More deletions * Remove missing icon * Updated the changeset * Add URL to the token return types * Remove wait for http callback page * Updated docs * More tidying * Type and import fix * Remove unused import * Some type fixes for the retrieve --------- Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
2025-05-07 19:55:29 +01:00
parent 4ec0395424
commit d23fa38a0a
21 changed files with 496 additions and 133 deletions
@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+When you create a Waitpoint token using `wait.createToken()` you get a URL back that can be used to complete it by making an HTTP POST request.
@@ -19,6 +19,7 @@ import { FunctionIcon } from "~/assets/icons/FunctionIcon";
 import { TriggerIcon } from "~/assets/icons/TriggerIcon";
 import { PythonLogoIcon } from "~/assets/icons/PythonLogoIcon";
 import { TraceIcon } from "~/assets/icons/TraceIcon";
+import { WaitpointTokenIcon } from "~/assets/icons/WaitpointTokenIcon";

 type TaskIconProps = {
  name: string | undefined;
@@ -75,6 +76,10 @@ export function RunIcon({ name, className, spanName }: TaskIconProps) {
      return <TriggerIcon className={cn(className, "text-orange-500")} />;
    case "python":
      return <PythonLogoIcon className={className} />;
+    case "wait-token":
+      return <WaitpointTokenIcon className={cn(className, "text-sky-500")} />;
+    case "function":
+      return <FunctionIcon className={cn(className, "text-text-dimmed")} />;
    //log levels
    case "debug":
    case "log":
@@ -11,6 +11,7 @@ import { v3WaitpointTokenPath, v3WaitpointTokensPath } from "~/utils/pathBuilder
 import { PacketDisplay } from "./PacketDisplay";
 import { WaitpointStatusCombo } from "./WaitpointStatus";
 import { RunTag } from "./RunTag";
+import { ClipboardField } from "~/components/primitives/ClipboardField";

 export function WaitpointDetailTable({
  waitpoint,
@@ -50,6 +51,14 @@ export function WaitpointDetailTable({
          )}
        </Property.Value>
      </Property.Item>
+      {waitpoint.type === "MANUAL" && (
+        <Property.Item>
+          <Property.Label>Callback URL</Property.Label>
+          <Property.Value className="my-1">
+            <ClipboardField value={waitpoint.url} variant={"secondary/small"} />
+          </Property.Value>
+        </Property.Item>
+      )}
      <Property.Item>
        <Property.Label>Idempotency key</Property.Label>
        <Property.Value>
@@ -1,16 +1,12 @@
-import { RuntimeEnvironmentType, WaitpointTokenStatus } from "@trigger.dev/core/v3";
+import { type RuntimeEnvironmentType, WaitpointTokenStatus } from "@trigger.dev/core/v3";
+import { type RunEngineVersion, type WaitpointResolver } from "@trigger.dev/database";
 import { z } from "zod";
-import { BasePresenter } from "./basePresenter.server";
 import { CoercedDate } from "~/utils/zod";
-import { AuthenticatedEnvironment } from "@internal/run-engine";
-import {
-  WaitpointTokenListOptions,
-  WaitpointTokenListPresenter,
-} from "./WaitpointTokenListPresenter.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
-import { RunEngineVersion } from "@trigger.dev/database";
+import { BasePresenter } from "./basePresenter.server";
+import { type WaitpointListOptions, WaitpointListPresenter } from "./WaitpointListPresenter.server";

-export const ApiWaitpointTokenListSearchParams = z.object({
+export const ApiWaitpointListSearchParams = z.object({
  "page[size]": z.coerce.number().int().positive().min(1).max(100).optional(),
  "page[after]": z.string().optional(),
  "page[before]": z.string().optional(),
@@ -61,9 +57,9 @@ export const ApiWaitpointTokenListSearchParams = z.object({
  "filter[createdAt][to]": CoercedDate,
 });

-type ApiWaitpointTokenListSearchParams = z.infer<typeof ApiWaitpointTokenListSearchParams>;
+type ApiWaitpointListSearchParams = z.infer<typeof ApiWaitpointListSearchParams>;

-export class ApiWaitpointTokenListPresenter extends BasePresenter {
+export class ApiWaitpointListPresenter extends BasePresenter {
  public async call(
    environment: {
      id: string;
@@ -72,11 +68,12 @@ export class ApiWaitpointTokenListPresenter extends BasePresenter {
        id: string;
        engine: RunEngineVersion;
      };
+      apiKey: string;
    },
-    searchParams: ApiWaitpointTokenListSearchParams
+    searchParams: ApiWaitpointListSearchParams
  ) {
    return this.trace("call", async (span) => {
-      const options: WaitpointTokenListOptions = {
+      const options: WaitpointListOptions = {
        environment,
      };

@@ -118,7 +115,7 @@ export class ApiWaitpointTokenListPresenter extends BasePresenter {
        options.to = searchParams["filter[createdAt][to]"].getTime();
      }

-      const presenter = new WaitpointTokenListPresenter();
+      const presenter = new WaitpointListPresenter();
      const result = await presenter.call(options);

      if (!result.success) {
@@ -2,8 +2,8 @@ import { logger, 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";
-import { waitpointStatusToApiStatus } from "./WaitpointTokenListPresenter.server";
+import { waitpointStatusToApiStatus } from "./WaitpointListPresenter.server";
+import { generateHttpCallbackUrl } from "~/services/httpCallback.server";

 export class ApiWaitpointPresenter extends BasePresenter {
  public async call(
@@ -14,6 +14,7 @@ export class ApiWaitpointPresenter extends BasePresenter {
        id: string;
        engine: RunEngineVersion;
      };
+      apiKey: string;
    },
    waitpointId: string
  ) {
@@ -24,6 +25,7 @@ export class ApiWaitpointPresenter extends BasePresenter {
          environmentId: environment.id,
        },
        select: {
+          id: true,
          friendlyId: true,
          type: true,
          status: true,
@@ -62,6 +64,7 @@ export class ApiWaitpointPresenter extends BasePresenter {
      return {
        id: waitpoint.friendlyId,
        type: waitpoint.type,
+        url: generateHttpCallbackUrl(waitpoint.id, environment.apiKey),
        status: waitpointStatusToApiStatus(waitpoint.status, waitpoint.outputIsError),
        idempotencyKey: waitpoint.idempotencyKey,
        userProvidedIdempotencyKey: waitpoint.userProvidedIdempotencyKey,
@@ -1,6 +1,7 @@
 import parse from "parse-duration";
 import {
  Prisma,
+  type WaitpointResolver,
  type RunEngineVersion,
  type RuntimeEnvironmentType,
  type WaitpointStatus,
@@ -11,10 +12,11 @@ import { BasePresenter } from "./basePresenter.server";
 import { type WaitpointSearchParams } from "~/components/runs/v3/WaitpointTokenFilters";
 import { determineEngineVersion } from "~/v3/engineVersion.server";
 import { type WaitpointTokenStatus, type WaitpointTokenItem } from "@trigger.dev/core/v3";
+import { generateHttpCallbackUrl } from "~/services/httpCallback.server";

 const DEFAULT_PAGE_SIZE = 25;

-export type WaitpointTokenListOptions = {
+export type WaitpointListOptions = {
  environment: {
    id: string;
    type: RuntimeEnvironmentType;
@@ -22,6 +24,7 @@ export type WaitpointTokenListOptions = {
      id: string;
      engine: RunEngineVersion;
    };
+    apiKey: string;
  };
  // filters
  id?: string;
@@ -63,7 +66,7 @@ type Result =
      filters: undefined;
    };

-export class WaitpointTokenListPresenter extends BasePresenter {
+export class WaitpointListPresenter extends BasePresenter {
  public async call({
    environment,
    id,
@@ -76,7 +79,7 @@ export class WaitpointTokenListPresenter extends BasePresenter {
    direction = "forward",
    cursor,
    pageSize = DEFAULT_PAGE_SIZE,
-  }: WaitpointTokenListOptions): Promise<Result> {
+  }: WaitpointListOptions): Promise<Result> {
    const engineVersion = await determineEngineVersion({ environment });
    if (engineVersion === "V1") {
      return {
@@ -165,8 +168,8 @@ export class WaitpointTokenListPresenter extends BasePresenter {
      ${sqlDatabaseSchema}."Waitpoint" w
    WHERE
      w."environmentId" = ${environment.id}
-      AND w.type = 'MANUAL'
-      -- cursor
+    AND w.type = 'MANUAL'      
+    -- cursor
      ${
        cursor
          ? direction === "forward"
@@ -263,6 +266,7 @@ export class WaitpointTokenListPresenter extends BasePresenter {
      success: true,
      tokens: tokensToReturn.map((token) => ({
        id: token.friendlyId,
+        url: generateHttpCallbackUrl(token.id, environment.apiKey),
        status: waitpointStatusToApiStatus(token.status, token.outputIsError),
        completedAt: token.completedAt ?? undefined,
        timeoutAt: token.completedAfter ?? undefined,
@@ -1,8 +1,9 @@
 import { isWaitpointOutputTimeout, prettyPrintPacket } from "@trigger.dev/core/v3";
+import { generateHttpCallbackUrl } from "~/services/httpCallback.server";
 import { logger } from "~/services/logger.server";
 import { BasePresenter } from "./basePresenter.server";
 import { type RunListItem, RunListPresenter } from "./RunListPresenter.server";
-import { waitpointStatusToApiStatus } from "./WaitpointTokenListPresenter.server";
+import { waitpointStatusToApiStatus } from "./WaitpointListPresenter.server";

 export type WaitpointDetail = NonNullable<Awaited<ReturnType<WaitpointPresenter["call"]>>>;

@@ -22,6 +23,7 @@ export class WaitpointPresenter extends BasePresenter {
        environmentId,
      },
      select: {
+        id: true,
        friendlyId: true,
        type: true,
        status: true,
@@ -42,6 +44,11 @@ export class WaitpointPresenter extends BasePresenter {
          take: 5,
        },
        tags: true,
+        environment: {
+          select: {
+            apiKey: true,
+          },
+        },
      },
    });

@@ -83,6 +90,7 @@ export class WaitpointPresenter extends BasePresenter {
    return {
      id: waitpoint.friendlyId,
      type: waitpoint.type,
+      url: generateHttpCallbackUrl(waitpoint.id, waitpoint.environment.apiKey),
      status: waitpointStatusToApiStatus(waitpoint.status, waitpoint.outputIsError),
      idempotencyKey: waitpoint.idempotencyKey,
      userProvidedIdempotencyKey: waitpoint.userProvidedIdempotencyKey,
@@ -7,6 +7,7 @@ import { NoWaitpointTokens } from "~/components/BlankStatePanels";
 import { MainCenteredContainer, PageBody, PageContainer } from "~/components/layout/AppLayout";
 import { ListPagination } from "~/components/ListPagination";
 import { LinkButton } from "~/components/primitives/Buttons";
+import { ClipboardField } from "~/components/primitives/ClipboardField";
 import { CopyableText } from "~/components/primitives/CopyableText";
 import { DateTime } from "~/components/primitives/DateTime";
 import { NavBar, PageAccessories, PageTitle } from "~/components/primitives/PageHeader";
@@ -36,7 +37,7 @@ import { useOrganization } from "~/hooks/useOrganizations";
 import { useProject } from "~/hooks/useProject";
 import { findProjectBySlug } from "~/models/project.server";
 import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
-import { WaitpointTokenListPresenter } from "~/presenters/v3/WaitpointTokenListPresenter.server";
+import { WaitpointListPresenter } from "~/presenters/v3/WaitpointListPresenter.server";
 import { requireUserId } from "~/services/session.server";
 import { docsPath, EnvironmentParamSchema, v3WaitpointTokenPath } from "~/utils/pathBuilder";

@@ -84,7 +85,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
  }

  try {
-    const presenter = new WaitpointTokenListPresenter();
+    const presenter = new WaitpointListPresenter();
    const result = await presenter.call({
      environment,
      ...searchParams,
@@ -143,6 +144,7 @@ export default function Page() {
                      <TableRow>
                        <TableHeaderCell className="w-[1%]">Created</TableHeaderCell>
                        <TableHeaderCell className="w-[20%]">ID</TableHeaderCell>
+                        <TableHeaderCell className="w-[20%]">Callback URL</TableHeaderCell>
                        <TableHeaderCell className="w-[20%]">Status</TableHeaderCell>
                        <TableHeaderCell className="w-[20%]">Completed</TableHeaderCell>
                        <TableHeaderCell className="w-[20%]">Idempotency Key</TableHeaderCell>
@@ -178,6 +180,9 @@ export default function Page() {
                              <TableCell to={path}>
                                <CopyableText value={token.id} className="font-mono" />
                              </TableCell>
+                              <TableCell to={path}>
+                                <ClipboardField value={token.url} variant={"secondary/small"} />
+                              </TableCell>
                              <TableCell to={path}>
                                <WaitpointStatusCombo status={token.status} className="text-xs" />
                              </TableCell>
@@ -0,0 +1,91 @@
+import { type ActionFunctionArgs, json } from "@remix-run/server-runtime";
+import {
+  type CompleteWaitpointTokenResponseBody,
+  conditionallyExportPacket,
+  stringifyIO,
+} from "@trigger.dev/core/v3";
+import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
+import { z } from "zod";
+import { $replica } from "~/db.server";
+import { env } from "~/env.server";
+import { verifyHttpCallbackHash } from "~/services/httpCallback.server";
+import { logger } from "~/services/logger.server";
+import { engine } from "~/v3/runEngine.server";
+
+const paramsSchema = z.object({
+  waitpointFriendlyId: z.string(),
+  hash: z.string(),
+});
+
+export async function action({ request, params }: ActionFunctionArgs) {
+  if (request.method.toUpperCase() !== "POST") {
+    return json({ error: "Method not allowed" }, { status: 405, headers: { Allow: "POST" } });
+  }
+
+  const contentLength = request.headers.get("content-length");
+  if (!contentLength) {
+    return json({ error: "Content-Length header is required" }, { status: 411 });
+  }
+
+  if (parseInt(contentLength) > env.TASK_PAYLOAD_MAXIMUM_SIZE) {
+    return json({ error: "Request body too large" }, { status: 413 });
+  }
+
+  const { waitpointFriendlyId, hash } = paramsSchema.parse(params);
+  const waitpointId = WaitpointId.toId(waitpointFriendlyId);
+
+  try {
+    const waitpoint = await $replica.waitpoint.findFirst({
+      where: {
+        id: waitpointId,
+      },
+      include: {
+        environment: {
+          select: {
+            apiKey: true,
+          },
+        },
+      },
+    });
+
+    if (!waitpoint) {
+      return json({ error: "Waitpoint not found" }, { status: 404 });
+    }
+
+    if (!verifyHttpCallbackHash(waitpoint.id, hash, waitpoint.environment.apiKey)) {
+      return json({ error: "Invalid URL, hash doesn't match" }, { status: 401 });
+    }
+
+    if (waitpoint.status === "COMPLETED") {
+      return json<CompleteWaitpointTokenResponseBody>({
+        success: true,
+      });
+    }
+
+    // If the request body is not valid JSON, return an empty object
+    const body = await request.json().catch(() => ({}));
+
+    const stringifiedData = await stringifyIO(body);
+    const finalData = await conditionallyExportPacket(
+      stringifiedData,
+      `${waitpointId}/waitpoint/http-callback`
+    );
+
+    const result = await engine.completeWaitpoint({
+      id: waitpointId,
+      output: finalData.data
+        ? { type: finalData.dataType, value: finalData.data, isError: false }
+        : undefined,
+    });
+
+    return json<CompleteWaitpointTokenResponseBody>(
+      {
+        success: true,
+      },
+      { status: 200 }
+    );
+  } catch (error) {
+    logger.error("Failed to complete HTTP callback", { error });
+    throw json({ error: "Failed to complete HTTP callback" }, { status: 500 });
+  }
+}
@@ -6,14 +6,15 @@ import {
 import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
 import { createWaitpointTag, MAX_TAGS_PER_WAITPOINT } from "~/models/waitpointTag.server";
 import {
-  ApiWaitpointTokenListPresenter,
-  ApiWaitpointTokenListSearchParams,
-} from "~/presenters/v3/ApiWaitpointTokenListPresenter.server";
+  ApiWaitpointListPresenter,
+  ApiWaitpointListSearchParams,
+} from "~/presenters/v3/ApiWaitpointListPresenter.server";
+import { type AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { generateHttpCallbackUrl } from "~/services/httpCallback.server";
 import {
  createActionApiRoute,
  createLoaderApiRoute,
 } from "~/services/routeBuilders/apiBuilder.server";
-import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { parseDelay } from "~/utils/delays";
 import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
 import { engine } from "~/v3/runEngine.server";
@@ -21,11 +22,11 @@ import { ServiceValidationError } from "~/v3/services/baseService.server";

 export const loader = createLoaderApiRoute(
  {
-    searchParams: ApiWaitpointTokenListSearchParams,
+    searchParams: ApiWaitpointListSearchParams,
    findResource: async () => 1, // This is a dummy function, we don't need to find a resource
  },
  async ({ searchParams, authentication }) => {
-    const presenter = new ApiWaitpointTokenListPresenter();
+    const presenter = new ApiWaitpointListPresenter();
    const result = await presenter.call(authentication.environment, searchParams);

    return json(result);
@@ -84,6 +85,7 @@ const { action } = createActionApiRoute(
        {
          id: WaitpointId.toFriendlyId(result.waitpoint.id),
          isCached: result.isCached,
+          url: generateHttpCallbackUrl(result.waitpoint.id, authentication.environment.apiKey),
        },
        { status: 200, headers: $responseHeaders }
      );
@@ -59,6 +59,7 @@ export const apiRateLimiter = authorizationRateLimitMiddleware({
    "/api/v1/usage/ingest",
    "/api/v1/auth/jwt/claims",
    /^\/api\/v1\/runs\/[^\/]+\/attempts$/, // /api/v1/runs/$runFriendlyId/attempts
+    /^\/api\/v1\/waitpoints\/tokens\/[^\/]+\/callback\/[^\/]+$/, // /api/v1/waitpoints/tokens/$waitpointFriendlyId/callback/$hash
  ],
  log: {
    rejections: env.API_RATE_LIMIT_REJECTION_LOGS_ENABLED === "1",
@@ -0,0 +1,30 @@
+import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
+import nodeCrypto from "node:crypto";
+import { env } from "~/env.server";
+
+export function generateHttpCallbackUrl(waitpointId: string, apiKey: string) {
+  const hash = generateHttpCallbackHash(waitpointId, apiKey);
+
+  return `${env.API_ORIGIN ?? env.APP_ORIGIN}/api/v1/waitpoints/tokens/${WaitpointId.toFriendlyId(
+    waitpointId
+  )}/callback/${hash}`;
+}
+
+function generateHttpCallbackHash(waitpointId: string, apiKey: string) {
+  const hmac = nodeCrypto.createHmac("sha256", apiKey);
+  hmac.update(waitpointId);
+  return hmac.digest("hex");
+}
+
+export function verifyHttpCallbackHash(waitpointId: string, hash: string, apiKey: string) {
+  const expectedHash = generateHttpCallbackHash(waitpointId, apiKey);
+
+  if (
+    hash.length === expectedHash.length &&
+    nodeCrypto.timingSafeEqual(Buffer.from(hash, "hex"), Buffer.from(expectedHash, "hex"))
+  ) {
+    return true;
+  }
+
+  return false;
+}
@@ -7,6 +7,8 @@ import UpgradeToV4Note from "/snippets/upgrade-to-v4-note.mdx";

 Waitpoint tokens pause task runs until you complete the token. They're commonly used for approval workflows and other scenarios where you need to wait for external confirmation, such as human-in-the-loop processes.

+You can complete a token using the SDK or by making a POST request to the token's URL.
+
 <UpgradeToV4Note />

 ## Usage
@@ -52,6 +54,29 @@ await wait.completeToken<ApprovalToken>(tokenId, {
 });
 ```

+Or you can make an HTTP POST request to the `url` it returns:
+
+```ts
+import { wait } from "@trigger.dev/sdk";
+
+const token = await wait.createToken({
+  timeout: "10m",
+});
+
+const call = await replicate.predictions.create({
+  version: "27b93a2413e7f36cd83da926f3656280b2931564ff050bf9575f1fdf9bcd7478",
+  input: {
+    prompt: "A painting of a cat by Andy Warhol",
+  },
+  // pass the provided URL to Replicate's webhook, so they can "callback"
+  webhook: token.url,
+  webhook_events_filter: ["completed"],
+});
+
+const prediction = await wait.forToken<Prediction>(token).unwrap();
+// unwrap() throws a timeout error or returns the result   👆
+```
+
 ## wait.createToken

 Create a waitpoint token.
@@ -85,6 +110,13 @@ The `createToken` function returns a token object with the following properties:
  The ID of the token. Starts with `waitpoint_`.
 </ParamField>

+<ParamField query="url" type="string">
+  The URL of the token. This is the URL you can make a POST request to in order to complete the token.
+
+The JSON body of the POST request will be used as the output of the token. If there's no body the output will be an empty object `{}`.
+
+</ParamField>
+
 <ParamField query="isCached" type="boolean">
  Whether the token is cached. Will return true if the token was created with an idempotency key and
  the same idempotency key was used again.
@@ -270,6 +302,18 @@ The `forToken` function returns a result object with the following properties:
  timeout error.
 </ParamField>

+### unwrap()
+
+We provide a handy `.unwrap()` method that will throw an error if the result is not ok. This means your happy path is a lot cleaner.
+
+```ts
+const approval = await wait.forToken<ApprovalToken>(tokenId).unwrap();
+// unwrap means an error will throw if the waitpoint times out 👆
+
+// This is the actual data you sent to the token now, not a result object
+console.log("Approval", approval);
+```
+
 ### Example

 ```ts
@@ -326,6 +370,13 @@ Each token is an object with the following properties:
  The ID of the token.
 </ParamField>

+<ParamField query="url" type="string">
+  The URL of the token. This is the URL you can make a POST request to in order to complete the token.
+
+The JSON body of the POST request will be used as the output of the token. If there's no body the output will be an empty object `{}`.
+
+</ParamField>
+
 <ParamField query="status" type="string">
  The status of the token.
 </ParamField>
@@ -392,6 +443,13 @@ The `retrieveToken` function returns a token object with the following propertie
  The ID of the token.
 </ParamField>

+<ParamField query="url" type="string">
+  The URL of the token. This is the URL you can make a POST request to in order to complete the token.
+
+The JSON body of the POST request will be used as the output of the token. If there's no body the output will be an empty object `{}`.
+
+</ParamField>
+
 <ParamField query="status" type="string">
  The status of the token.
 </ParamField>
@@ -4,14 +4,14 @@ sidebarTitle: "Overview"
 description: "During your run you can wait for a period of time or for something to happen."
 ---

-import PausedExecutionFree from "/snippets/paused-execution-free.mdx"
+import PausedExecutionFree from "/snippets/paused-execution-free.mdx";

-Waiting allows you to write complex tasks as a set of async code, without having to scheduled another task or poll for changes.
+Waiting allows you to write complex tasks as a set of async code, without having to schedule another task or poll for changes.

 <PausedExecutionFree />

-| Function                               | What it does                                                                              |
-| :--------------------------------------| :---------------------------------------------------------------------------------------- |
-| [wait.for()](/wait-for)                | Waits for a specific period of time, e.g. 1 day.                                          |
-| [wait.until()](/wait-until)            | Waits until the provided `Date`.                                                          |
-| [wait.forToken()](/wait-for-token)     | Pauses task runs until a token is completed.                                              |
+| Function                           | What it does                                     |
+| :--------------------------------- | :----------------------------------------------- |
+| [wait.for()](/wait-for)            | Waits for a specific period of time, e.g. 1 day. |
+| [wait.until()](/wait-until)        | Waits until the provided `Date`.                 |
+| [wait.forToken()](/wait-for-token) | Pauses runs until a token is completed.          |
@@ -2151,6 +2151,7 @@ model Waitpoint {
  updatedAt DateTime @updatedAt

  /// Denormized column that holds the raw tags
+  /// Denormalized column that holds the raw tags
  tags String[]

  /// Quickly find an idempotent waitpoint
@@ -16,9 +16,7 @@ import {
  CreateWaitpointTokenResponseBody,
  DeletedScheduleObject,
  EnvironmentVariableResponseBody,
-  EnvironmentVariableValue,
  EnvironmentVariableWithSecret,
-  EnvironmentVariables,
  ListQueueOptions,
  ListRunResponseItem,
  ListScheduleOptions,
@@ -42,8 +40,10 @@ import {
  WaitpointRetrieveTokenResponse,
  WaitpointTokenItem,
 } from "../schemas/index.js";
+import { AsyncIterableStream } from "../streams/asyncIterableStream.js";
 import { taskContext } from "../task-context-api.js";
 import { AnyRunTypes, TriggerJwtOptions } from "../types/tasks.js";
+import { Prettify } from "../types/utils.js";
 import {
  AnyZodFetchOptions,
  ApiPromise,
@@ -63,9 +63,9 @@ import {
  RunShape,
  RunStreamCallback,
  RunSubscription,
+  SSEStreamSubscriptionFactory,
  TaskRunShape,
  runShapeStream,
-  SSEStreamSubscriptionFactory,
 } from "./runStream.js";
 import {
  CreateEnvironmentVariableParams,
@@ -76,8 +76,6 @@ import {
  SubscribeToRunsQueryParams,
  UpdateEnvironmentVariableParams,
 } from "./types.js";
-import { AsyncIterableStream } from "../streams/asyncIterableStream.js";
-import { Prettify } from "../types/utils.js";

 export type CreateWaitpointTokenResponse = Prettify<
  CreateWaitpointTokenResponseBody & {
@@ -961,6 +961,7 @@ export type CreateWaitpointTokenRequestBody = z.infer<typeof CreateWaitpointToke
 export const CreateWaitpointTokenResponseBody = z.object({
  id: z.string(),
  isCached: z.boolean(),
+  url: z.string(),
 });
 export type CreateWaitpointTokenResponseBody = z.infer<typeof CreateWaitpointTokenResponseBody>;

@@ -970,6 +971,8 @@ export type WaitpointTokenStatus = z.infer<typeof WaitpointTokenStatus>;

 export const WaitpointTokenItem = z.object({
  id: z.string(),
+  /** If you make a POST request to this URL, it will complete the waitpoint. */
+  url: z.string(),
  status: WaitpointTokenStatus,
  completedAt: z.coerce.date().optional(),
  completedAfter: z.coerce.date().optional(),
@@ -1,29 +1,27 @@
+import { SpanStatusCode } from "@opentelemetry/api";
 import {
-  SemanticInternalAttributes,
  accessoryAttributes,
-  runtime,
  apiClientManager,
  ApiPromise,
  ApiRequestOptions,
-  CreateWaitpointTokenRequestBody,
-  CreateWaitpointTokenResponseBody,
-  mergeRequestOptions,
  CompleteWaitpointTokenResponseBody,
-  WaitpointTokenTypedResult,
-  Prettify,
-  taskContext,
-  ListWaitpointTokensQueryParams,
-  CursorPagePromise,
-  WaitpointTokenItem,
-  flattenAttributes,
-  WaitpointListTokenItem,
-  WaitpointTokenStatus,
-  WaitpointRetrieveTokenResponse,
+  CreateWaitpointTokenRequestBody,
  CreateWaitpointTokenResponse,
+  CreateWaitpointTokenResponseBody,
+  CursorPagePromise,
+  flattenAttributes,
+  ListWaitpointTokensQueryParams,
+  mergeRequestOptions,
+  runtime,
+  SemanticInternalAttributes,
+  taskContext,
+  WaitpointListTokenItem,
+  WaitpointRetrieveTokenResponse,
+  WaitpointTokenStatus,
+  WaitpointTokenTypedResult,
 } from "@trigger.dev/core/v3";
-import { tracer } from "./tracer.js";
 import { conditionallyImportAndParsePacket } from "@trigger.dev/core/v3/utils/ioSerialization";
-import { SpanStatusCode } from "@opentelemetry/api";
+import { tracer } from "./tracer.js";

 /**
 * This creates a waitpoint token.
@@ -31,6 +29,8 @@ import { SpanStatusCode } from "@opentelemetry/api";
 *
 * @example
 *
+ * **Manually completing a token**
+ *
 * ```ts
 * const token = await wait.createToken({
 *   idempotencyKey: `approve-document-${documentId}`,
@@ -45,6 +45,30 @@ import { SpanStatusCode } from "@opentelemetry/api";
 * });
 * ```
 *
+ * @example
+ *
+ * **Completing a token with a webhook**
+ *
+ * ```ts
+ * const token = await wait.createToken({
+ *   timeout: "10m",
+ *   tags: ["replicate"],
+ * });
+ *
+ * // Later, in a different part of your codebase, you can complete the waitpoint
+ * await replicate.predictions.create({
+ *   version: "27b93a2413e7f36cd83da926f3656280b2931564ff050bf9575f1fdf9bcd7478",
+ *   input: {
+ *     prompt: "A painting of a cat by Andy Warhol",
+ *   },
+ *   // pass the provided URL to Replicate's webhook, so they can "callback"
+ *   webhook: token.url,
+ *   webhook_events_filter: ["completed"],
+ * });
+ *
+ * const prediction = await wait.forToken<Prediction>(token).unwrap();
+ * ```
+ *
 * @param options - The options for the waitpoint token.
 * @param requestOptions - The request options for the waitpoint token.
 * @returns The waitpoint token.
@@ -73,6 +97,7 @@ function createToken(
      onResponseBody: (body: CreateWaitpointTokenResponseBody, span) => {
        span.setAttribute("id", body.id);
        span.setAttribute("isCached", body.isCached);
+        span.setAttribute("url", body.url);
      },
    },
    requestOptions
@@ -151,6 +176,8 @@ function listTokens(
 */
 export type WaitpointRetrievedToken<T> = {
  id: string;
+  /** A URL that you can make a POST request to in order to complete the waitpoint. */
+  url: string;
  status: WaitpointTokenStatus;
  completedAt?: Date;
  timeoutAt?: Date;
@@ -204,6 +231,7 @@ async function retrieveToken<T>(
      },
      onResponseBody: (body: WaitpointRetrieveTokenResponse, span) => {
        span.setAttribute("id", body.id);
+        span.setAttribute("url", body.url);
        span.setAttribute("status", body.status);
        if (body.completedAt) {
          span.setAttribute("completedAt", body.completedAt.toISOString());
@@ -244,6 +272,7 @@ async function retrieveToken<T>(

  return {
    id: result.id,
+    url: result.url,
    status: result.status,
    completedAt: result.completedAt,
    timeoutAt: result.timeoutAt,
@@ -377,6 +406,29 @@ function printWaitBelowThreshold() {
  );
 }

+class ManualWaitpointPromise<TOutput> extends Promise<WaitpointTokenTypedResult<TOutput>> {
+  constructor(
+    executor: (
+      resolve: (
+        value: WaitpointTokenTypedResult<TOutput> | PromiseLike<WaitpointTokenTypedResult<TOutput>>
+      ) => void,
+      reject: (reason?: any) => void
+    ) => void
+  ) {
+    super(executor);
+  }
+
+  unwrap(): Promise<TOutput> {
+    return this.then((result) => {
+      if (result.ok) {
+        return result.output;
+      } else {
+        throw new WaitpointTimeoutError(result.error.message);
+      }
+    });
+  }
+}
+
 export const wait = {
  for: async (options: WaitForOptions) => {
    const ctx = taskContext.ctx;
@@ -554,9 +606,9 @@ export const wait = {
   *
   * @param token - The token to wait for.
   * @param options - The options for the waitpoint token.
-   * @returns The waitpoint token.
+   * @returns A promise that resolves to the result of the waitpoint. You can use `.unwrap()` to get the result and an error will throw.
   */
-  forToken: async <T>(
+  forToken: <T>(
    /**
     * The token to wait for.
     * This can be a string token ID or an object with an `id` property.
@@ -575,76 +627,84 @@ export const wait = {
       */
      releaseConcurrency?: boolean;
    }
-  ): Promise<Prettify<WaitpointTokenTypedResult<T>>> => {
-    const ctx = taskContext.ctx;
+  ): ManualWaitpointPromise<T> => {
+    return new ManualWaitpointPromise<T>(async (resolve, reject) => {
+      try {
+        const ctx = taskContext.ctx;

-    if (!ctx) {
-      throw new Error("wait.forToken can only be used from inside a task.run()");
-    }
-
-    const apiClient = apiClientManager.clientOrThrow();
-
-    const tokenId = typeof token === "string" ? token : token.id;
-
-    return tracer.startActiveSpan(
-      `wait.forToken()`,
-      async (span) => {
-        const response = await apiClient.waitForWaitpointToken({
-          runFriendlyId: ctx.run.id,
-          waitpointFriendlyId: tokenId,
-          releaseConcurrency: options?.releaseConcurrency,
-        });
-
-        if (!response.success) {
-          throw new Error(`Failed to wait for wait token ${tokenId}`);
+        if (!ctx) {
+          throw new Error("wait.forToken can only be used from inside a task.run()");
        }

-        const result = await runtime.waitUntil(tokenId);
+        const apiClient = apiClientManager.clientOrThrow();

-        const data = result.output
-          ? await conditionallyImportAndParsePacket(
-              { data: result.output, dataType: result.outputType ?? "application/json" },
-              apiClient
-            )
-          : undefined;
+        const tokenId = typeof token === "string" ? token : token.id;

-        if (result.ok) {
-          return {
-            ok: result.ok,
-            output: data,
-          } as WaitpointTokenTypedResult<T>;
-        } else {
-          const error = new WaitpointTimeoutError(data.message);
+        const result = await tracer.startActiveSpan(
+          `wait.forToken()`,
+          async (span) => {
+            const response = await apiClient.waitForWaitpointToken({
+              runFriendlyId: ctx.run.id,
+              waitpointFriendlyId: tokenId,
+              releaseConcurrency: options?.releaseConcurrency,
+            });

-          span.recordException(error);
-          span.setStatus({
-            code: SpanStatusCode.ERROR,
-          });
+            if (!response.success) {
+              throw new Error(`Failed to wait for wait token ${tokenId}`);
+            }

-          return {
-            ok: result.ok,
-            error,
-          } as WaitpointTokenTypedResult<T>;
-        }
-      },
-      {
-        attributes: {
-          [SemanticInternalAttributes.STYLE_ICON]: "wait",
-          [SemanticInternalAttributes.ENTITY_TYPE]: "waitpoint",
-          [SemanticInternalAttributes.ENTITY_ID]: tokenId,
-          id: tokenId,
-          ...accessoryAttributes({
-            items: [
-              {
-                text: tokenId,
-                variant: "normal",
-              },
-            ],
-            style: "codepath",
-          }),
-        },
+            const result = await runtime.waitUntil(tokenId);
+
+            const data = result.output
+              ? await conditionallyImportAndParsePacket(
+                  { data: result.output, dataType: result.outputType ?? "application/json" },
+                  apiClient
+                )
+              : undefined;
+
+            if (result.ok) {
+              return {
+                ok: result.ok,
+                output: data,
+              } as WaitpointTokenTypedResult<T>;
+            } else {
+              const error = new WaitpointTimeoutError(data.message);
+
+              span.recordException(error);
+              span.setStatus({
+                code: SpanStatusCode.ERROR,
+              });
+
+              return {
+                ok: result.ok,
+                error,
+              } as WaitpointTokenTypedResult<T>;
+            }
+          },
+          {
+            attributes: {
+              [SemanticInternalAttributes.STYLE_ICON]: "wait",
+              [SemanticInternalAttributes.ENTITY_TYPE]: "waitpoint",
+              [SemanticInternalAttributes.ENTITY_ID]: tokenId,
+              id: tokenId,
+              ...accessoryAttributes({
+                items: [
+                  {
+                    text: tokenId,
+                    variant: "normal",
+                  },
+                ],
+                style: "codepath",
+              }),
+            },
+          }
+        );
+
+        resolve(result);
+      } catch (error) {
+        reject(error);
      }
-    );
+    });
  },
 };

@@ -711,8 +771,3 @@ function calculateDurationInMs(options: WaitForOptions): number {

  throw new Error("Invalid options");
 }
-
-type RequestOptions = {
-  to: (url: string) => Promise<void>;
-  timeout: WaitForOptions;
-};
@@ -1930,6 +1930,15 @@ importers:
      '@trigger.dev/sdk':
        specifier: workspace:*
        version: link:../../packages/trigger-sdk
+      openai:
+        specifier: ^4.97.0
+        version: 4.97.0(zod@3.23.8)
+      replicate:
+        specifier: ^1.0.1
+        version: 1.0.1
+      zod:
+        specifier: 3.23.8
+        version: 3.23.8
    devDependencies:
      trigger.dev:
        specifier: workspace:*
@@ -28886,6 +28895,30 @@ packages:
      - encoding
    dev: false

+  /openai@4.97.0(zod@3.23.8):
+    resolution: {integrity: sha512-LRoiy0zvEf819ZUEJhgfV8PfsE8G5WpQi4AwA1uCV8SKvvtXQkoWUFkepD6plqyJQRghy2+AEPQ07FrJFKHZ9Q==}
+    hasBin: true
+    peerDependencies:
+      ws: ^8.18.0
+      zod: ^3.23.8
+    peerDependenciesMeta:
+      ws:
+        optional: true
+      zod:
+        optional: true
+    dependencies:
+      '@types/node': 18.19.20
+      '@types/node-fetch': 2.6.12
+      abort-controller: 3.0.0
+      agentkeepalive: 4.5.0
+      form-data-encoder: 1.7.2
+      formdata-node: 4.4.1
+      node-fetch: 2.6.12
+      zod: 3.23.8
+    transitivePeerDependencies:
+      - encoding
+    dev: false
+
  /openapi-fetch@0.9.8:
    resolution: {integrity: sha512-zM6elH0EZStD/gSiNlcPrzXcVQ/pZo3BDvC6CDwRDUt1dDzxlshpmQnpD6cZaJ39THaSmwVCxxRrPKNM1hHrDg==}
    dependencies:
@@ -30178,7 +30211,7 @@ packages:
  /process@0.11.10:
    resolution: {integrity: sha512-cdGef/drWFoydD1JsMzuFf8100nZl+GT+yacc2bEced5f9Rjk4z+WtFUTBu9PhOi9j/jfmBPu0mMEY4wIdAF8A==}
    engines: {node: '>= 0.6.0'}
-    dev: true
+    requiresBuild: true

  /progress@2.0.3:
    resolution: {integrity: sha512-7PiHtLll5LdnKIMw100I+8xJXR5gW2QwWYkT6iJva0bXitZKa/XMrSbdmg3r2Xnaidz9Qumd0VPaMrZlF9V9sA==}
@@ -31149,7 +31182,6 @@ packages:
      events: 3.3.0
      process: 0.11.10
      string_decoder: 1.3.0
-    dev: true

  /readdir-glob@1.1.3:
    resolution: {integrity: sha512-v05I2k7xN8zXvPD9N+z/uhXPaj0sUFCe2rcWZIpBsqxfP7xXFQ0tipAd/wjj1YxWyWtUS5IDJpOG82JKt2EAVA==}
@@ -31526,6 +31558,13 @@ packages:
    engines: {node: '>=8'}
    dev: true

+  /replicate@1.0.1:
+    resolution: {integrity: sha512-EY+rK1YR5bKHcM9pd6WyaIbv6m2aRIvHfHDh51j/LahlHTLKemTYXF6ptif2sLa+YospupAsIoxw8Ndt5nI3vg==}
+    engines: {git: '>=2.11.0', node: '>=18.0.0', npm: '>=7.19.0', yarn: '>=1.7.0'}
+    optionalDependencies:
+      readable-stream: 4.5.2
+    dev: false
+
  /request@2.88.2:
    resolution: {integrity: sha512-MsvtOrfG9ZcrOwAW+Qi+F6HbD0CWXEh9ou77uOb7FM2WPhwT7smM833PzanhJLsgXjN89Ir6V2PczXNnMpwKhw==}
    engines: {node: '>= 6'}
@@ -6,7 +6,10 @@
    "trigger.dev": "workspace:*"
  },
  "dependencies": {
-    "@trigger.dev/sdk": "workspace:*"
+    "@trigger.dev/sdk": "workspace:*",
+    "openai": "^4.97.0",
+    "replicate": "^1.0.1",
+    "zod": "3.23.8"
  },
  "scripts": {
    "dev": "trigger dev"
@@ -1,5 +1,5 @@
-import { logger, wait, task, retry, idempotencyKeys, auth } from "@trigger.dev/sdk/v3";
-
+import { auth, idempotencyKeys, logger, retry, task, wait } from "@trigger.dev/sdk/v3";
+import Replicate, { Prediction } from "replicate";
 type Token = {
  status: "approved" | "pending" | "rejected";
 };
@@ -140,3 +140,49 @@ export const waitForDuration = task({
    );
  },
 });
+
+export const waitHttpCallback = task({
+  id: "wait-http-callback",
+  retry: {
+    maxAttempts: 1,
+  },
+  run: async () => {
+    if (process.env.REPLICATE_API_KEY) {
+      const replicate = new Replicate({
+        auth: process.env.REPLICATE_API_KEY,
+      });
+
+      const token = await wait.createToken({
+        timeout: "10m",
+        tags: ["replicate"],
+      });
+      logger.log("Create result", { token });
+
+      const call = await replicate.predictions.create({
+        version: "27b93a2413e7f36cd83da926f3656280b2931564ff050bf9575f1fdf9bcd7478",
+        input: {
+          prompt: "A painting of a cat by Any Warhol",
+        },
+        // pass the provided URL to Replicate's webhook, so they can "callback"
+        webhook: token.url,
+        webhook_events_filter: ["completed"],
+      });
+
+      const prediction = await wait.forToken<Prediction>(token);
+
+      if (!prediction.ok) {
+        throw new Error("Failed to create prediction");
+      }
+
+      logger.log("Prediction", prediction);
+
+      const imageUrl = prediction.output.output;
+      logger.log("Image URL", imageUrl);
+
+      //same again but with unwrapping
+      const result2 = await wait.forToken<Prediction>(token).unwrap();
+
+      logger.log("Result2", { result2 });
+    }
+  },
+});