# Files changed · PR \#24432 · pingdotgg/t3code

[View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/24432/diffs)

Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code)

Visibility: public

[fix\(server\): avoid reparsing streamed replies from the beginning](https://git.cafe/pingdotgg/t3code/pull/24432.md)

Version: 2

Head commit: 8de253eebffb79842ddaf2457b98142ce77931d4

Base commit: d53873b6a4799e87afd63c8d288ef0089f5a3748

Comparison base: d53873b6a4799e87afd63c8d288ef0089f5a3748

Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing.

## apps/server/src/orchestration-v2/assistantStreaming.test.ts

Status: modified; +53; −2

````diff
@@ -1,5 +1,8 @@
+import { MessageId, ProviderDriverKind, ThreadId } from "@t3tools/contracts";
+import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
+import { DateTime } from "effect";
+import { describe, expect, it, vi } from "vite-plus/test";
-import { describe, expect, it } from "vite-plus/test";
+import { makeAssistantStreamingFilter, splitBufferedAssistantText } from "./assistantStreaming.ts";
-import { splitBufferedAssistantText } from "./assistantStreaming.ts";
 
 describe("V2 assistant streaming", () => {
   it("keeps code fences intact", () => {
@@ -24,3 +27,51 @@
     rest: "Next",
   });
 });
+
+const assistantUpdate = (text: string): ProviderAdapter.ProviderAdapterV2Event => ({
+  type: "message.updated",
+  driver: ProviderDriverKind.make("codex"),
+  message: {
+    createdBy: "agent",
+    creationSource: "provider",
+    id: MessageId.make("message-1"),
+    threadId: ThreadId.make("thread-1"),
+    runId: null,
+    nodeId: null,
+    role: "assistant",
+    text,
+    attachments: [],
+    streaming: true,
+    createdAt: DateTime.makeUnsafe(0),
+    updatedAt: DateTime.makeUnsafe(0),
+  },
+});
+
+it("rescans a streamed reply the provider replaces", () => {
+  const filter = makeAssistantStreamingFilter("paragraph");
+  expect(filter(assistantUpdate("Intro\n\n```ts\nx()\n"), 0)).toMatchObject({
+    message: { text: "Intro\n\n" },
+  });
+  expect(filter(assistantUpdate("Retry\n\n- a\n- b"), 400)).toMatchObject({
+    message: { text: "Retry\n\n- a\n" },
+  });
+});
+
+it("scans each streamed line once while a long code fence is still open", () => {
+  const filter = makeAssistantStreamingFilter("paragraph");
+  const lineCount = 400;
+  let text = "Intro\n\n```ts\n";
+  const indexOf = vi.spyOn(String.prototype, "indexOf");
+  try {
+    for (let line = 0; line < lineCount; line++) {
+      text += `const value${line} = ${line};\n`;
+      filter(assistantUpdate(text), line * 400);
+    }
+    const newlineSearches = indexOf.mock.calls.filter(([search]) => search === "\n").length;
+    expect(newlineSearches).toBeLessThan(lineCount * 4);
+  } finally {
+    indexOf.mockRestore();
+  }
+  text += "```\n";
+  expect(filter(assistantUpdate(text), lineCount * 400)).toMatchObject({ message: { text } });
+});
````

## apps/server/src/orchestration-v2/assistantStreaming.ts

Status: modified; +119; −27

```diff
@@ -16,6 +16,65 @@
 // blank line between them. A bold line would continue the paragraph instead.
 const TOP_LEVEL_HEADING_PATTERN = /^#{1,6}(?:[ \t]|$)/;
 
+// Sticky helpers that inspect a streaming partial line in place, so a line that
+// grows without a newline is never copied or rescanned from its start.
+const INDENT_PATTERN = /[ \t]*/y;
+const LIST_MARKER_PATTERN = /(?:[-*+]|\d{1,9}[.)])[ \t]/y;
+const TRAILING_SPACE_PATTERN = /[ \t\r]*/y;
+
+interface AssistantTextScan {
+  text: string;
+  lineStart: number;
+  newlineSearchFrom: number;
+  openFence: { marker: string; indent: number } | null;
+  boundary: number;
+  titleAwaitingContent: boolean;
+  partialIndentEnd: number;
+  partialContentSearchFrom: number;
+}
+
+function startScan(): AssistantTextScan {
+  return {
+    text: "",
+    lineStart: 0,
+    newlineSearchFrom: 0,
+    openFence: null,
+    boundary: -1,
+    titleAwaitingContent: false,
+    partialIndentEnd: 0,
+    partialContentSearchFrom: 0,
+  };
+}
+
+function trimLineEnd(text: string, start: number, end: number): string {
+  while (end > start) {
+    const char = text.charCodeAt(end - 1);
+    if (char !== 32 && char !== 9 && char !== 13) break;
+    end--;
+  }
+  return text.slice(start, end);
+}
+
+function stickyMatchEnd(pattern: RegExp, text: string, from: number): number {
+  pattern.lastIndex = from;
+  return pattern.exec(text) === null ? -1 : pattern.lastIndex;
+}
+
+// Matches LIST_ITEM_START_PATTERN against the trimmed partial line. Every
+// search resumes where the previous call on this line stopped.
+function partialLineStartsListItem(scan: AssistantTextScan, text: string): boolean {
+  scan.partialIndentEnd = stickyMatchEnd(INDENT_PATTERN, text, scan.partialIndentEnd);
+  const markerEnd = stickyMatchEnd(LIST_MARKER_PATTERN, text, scan.partialIndentEnd);
+  if (markerEnd === -1) return false;
+  // Trimming would drop the marker's trailing space unless content follows it.
+  scan.partialContentSearchFrom = stickyMatchEnd(
+    TRAILING_SPACE_PATTERN,
+    text,
+    Math.max(markerEnd, scan.partialContentSearchFrom),
+  );
+  return scan.partialContentSearchFrom < text.length;
+}
+
 /**
  * Only fully terminated lines count, so a trailing partial line never leaks; a
  * list item start is the one lookahead that may sit on the partial line, since
@@ -24,35 +83,37 @@
  *
  * A section title holds the boundary until a content line follows it, so a
  * title never lands alone and waits above a block that is still streaming.
+ *
+ * Text that extends the previously scanned text resumes from its last partial
+ * line, so a streamed reply is scanned once rather than once per update.
  */
-export function splitBufferedAssistantText(text: string): { ready: string; rest: string } {
-  let openFence: { marker: string; indent: number } | null = null;
+function advanceScan(previous: AssistantTextScan, text: string): AssistantTextScan {
+  const scan = text.startsWith(previous.text) ? { ...previous } : startScan();
+  scan.text = text;
-  let boundary = -1;
-  let lineStart = 0;
-  let titleAwaitingContent = false;
   for (;;) {
-    const newline = text.indexOf("\n", lineStart);
-    const line = text
-      .slice(lineStart, newline === -1 ? text.length : newline)
+    const lineStart = scan.lineStart;
+    const newline = text.indexOf("\n", scan.newlineSearchFrom);
+    if (newline === -1) {
-      .replace(/[ \t\r]+$/, "");
+      scan.newlineSearchFrom = text.length;
+      break;
+    }
+    const line = trimLineEnd(text, lineStart, newline);
+    const { openFence, titleAwaitingContent } = scan;
     if (
       openFence === null &&
       lineStart > 0 &&
       !titleAwaitingContent &&
       LIST_ITEM_START_PATTERN.test(line)
     ) {
+      scan.boundary = lineStart;
-      boundary = lineStart;
     }
-    if (newline === -1) {
-      break;
-    }
     const fenceMatch = MARKDOWN_FENCE_PATTERN.exec(line);
     if (fenceMatch) {
       const indent = fenceMatch[1]!.length;
       const marker = fenceMatch[2]!;
       if (openFence === null) {
+        scan.openFence = { marker, indent };
-        openFence = { marker, indent };
-        titleAwaitingContent = false;
+        scan.titleAwaitingContent = false;
       } else if (
         marker[0] === openFence.marker[0] &&
         marker.length >= openFence.marker.length &&
@@ -60,21 +121,43 @@
         line.length === indent + marker.length
       ) {
         // CommonMark: a closing fence carries no info string.
+        scan.openFence = null;
-        openFence = null;
+        scan.boundary = newline + 1;
-        boundary = newline + 1;
       }
     } else if (openFence === null && BLANK_LINE_PATTERN.test(line) && lineStart > 0) {
       if (!titleAwaitingContent) {
-        boundary = newline + 1;
+        scan.boundary = newline + 1;
       }
     } else if (openFence === null) {
       if (lineStart > 0 && !titleAwaitingContent && TOP_LEVEL_HEADING_PATTERN.test(line)) {
-        boundary = lineStart;
+        scan.boundary = lineStart;
       }
-      titleAwaitingContent = SECTION_TITLE_PATTERN.test(line);
+      scan.titleAwaitingContent = SECTION_TITLE_PATTERN.test(line);
     }
+    scan.lineStart = newline + 1;
+    scan.newlineSearchFrom = newline + 1;
+    scan.partialIndentEnd = newline + 1;
+    scan.partialContentSearchFrom = newline + 1;
+  }
+  return scan;
+}
+
+// The partial line's list item start is not kept in the scan, because the
+// line may still be replaced before it ends.
+function readyBoundary(scan: AssistantTextScan): number {
+  if (
+    scan.openFence === null &&
+    scan.lineStart > 0 &&
-    lineStart = newline + 1;
+    !scan.titleAwaitingContent &&
+    partialLineStartsListItem(scan, scan.text)
+  ) {
+    return scan.lineStart;
   }
+  return scan.boundary;
+}
+
+export function splitBufferedAssistantText(text: string): { ready: string; rest: string } {
+  const boundary = readyBoundary(advanceScan(startScan(), text));
   if (boundary === -1) {
     return { ready: "", rest: text };
   }
@@ -82,7 +165,10 @@
 }
 
 export function makeAssistantStreamingFilter(mode: ResponseStreamingMode) {
-  const delivered = new Map<string, { text: string; at: number }>();
+  const streams = new Map<
+    string,
+    { scan: AssistantTextScan; delivered: { text: string; at: number } | null }
+  >();
   return (
     event: ProviderAdapter.ProviderAdapterV2Event,
     now: number,
@@ -103,15 +189,21 @@
     if (!message) return event;
     const key = `${event.type}:${message.id}`;
     if (!message.streaming) {
+      streams.delete(key);
-      delivered.delete(key);
       return event;
     }
     if (mode === "turn") return null;
+    const stream = streams.get(key);
-    const previous = delivered.get(key);
+    const previous = stream?.delivered ?? null;
     if (previous && now - previous.at < 400) return null;
-    const { ready } = splitBufferedAssistantText(message.text);
-    if (!ready || ready === previous?.text) return null;
+    const scan = advanceScan(stream?.scan ?? startScan(), message.text);
+    const boundary = readyBoundary(scan);
+    const ready = boundary === -1 ? "" : message.text.slice(0, boundary);
+    if (!ready || ready === previous?.text) {
+      streams.set(key, { scan, delivered: previous });
-    delivered.set(key, { text: ready, at: now });
+      return null;
+    }
+    streams.set(key, { scan, delivered: { text: ready, at: now } });
     if (event.type === "message.updated")
       return { ...event, message: { ...event.message, text: ready } };
     if (
```