input stream waitpoints and tests in the hello world reference project

.changeset/input-stream-wait.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/core": patch
+"@trigger.dev/sdk": patch
+---
+
+Add `.wait()` method to input streams for suspending tasks while waiting for data. Unlike `.once()` which keeps the task process alive, `.wait()` suspends the task entirely, freeing compute resources. The task resumes when data arrives via `.send()`.

.claude/skills/trigger-dev-tasks/realtime.md

@@ -128,7 +128,40 @@ export const approval = streams.input<{ approved: boolean; reviewer: string }>({
 
 ### Receiving Data Inside a Task
 
-#### `once()` — Wait for the next value
+#### `wait()` — Suspend until data arrives (recommended for long waits)
+
+Suspends the task entirely, freeing compute. Returns `ManualWaitpointPromise<TData>` (same as `wait.forToken()`).
+
+```ts
+import { task } from "@trigger.dev/sdk";
+import { approval } from "./streams";
+
+export const publishPost = task({
+  id: "publish-post",
+  run: async (payload: { postId: string }) => {
+    const draft = await prepareDraft(payload.postId);
+    await notifyReviewer(draft);
+
+    // Suspend — no compute cost while waiting
+    const result = await approval.wait({ timeout: "7d" });
+
+    if (result.ok) {
+      return { published: result.output.approved };
+    }
+    return { published: false, timedOut: true };
+  },
+});
+```
+
+Options: `timeout` (period string), `idempotencyKey`, `idempotencyKeyTTL`, `tags`.
+
+Use `.unwrap()` to throw on timeout: `const data = await approval.wait({ timeout: "24h" }).unwrap();`
+
+**Use `.wait()` when:** nothing to do until data arrives, wait could be long, want zero compute cost.
+
+#### `once()` — Wait for the next value (non-suspending)
+
+Keeps the task process alive. Use for short waits or when doing concurrent work.
 
 ```ts
 import { task } from "@trigger.dev/sdk";
@@ -150,6 +183,8 @@ export const draftEmailTask = task({
 
 Options: `once({ timeoutMs: 300_000 })` or `once({ signal: controller.signal })`.
 
+**Use `.once()` when:** wait is short, doing concurrent work, need AbortSignal support.
+
 #### `on()` — Listen for every value
 
 ```ts

.server-changes/input-stream-wait.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: feature
+---
+
+Add input stream `.wait()` support: new API route for creating input-stream-linked waitpoints, Redis cache for fast waitpoint lookup from `.send()`, and waitpoint completion bridging in the send route.

apps/webapp/app/routes/api.v1.runs.$runFriendlyId.input-streams.wait.ts

@@ -0,0 +1,170 @@
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import {
+  CreateInputStreamWaitpointRequestBody,
+  type CreateInputStreamWaitpointResponseBody,
+} from "@trigger.dev/core/v3";
+import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
+import { $replica } from "~/db.server";
+import { createWaitpointTag, MAX_TAGS_PER_WAITPOINT } from "~/models/waitpointTag.server";
+import {
+  deleteInputStreamWaitpoint,
+  setInputStreamWaitpoint,
+} from "~/services/inputStreamWaitpointCache.server";
+import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
+import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { parseDelay } from "~/utils/delays";
+import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
+import { engine } from "~/v3/runEngine.server";
+import { ServiceValidationError } from "~/v3/services/baseService.server";
+
+const ParamsSchema = z.object({
+  runFriendlyId: z.string(),
+});
+
+const { action, loader } = createActionApiRoute(
+  {
+    params: ParamsSchema,
+    body: CreateInputStreamWaitpointRequestBody,
+    maxContentLength: 1024 * 10, // 10KB
+    method: "POST",
+    allowJWT: true,
+    corsStrategy: "all",
+    authorization: {
+      action: "write",
+      resource: (params) => ({ inputStreams: params.runFriendlyId }),
+      superScopes: ["write:inputStreams", "write:all", "admin"],
+    },
+  },
+  async ({ authentication, body, params }) => {
+    try {
+      const run = await $replica.taskRun.findFirst({
+        where: {
+          friendlyId: params.runFriendlyId,
+          runtimeEnvironmentId: authentication.environment.id,
+        },
+        select: {
+          id: true,
+          friendlyId: true,
+          realtimeStreamsVersion: true,
+        },
+      });
+
+      if (!run) {
+        return json({ error: "Run not found" }, { status: 404 });
+      }
+
+      const idempotencyKeyExpiresAt = body.idempotencyKeyTTL
+        ? resolveIdempotencyKeyTTL(body.idempotencyKeyTTL)
+        : undefined;
+
+      const timeout = await parseDelay(body.timeout);
+
+      // Process tags (same pattern as api.v1.waitpoints.tokens.ts)
+      const bodyTags = typeof body.tags === "string" ? [body.tags] : body.tags;
+
+      if (bodyTags && bodyTags.length > MAX_TAGS_PER_WAITPOINT) {
+        throw new ServiceValidationError(
+          `Waitpoints can only have ${MAX_TAGS_PER_WAITPOINT} tags, you're trying to set ${bodyTags.length}.`
+        );
+      }
+
+      if (bodyTags && bodyTags.length > 0) {
+        for (const tag of bodyTags) {
+          await createWaitpointTag({
+            tag,
+            environmentId: authentication.environment.id,
+            projectId: authentication.environment.projectId,
+          });
+        }
+      }
+
+      // Step 1: Create the waitpoint
+      const result = await engine.createManualWaitpoint({
+        environmentId: authentication.environment.id,
+        projectId: authentication.environment.projectId,
+        idempotencyKey: body.idempotencyKey,
+        idempotencyKeyExpiresAt,
+        timeout,
+        tags: bodyTags,
+        inputStreamRunFriendlyId: run.friendlyId,
+        inputStreamId: body.streamId,
+      });
+
+      // Step 2: Cache the mapping in Redis for fast lookup from .send()
+      const ttlMs = timeout ? timeout.getTime() - Date.now() : undefined;
+      await setInputStreamWaitpoint(
+        run.friendlyId,
+        body.streamId,
+        result.waitpoint.id,
+        ttlMs && ttlMs > 0 ? ttlMs : undefined
+      );
+
+      // Step 3: Check if data was already sent to this input stream (race condition handling).
+      // If .send() landed before .wait(), the data is in the S2 stream but no waitpoint
+      // existed to complete. We check from the client's last known position.
+      if (!result.isCached) {
+        try {
+          const realtimeStream = getRealtimeStreamInstance(
+            authentication.environment,
+            run.realtimeStreamsVersion
+          );
+
+          if (realtimeStream.readRecords) {
+            const records = await realtimeStream.readRecords(
+              run.friendlyId,
+              "__input",
+              body.lastSeqNum
+            );
+
+            // Find the first record matching this input stream ID
+            for (const record of records) {
+              try {
+                const parsed = JSON.parse(record.data) as {
+                  stream: string;
+                  data: unknown;
+                };
+
+                if (parsed.stream === body.streamId) {
+                  // Data exists — complete the waitpoint immediately
+                  await engine.completeWaitpoint({
+                    id: result.waitpoint.id,
+                    output: {
+                      value: JSON.stringify(parsed.data),
+                      type: "application/json",
+                      isError: false,
+                    },
+                  });
+
+                  // Clean up the Redis cache since we completed it ourselves
+                  await deleteInputStreamWaitpoint(run.friendlyId, body.streamId);
+                  break;
+                }
+              } catch {
+                // Skip malformed records
+              }
+            }
+          }
+        } catch {
+          // Non-fatal: if the S2 check fails, the waitpoint is still PENDING.
+          // The next .send() will complete it via the Redis cache path.
+        }
+      }
+
+      return json<CreateInputStreamWaitpointResponseBody>({
+        waitpointId: WaitpointId.toFriendlyId(result.waitpoint.id),
+        isCached: result.isCached,
+      });
+    } catch (error) {
+      if (error instanceof ServiceValidationError) {
+        return json({ error: error.message }, { status: 422 });
+      } else if (error instanceof Error) {
+        return json({ error: error.message }, { status: 500 });
+      }
+
+      return json({ error: "Something went wrong" }, { status: 500 });
+    }
+  }
+);
+
+export { action, loader };

apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts

@@ -166,7 +166,7 @@ async function responseHeaders(
   const claims = {
     sub: environment.id,
     pub: true,
-    scopes: [`read:runs:${run.friendlyId}`],
+    scopes: [`read:runs:${run.friendlyId}`, `write:inputStreams:${run.friendlyId}`],
     realtime,
   };
 

apps/webapp/app/routes/realtime.v1.streams.$runId.input.$streamId.ts

@@ -1,8 +1,10 @@
 import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
-import { $replica, prisma } from "~/db.server";
+import { $replica } from "~/db.server";
+import { getAndDeleteInputStreamWaitpoint } from "~/services/inputStreamWaitpointCache.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
+import { engine } from "~/v3/runEngine.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
@@ -13,10 +15,17 @@ const BodySchema = z.object({
   data: z.unknown(),
 });
 
-const { action } = createActionApiRoute(
+const { action, loader } = createActionApiRoute(
   {
     params: ParamsSchema,
     maxContentLength: 1024 * 1024, // 1MB max
+    allowJWT: true,
+    corsStrategy: "all",
+    authorization: {
+      action: "write",
+      resource: (params) => ({ inputStreams: params.runId }),
+      superScopes: ["write:inputStreams", "write:all", "admin"],
+    },
   },
   async ({ request, params, authentication }) => {
     const run = await $replica.taskRun.findFirst({
@@ -28,7 +37,6 @@ const { action } = createActionApiRoute(
         id: true,
         friendlyId: true,
         completedAt: true,
-        hasInputStream: true,
         realtimeStreamsVersion: true,
       },
     });
@@ -55,16 +63,6 @@ const { action } = createActionApiRoute(
       run.realtimeStreamsVersion
     );
 
-    // Lazily create the input stream on first send
-    if (!run.hasInputStream) {
-      await prisma.taskRun.update({
-        where: { id: run.id },
-        data: { hasInputStream: true },
-      });
-
-      await realtimeStream.initializeStream(run.friendlyId, "__input");
-    }
-
     // Build the input stream record
     const recordId = `inp_${crypto.randomUUID().replace(/-/g, "").slice(0, 12)}`;
     const record = JSON.stringify({
@@ -74,11 +72,24 @@ const { action } = createActionApiRoute(
       id: recordId,
     });
 
-    // Append the record to the multiplexed __input stream
+    // Append the record to the multiplexed __input stream (auto-creates on first write)
     await realtimeStream.appendPart(record, recordId, run.friendlyId, "__input");
 
+    // Check Redis cache for a linked .wait() waitpoint (fast, no DB hit if none)
+    const waitpointId = await getAndDeleteInputStreamWaitpoint(params.runId, params.streamId);
+    if (waitpointId) {
+      await engine.completeWaitpoint({
+        id: waitpointId,
+        output: {
+          value: JSON.stringify(body.data.data),
+          type: "application/json",
+          isError: false,
+        },
+      });
+    }
+
     return json({ ok: true });
   }
 );
 
-export { action };
+export { action, loader };

apps/webapp/app/services/authorization.server.ts

@@ -1,6 +1,6 @@
 export type AuthorizationAction = "read" | "write" | string; // Add more actions as needed
 
-const ResourceTypes = ["tasks", "tags", "runs", "batch", "waitpoints", "deployments"] as const;
+const ResourceTypes = ["tasks", "tags", "runs", "batch", "waitpoints", "deployments", "inputStreams"] as const;
 
 export type AuthorizationResources = {
   [key in (typeof ResourceTypes)[number]]?: string | string[];

apps/webapp/app/services/inputStreamWaitpointCache.server.ts

@@ -0,0 +1,101 @@
+import { Redis } from "ioredis";
+import { env } from "~/env.server";
+import { singleton } from "~/utils/singleton";
+import { logger } from "./logger.server";
+
+const KEY_PREFIX = "isw:";
+const DEFAULT_TTL_MS = 7 * 24 * 60 * 60 * 1000; // 7 days
+
+function buildKey(runFriendlyId: string, streamId: string): string {
+  return `${KEY_PREFIX}${runFriendlyId}:${streamId}`;
+}
+
+function initializeRedis(): Redis | undefined {
+  const host = env.CACHE_REDIS_HOST;
+  if (!host) {
+    return undefined;
+  }
+
+  return new Redis({
+    connectionName: "inputStreamWaitpointCache",
+    host,
+    port: env.CACHE_REDIS_PORT,
+    username: env.CACHE_REDIS_USERNAME,
+    password: env.CACHE_REDIS_PASSWORD,
+    keyPrefix: "tr:",
+    enableAutoPipelining: true,
+    ...(env.CACHE_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
+  });
+}
+
+const redis = singleton("inputStreamWaitpointCache", initializeRedis);
+
+/**
+ * Store a mapping from input stream to waitpoint ID in Redis.
+ * Called when `.wait()` creates a new waitpoint.
+ */
+export async function setInputStreamWaitpoint(
+  runFriendlyId: string,
+  streamId: string,
+  waitpointId: string,
+  ttlMs?: number
+): Promise<void> {
+  if (!redis) return;
+
+  try {
+    const key = buildKey(runFriendlyId, streamId);
+    await redis.set(key, waitpointId, "PX", ttlMs ?? DEFAULT_TTL_MS);
+  } catch (error) {
+    logger.error("Failed to set input stream waitpoint cache", {
+      runFriendlyId,
+      streamId,
+      error,
+    });
+  }
+}
+
+/**
+ * Atomically get and delete the waitpoint ID for an input stream.
+ * Uses GETDEL for atomicity — only one concurrent `.send()` call will get the ID.
+ * Called from the `.send()` route.
+ */
+export async function getAndDeleteInputStreamWaitpoint(
+  runFriendlyId: string,
+  streamId: string
+): Promise<string | null> {
+  if (!redis) return null;
+
+  try {
+    const key = buildKey(runFriendlyId, streamId);
+    return await redis.getdel(key);
+  } catch (error) {
+    logger.error("Failed to get input stream waitpoint cache", {
+      runFriendlyId,
+      streamId,
+      error,
+    });
+    return null;
+  }
+}
+
+/**
+ * Delete the cache entry for an input stream waitpoint.
+ * Called when a waitpoint is completed or timed out.
+ */
+export async function deleteInputStreamWaitpoint(
+  runFriendlyId: string,
+  streamId: string
+): Promise<void> {
+  if (!redis) return;
+
+  try {
+    const key = buildKey(runFriendlyId, streamId);
+    await redis.del(key);
+  } catch (error) {
+    logger.error("Failed to delete input stream waitpoint cache", {
+      runFriendlyId,
+      streamId,
+      error,
+    });
+  }
+}

apps/webapp/app/services/realtime/s2realtimeStreams.server.ts

@@ -1,6 +1,6 @@
 // app/realtime/S2RealtimeStreams.ts
 import type { UnkeyCache } from "@internal/cache";
-import { StreamIngestor, StreamResponder, StreamResponseOptions } from "./types";
+import { StreamIngestor, StreamRecord, StreamResponder, StreamResponseOptions } from "./types";
 import { Logger, LogLevel } from "@trigger.dev/core/logger";
 import { randomUUID } from "node:crypto";
 
@@ -121,6 +121,87 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
     throw new Error("S2 streams are written to S2 via the client, not from the server");
   }
 
+  async readRecords(
+    runId: string,
+    streamId: string,
+    afterSeqNum?: number
+  ): Promise<StreamRecord[]> {
+    const s2Stream = this.toStreamName(runId, streamId);
+    const startSeq = afterSeqNum != null ? afterSeqNum + 1 : 0;
+
+    const qs = new URLSearchParams();
+    qs.set("seq_num", String(startSeq));
+    qs.set("clamp", "true");
+    qs.set("wait", "0"); // Non-blocking: return immediately with existing records
+
+    const res = await fetch(
+      `${this.baseUrl}/streams/${encodeURIComponent(s2Stream)}/records?${qs}`,
+      {
+        method: "GET",
+        headers: {
+          Authorization: `Bearer ${this.token}`,
+          Accept: "text/event-stream",
+          "S2-Format": "raw",
+        },
+      }
+    );
+
+    if (!res.ok) {
+      // Stream may not exist yet (no data sent)
+      if (res.status === 404) {
+        return [];
+      }
+      const text = await res.text().catch(() => "");
+      throw new Error(`S2 readRecords failed: ${res.status} ${res.statusText} ${text}`);
+    }
+
+    // Parse the SSE response body to extract records
+    const body = await res.text();
+    return this.parseSSEBatchRecords(body);
+  }
+
+  private parseSSEBatchRecords(sseText: string): StreamRecord[] {
+    const records: StreamRecord[] = [];
+
+    // SSE events are separated by double newlines
+    const events = sseText.split("\n\n").filter((e) => e.trim());
+
+    for (const event of events) {
+      const lines = event.split("\n");
+      let eventType: string | undefined;
+      let data: string | undefined;
+
+      for (const line of lines) {
+        if (line.startsWith("event:")) {
+          eventType = line.slice(6).trim();
+        } else if (line.startsWith("data:")) {
+          data = line.slice(5).trim();
+        }
+      }
+
+      if (eventType === "batch" && data) {
+        try {
+          const parsed = JSON.parse(data) as {
+            records: Array<{ body: string; seq_num: number; timestamp: number }>;
+          };
+
+          for (const record of parsed.records) {
+            const parsedBody = JSON.parse(record.body) as { data: string; id: string };
+            records.push({
+              data: parsedBody.data,
+              id: parsedBody.id,
+              seqNum: record.seq_num,
+            });
+          }
+        } catch {
+          // Skip malformed events
+        }
+      }
+    }
+
+    return records;
+  }
+
   // ---------- Serve SSE from S2 ----------
 
   async streamResponse(

apps/webapp/app/services/realtime/types.ts

@@ -1,3 +1,9 @@
+export type StreamRecord = {
+  data: string;
+  id: string;
+  seqNum: number;
+};
+
 // Interface for stream ingestion
 export interface StreamIngestor {
   initializeStream(
@@ -16,6 +22,17 @@ export interface StreamIngestor {
   appendPart(part: string, partId: string, runId: string, streamId: string): Promise<void>;
 
   getLastChunkIndex(runId: string, streamId: string, clientId: string): Promise<number>;
+
+  /**
+   * Read records from a stream starting after a given sequence number.
+   * Returns immediately with whatever records exist (non-blocking).
+   * Not all backends support this — returns undefined if unsupported.
+   */
+  readRecords?(
+    runId: string,
+    streamId: string,
+    afterSeqNum?: number
+  ): Promise<StreamRecord[] | undefined>;
 }
 
 export type StreamResponseOptions = {

internal-packages/database/prisma/migrations/20260222120000_add_has_input_stream/migration.sql

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

internal-packages/database/prisma/migrations/20260222130000_add_input_stream_waitpoint_columns/migration.sql

@@ -0,0 +1,3 @@
+-- AlterTable
+ALTER TABLE "Waitpoint" ADD COLUMN IF NOT EXISTS "inputStreamId" TEXT,
+ADD COLUMN IF NOT EXISTS "inputStreamRunFriendlyId" TEXT;

internal-packages/database/prisma/migrations/20260222130001_add_input_stream_waitpoint_index/migration.sql

@@ -0,0 +1,4 @@
+-- CreateIndex (CONCURRENTLY must be in its own migration)
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "Waitpoint_inputStream_idx"
+ON "Waitpoint" ("environmentId", "inputStreamRunFriendlyId", "inputStreamId", "status")
+WHERE "inputStreamRunFriendlyId" IS NOT NULL;

internal-packages/database/prisma/schema.prisma

@@ -778,8 +778,6 @@ model TaskRun {
   /// Store the stream keys that are being used by the run
   realtimeStreams        String[] @default([])
 
-  /// Whether this run has an active input stream (created lazily on first streams.input send)
-  hasInputStream Boolean @default(false)
 
   @@unique([oneTimeUseToken])
   @@unique([runtimeEnvironmentId, taskIdentifier, idempotencyKey])
@@ -1076,6 +1074,11 @@ model Waitpoint {
   createdAt DateTime @default(now())
   updatedAt DateTime @updatedAt
 
+  /// If this waitpoint is linked to an input stream .wait(), this is the run's friendlyId
+  inputStreamRunFriendlyId String?
+  /// If this waitpoint is linked to an input stream .wait(), this is the stream ID
+  inputStreamId            String?
+
   /// Denormized column that holds the raw tags
   /// Denormalized column that holds the raw tags
   tags String[]

internal-packages/run-engine/src/engine/index.ts

@@ -1481,6 +1481,20 @@ export class RunEngine {
     });
   }
 
+  /** Finds and completes a PENDING waitpoint linked to an input stream (DB fallback). */
+  async completeInputStreamWaitpoint(params: {
+    environmentId: string;
+    runFriendlyId: string;
+    streamId: string;
+    output?: {
+      value: string;
+      type?: string;
+      isError: boolean;
+    };
+  }): Promise<Waitpoint | null> {
+    return this.waitpointSystem.completeInputStreamWaitpoint(params);
+  }
+
   /**
    * This gets called AFTER the checkpoint has been created
    * The CPU/Memory checkpoint at this point exists in our snapshot storage

internal-packages/run-engine/src/engine/systems/waitpointSystem.ts

@@ -266,6 +266,8 @@ export class WaitpointSystem {
     idempotencyKeyExpiresAt,
     timeout,
     tags,
+    inputStreamRunFriendlyId,
+    inputStreamId,
   }: {
     environmentId: string;
     projectId: string;
@@ -273,6 +275,8 @@ export class WaitpointSystem {
     idempotencyKeyExpiresAt?: Date;
     timeout?: Date;
     tags?: string[];
+    inputStreamRunFriendlyId?: string;
+    inputStreamId?: string;
   }): Promise<{ waitpoint: Waitpoint; isCached: boolean }> {
     const existingWaitpoint = idempotencyKey
       ? await this.$.prisma.waitpoint.findFirst({
@@ -328,6 +332,8 @@ export class WaitpointSystem {
             projectId,
             completedAfter: timeout,
             tags,
+            inputStreamRunFriendlyId,
+            inputStreamId,
           },
           update: {},
         });
@@ -364,6 +370,40 @@ export class WaitpointSystem {
     throw new Error(`Failed to create waitpoint after ${maxRetries} attempts due to conflicts.`);
   }
 
+  /**
+   * Finds and completes a PENDING waitpoint linked to an input stream.
+   * This is the DB fallback path used when the Redis cache misses.
+   * Returns the completed waitpoint, or null if no matching waitpoint exists.
+   */
+  async completeInputStreamWaitpoint({
+    environmentId,
+    runFriendlyId,
+    streamId,
+    output,
+  }: {
+    environmentId: string;
+    runFriendlyId: string;
+    streamId: string;
+    output?: {
+      value: string;
+      type?: string;
+      isError: boolean;
+    };
+  }): Promise<Waitpoint | null> {
+    const waitpoint = await this.$.prisma.waitpoint.findFirst({
+      where: {
+        environmentId,
+        inputStreamRunFriendlyId: runFriendlyId,
+        inputStreamId: streamId,
+        status: "PENDING",
+      },
+    });
+
+    if (!waitpoint) return null;
+
+    return this.completeWaitpoint({ id: waitpoint.id, output });
+  }
+
   /**
    * Prevents a run from continuing until the waitpoint is completed.
    *

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

@@ -17,6 +17,8 @@ import {
   CreateBatchRequestBody,
   CreateBatchResponse,
   CreateEnvironmentVariableRequestBody,
+  CreateInputStreamWaitpointRequestBody,
+  CreateInputStreamWaitpointResponseBody,
   CreateScheduleOptions,
   CreateStreamResponseBody,
   CreateUploadPayloadUrlResponseBody,
@@ -1318,6 +1320,8 @@ export class ApiClient {
       onComplete?: () => void;
       onError?: (error: Error) => void;
       lastEventId?: string;
+      /** Called for each SSE event with the full event metadata (id, timestamp). */
+      onPart?: (part: SSEStreamPart<T>) => void;
     }
   ): Promise<AsyncIterableStream<T>> {
     const streamFactory = new SSEStreamSubscriptionFactory(options?.baseUrl ?? this.baseUrl, {
@@ -1334,10 +1338,14 @@ export class ApiClient {
 
     const stream = await subscription.subscribe();
 
+    const onPart = options?.onPart;
+
     return stream.pipeThrough(
       new TransformStream<SSEStreamPart, T>({
         transform(chunk, controller) {
-          controller.enqueue(chunk.chunk as T);
+          const data = chunk.chunk as T;
+          onPart?.(chunk as SSEStreamPart<T>);
+          controller.enqueue(data);
         },
       })
     );
@@ -1404,6 +1412,23 @@ export class ApiClient {
     );
   }
 
+  async createInputStreamWaitpoint(
+    runFriendlyId: string,
+    body: CreateInputStreamWaitpointRequestBody,
+    requestOptions?: ZodFetchOptions
+  ) {
+    return zodfetch(
+      CreateInputStreamWaitpointResponseBody,
+      `${this.baseUrl}/api/v1/runs/${runFriendlyId}/input-streams/wait`,
+      {
+        method: "POST",
+        headers: this.#getHeaders(false),
+        body: JSON.stringify(body),
+      },
+      mergeRequestOptions(this.defaultRequestOptions, requestOptions)
+    );
+  }
+
   async generateJWTClaims(requestOptions?: ZodFetchOptions): Promise<Record<string, any>> {
     return zodfetch(
       z.record(z.any()),

packages/core/src/v3/index.ts

@@ -21,6 +21,7 @@ export * from "./locals-api.js";
 export * from "./heartbeats-api.js";
 export * from "./realtime-streams-api.js";
 export * from "./input-streams-api.js";
+export * from "./waitpoints/index.js";
 export * from "./schemas/index.js";
 export { SemanticInternalAttributes } from "./semanticInternalAttributes.js";
 export * from "./resource-catalog-api.js";

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

@@ -47,6 +47,10 @@ export class InputStreamsAPI implements InputStreamManager {
     return this.#getManager().peek(streamId);
   }
 
+  public get lastSeqNum(): number | undefined {
+    return this.#getManager().lastSeqNum;
+  }
+
   public reset(): void {
     this.#getManager().reset();
   }

packages/core/src/v3/inputStreams/manager.ts

@@ -28,6 +28,7 @@ export class StandardInputStreamManager implements InputStreamManager {
   private tailPromise: Promise<void> | null = null;
   private currentRunId: string | null = null;
   private streamsVersion: string | undefined;
+  private _lastSeqNum: number | undefined;
 
   constructor(
     private apiClient: ApiClient,
@@ -35,6 +36,10 @@ export class StandardInputStreamManager implements InputStreamManager {
     private debug: boolean = false
   ) {}
 
+  get lastSeqNum(): number | undefined {
+    return this._lastSeqNum;
+  }
+
   setRunId(runId: string, streamsVersion?: string): void {
     this.currentRunId = runId;
     this.streamsVersion = streamsVersion;
@@ -159,6 +164,7 @@ export class StandardInputStreamManager implements InputStreamManager {
     this.disconnect();
     this.currentRunId = null;
     this.streamsVersion = undefined;
+    this._lastSeqNum = undefined;
     this.handlers.clear();
 
     // Reject all pending once waiters
@@ -196,8 +202,14 @@ export class StandardInputStreamManager implements InputStreamManager {
         {
           signal,
           baseUrl: this.baseUrl,
-          // Long timeout — we want to keep tailing for the duration of the run
-          timeoutInSeconds: 3600,
+          // Max allowed by the SSE endpoint is 600s; the tail will reconnect on close
+          timeoutInSeconds: 600,
+          onPart: (part) => {
+            const seqNum = parseInt(part.id, 10);
+            if (Number.isFinite(seqNum)) {
+              this._lastSeqNum = seqNum;
+            }
+          },
           onComplete: () => {
             if (this.debug) {
               console.log("[InputStreamManager] Tail stream completed");
@@ -213,7 +225,22 @@ export class StandardInputStreamManager implements InputStreamManager {
 
       for await (const record of stream) {
         if (signal.aborted) break;
-        this.#dispatchRecord(record);
+
+        // S2 SSE returns record bodies as JSON strings; parse into InputStreamRecord
+        let parsed: InputStreamRecord;
+        if (typeof record === "string") {
+          try {
+            parsed = JSON.parse(record) as InputStreamRecord;
+          } catch {
+            continue;
+          }
+        } else if (record.stream) {
+          parsed = record;
+        } else {
+          continue;
+        }
+
+        this.#dispatchRecord(parsed);
       }
     } catch (error) {
       // AbortError is expected when disconnecting

packages/core/src/v3/inputStreams/noopManager.ts

@@ -18,6 +18,10 @@ export class NoopInputStreamManager implements InputStreamManager {
     return undefined;
   }
 
+  get lastSeqNum(): number | undefined {
+    return undefined;
+  }
+
   reset(): void {}
   disconnect(): void {}
   connectTail(_runId: string, _fromSeq?: number): void {}

packages/core/src/v3/inputStreams/types.ts

@@ -23,6 +23,12 @@ export interface InputStreamManager {
    */
   peek(streamId: string): unknown | undefined;
 
+  /**
+   * The last S2 sequence number seen by the input stream tail.
+   * Used by `.wait()` to tell the server where to check for existing data.
+   */
+  readonly lastSeqNum: number | undefined;
+
   /**
    * Reset state between task executions.
    */

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

@@ -1,6 +1,7 @@
 import { AnyZodFetchOptions, ApiRequestOptions } from "../apiClient/core.js";
 import { AsyncIterableStream } from "../streams/asyncIterableStream.js";
 import { Prettify } from "../types/utils.js";
+import type { ManualWaitpointPromise } from "../waitpoints/index.js";
 
 export type RealtimeStreamOperationOptions = {
   signal?: AbortSignal;
@@ -169,6 +170,16 @@ export type RealtimeDefinedInputStream<TData> = {
    * Returns `undefined` if no data has been received yet.
    */
   peek: () => TData | undefined;
+  /**
+   * Suspend the task until data arrives on this input stream.
+   *
+   * Unlike `.once()` which keeps the task process alive while waiting,
+   * `.wait()` suspends the task entirely — freeing compute resources.
+   * The task resumes when data is sent via `.send()`.
+   *
+   * Uses a waitpoint token internally. Can only be called inside a task.run().
+   */
+  wait: (options?: InputStreamWaitOptions) => ManualWaitpointPromise<TData>;
   /**
    * Send data to this input stream on a specific run.
    * This is used from outside the task (e.g., from your backend or another task).
@@ -189,6 +200,37 @@ export type SendInputStreamOptions = {
   requestOptions?: ApiRequestOptions;
 };
 
+export type InputStreamWaitOptions = {
+  /**
+   * Maximum time to wait before the waitpoint times out.
+   * Uses the same period format as `wait.createToken()`.
+   * If the timeout is reached, the result will be `{ ok: false, error }`.
+   *
+   * @example "30s", "5m", "1h", "24h", "7d"
+   */
+  timeout?: string;
+
+  /**
+   * Idempotency key for the underlying waitpoint token.
+   * If the same key is used again (and hasn't expired), the existing
+   * waitpoint is reused. This means if the task retries, it will
+   * resume waiting on the same waitpoint rather than creating a new one.
+   */
+  idempotencyKey?: string;
+
+  /**
+   * TTL for the idempotency key. After this period, the same key
+   * will create a new waitpoint.
+   */
+  idempotencyKeyTTL?: string;
+
+  /**
+   * Tags for the underlying waitpoint token, useful for querying
+   * and filtering waitpoints via `wait.listTokens()`.
+   */
+  tags?: string[];
+};
+
 export type InferInputStreamType<T> = T extends RealtimeDefinedInputStream<infer TData>
   ? TData
   : unknown;

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

@@ -1370,6 +1370,31 @@ export const CreateWaitpointTokenResponseBody = z.object({
 });
 export type CreateWaitpointTokenResponseBody = z.infer<typeof CreateWaitpointTokenResponseBody>;
 
+export const CreateInputStreamWaitpointRequestBody = z.object({
+  streamId: z.string(),
+  timeout: z.string().optional(),
+  idempotencyKey: z.string().optional(),
+  idempotencyKeyTTL: z.string().optional(),
+  tags: z.union([z.string(), z.array(z.string())]).optional(),
+  /**
+   * The last S2 sequence number the client has seen on the __input stream.
+   * Used to check for data that arrived before .wait() was called.
+   * If undefined, the server checks from the beginning of the stream.
+   */
+  lastSeqNum: z.number().optional(),
+});
+export type CreateInputStreamWaitpointRequestBody = z.infer<
+  typeof CreateInputStreamWaitpointRequestBody
+>;
+
+export const CreateInputStreamWaitpointResponseBody = z.object({
+  waitpointId: z.string(),
+  isCached: z.boolean(),
+});
+export type CreateInputStreamWaitpointResponseBody = z.infer<
+  typeof CreateInputStreamWaitpointResponseBody
+>;
+
 export const waitpointTokenStatuses = ["WAITING", "COMPLETED", "TIMED_OUT"] as const;
 export const WaitpointTokenStatus = z.enum(waitpointTokenStatuses);
 export type WaitpointTokenStatus = z.infer<typeof WaitpointTokenStatus>;

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

@@ -0,0 +1,35 @@
+import type { WaitpointTokenTypedResult } from "../schemas/common.js";
+
+export class WaitpointTimeoutError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = "WaitpointTimeoutError";
+  }
+}
+
+export 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);
+      }
+    });
+  }
+}

packages/react-hooks/src/hooks/useInputStreamSend.ts

@@ -0,0 +1,60 @@
+"use client";
+
+import useSWRMutation from "swr/mutation";
+import { useApiClient, UseApiClientOptions } from "./useApiClient.js";
+
+export interface InputStreamSendInstance<TData> {
+  /** Send data to the input stream */
+  send: (data: TData) => void;
+  /** Whether a send is currently in progress */
+  isLoading: boolean;
+  /** Any error that occurred during the last send */
+  error?: Error;
+  /** Whether the hook is ready to send (has runId and access token) */
+  isReady: boolean;
+}
+
+/**
+ * Hook to send data to an input stream on a running task.
+ *
+ * @template TData - The type of data to send
+ * @param streamId - The input stream identifier
+ * @param runId - The run to send input stream data to
+ * @param options - API client options (e.g. accessToken)
+ *
+ * @example
+ * ```tsx
+ * const { send, isLoading } = useInputStreamSend("my-stream", runId, { accessToken });
+ * send({ message: "hello" });
+ * ```
+ */
+export function useInputStreamSend<TData>(
+  streamId: string,
+  runId?: string,
+  options?: UseApiClientOptions
+): InputStreamSendInstance<TData> {
+  const apiClient = useApiClient(options);
+
+  async function sendToStream(key: string, { arg }: { arg: { data: TData } }) {
+    if (!apiClient) {
+      throw new Error("Could not send to input stream: Missing access token");
+    }
+
+    if (!runId) {
+      throw new Error("Could not send to input stream: Missing run ID");
+    }
+
+    return await apiClient.sendInputStream(runId, streamId, arg.data);
+  }
+
+  const mutation = useSWRMutation(runId ? `input-stream:${runId}:${streamId}` : null, sendToStream);
+
+  return {
+    send: (data) => {
+      mutation.trigger({ data });
+    },
+    isLoading: mutation.isMutating,
+    isReady: !!runId && !!apiClient,
+    error: mutation.error,
+  };
+}

packages/react-hooks/src/index.ts

@@ -4,3 +4,4 @@ export * from "./hooks/useRun.js";
 export * from "./hooks/useRealtime.js";
 export * from "./hooks/useTaskTrigger.js";
 export * from "./hooks/useWaitToken.js";
+export * from "./hooks/useInputStreamSend.js";

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

@@ -62,6 +62,11 @@ type PublicTokenPermissionProperties = {
    * Grant access to specific waitpoints
    */
   waitpoints?: string | string[];
+
+  /**
+   * Grant access to send data to input streams on specific runs
+   */
+  inputStreams?: string | string[];
 };
 
 export type PublicTokenPermissions = {

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

@@ -16,12 +16,17 @@ import {
   AppendStreamOptions,
   RealtimeDefinedStream,
   InferStreamType,
+  ManualWaitpointPromise,
+  WaitpointTimeoutError,
+  runtime,
   type RealtimeDefinedInputStream,
   type InputStreamSubscription,
   type InputStreamOnceOptions,
+  type InputStreamWaitOptions,
   type SendInputStreamOptions,
   type InferInputStreamType,
 } from "@trigger.dev/core/v3";
+import { conditionallyImportAndParsePacket } from "@trigger.dev/core/v3/utils/ioSerialization";
 import { tracer } from "./tracer.js";
 import { SpanStatusCode } from "@opentelemetry/api";
 
@@ -703,6 +708,89 @@ function input<TData>(opts: { id: string }): RealtimeDefinedInputStream<TData> {
     peek() {
       return inputStreams.peek(opts.id) as TData | undefined;
     },
+    wait(options) {
+      return new ManualWaitpointPromise<TData>(async (resolve, reject) => {
+        try {
+          const ctx = taskContext.ctx;
+
+          if (!ctx) {
+            throw new Error("inputStream.wait() can only be used from inside a task.run()");
+          }
+
+          const apiClient = apiClientManager.clientOrThrow();
+
+          const result = await tracer.startActiveSpan(
+            `inputStream.wait()`,
+            async (span) => {
+              // 1. Create a waitpoint linked to this input stream
+              const response = await apiClient.createInputStreamWaitpoint(ctx.run.id, {
+                streamId: opts.id,
+                timeout: options?.timeout,
+                idempotencyKey: options?.idempotencyKey,
+                idempotencyKeyTTL: options?.idempotencyKeyTTL,
+                tags: options?.tags,
+                lastSeqNum: inputStreams.lastSeqNum,
+              });
+
+              // 2. Block the run on the waitpoint
+              const waitResponse = await apiClient.waitForWaitpointToken({
+                runFriendlyId: ctx.run.id,
+                waitpointFriendlyId: response.waitpointId,
+              });
+
+              if (!waitResponse.success) {
+                throw new Error("Failed to block on input stream waitpoint");
+              }
+
+              // 3. Suspend the task
+              const waitResult = await runtime.waitUntil(response.waitpointId);
+
+              // 4. Parse the output
+              const data = waitResult.output
+                ? await conditionallyImportAndParsePacket(
+                    {
+                      data: waitResult.output,
+                      dataType: waitResult.outputType ?? "application/json",
+                    },
+                    apiClient
+                  )
+                : undefined;
+
+              if (waitResult.ok) {
+                return { ok: true as const, output: data as TData };
+              } else {
+                const error = new WaitpointTimeoutError(data?.message ?? "Timed out");
+
+                span.recordException(error);
+                span.setStatus({ code: SpanStatusCode.ERROR });
+
+                return { ok: false as const, error };
+              }
+            },
+            {
+              attributes: {
+                [SemanticInternalAttributes.STYLE_ICON]: "wait",
+                [SemanticInternalAttributes.ENTITY_TYPE]: "waitpoint",
+                streamId: opts.id,
+                ...accessoryAttributes({
+                  items: [
+                    {
+                      text: `input:${opts.id}`,
+                      variant: "normal",
+                    },
+                  ],
+                  style: "codepath",
+                }),
+              },
+            }
+          );
+
+          resolve(result);
+        } catch (error) {
+          reject(error);
+        }
+      });
+    },
     async send(runId, data, options) {
       const apiClient = apiClientManager.clientOrThrow();
       await apiClient.sendInputStream(runId, opts.id, data, options?.requestOptions);

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

@@ -11,6 +11,7 @@ import {
   CursorPagePromise,
   flattenAttributes,
   ListWaitpointTokensQueryParams,
+  ManualWaitpointPromise,
   mergeRequestOptions,
   runtime,
   SemanticInternalAttributes,
@@ -19,6 +20,7 @@ import {
   WaitpointRetrieveTokenResponse,
   WaitpointTokenStatus,
   WaitpointTokenTypedResult,
+  WaitpointTimeoutError,
 } from "@trigger.dev/core/v3";
 import { conditionallyImportAndParsePacket } from "@trigger.dev/core/v3/utils/ioSerialization";
 import { tracer } from "./tracer.js";
@@ -378,12 +380,7 @@ type WaitPeriod =
       years: number;
     };
 
-export class WaitpointTimeoutError extends Error {
-  constructor(message: string) {
-    super(message);
-    this.name = "WaitpointTimeoutError";
-  }
-}
+export { WaitpointTimeoutError, ManualWaitpointPromise } from "@trigger.dev/core/v3";
 
 const DURATION_WAIT_CHARGE_THRESHOLD_MS = 5000;
 
@@ -393,29 +390,6 @@ 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;

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

@@ -0,0 +1,118 @@
+import { logger, runs, streams, task, wait } from "@trigger.dev/sdk/v3";
+
+// Define typed input streams
+const approvalStream = streams.input<{ approved: boolean; reviewer: string }>({
+  id: "approval",
+});
+
+const messageStream = streams.input<{ text: string }>({ id: "messages" });
+
+/**
+ * Coordinator task that exercises all input stream patterns end-to-end.
+ *
+ * 1. .once() — trigger a child, send it data via SSE tail, poll until complete
+ * 2. .on()   — trigger a child, send it multiple messages, poll until complete
+ * 3. .wait() — trigger a child, send it data (completes its waitpoint), poll until complete
+ * 4. .wait() race — send data before child calls .wait(), verify race handling
+ */
+export const inputStreamCoordinator = task({
+  id: "input-stream-coordinator",
+  run: async () => {
+    const results: Record<string, unknown> = {};
+
+    // --- Test 1: .once() ---
+    logger.info("Test 1: .once()");
+    const onceHandle = await inputStreamOnce.trigger({});
+    await wait.for({ seconds: 5 });
+    await approvalStream.send(onceHandle.id, { approved: true, reviewer: "coordinator-once" });
+    const onceRun = await runs.poll(onceHandle, { pollIntervalMs: 1000 });
+    results.once = onceRun.output;
+    logger.info("Test 1 passed", { output: onceRun.output });
+
+    // --- Test 2: .on() with multiple messages ---
+    logger.info("Test 2: .on()");
+    const onHandle = await inputStreamOn.trigger({ messageCount: 3 });
+    await wait.for({ seconds: 5 });
+    for (let i = 0; i < 3; i++) {
+      await messageStream.send(onHandle.id, { text: `message-${i}` });
+      await wait.for({ seconds: 1 });
+    }
+    const onRun = await runs.poll(onHandle, { pollIntervalMs: 1000 });
+    results.on = onRun.output;
+    logger.info("Test 2 passed", { output: onRun.output });
+
+    // --- Test 3: .wait() (waitpoint-based) ---
+    logger.info("Test 3: .wait()");
+    const waitHandle = await inputStreamWait.trigger({ timeout: "1m" });
+    await wait.for({ seconds: 5 });
+    await approvalStream.send(waitHandle.id, { approved: true, reviewer: "coordinator-wait" });
+    const waitRun = await runs.poll(waitHandle, { pollIntervalMs: 1000 });
+    results.wait = waitRun.output;
+    logger.info("Test 3 passed", { output: waitRun.output });
+
+    // --- Test 4: .wait() race condition (send before child calls .wait()) ---
+    logger.info("Test 4: .wait() race");
+    const raceHandle = await inputStreamWait.trigger({ timeout: "1m" });
+    await approvalStream.send(raceHandle.id, { approved: false, reviewer: "race-test" });
+    const raceRun = await runs.poll(raceHandle, { pollIntervalMs: 1000 });
+    results.race = raceRun.output;
+    logger.info("Test 4 passed", { output: raceRun.output });
+
+    logger.info("All input stream tests passed", { results });
+    return results;
+  },
+});
+
+/**
+ * Uses .once() to wait for a single input stream message.
+ */
+export const inputStreamOnce = task({
+  id: "input-stream-once",
+  run: async (_payload: Record<string, never>) => {
+    logger.info("Waiting for approval via .once()");
+    const approval = await approvalStream.once();
+    logger.info("Received approval", { approval });
+    return { approval };
+  },
+});
+
+/**
+ * Uses .on() to subscribe and collect multiple messages.
+ */
+export const inputStreamOn = task({
+  id: "input-stream-on",
+  run: async (payload: { messageCount?: number }) => {
+    const expected = payload.messageCount ?? 3;
+    const received: { text: string }[] = [];
+
+    logger.info("Subscribing to messages via .on()", { expected });
+
+    const { off } = messageStream.on((data) => {
+      logger.info("Received message", { data });
+      received.push(data);
+    });
+
+    while (received.length < expected) {
+      await wait.for({ seconds: 1 });
+    }
+
+    off();
+    logger.info("Done receiving messages", { count: received.length });
+    return { messages: received };
+  },
+});
+
+/**
+ * Uses .wait() to suspend the task via a waitpoint until data arrives.
+ */
+export const inputStreamWait = task({
+  id: "input-stream-wait",
+  run: async (payload: { timeout?: string }) => {
+    logger.info("Waiting for approval via .wait()");
+    const approval = await approvalStream.wait({
+      timeout: payload.timeout ?? "5m",
+    });
+    logger.info("Received approval via .wait()", { approval });
+    return { approval };
+  },
+});