From 5f23e9b61c2d17f7a083a73cf3a947c2690b12d7 Mon Sep 17 00:00:00 2001 From: Luxferre Date: Wed, 26 Aug 2026 08:16:43 +0300 Subject: [PATCH] 429 robustness --- architecture.md | 18 +- th3ist.go | 707 +++++++++++++++++++++++++++++++++++++++++------- 2 files changed, 621 insertions(+), 104 deletions(-) diff --git a/architecture.md b/architecture.md index 80d5951..75f5277 100644 --- a/architecture.md +++ b/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 ... 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 `...` tags and routing to `reasoning_content`. -2. `StreamToolCallFilter`: Stateful filter intercepting `...` tags, preventing raw XML leaks in `content` and emitting OpenAI `delta.tool_calls`. -3. Flushes `data: [DONE]` on stream termination. +1. `Runtime.addBinding`: Exposes `window.th3istStreamChunk` to the browser context, relaying `ReadableStream` chunks over CDP in real time without buffering. +2. `StreamLineBuffer`: Reassembles partial TCP packets into complete protocol lines across chunk boundaries. +3. `StreamThinkingFilter`: Stateful filter intercepting `...` tags and routing to `reasoning_content`. +4. `StreamToolCallFilter`: Stateful filter intercepting `...` 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: diff --git a/th3ist.go b/th3ist.go index e666011..b4fa8fd 100644 --- a/th3ist.go +++ b/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(); - const e = await fp.get(); - const comps = JSON.parse(JSON.stringify(e.components)); - const randSeed = Math.random(); + 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(); + if (e && e.components) { + 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") { - comps.audio.value = comps.audio.value + (randSeed * 0.01); + if (comps.audio) { + 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,90 +1780,307 @@ func (bb *BrowserBridge) ExecuteFetch(payload T3ChatPayload, deploymentID, clien } } - // If payload has no fresh hcaptcha token, generate one live in-browser - if payload.HcaptchaToken == "" { - tok, _ := bb.GetFreshHcaptchaToken() - if tok != "" { - payload.HcaptchaToken = tok - } - } + maxAttempts := 4 + var lastStatus int + var lastText string - payloadJSON, err := json.Marshal(payload) - if err != nil { - return 0, "", err - } - - fetchExpr := fmt.Sprintf(`(async () => { - try { - const resp = await fetch("https://t3.chat/api/chat", { - method: "POST", - headers: { - "Content-Type": "application/json", - "x-client-context": %q, - "x-deployment-id": %q - }, - body: JSON.stringify(%s) - }); - return { status: resp.status, text: await resp.text() }; - } catch (e) { - return { status: 500, text: e.message || String(e), error: true }; - } - })()`, clientContext, deploymentID, string(payloadJSON)) - - evalRes, err := sendCDPCommand(bb.conn, bb.reader, "Runtime.evaluate", map[string]interface{}{ - "expression": fetchExpr, - "awaitPromise": true, - "returnByValue": true, - }) - if err != nil { - return 0, "", fmt.Errorf("cdp evaluate error: %w", err) - } - - var status int - var text string - 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)) - text, _ = valObj["text"].(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 + } + + fetchExpr := fmt.Sprintf(`(async () => { + try { + const resp = await fetch("https://t3.chat/api/chat", { + method: "POST", + headers: { + "Content-Type": "application/json", + "x-client-context": %q, + "x-deployment-id": %q + }, + body: JSON.stringify(%s) + }); + return { status: resp.status, text: await resp.text() }; + } catch (e) { + return { status: 500, text: e.message || String(e), error: true }; + } + })()`, clientContext, deploymentID, string(payloadJSON)) + + evalRes, err := sendCDPCommand(bb.conn, bb.reader, "Runtime.evaluate", map[string]interface{}{ + "expression": fetchExpr, + "awaitPromise": true, + "returnByValue": true, + }) + if err != nil { + return 0, "", fmt.Errorf("cdp evaluate error: %w", err) + } + + var status int + var text string + 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 { + if stVal, ok := valObj["status"].(float64); ok { + status = int(stVal) + } + text, _ = valObj["text"].(string) + } + } + } + + lastStatus = status + lastText = text + + // 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) + } + + // 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 + } + + // 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 rate limited, auto-rotate hardware fingerprint and retry once - if status == 429 && strings.Contains(text, "ratelimit_hit") { - fmt.Println("Rate limit reached on current visitor identity; auto-rotating hardware fingerprint...") - if newID, rotErr := bb.RotateIdentity(); rotErr == nil && newID != "" { - fmt.Printf("Rotated hardware fingerprint to new identity: %s\n", newID) - if newTok, tokErr := bb.GetFreshHcaptchaToken(); tokErr == nil && newTok != "" { - payload.HcaptchaToken = newTok - if newPayloadJSON, pErr := json.Marshal(payload); pErr == nil { - retryExpr := fmt.Sprintf(`(async () => { - try { - const resp = await fetch("https://t3.chat/api/chat", { - method: "POST", - headers: { - "Content-Type": "application/json", - "x-client-context": %q, - "x-deployment-id": %q - }, - body: JSON.stringify(%s) - }); - return { status: resp.status, text: await resp.text() }; - } catch (e) { - return { status: 500, text: e.message || String(e), error: true }; - } - })()`, clientContext, deploymentID, string(newPayloadJSON)) + if lastStatus != 0 { + return lastStatus, lastText, nil + } - if retryRes, rErr := sendCDPCommand(bb.conn, bb.reader, "Runtime.evaluate", map[string]interface{}{ - "expression": retryExpr, - "awaitPromise": true, - "returnByValue": true, - }); rErr == nil { - if rMap, ok := retryRes["result"].(map[string]interface{}); ok { - if vMap, ok := rMap["result"].(map[string]interface{}); ok { - if vObj, ok := vMap["value"].(map[string]interface{}); ok { - return int(vObj["status"].(float64)), vObj["text"].(string), 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", + headers: { + "Content-Type": "application/json", + "x-client-context": %q, + "x-deployment-id": %q + }, + body: JSON.stringify(%s) + }); + + 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 }; + } + })()`, streamID, clientContext, deploymentID, string(payloadJSON)) + + cmdID := int(atomic.AddInt64(&cdpCmdCounter, 1)) + msg := map[string]interface{}{ + "id": cmdID, + "method": "Runtime.evaluate", + "params": map[string]interface{}{ + "expression": fetchExpr, + "awaitPromise": true, + "returnByValue": true, + }, + } + b, _ := json.Marshal(msg) + if err := sendWSFrame(bb.conn, b); err != 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 + } } } } @@ -1720,13 +2088,61 @@ func (bb *BrowserBridge) ExecuteFetch(payload T3ChatPayload, deploymentID, clien } } } + + lastStatus = streamStatus + lastText = streamErrText + + // If 200 OK, stream completed successfully in real time + if streamStatus == http.StatusOK { + return streamStatus, "", nil + } + + // 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 status != 0 { - return status, text, nil + if lastStatus != 0 { + return lastStatus, lastText, nil } - return 0, "", fmt.Errorf("invalid evaluate response: %v", evalRes) + 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) - } + g.handleNonStreamingResponse(w, strings.NewReader(responseBody), completionID, createdTime, modelName) return }