fix: gate input stream tail connection on v2 realtime streams

Input streams are backed by S2 which is only available with v2 realtime
streams. The lazy tail connection now checks that streamsVersion is "v2"
before connecting, so runs using v1 (Redis-backed) streams won't
attempt to open an S2 tail.

https://claude.ai/code/session_01SJHJts7r2yAxmoKLLz8vpc

Author: claude

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

@@ -391,7 +391,7 @@ const zodIpc = new ZodIpcConnection({
       }
 
       resetExecutionEnvironment();
-      standardInputStreamManager.setRunId(execution.run.id);
+      standardInputStreamManager.setRunId(execution.run.id, execution.run.realtimeStreamsVersion);
 
       standardTraceContextManager.traceContext = traceContext;
       standardRunTimelineMetricsManager.registerMetricsFromExecution(metrics, isWarmStart);

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

@@ -375,7 +375,7 @@ const zodIpc = new ZodIpcConnection({
       }
 
       resetExecutionEnvironment();
-      standardInputStreamManager.setRunId(execution.run.id);
+      standardInputStreamManager.setRunId(execution.run.id, execution.run.realtimeStreamsVersion);
 
       standardTraceContextManager.traceContext = traceContext;
 

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

@@ -28,8 +28,8 @@ export class InputStreamsAPI implements InputStreamManager {
     return getGlobal(API_NAME) ?? NOOP_MANAGER;
   }
 
-  public setRunId(runId: string): void {
-    this.#getManager().setRunId(runId);
+  public setRunId(runId: string, streamsVersion?: string): void {
+    this.#getManager().setRunId(runId, streamsVersion);
   }
 
   public on(

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

@@ -27,15 +27,17 @@ export class StandardInputStreamManager implements InputStreamManager {
   private tailAbortController: AbortController | null = null;
   private tailPromise: Promise<void> | null = null;
   private currentRunId: string | null = null;
+  private streamsVersion: string | undefined;
 
   constructor(
     private apiClient: ApiClient,
     private baseUrl: string,
     private debug: boolean = false
   ) {}
 
-  setRunId(runId: string): void {
+  setRunId(runId: string, streamsVersion?: string): void {
     this.currentRunId = runId;
+    this.streamsVersion = streamsVersion;
   }
 
   on(streamId: string, handler: InputStreamHandler): { off: () => void } {
@@ -152,6 +154,7 @@ export class StandardInputStreamManager implements InputStreamManager {
   reset(): void {
     this.disconnect();
     this.currentRunId = null;
+    this.streamsVersion = undefined;
     this.handlers.clear();
 
     // Reject all pending once waiters
@@ -168,7 +171,7 @@ export class StandardInputStreamManager implements InputStreamManager {
   }
 
   #ensureTailConnected(): void {
-    if (!this.tailAbortController && this.currentRunId) {
+    if (!this.tailAbortController && this.currentRunId && this.streamsVersion === "v2") {
       this.connectTail(this.currentRunId);
     }
   }

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

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

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

@@ -2,10 +2,11 @@ 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.
+   * Set the current run ID and streams version. The tail connection will be
+   * established lazily when `on()` or `once()` is first called, but only
+   * for v2 (S2-backed) realtime streams.
    */
-  setRunId(runId: string): void;
+  setRunId(runId: string, streamsVersion?: string): void;
 
   /**
    * Register a handler that fires every time data arrives on the given input stream.