refactor: connect input stream tail lazily from listener side

Instead of relying on an INPUT_STREAM_CREATED IPC message from the
platform (which was never wired up), the worker now connects the SSE
tail automatically when task code first calls .on() or .once() on an
input stream. The worker receives the run ID via setRunId() at
execution start, and #ensureTailConnected() opens the tail on demand.

This eliminates the need for platform→coordinator→executor→worker IPC
signaling and ensures no race conditions since the SSE tail reads from
the beginning of the __input stream.

https://claude.ai/code/session_01SJHJts7r2yAxmoKLLz8vpc

Author: claude

packages/cli-v3/src/entryPoints/dev-run-worker.ts

@@ -391,6 +391,7 @@ const zodIpc = new ZodIpcConnection({
       }
 
       resetExecutionEnvironment();
+      standardInputStreamManager.setRunId(execution.run.id);
 
       standardTraceContextManager.traceContext = traceContext;
       standardRunTimelineMetricsManager.registerMetricsFromExecution(metrics, isWarmStart);
@@ -645,9 +646,6 @@ const zodIpc = new ZodIpcConnection({
     RESOLVE_WAITPOINT: async ({ waitpoint }) => {
       _sharedWorkerRuntime?.resolveWaitpoints([waitpoint]);
     },
-    INPUT_STREAM_CREATED: async ({ runId }) => {
-      standardInputStreamManager.connectTail(runId);
-    },
   },
 });
 

packages/cli-v3/src/entryPoints/managed-run-worker.ts

@@ -375,6 +375,7 @@ const zodIpc = new ZodIpcConnection({
       }
 
       resetExecutionEnvironment();
+      standardInputStreamManager.setRunId(execution.run.id);
 
       standardTraceContextManager.traceContext = traceContext;
 
@@ -638,9 +639,6 @@ const zodIpc = new ZodIpcConnection({
     RESOLVE_WAITPOINT: async ({ waitpoint }) => {
       _sharedWorkerRuntime?.resolveWaitpoints([waitpoint]);
     },
-    INPUT_STREAM_CREATED: async ({ runId }) => {
-      standardInputStreamManager.connectTail(runId);
-    },
   },
 });
 

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

@@ -28,6 +28,10 @@ export class InputStreamsAPI implements InputStreamManager {
     return getGlobal(API_NAME) ?? NOOP_MANAGER;
   }
 
+  public setRunId(runId: string): void {
+    this.#getManager().setRunId(runId);
+  }
+
   public on(
     streamId: string,
     handler: (data: unknown) => void | Promise<void>

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

@@ -26,13 +26,18 @@ export class StandardInputStreamManager implements InputStreamManager {
   private buffer = new Map<string, unknown[]>();
   private tailAbortController: AbortController | null = null;
   private tailPromise: Promise<void> | null = null;
+  private currentRunId: string | null = null;
 
   constructor(
     private apiClient: ApiClient,
     private baseUrl: string,
     private debug: boolean = false
   ) {}
 
+  setRunId(runId: string): void {
+    this.currentRunId = runId;
+  }
+
   on(streamId: string, handler: InputStreamHandler): { off: () => void } {
     let handlerSet = this.handlers.get(streamId);
     if (!handlerSet) {
@@ -41,6 +46,9 @@ export class StandardInputStreamManager implements InputStreamManager {
     }
     handlerSet.add(handler);
 
+    // Lazily connect the tail on first listener registration
+    this.#ensureTailConnected();
+
     // Flush any buffered data for this stream
     const buffered = this.buffer.get(streamId);
     if (buffered && buffered.length > 0) {
@@ -61,6 +69,9 @@ export class StandardInputStreamManager implements InputStreamManager {
   }
 
   once(streamId: string, options?: InputStreamOnceOptions): Promise<unknown> {
+    // Lazily connect the tail on first listener registration
+    this.#ensureTailConnected();
+
     // Check buffer first
     const buffered = this.buffer.get(streamId);
     if (buffered && buffered.length > 0) {
@@ -140,6 +151,7 @@ export class StandardInputStreamManager implements InputStreamManager {
 
   reset(): void {
     this.disconnect();
+    this.currentRunId = null;
     this.handlers.clear();
 
     // Reject all pending once waiters
@@ -155,6 +167,12 @@ export class StandardInputStreamManager implements InputStreamManager {
     this.buffer.clear();
   }
 
+  #ensureTailConnected(): void {
+    if (!this.tailAbortController && this.currentRunId) {
+      this.connectTail(this.currentRunId);
+    }
+  }
+
   async #runTail(runId: string, signal: AbortSignal): Promise<void> {
     try {
       const stream = await this.apiClient.fetchStream<InputStreamRecord>(

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

@@ -2,6 +2,8 @@ import { InputStreamManager } from "./types.js";
 import { InputStreamOnceOptions } from "../realtimeStreams/types.js";
 
 export class NoopInputStreamManager implements InputStreamManager {
+  setRunId(_runId: string): void {}
+
   on(_streamId: string, _handler: (data: unknown) => void | Promise<void>): { off: () => void } {
     return { off: () => {} };
   }

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

@@ -1,6 +1,12 @@
 import { InputStreamOnceOptions } from "../realtimeStreams/types.js";
 
 export interface InputStreamManager {
+  /**
+   * Set the current run ID. The tail connection will be established lazily
+   * when `on()` or `once()` is first called.
+   */
+  setRunId(runId: string): void;
+
   /**
    * Register a handler that fires every time data arrives on the given input stream.
    */

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

@@ -229,12 +229,6 @@ export const WorkerToExecutorMessageCatalog = {
       waitpoint: CompletedWaitpoint,
     }),
   },
-  INPUT_STREAM_CREATED: {
-    message: z.object({
-      version: z.literal("v1").default("v1"),
-      runId: z.string(),
-    }),
-  },
 };
 
 export const ProviderToPlatformMessages = {