429 robustness

architecture.md

@@ -158,7 +158,11 @@ When the upstream model produces tool calls:
 
 #### Streaming pipeline architecture
 ```
-Stream Lines (0: "...", data: {...}, b: "...")
+Chromium ReadableStream (window.th3istStreamChunk)
+  │
+  ├─► CDP Runtime.bindingCalled Event
+  │
+  ├─► StreamLineBuffer (Reassembles complete stream lines across TCP chunks)
   │
   ├─► StreamThinkingFilter (Stateful sliding window)
   │     ├─► Emits reasoning content to delta.reasoning_content
@@ -167,11 +171,13 @@ Stream Lines (0: "...", data: {...}, b: "...")
   └─► StreamToolCallFilter (Stateful tool tag parser)
         ├─► Buffers & parses <tool_call>...</tool_call> blocks
         ├─► Emits structured OpenAI delta.tool_calls
-        └─► Emits clean assistant text to delta.content
+        └─► Emits clean assistant text to delta.content in real time
 ```
-1. `StreamThinkingFilter`: Stateful filter intercepting `<think>...</think>` tags and routing to `reasoning_content`.
+1. `Runtime.addBinding`: Exposes `window.th3istStreamChunk` to the browser context, relaying `ReadableStream` chunks over CDP in real time without buffering.
-2. `StreamToolCallFilter`: Stateful filter intercepting `<tool_call>...</tool_call>` tags, preventing raw XML leaks in `content` and emitting OpenAI `delta.tool_calls`.
+2. `StreamLineBuffer`: Reassembles partial TCP packets into complete protocol lines across chunk boundaries.
-3. Flushes `data: [DONE]` on stream termination.
+3. `StreamThinkingFilter`: Stateful filter intercepting `<think>...</think>` tags and routing to `reasoning_content`.
+4. `StreamToolCallFilter`: Stateful filter intercepting `<tool_call>...</tool_call>` tags, preventing raw XML leaks in `content` and emitting OpenAI `delta.tool_calls`.
+5. Flushes `data: [DONE]` on stream termination.
 
 ---
 
@@ -239,7 +245,7 @@ Because `th3ist` controls Chromium via CDP, `BrowserBridge.RotateIdentity()` mut
     "confidence": 1
   }
   ```
-- **Zero-downtime auto-recovery**: If `ExecuteFetch()` receives an `HTTP 429 ratelimit_hit`, it automatically invokes `RotateIdentity()`, mints a fresh hCaptcha token, and retries the completion request transparently.
+- **Zero-downtime auto-recovery**: If `ExecuteFetch()` receives an `HTTP 429` or rate limit error, it immediately invokes `RotateIdentity()`, mints a fresh hCaptcha token, resets session IDs, and retries the completion request transparently (up to 4 attempts) without returning 429 errors to the client right away.
 
 #### 3. Bypassing guest limits via BYOK (bring your own key)
 `th3ist` also supports passing custom provider API keys directly in standard OpenAI format:

th3ist.go

@@ -342,6 +342,19 @@ func EffectiveUserAgent(r *http.Request) string {
   return DefaultUserAgent
 }
 
+func isRateLimited(status int, body string) bool {
+  if status == http.StatusTooManyRequests {
+    return true
+  }
+  lower := strings.ToLower(body)
+  return strings.Contains(lower, "ratelimit") ||
+    strings.Contains(lower, "rate_limit") ||
+    strings.Contains(lower, "rate limit") ||
+    strings.Contains(lower, "too many requests") ||
+    strings.Contains(lower, "quota_exceeded") ||
+    strings.Contains(lower, "quota exceeded")
+}
+
 // ---------------------------------------------------------------------------
 // Tool and Message Processing
 // ---------------------------------------------------------------------------
@@ -897,6 +910,34 @@ func sendStreamChunk(w http.ResponseWriter, flusher http.Flusher, completionID s
   }
 }
 
+// ---------------------------------------------------------------------------
+// Stateful Line Buffer for Chunked Streaming
+// ---------------------------------------------------------------------------
+
+type StreamLineBuffer struct {
+  buf string
+}
+
+func (b *StreamLineBuffer) Feed(chunk string, onLine func(string)) {
+  b.buf += chunk
+  for {
+    idx := strings.Index(b.buf, "\n")
+    if idx == -1 {
+      break
+    }
+    line := b.buf[:idx]
+    b.buf = b.buf[idx+1:]
+    onLine(line)
+  }
+}
+
+func (b *StreamLineBuffer) Flush(onLine func(string)) {
+  if b.buf != "" {
+    onLine(b.buf)
+    b.buf = ""
+  }
+}
+
 // ---------------------------------------------------------------------------
 // Stateful Thinking Tag Filter for Streaming
 // ---------------------------------------------------------------------------
@@ -1414,6 +1455,14 @@ func (bb *BrowserBridge) Start() error {
   bb.conn = conn
   bb.reader = reader
 
+  // Enable Runtime domain for binding events
+  _, _ = sendCDPCommand(conn, reader, "Runtime.enable", nil)
+
+  // Add binding for real-time streaming chunks
+  _, _ = sendCDPCommand(conn, reader, "Runtime.addBinding", map[string]interface{}{
+    "name": "th3istStreamChunk",
+  })
+
   // Inject stealth scripts
   _, _ = sendCDPCommand(conn, reader, "Page.addScriptToEvaluateOnNewDocument", map[string]interface{}{
     "source": `
@@ -1562,19 +1611,112 @@ func (bb *BrowserBridge) RotateIdentity() (string, error) {
 
   rotateExpr := `(async () => {
     try {
-      const fp = await (await import("./assets/fp.esm-Bp3Vx1Qv.js")).default.load();
+      let fpModule = null;
+      try {
+        const scripts = Array.from(document.querySelectorAll('script[src], link[href]'));
+        for (const s of scripts) {
+          const u = s.src || s.href || '';
+          if (u.includes('fp.esm-') || u.includes('/fp.') || u.includes('fingerprint')) {
+            fpModule = (await import(u)).default;
+            break;
+          }
+        }
+      } catch (e) {}
+
+      if (!fpModule) {
+        try {
+          const entries = (typeof performance !== "undefined" && performance.getEntriesByType) ? performance.getEntriesByType('resource') : [];
+          for (const e of entries) {
+            if (e.name && (e.name.includes('fp.esm-') || e.name.includes('/fp.') || e.name.includes('fingerprint'))) {
+              fpModule = (await import(e.name)).default;
+              break;
+            }
+          }
+        } catch (e) {}
+      }
+
+      if (!fpModule) {
+        try {
+          fpModule = (await import("./assets/fp.esm-Bp3Vx1Qv.js")).default;
+        } catch (e) {}
+      }
+
+      let comps = {};
+      let version = "3.4.2";
+      if (fpModule && typeof fpModule.load === "function") {
+        try {
+          const fp = await fpModule.load();
           const e = await fp.get();
-      const comps = JSON.parse(JSON.stringify(e.components));
+          if (e && e.components) {
-      const randSeed = Math.random();
+            comps = JSON.parse(JSON.stringify(e.components));
+          }
+          if (e && e.version) {
+            version = e.version;
+          }
+        } catch (e) {}
+      }
+
+      const randHex = (len) => {
+        const chars = "0123456789abcdef";
+        let out = "";
+        for (let i = 0; i < len; i++) {
+          out += chars[Math.floor(Math.random() * chars.length)];
+        }
+        return out;
+      };
+
+      const randInt = (min, max) => Math.floor(Math.random() * (max - min + 1)) + min;
+
+      if (!comps.canvas || !comps.canvas.value) {
+        comps.canvas = { value: { geometry: randHex(32), text: randHex(32) }, duration: randInt(1, 4) };
+      }
+      if (!comps.audio) {
+        comps.audio = { value: Math.random() * 80 + 10, duration: randInt(1, 4) };
+      }
+      if (!comps.platform) {
+        comps.platform = { value: "Linux x86_64", duration: 0 };
+      }
+      if (!comps.vendor) {
+        comps.vendor = { value: "Google Inc.", duration: 0 };
+      }
+      if (!comps.timezone) {
+        comps.timezone = { value: "UTC", duration: 0 };
+      }
+      if (!comps.languages) {
+        comps.languages = { value: [["en-US", "en"]], duration: 0 };
+      }
 
       if (comps.canvas && comps.canvas.value) {
-        comps.canvas.value.geometry = (comps.canvas.value.geometry || "").slice(0, -3) + Math.floor(randSeed * 900 + 100);
+        comps.canvas.value.geometry = randHex(32);
+        comps.canvas.value.text = randHex(32);
       }
-      if (comps.audio && typeof comps.audio.value === "number") {
+      if (comps.audio) {
-        comps.audio.value = comps.audio.value + (randSeed * 0.01);
+        comps.audio.value = Math.random() * 100 + 0.1;
       }
-      comps.hardwareConcurrency = { value: [4, 6, 8, 12, 16][Math.floor(randSeed * 5)], duration: 0 };
+
-      comps.screenResolution = { value: [[1920, 1080], [2560, 1440], [1680, 1050], [1920, 1200]][Math.floor(randSeed * 4)], duration: 0 };
+      const cpuCores = [4, 6, 8, 12, 16, 24, 32];
+      comps.hardwareConcurrency = { value: cpuCores[Math.floor(Math.random() * cpuCores.length)], duration: 0 };
+
+      const resolutions = [
+        [1920, 1080], [2560, 1440], [1680, 1050], [1920, 1200],
+        [1440, 900], [3840, 2160], [1366, 768], [1536, 864]
+      ];
+      comps.screenResolution = { value: resolutions[Math.floor(Math.random() * resolutions.length)], duration: 0 };
+
+      const devMems = [4, 8, 16, 32];
+      comps.deviceMemory = { value: devMems[Math.floor(Math.random() * devMems.length)], duration: 0 };
+
+      const freshVisitorId = "visitor_" + randHex(16) + "_" + Date.now();
+
+      try {
+        if (typeof localStorage !== "undefined") {
+          localStorage.removeItem("t3-visitor-id");
+          localStorage.removeItem("t3-anon-visitor");
+        }
+        if (typeof sessionStorage !== "undefined") {
+          sessionStorage.clear();
+        }
+      } catch (e) {}
 
       const t = await fetch("/api/identity", {
         method: "POST",
@@ -1582,15 +1724,21 @@ func (bb *BrowserBridge) RotateIdentity() (string, error) {
         headers: { "content-type": "application/json" },
         body: JSON.stringify({
           fingerprint: {
-            visitorId: "rand_" + Date.now(),
+            visitorId: freshVisitorId,
             confidence: { score: 0.99 },
             components: comps,
-            version: e.version
+            version: version
           }
         })
       });
+
+      if (!t.ok) {
+        const errText = await t.text();
+        return { success: false, error: "HTTP " + t.status + ": " + errText };
+      }
+
       const data = await t.json();
-      return { success: true, visitorId: data.visitorId, requiresSignIn: data.requiresSignIn };
+      return { success: true, visitorId: data.visitorId || freshVisitorId, requiresSignIn: data.requiresSignIn };
     } catch (err) {
       return { success: false, error: err.message || String(err) };
     }
@@ -1612,6 +1760,9 @@ func (bb *BrowserBridge) RotateIdentity() (string, error) {
           visitorID, _ := valObj["visitorId"].(string)
           return visitorID, nil
         }
+        if errMsg, _ := valObj["error"].(string); errMsg != "" {
+          return "", fmt.Errorf("identity rotation endpoint error: %s", errMsg)
+        }
       }
     }
   }
@@ -1629,6 +1780,11 @@ func (bb *BrowserBridge) ExecuteFetch(payload T3ChatPayload, deploymentID, clien
     }
   }
 
+  maxAttempts := 4
+  var lastStatus int
+  var lastText string
+
+  for attempt := 1; attempt <= maxAttempts; attempt++ {
     // If payload has no fresh hcaptcha token, generate one live in-browser
     if payload.HcaptchaToken == "" {
       tok, _ := bb.GetFreshHcaptchaToken()
@@ -1673,21 +1829,112 @@ func (bb *BrowserBridge) ExecuteFetch(payload T3ChatPayload, deploymentID, clien
     if resMap, ok := evalRes["result"].(map[string]interface{}); ok {
       if valMap, ok := resMap["result"].(map[string]interface{}); ok {
         if valObj, ok := valMap["value"].(map[string]interface{}); ok {
-        status = int(valObj["status"].(float64))
+          if stVal, ok := valObj["status"].(float64); ok {
+            status = int(stVal)
+          }
           text, _ = valObj["text"].(string)
         }
       }
     }
 
-  // If rate limited, auto-rotate hardware fingerprint and retry once
+    lastStatus = status
-  if status == 429 && strings.Contains(text, "ratelimit_hit") {
+    lastText = text
-    fmt.Println("Rate limit reached on current visitor identity; auto-rotating hardware fingerprint...")
+
-    if newID, rotErr := bb.RotateIdentity(); rotErr == nil && newID != "" {
+    // Return immediately on success
+    if status == http.StatusOK {
+      return status, text, nil
+    }
+
+    // If rate limited (429 or quota exceeded), auto-rotate hardware identity and retry immediately
+    if isRateLimited(status, text) {
+      if attempt < maxAttempts {
+        fmt.Printf("Rate limit / 429 encountered (attempt %d/%d); immediately rotating identity and retrying...\n", attempt, maxAttempts)
+        newID, rotErr := bb.RotateIdentity()
+        if rotErr != nil {
+          fmt.Fprintf(os.Stderr, "warning: identity rotation failed on attempt %d: %v\n", attempt, rotErr)
+        } else {
           fmt.Printf("Rotated hardware fingerprint to new identity: %s\n", newID)
-      if newTok, tokErr := bb.GetFreshHcaptchaToken(); tokErr == nil && newTok != "" {
+        }
+
+        // Mint a fresh hCaptcha token for the new visitor identity
+        newTok, tokErr := bb.GetFreshHcaptchaToken()
+        if tokErr != nil {
+          fmt.Fprintf(os.Stderr, "warning: token minting failed after identity rotation: %v\n", tokErr)
+          payload.HcaptchaToken = ""
+        } else {
           payload.HcaptchaToken = newTok
-        if newPayloadJSON, pErr := json.Marshal(payload); pErr == nil {
+        }
-          retryExpr := fmt.Sprintf(`(async () => {
+
+        // Regenerate session & response IDs for a clean turn
+        payload.ConvexSessionID = GenerateUUID()
+        payload.ResponseMessageID = GenerateUUID()
+
+        time.Sleep(100 * time.Millisecond)
+        continue
+      }
+    } else if (status == http.StatusForbidden || strings.Contains(strings.ToLower(text), "captcha")) && attempt < maxAttempts {
+      // Captcha verification failed; mint fresh token and retry immediately
+      fmt.Printf("Captcha validation failure encountered (attempt %d/%d); refreshing hCaptcha token and retrying...\n", attempt, maxAttempts)
+      newTok, tokErr := bb.GetFreshHcaptchaToken()
+      if tokErr == nil && newTok != "" {
+        payload.HcaptchaToken = newTok
+        payload.ResponseMessageID = GenerateUUID()
+        time.Sleep(100 * time.Millisecond)
+        continue
+      }
+    }
+
+    // If non-retryable client error (e.g. 400 bad request), return immediately
+    if status != 0 {
+      return status, text, nil
+    }
+  }
+
+  if lastStatus != 0 {
+    return lastStatus, lastText, nil
+  }
+
+  return 0, "", fmt.Errorf("evaluate failed without valid status: %s", lastText)
+}
+
+func (bb *BrowserBridge) ExecuteStreamFetch(payload T3ChatPayload, deploymentID, clientContext string, onStatus func(int), onChunk func(string)) (int, string, error) {
+  bb.mu.Lock()
+  defer bb.mu.Unlock()
+
+  if bb.conn == nil {
+    if err := bb.Start(); err != nil {
+      return 0, "", fmt.Errorf("failed to start browser bridge: %w", err)
+    }
+  }
+
+  // Ensure runtime and binding are enabled
+  _, _ = sendCDPCommand(bb.conn, bb.reader, "Runtime.enable", nil)
+  _, _ = sendCDPCommand(bb.conn, bb.reader, "Runtime.addBinding", map[string]interface{}{
+    "name": "th3istStreamChunk",
+  })
+
+  maxAttempts := 4
+  var lastStatus int
+  var lastText string
+
+  for attempt := 1; attempt <= maxAttempts; attempt++ {
+    // If payload has no fresh hcaptcha token, generate one live in-browser
+    if payload.HcaptchaToken == "" {
+      tok, _ := bb.GetFreshHcaptchaToken()
+      if tok != "" {
+        payload.HcaptchaToken = tok
+      }
+    }
+
+    payloadJSON, err := json.Marshal(payload)
+    if err != nil {
+      return 0, "", err
+    }
+
+    streamID := GenerateUUID()
+
+    fetchExpr := fmt.Sprintf(`(async () => {
+      const streamId = %q;
       try {
         const resp = await fetch("https://t3.chat/api/chat", {
           method: "POST",
@@ -1698,21 +1945,141 @@ func (bb *BrowserBridge) ExecuteFetch(payload T3ChatPayload, deploymentID, clien
           },
           body: JSON.stringify(%s)
         });
-              return { status: resp.status, text: await resp.text() };
+
+        if (!resp.ok) {
+          const errText = await resp.text();
+          if (window.th3istStreamChunk) {
+            window.th3istStreamChunk(JSON.stringify({
+              streamId: streamId,
+              type: "error",
+              status: resp.status,
+              text: errText
+            }));
+          }
+          return { status: resp.status, text: errText, error: true };
+        }
+
+        if (window.th3istStreamChunk) {
+          window.th3istStreamChunk(JSON.stringify({
+            streamId: streamId,
+            type: "status",
+            status: resp.status
+          }));
+        }
+
+        const reader = resp.body.getReader();
+        const decoder = new TextDecoder();
+        while (true) {
+          const { done, value } = await reader.read();
+          if (done) break;
+          const chunkStr = decoder.decode(value, { stream: true });
+          if (window.th3istStreamChunk) {
+            window.th3istStreamChunk(JSON.stringify({
+              streamId: streamId,
+              type: "chunk",
+              data: chunkStr
+            }));
+          }
+        }
+
+        if (window.th3istStreamChunk) {
+          window.th3istStreamChunk(JSON.stringify({
+            streamId: streamId,
+            type: "done"
+          }));
+        }
+        return { status: 200, done: true };
       } catch (e) {
+        if (window.th3istStreamChunk) {
+          window.th3istStreamChunk(JSON.stringify({
+            streamId: streamId,
+            type: "error",
+            status: 500,
+            text: e.message || String(e)
+          }));
+        }
         return { status: 500, text: e.message || String(e), error: true };
       }
-          })()`, clientContext, deploymentID, string(newPayloadJSON))
+    })()`, streamID, clientContext, deploymentID, string(payloadJSON))
 
-          if retryRes, rErr := sendCDPCommand(bb.conn, bb.reader, "Runtime.evaluate", map[string]interface{}{
+    cmdID := int(atomic.AddInt64(&cdpCmdCounter, 1))
-            "expression":    retryExpr,
+    msg := map[string]interface{}{
+      "id":     cmdID,
+      "method": "Runtime.evaluate",
+      "params": map[string]interface{}{
+        "expression":    fetchExpr,
         "awaitPromise":  true,
         "returnByValue": true,
-          }); rErr == nil {
+      },
-            if rMap, ok := retryRes["result"].(map[string]interface{}); ok {
+    }
-              if vMap, ok := rMap["result"].(map[string]interface{}); ok {
+    b, _ := json.Marshal(msg)
-                if vObj, ok := vMap["value"].(map[string]interface{}); ok {
+    if err := sendWSFrame(bb.conn, b); err != nil {
-                  return int(vObj["status"].(float64)), vObj["text"].(string), nil
+      return 0, "", err
+    }
+
+    var streamStatus int
+    var streamErrText string
+    evalDone := false
+    streamDone := false
+
+    for !evalDone || !streamDone {
+      frame, err := readWSFrame(bb.conn, bb.reader)
+      if err != nil {
+        return 0, "", err
+      }
+
+      var res map[string]interface{}
+      if err := json.Unmarshal(frame, &res); err != nil {
+        continue
+      }
+
+      if idVal, ok := res["id"].(float64); ok && int(idVal) == cmdID {
+        evalDone = true
+        if resMap, ok := res["result"].(map[string]interface{}); ok {
+          if valMap, ok := resMap["result"].(map[string]interface{}); ok {
+            if valObj, ok := valMap["value"].(map[string]interface{}); ok {
+              if stVal, ok := valObj["status"].(float64); ok {
+                if streamStatus == 0 {
+                  streamStatus = int(stVal)
+                }
+              }
+              if txt, ok := valObj["text"].(string); ok && streamErrText == "" {
+                streamErrText = txt
+              }
+            }
+          }
+        }
+      }
+
+      if method, ok := res["method"].(string); ok && method == "Runtime.bindingCalled" {
+        if params, ok := res["params"].(map[string]interface{}); ok {
+          if name, _ := params["name"].(string); name == "th3istStreamChunk" {
+            if payloadStr, ok := params["payload"].(string); ok {
+              var ev struct {
+                StreamID string `json:"streamId"`
+                Type     string `json:"type"`
+                Status   int    `json:"status"`
+                Data     string `json:"data"`
+                Text     string `json:"text"`
+              }
+              if json.Unmarshal([]byte(payloadStr), &ev) == nil {
+                if ev.StreamID == streamID {
+                  switch ev.Type {
+                  case "status":
+                    streamStatus = ev.Status
+                    if onStatus != nil {
+                      onStatus(ev.Status)
+                    }
+                  case "chunk":
+                    if ev.Data != "" && onChunk != nil {
+                      onChunk(ev.Data)
+                    }
+                  case "done":
+                    streamDone = true
+                  case "error":
+                    streamStatus = ev.Status
+                    streamErrText = ev.Text
+                    streamDone = true
                   }
                 }
               }
@@ -1722,11 +2089,60 @@ func (bb *BrowserBridge) ExecuteFetch(payload T3ChatPayload, deploymentID, clien
       }
     }
 
-  if status != 0 {
+    lastStatus = streamStatus
-    return status, text, nil
+    lastText = streamErrText
+
+    // If 200 OK, stream completed successfully in real time
+    if streamStatus == http.StatusOK {
+      return streamStatus, "", nil
     }
 
-  return 0, "", fmt.Errorf("invalid evaluate response: %v", evalRes)
+    // Rate limit / 429 auto-rotation retry
+    if isRateLimited(streamStatus, streamErrText) {
+      if attempt < maxAttempts {
+        fmt.Printf("Rate limit / 429 encountered in streaming (attempt %d/%d); immediately rotating identity and retrying...\n", attempt, maxAttempts)
+        newID, rotErr := bb.RotateIdentity()
+        if rotErr != nil {
+          fmt.Fprintf(os.Stderr, "warning: identity rotation failed on attempt %d: %v\n", attempt, rotErr)
+        } else {
+          fmt.Printf("Rotated hardware fingerprint to new identity: %s\n", newID)
+        }
+
+        newTok, tokErr := bb.GetFreshHcaptchaToken()
+        if tokErr != nil {
+          fmt.Fprintf(os.Stderr, "warning: token minting failed after identity rotation: %v\n", tokErr)
+          payload.HcaptchaToken = ""
+        } else {
+          payload.HcaptchaToken = newTok
+        }
+
+        payload.ConvexSessionID = GenerateUUID()
+        payload.ResponseMessageID = GenerateUUID()
+
+        time.Sleep(100 * time.Millisecond)
+        continue
+      }
+    } else if (streamStatus == http.StatusForbidden || strings.Contains(strings.ToLower(streamErrText), "captcha")) && attempt < maxAttempts {
+      fmt.Printf("Captcha validation failure encountered in streaming (attempt %d/%d); refreshing token and retrying...\n", attempt, maxAttempts)
+      newTok, tokErr := bb.GetFreshHcaptchaToken()
+      if tokErr == nil && newTok != "" {
+        payload.HcaptchaToken = newTok
+        payload.ResponseMessageID = GenerateUUID()
+        time.Sleep(100 * time.Millisecond)
+        continue
+      }
+    }
+
+    if streamStatus != 0 {
+      return streamStatus, streamErrText, nil
+    }
+  }
+
+  if lastStatus != 0 {
+    return lastStatus, lastText, nil
+  }
+
+  return 0, "", fmt.Errorf("stream failed without status: %s", lastText)
 }
 
 // ---------------------------------------------------------------------------
@@ -2043,6 +2459,105 @@ func (g *T3Gateway) HandleChatCompletions(w http.ResponseWriter, r *http.Request
 
   // 1. Browser Bridge Execution (TLS & Vercel bypass + auto-generated fresh hCaptcha token)
   if g.bridge != nil {
+    if req.Stream {
+      var streamer *Streamer
+      initStreamer := func() {
+        if streamer == nil {
+          flusher, _ := w.(http.Flusher)
+          streamer = NewStreamer(w, flusher, completionID, createdTime, modelName)
+          streamer.Role()
+        }
+      }
+
+      thinkingFilter := NewStreamThinkingFilter()
+      toolFilter := NewStreamToolCallFilter()
+      var fullAccumulatedText strings.Builder
+      lastToolCallArgs := map[string]string{}
+      emittedToolCallIDs := map[string]bool{}
+      lineBuf := &StreamLineBuffer{}
+
+      status, errBody, err := g.bridge.ExecuteStreamFetch(
+        t3Payload,
+        effDeploymentID,
+        effClientContext,
+        func(st int) {
+          if st == http.StatusOK {
+            initStreamer()
+          }
+        },
+        func(rawChunk string) {
+          initStreamer()
+          lineBuf.Feed(rawChunk, func(line string) {
+            g.processStreamLine(line, streamer, thinkingFilter, toolFilter, &fullAccumulatedText, lastToolCallArgs, emittedToolCallIDs)
+          })
+        },
+      )
+
+      if err != nil {
+        if streamer == nil {
+          http.Error(w, fmt.Sprintf(`{"error":"Browser bridge streaming error: %v"}`, err), http.StatusBadGateway)
+        }
+        return
+      }
+
+      if status != http.StatusOK {
+        if streamer == nil {
+          w.Header().Set("Content-Type", "application/json")
+          w.WriteHeader(status)
+          w.Write([]byte(errBody))
+        }
+        return
+      }
+
+      initStreamer()
+
+      lineBuf.Flush(func(line string) {
+        g.processStreamLine(line, streamer, thinkingFilter, toolFilter, &fullAccumulatedText, lastToolCallArgs, emittedToolCallIDs)
+      })
+
+      thinkingFilter.Flush(
+        func(text string) {
+          toolFilter.Feed(text,
+            func(t string) {
+              fullAccumulatedText.WriteString(t)
+              streamer.Content(t)
+            },
+            func(tc ToolCall) {
+              emittedToolCallIDs[tc.ID] = true
+              streamer.ToolCallDelta(tc)
+            },
+          )
+        },
+        func(reasoning string) {
+          streamer.Reasoning(reasoning)
+        },
+      )
+
+      toolFilter.Flush(
+        func(text string) {
+          fullAccumulatedText.WriteString(text)
+          streamer.Content(text)
+        },
+        func(tc ToolCall) {
+          emittedToolCallIDs[tc.ID] = true
+          streamer.ToolCallDelta(tc)
+        },
+      )
+
+      finishReason := "stop"
+      if len(emittedToolCallIDs) > 0 || toolFilter.emittedCall {
+        finishReason = "tool_calls"
+      } else {
+        if tc, _, found := DetectToolCalls(fullAccumulatedText.String()); found && len(tc) > 0 {
+          finishReason = "tool_calls"
+        }
+      }
+
+      streamer.Finish(finishReason)
+      streamer.Done()
+      return
+    }
+
     status, responseBody, err := g.bridge.ExecuteFetch(t3Payload, effDeploymentID, effClientContext)
     if err != nil {
       http.Error(w, fmt.Sprintf(`{"error":"Browser bridge error: %v"}`, err), http.StatusBadGateway)
@@ -2055,11 +2570,7 @@ func (g *T3Gateway) HandleChatCompletions(w http.ResponseWriter, r *http.Request
       return
     }
 
-    if req.Stream {
-      g.handleStreamingResponse(w, strings.NewReader(responseBody), completionID, createdTime, modelName)
-    } else {
     g.handleNonStreamingResponse(w, strings.NewReader(responseBody), completionID, createdTime, modelName)
-    }
     return
   }