fix(bridge): stop replaying explicit terminal provider failures

apps/desktop/assets/codex-bridge/bridge.py

@@ -32,6 +32,7 @@ import time
 import urllib.error
 import urllib.request
 import gecis_context
+import terminal_stream
 
 HERE = os.path.dirname(os.path.abspath(__file__))
 SURUM = "17"         # AITURK IDE'nin bekledigi kopru surumu (codex-link-config.ts CODEX_BRIDGE_VERSION)
@@ -348,6 +349,36 @@ class H(http.server.BaseHTTPRequestHandler):
             self.end_headers()
             self.wfile.write(hata)
             return True
+        observer = None
+        first_chunks = []
+        failure_key = getattr(self, "_failure_key", None)
+        if (hedef == KOPRU and failure_key is not None and durum == 200
+                and "text/event-stream" in yanit.headers.get("Content-Type", "")
+                and yanit.headers.get("Content-Encoding", "identity") == "identity"):
+            observer = terminal_stream.Observer()
+            if not started:
+                # Inspect only until the first SSE frame/comment. Keep normal
+                # streaming and heartbeats; an error-only response can still
+                # return a truthful HTTP failure before headers are committed.
+                try:
+                    while True:
+                        first = yanit.read1(4096)
+                        if not first:
+                            break
+                        first_chunks.append(first)
+                        observer.feed(first)
+                        if observer.error and not observer.payload_seen:
+                            yanit.close()
+                            self._json(400, terminal_stream.error_body(observer.error))
+                            return True
+                        if observer.comment_seen or observer.payload_seen or sum(map(len, first_chunks)) >= observer.LIMIT:
+                            break
+                except (OSError, http.client.HTTPException):
+                    # A transport interruption without an explicit provider
+                    # failure keeps the client's existing recovery behavior.
+                    yanit.close()
+                    self.close_connection = True
+                    return True
         if not started:
             self.send_response(durum)
             for k, v in yanit.headers.items():
@@ -359,6 +390,10 @@ class H(http.server.BaseHTTPRequestHandler):
         event_tail = b""
         completed = False
         try:
+            for first in first_chunks:
+                toplam += len(first)
+                self.wfile.write(b"%x\r\n" % len(first) + first + b"\r\n")
+                self.wfile.flush()
             while True:
                 try:
                     parca = yanit.read1(4096)      # gelen kadarini hemen verir; bekletmez
@@ -370,6 +405,8 @@ class H(http.server.BaseHTTPRequestHandler):
                 if not parca:
                     break
                 toplam += len(parca)
+                if observer is not None:
+                    observer.feed(parca)
                 if getattr(self, "_native_return_scope", None):
                     event_tail += parca
                     while b"\n" in event_tail:
@@ -382,6 +419,9 @@ class H(http.server.BaseHTTPRequestHandler):
                                 pass
                 self.wfile.write(b"%x\r\n" % len(parca) + parca + b"\r\n")
                 self.wfile.flush()
+            if observer is not None and observer.error and not observer.completed:
+                self.server.failure_replays.remember(failure_key, observer.error)
+                log(f"{tag} terminal worker error; identical automatic replay will return HTTP 400")
             self.wfile.write(b"0\r\n\r\n")
             self.wfile.flush()
         except (ConnectionAbortedError, BrokenPipeError, ConnectionResetError):
@@ -453,6 +493,7 @@ class H(http.server.BaseHTTPRequestHandler):
 
     def _isle(self):
         self._native_return_scope = None
+        self._failure_key = None
         with kilit:
             sira[0] += 1
             n = sira[0]
@@ -479,6 +520,12 @@ class H(http.server.BaseHTTPRequestHandler):
                     req = {}
             except Exception:
                 req = {}
+            if external:
+                self._failure_key = terminal_stream.request_key(req or {"raw": govde.hex()}, self.headers)
+                previous_failure = self.server.failure_replays.take(self._failure_key)
+                if previous_failure:
+                    log(f"[{n} {model}] terminal worker error replay stopped before upstream request")
+                    return self._json(400, terminal_stream.error_body(previous_failure))
             thread = str(req.get("prompt_cache_key") or "")
             scope = hashlib.sha256((str(self.headers.get("chatgpt-account-id") or "") + "|" + thread).encode()).hexdigest()
             source = TRANSFERS.get(scope) if thread else None
@@ -541,6 +588,10 @@ class H(http.server.BaseHTTPRequestHandler):
 
 
 class Sunucu(http.server.ThreadingHTTPServer):
+    def __init__(self, *args, **kwargs):
+        self.failure_replays = terminal_stream.FailureReplays()
+        super().__init__(*args, **kwargs)
+
     daemon_threads = True
     # Shipped Python 3.11 defaults to only five pending connections. Model
     # discovery, voice and simultaneous chats can arrive in the same burst.

apps/desktop/assets/codex-bridge/terminal_stream.py

@@ -0,0 +1,119 @@
+"""Transport-only handling of a worker's explicit terminal failure.
+
+Never selects a model, evaluates content, or changes worker/Jev decisions.
+If HTTP headers have already gone out, remember the failure for exactly one
+identical retry. Codex then receives an HTTP error instead of re-running the
+whole exhausted provider chain. Ordinary network interruptions remain retries.
+"""
+import hashlib
+import json
+import threading
+import time
+from collections import OrderedDict
+
+
+def request_key(request, headers):
+    principal = "\0".join(headers.get(name, "") for name in
+                          ("Authorization", "ChatGPT-Account-Id", "session_id"))
+    body = json.dumps(request, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
+    return hashlib.sha256((principal + "\0" + body).encode()).digest()
+
+
+class FailureReplays:
+    def __init__(self, ttl=120, capacity=128, clock=time.monotonic):
+        self.ttl, self.capacity, self.clock = ttl, capacity, clock
+        self.items = OrderedDict()
+        self.lock = threading.Lock()
+
+    def remember(self, key, error):
+        with self.lock:
+            self.items[key] = (self.clock() + self.ttl, error)
+            self.items.move_to_end(key)
+            while len(self.items) > self.capacity:
+                self.items.popitem(last=False)
+
+    def take(self, key):
+        with self.lock:
+            entry = self.items.pop(key, None)
+        return entry[1] if entry and entry[0] > self.clock() else None
+
+
+def error_body(error):
+    # Preserve the actual provider failure; don't relabel it invalid_prompt,
+    # fabricate response.completed, or expose a rejected model response.
+    return {"error": {"type": "upstream_terminal_error", "code": error["code"],
+                      "message": "Model yanıtı hata ile sonlandı. " + error["message"]}}
+
+
+class Observer:
+    """Bounded SSE observer. Forwarded stream bytes are never rewritten."""
+    LIMIT = 65536
+
+    def __init__(self):
+        self.tail = b""
+        self.data = []
+        self.size = 0
+        self.discard = False
+        self.discard_line = False
+        self.comment_seen = False
+        self.payload_seen = False
+        self.completed = False
+        self.error = None
+
+    def feed(self, chunk):
+        # read1() supplies small chunks, but also bound arbitrary caller input.
+        for offset in range(0, len(chunk), 4096):
+            self.tail += chunk[offset:offset + 4096]
+            while b"\n" in self.tail:
+                line, self.tail = self.tail.split(b"\n", 1)
+                if self.discard_line:
+                    self.discard_line = False
+                    continue
+                self._line(line.rstrip(b"\r"))
+            if len(self.tail) > self.LIMIT:
+                self.tail = b""
+                self.discard = self.discard_line = True
+
+    def _line(self, line):
+        if line.startswith(b":"):
+            self.comment_seen = True
+        elif line.startswith(b"data:"):
+            self.size += len(line)
+            if self.size > self.LIMIT:
+                self.discard = True
+                self.data.clear()
+            elif not self.discard:
+                self.data.append(line[5:].lstrip())
+        elif not line:
+            if self.data and not self.discard:
+                self._event(b"\n".join(self.data))
+            elif self.discard:
+                self.payload_seen = True
+            self.data, self.size, self.discard = [], 0, False
+
+    def _event(self, data):
+        try:
+            event = json.loads(data)
+        except (ValueError, UnicodeError):
+            self.payload_seen = True
+            return
+        if not isinstance(event, dict):
+            self.payload_seen = True
+            return
+        kind = event.get("type")
+        response = event.get("response")
+        error = event.get("error") if kind == "error" else response.get("error") if isinstance(response, dict) else None
+        # Known worker terminal errors only. Preserve context/rate-limit and
+        # other native error classifications rather than guessing semantics.
+        is_failure = (kind == "error" and isinstance(error, dict)
+                      and error.get("type") == "upstream_error")
+        is_failure |= (kind == "response.failed" and isinstance(error, dict)
+                       and error.get("code") == "jev_no_fallback")
+        if is_failure:
+            self.error = {"code": str(error.get("code") or "upstream_error")[:100],
+                          "message": str(error.get("message") or "Sağlayıcı yanıtı tamamlanamadı.")[:2000]}
+        else:
+            self.payload_seen = True
+        if kind == "response.completed":
+            self.completed = True
+            self.error = None

apps/desktop/assets/codex-bridge/test_terminal_stream.py

@@ -0,0 +1,143 @@
+"""Real-socket regressions for terminal worker errors (no provider calls)."""
+import json
+import unittest
+
+import test_ayirici as fixtures
+import terminal_stream
+
+
+ERROR = b'event: error\ndata: {"type":"error","error":{"type":"upstream_error","message":"upstream 503"}}\n\n'
+COMPLETE = b'event: response.completed\ndata: {"type":"response.completed","response":{"id":"resp_ok","status":"completed","output":[]}}\n\n'
+
+
+class TerminalStream(unittest.TestCase):
+    setUp = fixtures.Ayirici.setUp
+    tearDown = fixtures.Ayirici.tearDown
+    iste = fixtures.Ayirici.iste
+
+    def request(self, token="a", text="check"):
+        body = json.dumps({"model": "ide/test", "input": [{"role": "user", "content": text}]})
+        return self.iste("POST", "/backend-api/codex/responses", body,
+                         {"Authorization": "Bearer " + token})
+
+    def worker(self, payload):
+        def respond(h, body):
+            if h.path.endswith("/prepare-transfer"):
+                return fixtures.duz(200, json.dumps({"request": json.loads(body)["request"]}).encode())(h, body)
+            return fixtures.duz(200, payload, "text/event-stream")(h, body)
+        self.kopru.cevap = respond
+
+    def inference_count(self):
+        return sum(x["yol"].endswith("/responses") for x in self.kopru.gelen)
+
+    def test_error_only_stream_is_an_honest_http_failure(self):
+        self.worker(ERROR)
+        status, body, _ = self.request()
+        self.assertEqual(status, 400)
+        self.assertIn("upstream 503", json.loads(body)["error"]["message"])
+        self.assertNotIn(b"response.completed", body)
+        self.assertEqual(self.inference_count(), 1)
+
+    def test_error_after_heartbeat_stops_automatic_replay_once(self):
+        def respond(h, body):
+            if h.path.endswith("/prepare-transfer"):
+                return fixtures.duz(200, json.dumps({"request": json.loads(body)["request"]}).encode())(h, body)
+            h.send_response(200)
+            h.send_header("Content-Type", "text/event-stream")
+            h.send_header("Transfer-Encoding", "chunked")
+            h.end_headers()
+            for chunk in (b": keepalive\n\n", ERROR):
+                h.wfile.write(b"%x\r\n" % len(chunk) + chunk + b"\r\n")
+                h.wfile.flush()
+                fixtures.time.sleep(.08)
+            h.wfile.write(b"0\r\n\r\n")
+        self.kopru.cevap = respond
+        status, body, _ = self.request()
+        self.assertEqual((status, body), (200, b": keepalive\n\n" + ERROR))
+        status, body, _ = self.request()
+        self.assertEqual(status, 400)
+        self.assertIn(b"upstream 503", body)
+        self.assertEqual(self.inference_count(), 1)
+        # A later explicit retry is allowed; the one-shot guard is consumed.
+        self.worker(COMPLETE)
+        self.assertEqual(self.request()[:2], (200, COMPLETE))
+        self.assertEqual(self.inference_count(), 2)
+
+    def test_native_error_stream_is_not_intercepted(self):
+        self.ozgun.cevap = fixtures.duz(200, ERROR, "text/event-stream")
+        body = json.dumps({"model": "gpt-6-astra", "input": []})
+        self.assertEqual(self.iste("POST", "/backend-api/codex/responses", body)[:2], (200, ERROR))
+
+    def test_success_stream_is_byte_identical(self):
+        self.worker(COMPLETE)
+        self.assertEqual(self.request()[:2], (200, COMPLETE))
+        self.assertEqual(self.request()[:2], (200, COMPLETE))
+        self.assertEqual(self.inference_count(), 2)
+
+    def test_eof_without_upstream_error_keeps_network_retry(self):
+        self.worker(b": keepalive\n\n")
+        self.assertEqual(self.request()[0], 200)
+        self.assertEqual(self.request()[0], 200)
+        self.assertEqual(self.inference_count(), 2)
+
+    def test_worker_exhausted_failure_is_terminal_without_changing_its_reason(self):
+        self.worker(b'event: response.failed\ndata: {"type":"response.failed","response":{"status":"failed","error":{"code":"jev_no_fallback","message":"No accepted response"}}}\n\n')
+        status, body, _ = self.request()
+        self.assertEqual(status, 400)
+        self.assertEqual(json.loads(body)["error"]["code"], "jev_no_fallback")
+
+    def test_existing_context_error_classification_is_preserved(self):
+        data = b'event: response.failed\ndata: {"type":"response.failed","response":{"error":{"code":"context_length_exceeded","message":"Context full"}}}\n\n'
+        self.worker(data)
+        self.assertEqual(self.request()[:2], (200, data))
+
+
+class ObserverTests(unittest.TestCase):
+    def test_fragmented_crlf_and_multiline_json(self):
+        data = b'event: error\r\ndata: {"type":"error",\r\ndata: "error":{"type":"upstream_error","message":"503"}}\r\n\r\n'
+        observer = terminal_stream.Observer()
+        for byte in data:
+            observer.feed(bytes([byte]))
+        self.assertEqual(observer.error['message'], '503')
+
+    def test_large_and_malformed_frames_are_bounded_and_do_not_crash(self):
+        observer = terminal_stream.Observer()
+        observer.feed(b'data: ' + b'x'*200000)
+        self.assertLessEqual(len(observer.tail), observer.LIMIT)
+        observer.feed(b'\n\ndata: {"response":[]}\n\n' + ERROR)
+        self.assertEqual(observer.error['message'], 'upstream 503')
+
+    def test_completed_stream_does_not_poison_next_request(self):
+        observer = terminal_stream.Observer()
+        observer.feed(ERROR + COMPLETE)
+        self.assertIsNone(observer.error)
+        self.assertTrue(observer.completed)
+
+    def test_one_shot_expiry_and_capacity(self):
+        now = [10]
+        cache = terminal_stream.FailureReplays(ttl=5, capacity=2, clock=lambda: now[0])
+        cache.remember(b'a', {'code': 'a'})
+        self.assertEqual(cache.take(b'a'), {'code': 'a'})
+        self.assertIsNone(cache.take(b'a'))
+        cache.remember(b'a', {})
+        now[0] = 16
+        self.assertIsNone(cache.take(b'a'))
+        for key in (b'a', b'b', b'c'):
+            cache.remember(key, {'code': 'error'})
+        self.assertIsNone(cache.take(b'a'))
+        self.assertEqual(len(cache.items), 2)
+
+    def test_request_identity_separates_accounts_sessions_models_and_inputs(self):
+        base = {'model': 'ide/a', 'input': []}
+        headers = {'Authorization': 'Bearer a', 'session_id': 'one'}
+        key = terminal_stream.request_key(base, headers)
+        self.assertEqual(key, terminal_stream.request_key(dict(reversed(list(base.items()))), headers))
+        for request, other in [({'model': 'ide/b', 'input': []}, headers),
+                               ({'model': 'ide/a', 'input': ['next']}, headers),
+                               (base, dict(headers, Authorization='Bearer b')),
+                               (base, dict(headers, session_id='two'))]:
+            self.assertNotEqual(key, terminal_stream.request_key(request, other))
+
+
+if __name__ == "__main__":
+    unittest.main()