From 4924781e2dde34e9c5a6be7f0bb8424dca203450 Mon Sep 17 00:00:00 2001 From: yilsem Date: Tue, 13 Oct 2026 17:15:57 +0300 Subject: [PATCH] fix(bridge): stop replaying explicit terminal provider failures --- apps/desktop/assets/codex-bridge/bridge.py | 51 +++++++ .../assets/codex-bridge/terminal_stream.py | 119 +++++++++++++++ .../codex-bridge/test_terminal_stream.py | 143 ++++++++++++++++++ 3 files changed, 313 insertions(+) create mode 100644 apps/desktop/assets/codex-bridge/terminal_stream.py create mode 100644 apps/desktop/assets/codex-bridge/test_terminal_stream.py diff --git a/apps/desktop/assets/codex-bridge/bridge.py b/apps/desktop/assets/codex-bridge/bridge.py index 82ca46b..e1c6006 100644 --- a/apps/desktop/assets/codex-bridge/bridge.py +++ b/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. diff --git a/apps/desktop/assets/codex-bridge/terminal_stream.py b/apps/desktop/assets/codex-bridge/terminal_stream.py new file mode 100644 index 0000000..e5e9b38 --- /dev/null +++ b/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 diff --git a/apps/desktop/assets/codex-bridge/test_terminal_stream.py b/apps/desktop/assets/codex-bridge/test_terminal_stream.py new file mode 100644 index 0000000..d240d20 --- /dev/null +++ b/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()