yilsemandClaude Opus 5.5
adf73eeedd
Model switch lost the conversation: on 10 Oct a 1 MB thread was compacted by the newly selected small model into 2201 characters. The compaction summary for a bridge model is now written by a subscription model (ozet_kopru.ozet_modeli, default gpt-6.1-sol); if that request is refused before it starts, the target model is used as before. A summary far shorter than the conversation warrants is not accepted: Codex gets response.failed and keeps the history. The line under each added model no longer repeats its name. Teams show their members, fallback chains their order, single models their provider and the traits readable from the name; the text fits one phone row (30 characters). Also normalizes tool call ids in the streamed response so Codex persists ids the subscription backend accepts (fc_ -> ctc_), and bumps the bridge protocol to 6. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
171 lines
9.5 KiB
Python
171 lines
9.5 KiB
Python
"""ozet_kopru testleri: 9Router icin sikistirma bagdastiricisi ve saglayicilar arasi ozet aktarimi."""
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
|
|
import ozet_kopru as o
|
|
|
|
|
|
def msg(rol, metin):
|
|
return {"type": "message", "role": rol, "content": [{"type": "input_text", "text": metin}]}
|
|
|
|
|
|
def sse(*olaylar):
|
|
return b"".join(b"event: " + e["type"].encode() + b"\ndata: " + json.dumps(e, ensure_ascii=False).encode() + b"\n\n" for e in olaylar)
|
|
|
|
|
|
def coz(veri):
|
|
return [json.loads(s[5:]) for blok in veri.split(b"\n\n") for s in blok.split(b"\n") if s.startswith(b"data:")]
|
|
|
|
|
|
OPENAI_OZETI = {"type": "compaction", "id": "cmp_0123", "encrypted_content": "gAAAAABsahte", "internal_chat_message_metadata_passthrough": {"turn_id": "t"}}
|
|
|
|
|
|
class Istek(unittest.TestCase):
|
|
def test_tetik_algilanir_ve_ozet_istegine_cevrilir(self):
|
|
req = {"model": "m", "tools": [{"type": "function", "name": "x"}], "input": [msg("user", "KARAR-1: 14 Kasim"), {"type": "compaction_trigger"}]}
|
|
self.assertTrue(o.tetik_mi(req))
|
|
self.assertFalse(o.tetik_mi({"input": [msg("user", "normal")]}))
|
|
self.assertFalse(o.tetik_mi({"input": []}))
|
|
o.ozet_istegi_hazirla(req)
|
|
self.assertNotIn("compaction_trigger", json.dumps(req))
|
|
self.assertEqual(req["input"][0]["content"][0]["text"], "KARAR-1: 14 Kasim") # gecmis aynen kalir
|
|
self.assertEqual(req["input"][-1]["role"], "user")
|
|
istem = req["input"][-1]["content"][0]["text"]
|
|
for baslik in ("KARARLAR VE GEREKCELERI", "ACIK ISLER", "ONEMLI SONUCLAR", "KULLANICININ ISTEGI"):
|
|
self.assertIn(baslik, istem)
|
|
self.assertEqual(req["tool_choice"], "none")
|
|
|
|
def test_kopru_ozeti_her_hedefte_duz_metne_acilir(self):
|
|
oge = o.ozet_ogesi("KARAR-1: teslim 14 Kasim\nACIK-IS-1: fatura onayi")
|
|
self.assertTrue(oge["id"].startswith("cmp_astrabridge_"))
|
|
self.assertNotIn("KARAR-1", json.dumps(oge)) # ogede duz metin gorunmez (base64)
|
|
for yabanci in (False, True):
|
|
req = {"input": [msg("user", "once"), dict(oge), msg("user", "soru")]}
|
|
b = o.ozetleri_ac(req, yabanci_hedef=yabanci)
|
|
self.assertEqual(b["acilan"], 1)
|
|
metin = req["input"][1]["content"][0]["text"]
|
|
self.assertEqual(req["input"][1]["role"], "user")
|
|
self.assertIn("KARAR-1: teslim 14 Kasim", metin)
|
|
self.assertTrue(metin.startswith(o.OZET_BASLIK))
|
|
self.assertNotIn('"compaction"', json.dumps(req))
|
|
|
|
def test_openai_ozeti_openai_hedefinde_dokunulmaz(self):
|
|
req = {"input": [dict(OPENAI_OZETI), msg("user", "soru")]}
|
|
once = json.dumps(req, sort_keys=True)
|
|
b = o.ozetleri_ac(req, yabanci_hedef=False)
|
|
self.assertEqual(json.dumps(req, sort_keys=True), once)
|
|
self.assertEqual(b, {"acilan": 0, "cozulen": 0, "okunamayan": 0})
|
|
|
|
def test_openai_ozeti_yabanci_hedefte_duz_metne_cevrilir_ya_da_not_dusulur(self):
|
|
req = {"input": [dict(OPENAI_OZETI), msg("user", "soru")]}
|
|
b = o.ozetleri_ac(req, yabanci_hedef=True, coz=lambda oge: "KARAR-2: saglayici degismeyecek")
|
|
self.assertEqual(b["cozulen"], 1)
|
|
self.assertIn("KARAR-2", req["input"][0]["content"][0]["text"])
|
|
for bozuk in (None, lambda oge: None, lambda oge: 1 / 0):
|
|
req = {"input": [dict(OPENAI_OZETI), msg("user", "soru")]}
|
|
b = o.ozetleri_ac(req, yabanci_hedef=True, coz=bozuk)
|
|
self.assertEqual(b["okunamayan"], 1)
|
|
self.assertEqual(req["input"][0]["content"][0]["text"], o.OKUNAMADI_NOTU) # sessiz kayip yok
|
|
self.assertNotIn("gAAAAAB", json.dumps(req))
|
|
|
|
def test_coz_istegi_yalniz_gerekli_alanlari_tasir(self):
|
|
g = o.coz_istegi(OPENAI_OZETI, "gpt-6-luna")
|
|
self.assertEqual(g["input"][0], {"type": "compaction", "id": "cmp_0123", "encrypted_content": "gAAAAABsahte"})
|
|
self.assertEqual(g["model"], "gpt-6-luna")
|
|
self.assertFalse(g["store"])
|
|
|
|
|
|
class Akis(unittest.TestCase):
|
|
def _mesaj(self, metin, kimlik="msg_1"):
|
|
return {"id": kimlik, "type": "message", "role": "assistant", "content": [{"type": "output_text", "text": metin}]}
|
|
|
|
def test_metin_tek_compaction_ogesine_sarilir(self):
|
|
ozet = "1. KULLANICININ ISTEGI: x\n2. KARARLAR: KARAR-1 14 Kasim"
|
|
ham = sse({"type": "response.created", "response": {"id": "r1"}},
|
|
{"type": "response.output_item.added", "item": {"id": "msg_1", "type": "message"}},
|
|
{"type": "response.output_text.delta", "item_id": "msg_1", "delta": ozet[:10]},
|
|
{"type": "response.output_text.delta", "item_id": "msg_1", "delta": ozet[10:]},
|
|
{"type": "response.output_item.done", "item": self._mesaj(ozet)},
|
|
{"type": "response.completed", "response": {"id": "r1", "output": [self._mesaj(ozet)], "usage": {"input_tokens": 5}}})
|
|
a = o.OzetAkisi(tohum="7")
|
|
out = b"".join(a.feed(ham[i:i + 53]) for i in range(0, len(ham), 53)) + a.flush()
|
|
olaylar = coz(out)
|
|
self.assertEqual([e["type"] for e in olaylar], ["response.created", "response.output_item.done", "response.completed"])
|
|
oge = olaylar[1]["item"]
|
|
self.assertEqual(oge["type"], "compaction")
|
|
self.assertEqual(o.ozet_metni(oge), ozet)
|
|
self.assertEqual(olaylar[2]["response"]["output"], [oge]) # Codex tam bir compaction ogesi gorur
|
|
self.assertEqual(olaylar[2]["response"]["usage"], {"input_tokens": 5})
|
|
self.assertEqual(a.metin, ozet)
|
|
self.assertNotIn("KARAR-1", out.decode()) # ozet metni akista duz gorunmez
|
|
|
|
def test_yalniz_delta_gelirse_de_toplanir(self):
|
|
ham = sse({"type": "response.created", "response": {}},
|
|
{"type": "response.output_text.delta", "item_id": "a", "delta": "ozet "},
|
|
{"type": "response.output_text.delta", "item_id": "a", "delta": "metni"},
|
|
{"type": "response.completed", "response": {}})
|
|
a = o.OzetAkisi()
|
|
self.assertEqual(o.ozet_metni(coz(a.feed(ham))[1]["item"]), "ozet metni")
|
|
|
|
def test_anormal_kisa_ozet_gecmisin_yerine_konmaz(self):
|
|
# 10 Ekim 2026: 1.018.151 baytlik sohbet 2201 karaktere indi; bu ozet kabul edilmemeli.
|
|
self.assertEqual(o.en_az_karakter(1018151), 4000)
|
|
self.assertEqual(o.en_az_karakter(100000), 400)
|
|
self.assertEqual(o.en_az_karakter(0), 0)
|
|
self.assertEqual(o.en_az_karakter(1018151, {"en_az_tavan": 1500}), 1500)
|
|
ham = sse({"type": "response.created", "response": {"id": "r1"}},
|
|
{"type": "response.output_item.done", "item": self._mesaj("k" * 2201)},
|
|
{"type": "response.completed", "response": {"id": "r1", "status": "completed"}})
|
|
a = o.OzetAkisi(tohum="7", en_az=o.en_az_karakter(1018151))
|
|
olaylar = coz(a.feed(ham) + a.flush())
|
|
self.assertEqual([e["type"] for e in olaylar], ["response.created", "response.failed"])
|
|
self.assertNotIn("compaction", json.dumps(olaylar))
|
|
self.assertEqual(olaylar[1]["response"]["error"]["code"], "astrabridge_ozet_kisa")
|
|
self.assertTrue(a.kisa)
|
|
self.assertIn("REDDEDILDI", a.sonuc_notu())
|
|
# Yeterli uzunlukta ozet ayni esikle kabul edilir; kucuk sohbetin kisa ozeti de reddedilmez.
|
|
for metin, esik in (("u" * 4000, o.en_az_karakter(1018151)), ("kisa ama dogru ozet", o.en_az_karakter(3000))):
|
|
b = o.OzetAkisi(tohum="8", en_az=esik)
|
|
ham2 = sse({"type": "response.output_item.done", "item": self._mesaj(metin)},
|
|
{"type": "response.completed", "response": {"id": "r2"}})
|
|
cikti = coz(b.feed(ham2))
|
|
self.assertEqual(o.ozet_metni(cikti[0]["item"]), metin)
|
|
self.assertFalse(b.kisa)
|
|
|
|
def test_metin_yoksa_compaction_uretilmez_hata_gorunur(self):
|
|
ham = sse({"type": "response.created", "response": {}},
|
|
{"type": "response.output_item.done", "item": {"type": "function_call", "name": "x"}},
|
|
{"type": "response.completed", "response": {"output": []}})
|
|
a = o.OzetAkisi()
|
|
olaylar = coz(a.feed(ham))
|
|
self.assertNotIn("compaction", json.dumps(olaylar))
|
|
self.assertFalse(a.metin)
|
|
hata = sse({"type": "response.created", "response": {}}, {"type": "response.failed", "response": {"error": {"message": "x"}}})
|
|
self.assertIn("response.failed", o.OzetAkisi().feed(hata).decode())
|
|
|
|
def test_onbellek(self):
|
|
klasor = os.path.join(tempfile.mkdtemp(), "c")
|
|
self.assertIsNone(o.onbellekten(klasor, OPENAI_OZETI))
|
|
o.onbellege(klasor, OPENAI_OZETI, "duz metin ozet")
|
|
self.assertEqual(o.onbellekten(klasor, OPENAI_OZETI), "duz metin ozet")
|
|
|
|
def test_sse_metni(self):
|
|
ham = sse({"type": "response.created", "response": {}}, {"type": "response.output_item.done", "item": self._mesaj("cozulmus ozet")},
|
|
{"type": "response.completed", "response": {}})
|
|
self.assertEqual(o.sse_metni(ham), "cozulmus ozet")
|
|
|
|
|
|
class Fren(unittest.TestCase):
|
|
def test_ayni_sohbette_ust_uste_ozet_durdurulur(self):
|
|
f = o.DonguFreni(sinir=3, pencere_sn=600)
|
|
self.assertEqual([f.izin("sohbet-a", t) for t in (0, 10, 20, 30, 40)], [True, True, True, False, False])
|
|
self.assertTrue(f.izin("sohbet-b", 40)) # baska sohbet etkilenmez
|
|
self.assertTrue(f.izin("sohbet-a", 700)) # pencere gecince yeniden izin
|
|
self.assertIn("Dongu durduruldu", o.DONGU_MESAJI % 3)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|