perf(bridge-json): scan stream deltas instead of reparsing the buffer

src/proxy/bridge-json.ts

@@ -56,7 +56,15 @@ export type BridgeStreamDecision =
 export class BridgeJsonStreamDetector {
   private state: "undecided" | "candidate" | "passthrough" = "undecided";
   private buffer = "";
+  /** True until a fenced candidate's info line is complete and accepted. */
+  private fenceInfoPending = false;
   private readonly tracker = new MixedDeltaTracker();
+  // Incremental bracket/string scan of the candidate. A parseable envelope must
+  // end balanced and outside a string, so the full-buffer parse waits for that
+  // instead of rerunning on every streamed delta.
+  private depth = 0;
+  private inString = false;
+  private escaped = false;
 
   constructor(
     private readonly allowedToolNames: Set<string>,
@@ -86,6 +94,8 @@ export class BridgeJsonStreamDetector {
       // `[` alone is common prose (markdown links); only `[{` starts an envelope array.
       if (/^(\{|```|\[\s*\{)/.test(meaningful)) {
         this.state = "candidate";
+        this.fenceInfoPending = meaningful.startsWith("```");
+        this.scan(this.buffer);
       } else {
         const withheld = this.buffer;
         this.buffer = "";
@@ -94,10 +104,12 @@ export class BridgeJsonStreamDetector {
           ? { action: "passthrough", text: withheld }
           : { action: "passthrough" };
       }
+    } else {
+      this.scan(delta);
     }
 
-    const trimmed = this.buffer.trimStart();
-    if (trimmed.startsWith("```")) {
+    if (this.fenceInfoPending) {
+      const trimmed = this.buffer.trimStart();
       const infoLineEnd = trimmed.indexOf("\n", 3);
       if (infoLineEnd < 0) {
         return { action: "buffer" };
@@ -106,10 +118,13 @@ export class BridgeJsonStreamDetector {
       if (info && info.toLowerCase() !== "json") {
         return this.releaseBuffer();
       }
+      this.fenceInfoPending = false;
+    }
+
+    if (this.depth !== 0 || this.inString) {
+      return { action: "buffer" };
     }
 
-    // ponytail: bridge responses are small; reparse the accumulated candidate.
-    // If envelopes become large, replace this O(n²) path with an incremental parser.
     const toolCalls = extractBridgeToolCallsFromText(
       this.buffer,
       this.allowedToolNames,
@@ -140,7 +155,11 @@ export class BridgeJsonStreamDetector {
   reset(): void {
     this.state = "undecided";
     this.buffer = "";
+    this.fenceInfoPending = false;
     this.tracker.reset();
+    this.depth = 0;
+    this.inString = false;
+    this.escaped = false;
   }
 
   private releaseBuffer(): BridgeStreamDecision {
@@ -149,6 +168,28 @@ export class BridgeJsonStreamDetector {
     this.state = "passthrough";
     return { action: "passthrough", text };
   }
+
+  // Each delta is scanned once as it arrives, keeping detection linear in envelope size.
+  private scan(text: string): void {
+    for (let i = 0; i < text.length; i++) {
+      const ch = text[i];
+      if (this.inString) {
+        if (this.escaped) {
+          this.escaped = false;
+        } else if (ch === "\\") {
+          this.escaped = true;
+        } else if (ch === '"') {
+          this.inString = false;
+        }
+      } else if (ch === '"') {
+        this.inString = true;
+      } else if (ch === "{" || ch === "[") {
+        this.depth++;
+      } else if (ch === "}" || ch === "]") {
+        this.depth--;
+      }
+    }
+  }
 }
 
 export function isBridgeJsonEnabled(env: Record<string, string | undefined> = process.env): boolean {

tests/unit/proxy/bridge-json.test.ts

@@ -561,5 +561,47 @@ describe("proxy/bridge-json", () => {
         text: write,
       });
     });
+
+    it("parses a streamed envelope once, not on every delta", () => {
+      const detector = new BridgeJsonStreamDetector(new Set(["write"]));
+      const content = 'function f() {\n  return { a: "}]\\\\" };\n}\n'.repeat(200);
+      const envelope = JSON.stringify({ name: "write", arguments: { path: "src/f.ts", content } });
+      const originalParse = JSON.parse;
+      let parseCalls = 0;
+      JSON.parse = ((...args: Parameters<typeof JSON.parse>) => {
+        parseCalls++;
+        return originalParse(...args);
+      }) as typeof JSON.parse;
+
+      let decision;
+      try {
+        for (let i = 0; i < envelope.length; i += 7) {
+          decision = detector.push(delta(envelope.slice(i, i + 7)));
+          if (i + 7 < envelope.length) {
+            expect(decision).toEqual({ action: "buffer" });
+          }
+        }
+      } finally {
+        JSON.parse = originalParse;
+      }
+
+      expect(decision?.action).toBe("tool_call");
+      if (decision?.action === "tool_call") {
+        expect(JSON.parse(decision.toolCalls[0].function.arguments).content).toBe(content);
+      }
+      expect(parseCalls).toBe(1);
+    });
+
+    it("extracts a streamed fenced envelope after the closing fence", () => {
+      const detector = new BridgeJsonStreamDetector(new Set(["task"]));
+      const fenced = `\`\`\`json\n${TASK_JSON}\n\`\`\``;
+
+      for (let i = 0; i < fenced.length - 3; i += 5) {
+        expect(detector.push(delta(fenced.slice(i, Math.min(i + 5, fenced.length - 3))))).toEqual({
+          action: "buffer",
+        });
+      }
+      expect(detector.push(delta("```")).action).toBe("tool_call");
+    });
   });
 });