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
}