fix(core): avoid ReadableStream lock after fullStream startup probe (#1103)

* fix(core): avoid fullStream lock after startup probe

* test(core): tighten teeing fullStream regression assertions

Author: omeraplak

.changeset/witty-cameras-smile.md

@@ -0,0 +1,7 @@
+---
+"@voltagent/core": patch
+---
+
+fix: preserve getter-based `fullStream` tee behavior after startup probing in `streamText`/`streamObject`
+
+This prevents `TypeError [ERR_INVALID_STATE]: Invalid state: ReadableStream is locked` when SDK consumers iterate `result.fullStream` while other result accessors (such as `result.text` or UI stream helpers) are also consuming the stream.

packages/core/src/agent/agent.spec.ts

@@ -1056,6 +1056,92 @@ Use pandas and summarize findings.`.split("\n"),
       expect(emittedTypes).toContain("text-delta");
       expect(emittedTypes).toContain("finish");
     });
+
+    it("does not lock getter-based teeing fullStream after probe", async () => {
+      const agent = new Agent({
+        name: "TestAgent",
+        instructions: "You are a helpful assistant",
+        model: mockModel as any,
+      });
+
+      const streamParts = [
+        { type: "start" as const },
+        { type: "text-start" as const, id: "text-1" },
+        { type: "text-delta" as const, id: "text-1", delta: "Hello " },
+        { type: "text-delta" as const, id: "text-1", delta: "world" },
+        { type: "text-end" as const, id: "text-1" },
+        { type: "finish" as const, finishReason: "stop", totalUsage: {} },
+      ];
+
+      class TeeingStreamResult {
+        private baseStream: ReadableStream<any>;
+        private readonly cachedText: Promise<string>;
+
+        constructor(parts: ReadonlyArray<any>) {
+          this.baseStream = convertArrayToReadableStream([...parts]);
+          this.cachedText = this.consumeText();
+        }
+
+        private teeStream(): ai.AsyncIterableStream<any> {
+          const [probeStream, passthroughStream] = this.baseStream.tee();
+          this.baseStream = passthroughStream;
+          return toAsyncIterableStream(probeStream);
+        }
+
+        get fullStream(): ai.AsyncIterableStream<any> {
+          return this.teeStream();
+        }
+
+        get text(): Promise<string> {
+          return this.cachedText;
+        }
+
+        get textStream(): AsyncIterable<string> {
+          return (async function* () {
+            yield "Hello world";
+          })();
+        }
+
+        readonly usage = Promise.resolve({
+          inputTokens: 10,
+          outputTokens: 2,
+          totalTokens: 12,
+        });
+        readonly finishReason = Promise.resolve("stop");
+        readonly warnings: never[] = [];
+        readonly toUIMessageStream = vi.fn();
+        readonly toUIMessageStreamResponse = vi.fn();
+        readonly pipeUIMessageStreamToResponse = vi.fn();
+        readonly pipeTextStreamToResponse = vi.fn();
+        readonly toTextStreamResponse = vi.fn();
+        readonly partialOutputStream = undefined;
+
+        private async consumeText(): Promise<string> {
+          let text = "";
+          for await (const part of this.fullStream) {
+            if (part.type === "text-delta" && typeof part.delta === "string") {
+              text += part.delta;
+            }
+          }
+          return text;
+        }
+      }
+
+      const mockStream = new TeeingStreamResult(streamParts);
+      vi.mocked(ai.streamText).mockReturnValue(mockStream as any);
+
+      const result = await agent.streamText("answer me");
+      const emittedTypes: string[] = [];
+
+      for await (const part of result.fullStream as AsyncIterable<{ type: string }>) {
+        emittedTypes.push(part.type);
+      }
+
+      expect(emittedTypes).toContain("start");
+      expect(emittedTypes).toContain("text-delta");
+      expect(emittedTypes).toContain("finish");
+      await expect(result.text).resolves.toBe("Hello world");
+    });
   });
 
   describe("Tool Management", () => {

packages/core/src/agent/agent.ts

@@ -2010,11 +2010,13 @@ export class Agent {
               },
             });
 
-            const probeResult = await this.probeStreamStart(streamResult.fullStream, attemptState);
-            const streamResultForConsumption =
-              probeResult.stream === streamResult.fullStream
-                ? streamResult
-                : this.cloneResultWithFullStream(streamResult, probeResult.stream);
+            const originalFullStream = streamResult.fullStream;
+            const probeResult = await this.probeStreamStart(originalFullStream, attemptState);
+            const streamResultForConsumption = this.withProbedFullStream(
+              streamResult,
+              originalFullStream,
+              probeResult.stream,
+            );
 
             if (probeResult.status === "error") {
               this.discardStream(streamResultForConsumption.fullStream);
@@ -3136,11 +3138,13 @@ export class Agent {
               },
             });
 
-            const probeResult = await this.probeStreamStart(streamResult.fullStream, attemptState);
-            const streamResultForConsumption =
-              probeResult.stream === streamResult.fullStream
-                ? streamResult
-                : this.cloneResultWithFullStream(streamResult, probeResult.stream);
+            const originalFullStream = streamResult.fullStream;
+            const probeResult = await this.probeStreamStart(originalFullStream, attemptState);
+            const streamResultForConsumption = this.withProbedFullStream(
+              streamResult,
+              originalFullStream,
+              probeResult.stream,
+            );
 
             if (probeResult.status === "error") {
               this.discardStream(streamResultForConsumption.fullStream);
@@ -5317,6 +5321,53 @@ export class Agent {
     return clone;
   }
 
+  private withProbedFullStream<
+    TResult extends {
+      fullStream: AsyncIterableStream<unknown>;
+    },
+  >(
+    result: TResult,
+    originalFullStream: TResult["fullStream"],
+    probedFullStream: TResult["fullStream"],
+  ): TResult {
+    if (probedFullStream === originalFullStream) {
+      return result;
+    }
+
+    if (this.usesGetterBasedTeeingFullStream(result)) {
+      // AI SDK stream results expose fullStream via a teeing getter.
+      // Preserving the original instance keeps that multi-consumer behavior intact.
+      return result;
+    }
+
+    return this.cloneResultWithFullStream(result, probedFullStream);
+  }
+
+  private usesGetterBasedTeeingFullStream(result: {
+    fullStream: AsyncIterableStream<unknown>;
+  }): boolean {
+    const descriptor = this.findPropertyDescriptor(result, "fullStream");
+    return (
+      typeof descriptor?.get === "function" &&
+      typeof (result as { teeStream?: unknown }).teeStream === "function"
+    );
+  }
+
+  private findPropertyDescriptor(
+    target: object,
+    propertyName: string,
+  ): PropertyDescriptor | undefined {
+    let current: object | null = target;
+    while (current) {
+      const descriptor = Object.getOwnPropertyDescriptor(current, propertyName);
+      if (descriptor) {
+        return descriptor;
+      }
+      current = Object.getPrototypeOf(current);
+    }
+    return undefined;
+  }
+
   private toAsyncIterableStream<T>(stream: ReadableStream<T>): AsyncIterableStream<T> {
     const asyncStream = stream as AsyncIterableStream<T>;