diff --git a/.changeset/input-stream-wait.md b/.changeset/input-stream-wait.md new file mode 100644 index 000000000..96aeba1cd --- /dev/null +++ b/.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()`. diff --git a/.claude/skills/trigger-dev-tasks/realtime.md b/.claude/skills/trigger-dev-tasks/realtime.md index 0e3d0747b..811f13ebb 100644 --- a/.claude/skills/trigger-dev-tasks/realtime.md +++ b/.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` (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 diff --git a/.server-changes/input-stream-wait.md b/.server-changes/input-stream-wait.md new file mode 100644 index 000000000..dc3f05b28 --- /dev/null +++ b/.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. diff --git a/apps/webapp/app/routes/api.v1.runs.$runFriendlyId.input-streams.wait.ts b/apps/webapp/app/routes/api.v1.runs.$runFriendlyId.input-streams.wait.ts new file mode 100644 index 000000000..c6268205d --- /dev/null +++ b/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({ + 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 }; diff --git a/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts b/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts index 0c188c177..d55c3659e 100644 --- a/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts +++ b/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, }; diff --git a/apps/webapp/app/routes/realtime.v1.streams.$runId.input.$streamId.ts b/apps/webapp/app/routes/realtime.v1.streams.$runId.input.$streamId.ts index 825184ac6..4eab5b687 100644 --- a/apps/webapp/app/routes/realtime.v1.streams.$runId.input.$streamId.ts +++ b/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 }; diff --git a/apps/webapp/app/services/authorization.server.ts b/apps/webapp/app/services/authorization.server.ts index 15f85cc32..2ea410e2c 100644 --- a/apps/webapp/app/services/authorization.server.ts +++ b/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[]; diff --git a/apps/webapp/app/services/inputStreamWaitpointCache.server.ts b/apps/webapp/app/services/inputStreamWaitpointCache.server.ts new file mode 100644 index 000000000..e272b117b --- /dev/null +++ b/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 { + 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 { + 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 { + 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, + }); + } +} diff --git a/apps/webapp/app/services/realtime/s2realtimeStreams.server.ts b/apps/webapp/app/services/realtime/s2realtimeStreams.server.ts index 1d03116ff..918c32ac7 100644 --- a/apps/webapp/app/services/realtime/s2realtimeStreams.server.ts +++ b/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 { + 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( diff --git a/apps/webapp/app/services/realtime/types.ts b/apps/webapp/app/services/realtime/types.ts index 912711019..208e9f524 100644 --- a/apps/webapp/app/services/realtime/types.ts +++ b/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; getLastChunkIndex(runId: string, streamId: string, clientId: string): Promise; + + /** + * 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; } export type StreamResponseOptions = { diff --git a/internal-packages/database/prisma/migrations/20260222120000_add_has_input_stream/migration.sql b/internal-packages/database/prisma/migrations/20260222120000_add_has_input_stream/migration.sql deleted file mode 100644 index f355e1add..000000000 --- a/internal-packages/database/prisma/migrations/20260222120000_add_has_input_stream/migration.sql +++ /dev/null @@ -1,2 +0,0 @@ --- AlterTable -ALTER TABLE "TaskRun" ADD COLUMN "hasInputStream" BOOLEAN NOT NULL DEFAULT false; diff --git a/internal-packages/database/prisma/migrations/20260222130000_add_input_stream_waitpoint_columns/migration.sql b/internal-packages/database/prisma/migrations/20260222130000_add_input_stream_waitpoint_columns/migration.sql new file mode 100644 index 000000000..c3af224ea --- /dev/null +++ b/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; diff --git a/internal-packages/database/prisma/migrations/20260222130001_add_input_stream_waitpoint_index/migration.sql b/internal-packages/database/prisma/migrations/20260222130001_add_input_stream_waitpoint_index/migration.sql new file mode 100644 index 000000000..94a4efe10 --- /dev/null +++ b/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; diff --git a/internal-packages/database/prisma/schema.prisma b/internal-packages/database/prisma/schema.prisma index 8716072a0..c3d72b82d 100644 --- a/internal-packages/database/prisma/schema.prisma +++ b/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[] diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts index 846252398..727d36138 100644 --- a/internal-packages/run-engine/src/engine/index.ts +++ b/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 { + 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 diff --git a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts index f2ad85cc9..6c9342696 100644 --- a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts +++ b/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 { + 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. * diff --git a/packages/core/src/v3/apiClient/index.ts b/packages/core/src/v3/apiClient/index.ts index 016b64ffb..7de6e275f 100644 --- a/packages/core/src/v3/apiClient/index.ts +++ b/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) => void; } ): Promise> { 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({ transform(chunk, controller) { - controller.enqueue(chunk.chunk as T); + const data = chunk.chunk as T; + onPart?.(chunk as SSEStreamPart); + 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> { return zodfetch( z.record(z.any()), diff --git a/packages/core/src/v3/index.ts b/packages/core/src/v3/index.ts index 61cb7faa7..2757363f4 100644 --- a/packages/core/src/v3/index.ts +++ b/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"; diff --git a/packages/core/src/v3/inputStreams/index.ts b/packages/core/src/v3/inputStreams/index.ts index 0a7f2930b..df0c29b0f 100644 --- a/packages/core/src/v3/inputStreams/index.ts +++ b/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(); } diff --git a/packages/core/src/v3/inputStreams/manager.ts b/packages/core/src/v3/inputStreams/manager.ts index 25ad90b96..526d72c7d 100644 --- a/packages/core/src/v3/inputStreams/manager.ts +++ b/packages/core/src/v3/inputStreams/manager.ts @@ -28,6 +28,7 @@ export class StandardInputStreamManager implements InputStreamManager { private tailPromise: Promise | 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 diff --git a/packages/core/src/v3/inputStreams/noopManager.ts b/packages/core/src/v3/inputStreams/noopManager.ts index 5b5d3df53..76c01ad98 100644 --- a/packages/core/src/v3/inputStreams/noopManager.ts +++ b/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 {} diff --git a/packages/core/src/v3/inputStreams/types.ts b/packages/core/src/v3/inputStreams/types.ts index 0bbafa49a..b633f6deb 100644 --- a/packages/core/src/v3/inputStreams/types.ts +++ b/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. */ diff --git a/packages/core/src/v3/realtimeStreams/types.ts b/packages/core/src/v3/realtimeStreams/types.ts index 4962c4783..33b77c9af 100644 --- a/packages/core/src/v3/realtimeStreams/types.ts +++ b/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 = { * 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; /** * 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 extends RealtimeDefinedInputStream ? TData : unknown; diff --git a/packages/core/src/v3/schemas/api.ts b/packages/core/src/v3/schemas/api.ts index 6c932ae08..e7d3a021f 100644 --- a/packages/core/src/v3/schemas/api.ts +++ b/packages/core/src/v3/schemas/api.ts @@ -1370,6 +1370,31 @@ export const CreateWaitpointTokenResponseBody = z.object({ }); export type CreateWaitpointTokenResponseBody = z.infer; +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; diff --git a/packages/core/src/v3/waitpoints/index.ts b/packages/core/src/v3/waitpoints/index.ts new file mode 100644 index 000000000..fc70ff57f --- /dev/null +++ b/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 extends Promise< + WaitpointTokenTypedResult +> { + constructor( + executor: ( + resolve: ( + value: + | WaitpointTokenTypedResult + | PromiseLike> + ) => void, + reject: (reason?: any) => void + ) => void + ) { + super(executor); + } + + unwrap(): Promise { + return this.then((result) => { + if (result.ok) { + return result.output; + } else { + throw new WaitpointTimeoutError(result.error.message); + } + }); + } +} diff --git a/packages/react-hooks/src/hooks/useInputStreamSend.ts b/packages/react-hooks/src/hooks/useInputStreamSend.ts new file mode 100644 index 000000000..d3f816f30 --- /dev/null +++ b/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 { + /** 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( + streamId: string, + runId?: string, + options?: UseApiClientOptions +): InputStreamSendInstance { + 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, + }; +} diff --git a/packages/react-hooks/src/index.ts b/packages/react-hooks/src/index.ts index bc20cf837..23c8ca947 100644 --- a/packages/react-hooks/src/index.ts +++ b/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"; diff --git a/packages/trigger-sdk/src/v3/auth.ts b/packages/trigger-sdk/src/v3/auth.ts index ddcf92569..1f2df463b 100644 --- a/packages/trigger-sdk/src/v3/auth.ts +++ b/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 = { diff --git a/packages/trigger-sdk/src/v3/streams.ts b/packages/trigger-sdk/src/v3/streams.ts index 9d4ebf37a..1426bd247 100644 --- a/packages/trigger-sdk/src/v3/streams.ts +++ b/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(opts: { id: string }): RealtimeDefinedInputStream { peek() { return inputStreams.peek(opts.id) as TData | undefined; }, + wait(options) { + return new ManualWaitpointPromise(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); diff --git a/packages/trigger-sdk/src/v3/wait.ts b/packages/trigger-sdk/src/v3/wait.ts index 7ac85b034..aab0797f4 100644 --- a/packages/trigger-sdk/src/v3/wait.ts +++ b/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 extends Promise> { - constructor( - executor: ( - resolve: ( - value: WaitpointTokenTypedResult | PromiseLike> - ) => void, - reject: (reason?: any) => void - ) => void - ) { - super(executor); - } - - unwrap(): Promise { - 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; diff --git a/references/hello-world/src/trigger/inputStreams.ts b/references/hello-world/src/trigger/inputStreams.ts new file mode 100644 index 000000000..8d9ff4261 --- /dev/null +++ b/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 = {}; + + // --- 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) => { + 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 }; + }, +});