fix: address CodeRabbit review feedback on input streams PR

- Prevent waitpoint linkage loss: get before complete, then delete
- Fix timeout header validation for 0 and NaN values
- Allow skipAccessTokens as alternative to access token for v2 gating
- Clear timeout timer on abort in once() to prevent leaks
- Clean up tail state after error/completion for reconnection
- Use strict undefined check for falsy input payloads in wait()

Author: ericallam

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

@@ -1,7 +1,10 @@
 import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { $replica } from "~/db.server";
-import { getAndDeleteInputStreamWaitpoint } from "~/services/inputStreamWaitpointCache.server";
+import {
+  getInputStreamWaitpoint,
+  deleteInputStreamWaitpoint,
+} from "~/services/inputStreamWaitpointCache.server";
 import {
   createActionApiRoute,
   createLoaderApiRoute,
@@ -80,7 +83,8 @@ const { action } = createActionApiRoute(
     );
 
     // Check Redis cache for a linked .wait() waitpoint (fast, no DB hit if none)
-    const waitpointId = await getAndDeleteInputStreamWaitpoint(params.runId, params.streamId);
+    // Get first, complete, then delete — so the mapping survives if completeWaitpoint throws
+    const waitpointId = await getInputStreamWaitpoint(params.runId, params.streamId);
     if (waitpointId) {
       await engine.completeWaitpoint({
         id: waitpointId,
@@ -90,6 +94,7 @@ const { action } = createActionApiRoute(
           isError: false,
         },
       });
+      await deleteInputStreamWaitpoint(params.runId, params.streamId);
     }
 
     return json({ ok: true });
@@ -132,17 +137,18 @@ const loader = createLoaderApiRoute(
     const lastEventId = request.headers.get("Last-Event-ID") || undefined;
 
     const timeoutInSecondsRaw = request.headers.get("Timeout-Seconds") ?? undefined;
-    const timeoutInSeconds = timeoutInSecondsRaw ? parseInt(timeoutInSecondsRaw) : undefined;
+    const timeoutInSeconds =
+      timeoutInSecondsRaw !== undefined ? parseInt(timeoutInSecondsRaw, 10) : undefined;
 
-    if (timeoutInSeconds && isNaN(timeoutInSeconds)) {
+    if (timeoutInSeconds !== undefined && isNaN(timeoutInSeconds)) {
       return new Response("Invalid timeout seconds", { status: 400 });
     }
 
-    if (timeoutInSeconds && timeoutInSeconds < 1) {
+    if (timeoutInSeconds !== undefined && timeoutInSeconds < 1) {
       return new Response("Timeout seconds must be greater than 0", { status: 400 });
     }
 
-    if (timeoutInSeconds && timeoutInSeconds > 600) {
+    if (timeoutInSeconds !== undefined && timeoutInSeconds > 600) {
       return new Response("Timeout seconds must be less than 600", { status: 400 });
     }
 

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

@@ -54,6 +54,29 @@ export async function setInputStreamWaitpoint(
   }
 }
 
+/**
+ * Get the waitpoint ID for an input stream without deleting it.
+ * Called from the `.send()` route before completing the waitpoint.
+ */
+export async function getInputStreamWaitpoint(
+  runFriendlyId: string,
+  streamId: string
+): Promise<string | null> {
+  if (!redis) return null;
+
+  try {
+    const key = buildKey(runFriendlyId, streamId);
+    return await redis.get(key);
+  } catch (error) {
+    logger.error("Failed to get input stream waitpoint cache", {
+      runFriendlyId,
+      streamId,
+      error,
+    });
+    return null;
+  }
+}
+
 /**
  * Atomically get and delete the waitpoint ID for an input stream.
  * Uses GETDEL for atomicity — only one concurrent `.send()` call will get the ID.

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

@@ -36,7 +36,11 @@ export function getRealtimeStreamInstance(
   if (streamVersion === "v1") {
     return v1RealtimeStreams;
   } else {
-    if (env.REALTIME_STREAMS_S2_BASIN && env.REALTIME_STREAMS_S2_ACCESS_TOKEN) {
+    if (
+      env.REALTIME_STREAMS_S2_BASIN &&
+      (env.REALTIME_STREAMS_S2_ACCESS_TOKEN ||
+        env.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS === "true")
+    ) {
       return new S2RealtimeStreams({
         basin: env.REALTIME_STREAMS_S2_BASIN,
         accessToken: env.REALTIME_STREAMS_S2_ACCESS_TOKEN,
@@ -70,7 +74,7 @@ export function determineRealtimeStreamsVersion(streamVersion?: string): "v1" |
   if (
     streamVersion === "v2" &&
     env.REALTIME_STREAMS_S2_BASIN &&
-    env.REALTIME_STREAMS_S2_ACCESS_TOKEN
+    (env.REALTIME_STREAMS_S2_ACCESS_TOKEN || env.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS === "true")
   ) {
     return "v2";
   }

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

@@ -100,6 +100,9 @@ export class StandardInputStreamManager implements InputStreamManager {
         options.signal.addEventListener(
           "abort",
           () => {
+            if (waiter.timeoutHandle) {
+              clearTimeout(waiter.timeoutHandle);
+            }
             this.#removeOnceWaiter(streamId, waiter);
             reject(new Error("Aborted"));
           },
@@ -174,13 +177,15 @@ export class StandardInputStreamManager implements InputStreamManager {
   #ensureStreamTailConnected(streamId: string): void {
     if (!this.tails.has(streamId) && this.currentRunId) {
       const abortController = new AbortController();
-      const promise = this.#runTail(this.currentRunId, streamId, abortController.signal).catch(
-        (error) => {
+      const promise = this.#runTail(this.currentRunId, streamId, abortController.signal)
+        .catch((error) => {
           if (this.debug) {
             console.error(`[InputStreamManager] Tail error for "${streamId}":`, error);
           }
-        }
-      );
+        })
+        .finally(() => {
+          this.tails.delete(streamId);
+        });
       this.tails.set(streamId, { abortController, promise });
     }
   }

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

@@ -771,15 +771,16 @@ function input<TData>(opts: { id: string }): RealtimeDefinedInputStream<TData> {
               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;
+              const data =
+                waitResult.output !== undefined
+                  ? await conditionallyImportAndParsePacket(
+                      {
+                        data: waitResult.output,
+                        dataType: waitResult.outputType ?? "application/json",
+                      },
+                      apiClient
+                    )
+                  : undefined;
 
               if (waitResult.ok) {
                 return { ok: true as const, output: data as TData };

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

@@ -20,7 +20,7 @@ export const inputStreamCoordinator = task({
   run: async () => {
     const results: Record<string, unknown> = {};
 
-    // --- Test 1: .once() ---
+    // --- Test 1: .once() ----
     logger.info("Test 1: .once()");
     const onceHandle = await inputStreamOnce.trigger({});
     await wait.for({ seconds: 5 });