215 lines
11 KiB
Python
215 lines
11 KiB
Python
"""The splitter (bridge.py): Codex's own models go to chatgpt.com byte for byte, added models to the bridge process.
|
||
|
||
Real sockets, no network: a fake chatgpt.com and a fake bridge process record what reaches them."""
|
||
import gzip
|
||
import hashlib
|
||
import http.client
|
||
import http.server
|
||
import json
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
import unittest
|
||
|
||
import bridge
|
||
|
||
|
||
class Kayit(http.server.BaseHTTPRequestHandler):
|
||
"""Records every request; answers from the server's `cevap` function."""
|
||
protocol_version = "HTTP/1.1"
|
||
|
||
def log_message(self, *a):
|
||
pass
|
||
|
||
def _her(self):
|
||
n = int(self.headers.get("Content-Length") or 0)
|
||
govde = self.rfile.read(n) if n else b""
|
||
self.server.gelen.append({"yontem": self.command, "yol": self.path, "baslik": dict(self.headers), "govde": govde})
|
||
self.server.cevap(self, govde)
|
||
|
||
do_GET = do_POST = _her
|
||
|
||
|
||
def sunucu(cevap):
|
||
s = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Kayit)
|
||
s.daemon_threads = True
|
||
s.gelen, s.cevap = [], cevap
|
||
threading.Thread(target=s.serve_forever, daemon=True).start()
|
||
return s
|
||
|
||
|
||
def duz(durum, govde, tur="application/json"):
|
||
def f(h, _):
|
||
h.send_response(durum)
|
||
h.send_header("Content-Type", tur)
|
||
h.send_header("Content-Length", str(len(govde)))
|
||
h.end_headers()
|
||
h.wfile.write(govde)
|
||
return f
|
||
|
||
|
||
def zstd(veri):
|
||
try:
|
||
from compression import zstd as z
|
||
return z.compress(veri)
|
||
except ImportError:
|
||
import zstandard
|
||
return zstandard.ZstdCompressor().compress(veri)
|
||
|
||
|
||
class Ayirici(unittest.TestCase):
|
||
def setUp(self):
|
||
self.ozgun = sunucu(duz(200, b'event: response.completed\ndata: {"ok":1}\n\n', "text/event-stream"))
|
||
self.kopru = sunucu(duz(200, b'event: response.completed\ndata: {"kopru":1}\n\n', "text/event-stream"))
|
||
self.on = bridge.Sunucu(("127.0.0.1", 0), bridge.H)
|
||
threading.Thread(target=self.on.serve_forever, daemon=True).start()
|
||
self.eski = (bridge.CHATGPT, bridge.KOPRU, bridge.LOGDIR, dict(bridge.CFG))
|
||
self.gecici = tempfile.TemporaryDirectory()
|
||
bridge.CHATGPT = f"http://127.0.0.1:{self.ozgun.server_address[1]}"
|
||
bridge.KOPRU = f"http://127.0.0.1:{self.kopru.server_address[1]}"
|
||
bridge.LOGDIR = self.gecici.name
|
||
bridge.CFG.clear()
|
||
|
||
def tearDown(self):
|
||
for s in (self.on, self.ozgun, self.kopru):
|
||
s.shutdown()
|
||
s.server_close()
|
||
bridge.CHATGPT, bridge.KOPRU, bridge.LOGDIR, cfg = self.eski
|
||
bridge.CFG.clear()
|
||
bridge.CFG.update(cfg)
|
||
self.gecici.cleanup()
|
||
|
||
def iste(self, yontem, yol, govde=None, baslik=None):
|
||
b = http.client.HTTPConnection("127.0.0.1", self.on.server_address[1], timeout=10)
|
||
b.request(yontem, yol, body=govde, headers=baslik or {})
|
||
y = b.getresponse()
|
||
veri = y.read()
|
||
b.close()
|
||
return y.status, veri, dict(y.getheaders())
|
||
|
||
def test_codex_own_model_reaches_chatgpt_byte_for_byte(self):
|
||
"""Compressed or not, with a command, a lone dot or a compaction trigger in it: nothing is changed."""
|
||
istekler = [{"model": "gpt-6-astra", "input": [{"role": "user", "content": "modeli opus 5 5 yap"}]},
|
||
{"model": "gpt-6-luna", "input": [{"role": "user", "content": "."}]},
|
||
{"model": "gpt-6-astra", "prompt_cache_key": "s", "input": [{"type": "compaction_trigger"}]},
|
||
{"model": "gpt-6-astra", "input": [{"type": "compaction", "encrypted_content": "ASTRABRIDGE-OZET-1:eA=="},
|
||
{"type": "reasoning", "id": "rs_chatcmpl-x"}]}]
|
||
for veri in istekler:
|
||
ham = json.dumps(veri, ensure_ascii=False).encode("utf-8")
|
||
for kodlama, govde in (("zstd", zstd(ham)), ("gzip", gzip.compress(ham)), ("", ham)):
|
||
with self.subTest(model=veri["model"], kodlama=kodlama):
|
||
baslik = {"Authorization": "Bearer sahte", "originator": "Codex Desktop", "chatgpt-account-id": "h1"}
|
||
if kodlama:
|
||
baslik["Content-Encoding"] = kodlama
|
||
durum, cevap, _ = self.iste("POST", "/backend-api/codex/responses", govde, baslik)
|
||
giden = self.ozgun.gelen[-1]
|
||
self.assertEqual(hashlib.sha256(giden["govde"]).hexdigest(), hashlib.sha256(govde).hexdigest())
|
||
basliklar = {k.lower(): v for k, v in giden["baslik"].items()}
|
||
self.assertEqual(basliklar.get("content-encoding", ""), kodlama)
|
||
self.assertEqual((basliklar["authorization"], basliklar["originator"], basliklar["chatgpt-account-id"]),
|
||
("Bearer sahte", "Codex Desktop", "h1"))
|
||
self.assertEqual(giden["yol"], "/backend-api/codex/responses")
|
||
self.assertEqual((durum, cevap), (200, b'event: response.completed\ndata: {"ok":1}\n\n'))
|
||
self.assertEqual(self.kopru.gelen, []) # the bridge process never saw any of them
|
||
|
||
def test_added_models_go_to_the_bridge_process_and_only_there(self):
|
||
for model in ("ide/Opus-5", "9r/cc/claude-opus-5-5"):
|
||
govde = zstd(json.dumps({"model": model, "input": []}).encode())
|
||
durum, cevap, _ = self.iste("POST", "/backend-api/codex/responses", govde,
|
||
{"Content-Encoding": "zstd", "originator": "Codex Desktop"})
|
||
self.assertEqual((durum, cevap), (200, b'event: response.completed\ndata: {"kopru":1}\n\n'))
|
||
self.assertEqual(self.kopru.gelen[-1]["govde"], govde)
|
||
basliklar = {k.lower(): v for k, v in self.kopru.gelen[-1]["baslik"].items()}
|
||
self.assertEqual((basliklar["originator"], basliklar["content-encoding"]), ("Codex Desktop", "zstd"))
|
||
bridge.CFG["model_map"] = {"ozel-model": ["rota", "hedef"]}
|
||
self.iste("POST", "/backend-api/codex/responses", json.dumps({"model": "ozel-model"}).encode())
|
||
self.assertEqual(len(self.kopru.gelen), 3)
|
||
self.assertEqual(self.ozgun.gelen, [])
|
||
|
||
def test_a_body_that_cannot_be_read_takes_codex_own_path_untouched(self):
|
||
for govde, baslik in ((b"\x00\x01bozuk", {"Content-Encoding": "zstd"}), (b"json degil", {}), (b"[1,2]", {})):
|
||
self.iste("POST", "/backend-api/codex/responses", govde, baslik)
|
||
self.assertEqual(self.ozgun.gelen[-1]["govde"], govde)
|
||
self.assertEqual(self.kopru.gelen, [])
|
||
|
||
def test_codex_own_model_keeps_working_while_the_bridge_process_is_down(self):
|
||
self.kopru.shutdown()
|
||
self.kopru.server_close()
|
||
govde = json.dumps({"model": "gpt-6-astra", "input": []}).encode()
|
||
self.assertEqual(self.iste("POST", "/backend-api/codex/responses", govde)[0], 200)
|
||
# The model list falls back to chatgpt.com's own list, unchanged.
|
||
self.ozgun.cevap = duz(200, b'{"models":[{"slug":"gpt-6-astra"}]}')
|
||
durum, liste, _ = self.iste("GET", "/backend-api/codex/models?client_version=1", None, {"If-None-Match": "x"})
|
||
self.assertEqual((durum, liste), (200, b'{"models":[{"slug":"gpt-6-astra"}]}'))
|
||
# An added model says so instead of hanging or going to chatgpt.com under another name.
|
||
durum, cevap, _ = self.iste("POST", "/backend-api/codex/responses", json.dumps({"model": "ide/Opus-5"}).encode())
|
||
self.assertEqual(durum, 503)
|
||
self.assertIn("köprüsü şu an çalışmıyor", json.loads(cevap)["error"]["message"])
|
||
self.assertEqual([g["yol"].split("?")[0] for g in self.ozgun.gelen],
|
||
["/backend-api/codex/responses", "/backend-api/codex/models"])
|
||
# Status still answers in the shape AITURK IDE expects.
|
||
durum, cevap, _ = self.iste("GET", "/aiturk/codex-baglama/status")
|
||
d = json.loads(cevap)
|
||
self.assertEqual((durum, d["surum"], d["kopru"], d["ayirici"]), (200, bridge.SURUM, "kapali", True))
|
||
|
||
def test_model_list_comes_from_the_bridge_and_falls_back_when_it_fails(self):
|
||
self.kopru.cevap = duz(200, b'{"models":[{"slug":"gpt-6-astra"},{"slug":"ide/Opus-5"}]}')
|
||
self.assertIn(b"ide/Opus-5", self.iste("GET", "/backend-api/codex/models")[1])
|
||
self.assertEqual(self.ozgun.gelen, [])
|
||
self.kopru.cevap = duz(500, b'{"error":"x"}')
|
||
self.ozgun.cevap = duz(200, b'{"models":[{"slug":"gpt-6-astra"}]}')
|
||
self.assertEqual(self.iste("GET", "/backend-api/codex/models")[1], b'{"models":[{"slug":"gpt-6-astra"}]}')
|
||
|
||
def test_everything_else_and_errors_pass_through_as_they_are(self):
|
||
self.ozgun.cevap = duz(200, b'{"sonuc":1}')
|
||
self.assertEqual(self.iste("POST", "/backend-api/codex/alpha/search", b'{"q":"x"}')[1], b'{"sonuc":1}')
|
||
self.assertEqual(self.ozgun.gelen[-1]["govde"], b'{"q":"x"}')
|
||
self.ozgun.cevap = duz(400, b'{"error":{"message":"gecersiz"}}')
|
||
durum, cevap, _ = self.iste("POST", "/backend-api/codex/responses", json.dumps({"model": "gpt-6-astra"}).encode())
|
||
self.assertEqual((durum, cevap), (400, b'{"error":{"message":"gecersiz"}}'))
|
||
self.assertEqual(self.kopru.gelen, [])
|
||
|
||
def test_a_long_answer_streams_piece_by_piece_without_waiting_for_the_end(self):
|
||
"""This morning's failure was an idle stream: Codex got no bytes for five minutes. Bytes must flow as they come."""
|
||
def yavas(h, _):
|
||
h.send_response(200)
|
||
h.send_header("Content-Type", "text/event-stream")
|
||
h.send_header("Transfer-Encoding", "chunked")
|
||
h.end_headers()
|
||
for parca in (b"data: bir\n\n", b"data: iki\n\n"):
|
||
h.wfile.write(b"%x\r\n" % len(parca) + parca + b"\r\n")
|
||
h.wfile.flush()
|
||
time.sleep(1.0)
|
||
h.wfile.write(b"0\r\n\r\n")
|
||
self.ozgun.cevap = yavas
|
||
b = http.client.HTTPConnection("127.0.0.1", self.on.server_address[1], timeout=10)
|
||
bas = time.time()
|
||
b.request("POST", "/backend-api/codex/responses", body=json.dumps({"model": "gpt-6-astra"}).encode())
|
||
y = b.getresponse()
|
||
ilk = y.read1(64)
|
||
ilk_sure = time.time() - bas
|
||
kalan = y.read()
|
||
b.close()
|
||
self.assertEqual(ilk, b"data: bir\n\n")
|
||
self.assertLess(ilk_sure, 0.8) # the first piece arrived before the answer ended
|
||
self.assertEqual(kalan, b"data: iki\n\n")
|
||
|
||
def test_status_is_the_bridge_status_plus_the_splitter_fields(self):
|
||
self.kopru.cevap = duz(200, b'{"ok": true, "surum": "8", "kod": "abc", "ide_durum": "tamam", "ide_model": 147}')
|
||
d = json.loads(self.iste("GET", "/aiturk/codex-baglama/status")[1])
|
||
self.assertEqual((d["surum"], d["ide_model"], d["kopru"], d["ayirici"]), ("8", 147, "calisiyor", True))
|
||
|
||
def test_a_persistent_connection_is_refused_the_way_it_is_today(self):
|
||
durum, _, _ = self.iste("GET", "/backend-api/codex/responses", None, {"Upgrade": "websocket", "Connection": "Upgrade"})
|
||
self.assertEqual(durum, 426)
|
||
self.assertEqual(self.ozgun.gelen + self.kopru.gelen, [])
|
||
|
||
def test_which_side_a_model_is_on(self):
|
||
for slug, ek in (("ide/Opus-5", True), ("9r/ag/gemini", True), ("gpt-6-astra", False), ("gpt-6.1-sol", False),
|
||
("", False), (None, False), ("idea", False)):
|
||
self.assertEqual(bridge.ek_model_mu(slug), ek, slug)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|