fix(streaming): flush buffered deltas before finishing the stream externally

src/vercel/client/streamText.test.ts

@@ -91,9 +91,30 @@ export const streamTextEmptyReturnImmediately = action({
   },
 });
 
+export const streamTextThrottled = action({
+  args: { threadId: v.string() },
+  handler: async (ctx, { threadId }) => {
+    const result = await agent.streamText(
+      ctx,
+      { threadId },
+      { prompt: "Test" },
+      {
+        saveStreamDeltas: {
+          returnImmediately: true,
+          chunking: "word",
+          throttleMs: 60_000,
+        },
+      },
+    );
+    await result.consumeStream();
+    return { ok: true };
+  },
+});
+
 const testApi: ApiFromModules<{
   fns: {
     streamTextReturnImmediately: typeof streamTextReturnImmediately;
+    streamTextThrottled: typeof streamTextThrottled;
     streamTextEmptyAwaited: typeof streamTextEmptyAwaited;
     streamTextEmptyReturnImmediately: typeof streamTextEmptyReturnImmediately;
   };
@@ -221,3 +242,53 @@ describe("streamText with an empty final step (issue #274)", () => {
     },
   );
 });
+
+describe("saveStreamDeltas flushes buffered parts (issue #323)", () => {
+  test("deltas hold the full text when the generation outpaces the throttle", async () => {
+    const t = initConvexTest(schema);
+    const threadId = await t.run(async (ctx) =>
+      createThread(ctx, components.agent, { userId: "u1" }),
+    );
+
+    await t.action(testApi.streamTextThrottled, { threadId });
+    await t.finishAllScheduledFunctions(() => {});
+
+    const streams = await t.run(async (ctx) =>
+      ctx.runQuery(components.agent.streams.list, {
+        threadId,
+        statuses: ["streaming", "finished", "aborted"],
+      }),
+    );
+    const deltas = await t.run(async (ctx) =>
+      ctx.runQuery(components.agent.streams.listDeltas, {
+        threadId,
+        cursors: streams.map((s) => ({ streamId: s.streamId, cursor: 0 })),
+      }),
+    );
+
+    expect(streams).toHaveLength(1);
+    expect(streams[0].status).toBe("finished");
+
+    let cursor = 0;
+    for (const delta of deltas) {
+      expect(delta.start).toBe(cursor);
+      cursor = delta.end;
+    }
+
+    const parts = deltas.flatMap((d) => d.parts);
+    const types = parts.map((p) => p.type);
+    expect(types.at(0)).toBe("start");
+    expect(types).toContain("text-start");
+    expect(types).toContain("text-end");
+    // The stream-level "finish" chunk is emitted after the last step ends, so
+    // it cannot exist yet; the row's finished status carries that instead.
+    expect(types.at(-1)).toBe("finish-step");
+    expect(types).not.toContain("finish");
+    expect(
+      parts
+        .filter((p) => p.type === "text-delta")
+        .map((p) => (p as { delta?: string }).delta ?? "")
+        .join(""),
+    ).toBe(FINAL_TEXT);
+  });
+});

src/vercel/client/streamText.ts

@@ -215,7 +215,7 @@ export async function streamText<
       const createPendingMessage = await willContinue(steps, args.stopWhen);
       if (!createPendingMessage && streamer) {
         // Final step with streaming enabled.
-        streamer.markFinishedExternally();
+        await streamer.markFinishedExternally();
         if (willAwaitStream) {
           // We're about to `await stream` below — defer the save so it
           // happens atomically with stream finish (issue #181).

src/vercel/client/streaming.integration.test.ts

@@ -200,7 +200,7 @@ describe("HTTP Streaming Initiation", () => {
       );
 
       await streamer.getStreamId();
-      streamer.markFinishedExternally();
+      await streamer.markFinishedExternally();
 
       const result = streamText({
         model: mockModel({

src/vercel/client/streaming.ts

@@ -301,6 +301,9 @@ export class DeltaStreamer<T> {
       return;
     }
     await this.getStreamId();
+    if (this.#finishedExternally || this.abortController.signal.aborted) {
+      return;
+    }
     this.#nextParts.push(...parts);
     if (
       !this.#ongoingWrite &&
@@ -340,7 +343,20 @@ export class DeltaStreamer<T> {
    * When called, consumeStream() will skip calling finish() since it will be
    * handled elsewhere in the same mutation as message saving.
    */
-  public markFinishedExternally(): void {
+  public async markFinishedExternally(): Promise<void> {
+    while (!this.abortController.signal.aborted) {
+      const inFlight = this.#ongoingWrite;
+      await inFlight;
+      // #sendDelta reassigns #ongoingWrite from its own tail, so a write can
+      // still be live even though the buffer it drained is now empty.
+      if (this.#ongoingWrite !== inFlight) {
+        continue;
+      }
+      if (this.#nextParts.length === 0) {
+        break;
+      }
+      this.#ongoingWrite = this.#sendDelta();
+    }
     this.#finishedExternally = true;
   }