yilsemandClaude Opus 5.5
be0b33e89b
Katman-1..3 answers are checked by Jev each step; a blocked step is redone by the next layer combo. Katman-4 is delivered unasked. The technical-obstacle, question and third-party exemptions and the settings overrides are gone. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
215 lines
8.7 KiB
Python
215 lines
8.7 KiB
Python
"""Layer check (11 Oct 2026): Jev passes or blocks each step of Katman-1..3; a blocked step is redone by the next
|
||
layer. Katman-4 is never asked. No exceptions."""
|
||
import ast
|
||
import io
|
||
import json
|
||
import os
|
||
import tempfile
|
||
import time
|
||
import unittest
|
||
from pathlib import Path
|
||
from unittest.mock import Mock
|
||
|
||
import jev_denetim
|
||
import yarim
|
||
|
||
SOURCE = Path(__file__).with_name("kopru.py").read_text(encoding="utf-8")
|
||
tree = ast.parse(SOURCE)
|
||
handler = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == "H")
|
||
|
||
|
||
def method(name):
|
||
return next(n for n in handler.body if isinstance(n, ast.FunctionDef) and n.name == name)
|
||
|
||
|
||
def user(text):
|
||
return {"type": "message", "role": "user", "content": [{"type": "input_text", "text": text}]}
|
||
|
||
|
||
def sse(*deltas):
|
||
out = b""
|
||
for d in deltas:
|
||
out += b"event: response.output_text.delta\ndata: " + json.dumps(
|
||
{"type": "response.output_text.delta", "delta": d}).encode() + b"\n\n"
|
||
return out + b'event: response.completed\ndata: {"type": "response.completed"}\n\n'
|
||
|
||
|
||
class Yanit:
|
||
status = 200
|
||
|
||
def __init__(self, body):
|
||
self.parts = [body[:len(body) // 2], body[len(body) // 2:]]
|
||
self.headers = {"Content-Type": "text/event-stream"}
|
||
|
||
def read1(self, n):
|
||
return self.parts.pop(0) if self.parts else b""
|
||
|
||
|
||
class Karar(unittest.TestCase):
|
||
def test_refusal_or_own_conditions_block(self):
|
||
self.assertTrue(jev_denetim.karar({"ret": 0.84, "sart": 0.1}))
|
||
self.assertTrue(jev_denetim.karar({"ret": 0.41, "sart": 0.95}))
|
||
self.assertTrue(jev_denetim.karar({"ret": 0.5, "sart": 0.0}))
|
||
|
||
def test_normal_answer_passes(self):
|
||
self.assertFalse(jev_denetim.karar({"ret": 0.05, "sart": 0.3}))
|
||
|
||
def test_no_exception_questions_exist(self):
|
||
self.assertEqual(set(jev_denetim.SORU), {"ret", "sart"})
|
||
|
||
|
||
class Katman(unittest.TestCase):
|
||
def test_chain(self):
|
||
self.assertEqual(jev_denetim.sonrakiler("Katman-1"), ["Katman-2", "Katman-3", "Katman-4"])
|
||
self.assertEqual(jev_denetim.sonrakiler("Katman-2"), ["Katman-3", "Katman-4"])
|
||
self.assertEqual(jev_denetim.sonrakiler("Katman-3"), ["Katman-4"])
|
||
|
||
def test_last_layer_and_other_models_are_not_asked(self):
|
||
self.assertEqual(jev_denetim.sonrakiler("Katman-4"), [])
|
||
self.assertEqual(jev_denetim.sonrakiler("Uncensored-Uygulama"), [])
|
||
self.assertEqual(jev_denetim.sonrakiler(None), [])
|
||
|
||
|
||
class Istek(unittest.TestCase):
|
||
def test_last_user_request_is_taken_and_wrappers_are_skipped(self):
|
||
voice = user("<realtime_delegation>\n <input>siteyi yap</input>\n</realtime_delegation>")
|
||
self.assertEqual(jev_denetim.son_istek({"input": [user("eski"), voice]}), "siteyi yap")
|
||
hook = user("<hook_context>pano</hook_context>")
|
||
call = {"type": "function_call", "call_id": "c", "name": "shell", "arguments": "{}"}
|
||
self.assertEqual(jev_denetim.son_istek({"input": [user("README'yi düzelt"), call, hook]}), "README'yi düzelt")
|
||
self.assertEqual(jev_denetim.son_istek({"input": [call]}), "")
|
||
|
||
|
||
class Sor(unittest.TestCase):
|
||
def setUp(self):
|
||
jev_denetim._DEVRE.update(hata=0, kadar=0.0)
|
||
|
||
def test_request_shape_and_scores(self):
|
||
seen = {}
|
||
|
||
def gonder(govde, key, sure):
|
||
seen.update(govde=govde, key=key)
|
||
return {"answers": {k: {"type": "noul", "noul": 0.25} for k in jev_denetim.SORU}}
|
||
self.assertEqual(jev_denetim.sor("istek", "c" * 9000, "anahtar", gonder=gonder), {"ret": 0.25, "sart": 0.25})
|
||
self.assertEqual(seen["key"], "anahtar")
|
||
self.assertEqual(seen["govde"]["state"]["user_request"], "istek")
|
||
self.assertLess(len(seen["govde"]["state"]["assistant_reply"]), 6100)
|
||
|
||
def test_failures_return_none_and_open_the_breaker(self):
|
||
calls = []
|
||
|
||
def bozuk(govde, key, sure):
|
||
calls.append(1)
|
||
raise OSError("ulasilamadi")
|
||
for _ in range(jev_denetim.DEVRE_HATA):
|
||
self.assertIsNone(jev_denetim.sor("i", "c", "k", gonder=bozuk, simdi=1000.0))
|
||
self.assertIsNone(jev_denetim.sor("i", "c", "k", gonder=bozuk, simdi=1001.0))
|
||
self.assertEqual(len(calls), jev_denetim.DEVRE_HATA)
|
||
|
||
def test_key_is_read_from_env_or_file(self):
|
||
self.assertEqual(jev_denetim.anahtar({"TYPESAFE_API_KEY": " a "}), "a")
|
||
with tempfile.TemporaryDirectory() as d:
|
||
yol = os.path.join(d, "typesafe.env")
|
||
Path(yol).write_text('TYPESAFE_API_KEY="b"\n', encoding="utf-8")
|
||
self.assertEqual(jev_denetim.anahtar({}, [yol]), "b")
|
||
self.assertEqual(jev_denetim.anahtar({}, [os.path.join(d, "yok.env")]), "")
|
||
|
||
|
||
class Akis(unittest.TestCase):
|
||
"""_denetimli_akis through a fake handler: what reaches the client."""
|
||
|
||
def setUp(self):
|
||
self.logs, self.asked, self.opened = [], [], []
|
||
self.puanlar = [] # one verdict per asked answer; empty -> pass
|
||
sahte = Mock()
|
||
sahte.anahtar = lambda: "k"
|
||
sahte.sor = self.sor
|
||
sahte.karar = jev_denetim.karar
|
||
env = {"jev_denetim": sahte, "yarim": yarim, "time": time, "http": __import__("http.client"),
|
||
"log": self.logs.append, "RESP_HOP": set(), "json": json}
|
||
adlar = ("_akis_basligi", "_denetimli_akis", "_stream")
|
||
exec(compile(ast.Module(body=[method(a) for a in adlar], type_ignores=[]), "kopru.py", "exec"), env)
|
||
self.h = type("H", (), {a: env[a] for a in adlar})()
|
||
self.h.wfile = io.BytesIO()
|
||
self.h.send_response = Mock()
|
||
self.h.send_header = Mock()
|
||
self.h.end_headers = Mock()
|
||
self.cevap = {"Katman-2": sse("iki"), "Katman-3": sse("uc"), "Katman-4": sse("dort")}
|
||
|
||
def sor(self, istek, metin, key):
|
||
self.asked.append(metin)
|
||
return self.puanlar.pop(0) if self.puanlar else {"ret": 0.0, "sart": 0.0}
|
||
|
||
def ac(self, ad):
|
||
self.opened.append(ad)
|
||
if self.cevap[ad] is None:
|
||
raise OSError("acilamadi")
|
||
return Yanit(self.cevap[ad]), None, f"[{ad}]"
|
||
|
||
def sent(self):
|
||
raw, out = self.h.wfile.getvalue(), b""
|
||
while raw:
|
||
size, _, rest = raw.partition(b"\r\n")
|
||
n = int(size, 16)
|
||
if n == 0:
|
||
break
|
||
out += rest[:n]
|
||
raw = rest[n + 2:]
|
||
return out
|
||
|
||
def calistir(self, body, sira=("Katman-2", "Katman-3", "Katman-4")):
|
||
self.h._denetimli_akis(Yanit(body), "[1]", None, None, "yap", list(sira), self.ac)
|
||
|
||
def test_a_passed_step_is_delivered_unchanged(self):
|
||
body = sse("Fenerbahçe'nin ", "13 puanı var.")
|
||
self.calistir(body)
|
||
self.assertEqual(self.sent(), body)
|
||
self.assertEqual(self.asked, ["Fenerbahçe'nin 13 puanı var."])
|
||
self.assertEqual(self.opened, [])
|
||
|
||
def test_blocked_layer_1_goes_to_layer_2_which_is_checked_too(self):
|
||
self.puanlar = [{"ret": 0.9, "sart": 0.0}]
|
||
self.calistir(sse("Bunu yapamam."))
|
||
self.assertEqual(self.sent(), self.cevap["Katman-2"])
|
||
self.assertNotIn(b"yapamam", self.h.wfile.getvalue())
|
||
self.assertEqual(self.opened, ["Katman-2"])
|
||
self.assertEqual(self.asked, ["Bunu yapamam.", "iki"])
|
||
self.h.send_response.assert_called_once()
|
||
|
||
def test_every_layer_blocked_ends_at_layer_4_unasked(self):
|
||
self.puanlar = [{"ret": 0.9, "sart": 0.0}] * 3
|
||
self.calistir(sse("hayir"))
|
||
self.assertEqual(self.sent(), self.cevap["Katman-4"])
|
||
self.assertEqual(self.opened, ["Katman-2", "Katman-3", "Katman-4"])
|
||
self.assertEqual(self.asked, ["hayir", "iki", "uc"]) # layer 4 is never asked
|
||
|
||
def test_blocked_layer_3_goes_straight_to_layer_4(self):
|
||
self.puanlar = [{"ret": 0.0, "sart": 0.8}]
|
||
self.calistir(sse("kendi sartimla"), sira=("Katman-4",))
|
||
self.assertEqual(self.sent(), self.cevap["Katman-4"])
|
||
self.assertEqual(self.asked, ["kendi sartimla"])
|
||
|
||
def test_a_layer_that_cannot_be_opened_is_skipped(self):
|
||
self.puanlar = [{"ret": 0.9, "sart": 0.0}]
|
||
self.cevap["Katman-2"] = None
|
||
self.calistir(sse("hayir"))
|
||
self.assertEqual(self.opened, ["Katman-2", "Katman-3"])
|
||
self.assertEqual(self.sent(), self.cevap["Katman-3"])
|
||
|
||
def test_a_step_without_text_is_not_sent_to_jev(self):
|
||
body = b'event: response.output_item.done\ndata: {"type": "response.output_item.done", "item": {"type": "function_call"}}\n\n' + sse()
|
||
self.calistir(body)
|
||
self.assertEqual(self.sent(), body)
|
||
self.assertEqual(self.asked, [])
|
||
|
||
def test_unreachable_jev_keeps_the_answer(self):
|
||
self.puanlar = [None]
|
||
body = sse("Bunu yapamam.")
|
||
self.calistir(body)
|
||
self.assertEqual(self.sent(), body)
|
||
self.assertEqual(self.opened, [])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|