No more input stream multiplexing, no naming collision risk, removed unnecessary waitpoint added columns.

Author: ericallam

.gitignore

@@ -67,4 +67,5 @@ apps/**/public/build
 **/.claude/settings.local.json
 .mcp.log
 .mcp.json
-.cursor/debug.log
+.cursor/debug.log
+ailogger-output.log

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

@@ -28,13 +28,6 @@ const { action, loader } = createActionApiRoute(
     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 {
@@ -87,8 +80,6 @@ const { action, loader } = createActionApiRoute(
         idempotencyKeyExpiresAt,
         timeout,
         tags: bodyTags,
-        inputStreamRunFriendlyId: run.friendlyId,
-        inputStreamId: body.streamId,
       });
 
       // Step 2: Cache the mapping in Redis for fast lookup from .send()
@@ -113,33 +104,27 @@ const { action, loader } = createActionApiRoute(
           if (realtimeStream.readRecords) {
             const records = await realtimeStream.readRecords(
               run.friendlyId,
-              "__input",
+              `$trigger.input:${body.streamId}`,
               body.lastSeqNum
             );
 
-            // Find the first record matching this input stream ID
-            for (const record of records) {
+            if (records.length > 0) {
+              const record = records[0]!;
               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;
-                }
+                const parsed = JSON.parse(record.data) as { data: unknown };
+
+                // 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);
               } catch {
                 // Skip malformed records
               }

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

@@ -2,7 +2,10 @@ import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { $replica } from "~/db.server";
 import { getAndDeleteInputStreamWaitpoint } from "~/services/inputStreamWaitpointCache.server";
-import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import {
+  createActionApiRoute,
+  createLoaderApiRoute,
+} from "~/services/routeBuilders/apiBuilder.server";
 import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
 import { engine } from "~/v3/runEngine.server";
 
@@ -15,7 +18,8 @@ const BodySchema = z.object({
   data: z.unknown(),
 });
 
-const { action, loader } = createActionApiRoute(
+// POST: Send data to an input stream
+const { action } = createActionApiRoute(
   {
     params: ParamsSchema,
     maxContentLength: 1024 * 1024, // 1MB max
@@ -66,14 +70,18 @@ const { action, loader } = createActionApiRoute(
     // Build the input stream record
     const recordId = `inp_${crypto.randomUUID().replace(/-/g, "").slice(0, 12)}`;
     const record = JSON.stringify({
-      stream: params.streamId,
       data: body.data.data,
       ts: Date.now(),
       id: recordId,
     });
 
-    // Append the record to the multiplexed __input stream (auto-creates on first write)
-    await realtimeStream.appendPart(record, recordId, run.friendlyId, "__input");
+    // Append the record to the per-stream S2 stream (auto-creates on first write)
+    await realtimeStream.appendPart(
+      record,
+      recordId,
+      run.friendlyId,
+      `$trigger.input:${params.streamId}`
+    );
 
     // Check Redis cache for a linked .wait() waitpoint (fast, no DB hit if none)
     const waitpointId = await getAndDeleteInputStreamWaitpoint(params.runId, params.streamId);
@@ -92,4 +100,73 @@ const { action, loader } = createActionApiRoute(
   }
 );
 
+// GET: SSE stream for reading input stream data (used by the in-task SSE tail)
+const loader = createLoaderApiRoute(
+  {
+    params: ParamsSchema,
+    allowJWT: true,
+    corsStrategy: "all",
+    findResource: async (params, auth) => {
+      return $replica.taskRun.findFirst({
+        where: {
+          friendlyId: params.runId,
+          runtimeEnvironmentId: auth.environment.id,
+        },
+        include: {
+          batch: {
+            select: {
+              friendlyId: true,
+            },
+          },
+        },
+      });
+    },
+    authorization: {
+      action: "read",
+      resource: (run) => ({
+        runs: run.friendlyId,
+        tags: run.runTags,
+        batch: run.batch?.friendlyId,
+        tasks: run.taskIdentifier,
+      }),
+      superScopes: ["read:runs", "read:all", "admin"],
+    },
+  },
+  async ({ params, request, resource: run, authentication }) => {
+    const lastEventId = request.headers.get("Last-Event-ID") || undefined;
+
+    const timeoutInSecondsRaw = request.headers.get("Timeout-Seconds") ?? undefined;
+    const timeoutInSeconds = timeoutInSecondsRaw ? parseInt(timeoutInSecondsRaw) : undefined;
+
+    if (timeoutInSeconds && isNaN(timeoutInSeconds)) {
+      return new Response("Invalid timeout seconds", { status: 400 });
+    }
+
+    if (timeoutInSeconds && timeoutInSeconds < 1) {
+      return new Response("Timeout seconds must be greater than 0", { status: 400 });
+    }
+
+    if (timeoutInSeconds && timeoutInSeconds > 600) {
+      return new Response("Timeout seconds must be less than 600", { status: 400 });
+    }
+
+    const realtimeStream = getRealtimeStreamInstance(
+      authentication.environment,
+      run.realtimeStreamsVersion
+    );
+
+    // Read from the internal S2 stream name (prefixed to avoid user stream collisions)
+    return realtimeStream.streamResponse(
+      request,
+      run.friendlyId,
+      `$trigger.input:${params.streamId}`,
+      request.signal,
+      {
+        lastEventId,
+        timeoutInSeconds,
+      }
+    );
+  }
+);
+
 export { action, loader };

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

@@ -1074,11 +1074,6 @@ 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/database/prisma/migrations/20260222130001_add_input_stream_waitpoint_index/migration.sql

@@ -1481,20 +1481,6 @@ 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/database/prisma/schema.prisma

@@ -266,17 +266,13 @@ export class WaitpointSystem {
     idempotencyKeyExpiresAt,
     timeout,
     tags,
-    inputStreamRunFriendlyId,
-    inputStreamId,
   }: {
     environmentId: string;
     projectId: string;
     idempotencyKey?: string;
     idempotencyKeyExpiresAt?: Date;
     timeout?: Date;
     tags?: string[];
-    inputStreamRunFriendlyId?: string;
-    inputStreamId?: string;
   }): Promise<{ waitpoint: Waitpoint; isCached: boolean }> {
     const existingWaitpoint = idempotencyKey
       ? await this.$.prisma.waitpoint.findFirst({
@@ -332,8 +328,6 @@ export class WaitpointSystem {
             projectId,
             completedAfter: timeout,
             tags,
-            inputStreamRunFriendlyId,
-            inputStreamId,
           },
           update: {},
         });
@@ -370,40 +364,6 @@ 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.
    *

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

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