yilsemandClaude Opus 4.8
1f4f2de428
User, 11 Oct 2026: 'Jev geçmek zor olsun.' Lower thresholds (ret 0.5, sart 0.7) and only let a refusal pass on a genuinely high technical obstacle, so a fake technical excuse no longer passes. One carve-out stated to the user first: when the request is to hack, intrude on or surveil a third party's device or account or to build malware, a refusal is NOT routed to the uncensored fallback; the refusal stands. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
208 lines
9.7 KiB
Python
208 lines
9.7 KiB
Python
"""Step check (11 Oct 2026): Jev passes or blocks each step of an added model; a blocked step is redone by the
|
||
fallback model, the next step goes back to the selected model."""
|
||
import ast
|
||
import http.client
|
||
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, done=True):
|
||
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"
|
||
if done:
|
||
out += b'event: response.completed\ndata: {"type": "response.completed"}\n\n'
|
||
return out
|
||
|
||
|
||
class Yanit:
|
||
"""Fake upstream response: read1 returns the body in two pieces."""
|
||
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""
|
||
|
||
|
||
def scores(ret=0.0, sart=0.0, teknik=0.0, soru=0.0, zarar=0.0):
|
||
return {"ret": ret, "sart": sart, "teknik": teknik, "soru": soru, "zarar": zarar}
|
||
|
||
|
||
class Karar(unittest.TestCase):
|
||
def test_refusals_and_silent_conditions_are_blocked(self):
|
||
self.assertTrue(jev_denetim.karar(scores(ret=0.84))) # "böyle bir hüküm vermiyorum"
|
||
self.assertTrue(jev_denetim.karar(scores(ret=0.41, sart=0.95))) # did it with its own conditions
|
||
self.assertTrue(jev_denetim.karar(scores(ret=0.93, sart=0.72)))
|
||
self.assertTrue(jev_denetim.karar(scores(ret=0.62, sart=0.91, soru=0.93))) # "I can't; shall I do less?"
|
||
self.assertTrue(jev_denetim.karar(scores(ret=0.70, sart=0.90, teknik=0.23, soru=0.89))) # measured 11 Oct 2026
|
||
self.assertTrue(jev_denetim.karar(scores(ret=0.55, teknik=0.60))) # a fake technical excuse no longer passes
|
||
|
||
def test_normal_answers_real_obstacles_and_questions_pass(self):
|
||
self.assertFalse(jev_denetim.karar(scores(ret=0.05, sart=0.3)))
|
||
self.assertFalse(jev_denetim.karar(scores(ret=0.85, teknik=0.97))) # genuine "SSH key rejected, could not read"
|
||
self.assertFalse(jev_denetim.karar(scores(ret=0.48, sart=0.44, soru=0.96))) # a clarifying question without a refusal
|
||
|
||
def test_attack_on_a_third_party_stays_refused_not_sent_to_fallback(self):
|
||
# User decision 11 Oct 2026: making Jev strict must not turn a refusal of third-party intrusion/malware into
|
||
# a request the uncensored fallback fulfils. Such a refusal passes (the refusal stands), it is not blocked.
|
||
self.assertFalse(jev_denetim.karar(scores(ret=0.94, sart=0.92, teknik=0.58, soru=0.85, zarar=0.93)))
|
||
self.assertTrue(jev_denetim.karar(scores(ret=0.94, sart=0.92, zarar=0.2))) # ordinary refusal still blocked
|
||
|
||
def test_thresholds_come_from_settings(self):
|
||
self.assertFalse(jev_denetim.karar(scores(ret=0.45))) # just under the default
|
||
self.assertTrue(jev_denetim.karar(scores(ret=0.45), {"ret": 0.4}))
|
||
self.assertFalse(jev_denetim.karar(scores(ret=0.9, teknik=0.9), {"teknik_muaf": 0.85}))
|
||
|
||
|
||
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 <transcript_delta>x</transcript_delta>\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}}
|
||
puan = jev_denetim.sor("istek", "c" * 9000, "anahtar", gonder=gonder)
|
||
self.assertEqual(puan, scores(0.25, 0.25, 0.25, 0.25, 0.25))
|
||
self.assertEqual(seen["key"], "anahtar")
|
||
self.assertEqual(seen["govde"]["state"]["user_request"], "istek")
|
||
self.assertLess(len(seen["govde"]["state"]["assistant_reply"]), 6100) # long answers are cut
|
||
self.assertEqual(set(seen["govde"]["questions"]), {"ret", "sart", "teknik", "soru", "zarar"})
|
||
|
||
def test_failures_pass_the_answer_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)) # breaker open: not called
|
||
self.assertEqual(len(calls), jev_denetim.DEVRE_HATA)
|
||
self.assertIsNone(jev_denetim.sor("i", "c", "k", gonder=bozuk, simdi=1000.0 + jev_denetim.DEVRE_SN + 1))
|
||
self.assertEqual(len(calls), jev_denetim.DEVRE_HATA + 1)
|
||
|
||
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")]), "")
|
||
|
||
def test_settings(self):
|
||
self.assertEqual(jev_denetim.ayar_oku(None)[:2], (True, "ide/Uncensored-Uygulama"))
|
||
self.assertEqual(jev_denetim.ayar_oku({"enabled": False, "hedef": "ide/x"})[:2], (False, "ide/x"))
|
||
|
||
|
||
class Akis(unittest.TestCase):
|
||
"""_denetimli_akis through a fake handler: what reaches the client."""
|
||
|
||
def setUp(self):
|
||
self.logs = []
|
||
self.asked = []
|
||
self.puan = scores()
|
||
sahte = Mock()
|
||
sahte.SURE = 6.0
|
||
sahte.anahtar = lambda: "k"
|
||
sahte.sor = lambda istek, metin, key, sure: (self.asked.append((istek, metin)), self.puan)[1]
|
||
sahte.karar = jev_denetim.karar
|
||
env = {"jev_denetim": sahte, "yarim": yarim, "time": time, "http": __import__("http"), "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()
|
||
|
||
def sent(self):
|
||
"""Chunked body the client received, decoded."""
|
||
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 test_a_passed_step_is_delivered_unchanged(self):
|
||
body = sse("Fenerbahçe'nin ", "13 puanı var.")
|
||
self.h._denetimli_akis(Yanit(body), "[1]", None, None, "kaç puan", {}, lambda: self.fail("fallback opened"))
|
||
self.assertEqual(self.sent(), body)
|
||
self.assertEqual(self.asked, [("kaç puan", "Fenerbahçe'nin 13 puanı var.")])
|
||
self.assertIn("gecti", self.logs[0])
|
||
|
||
def test_a_blocked_step_is_dropped_and_redone_by_the_fallback(self):
|
||
self.puan = scores(ret=0.9)
|
||
yedek = sse("Tamam, yaptım.")
|
||
self.h._denetimli_akis(Yanit(sse("Bunu yapamam.")), "[1]", None, None, "yap", {},
|
||
lambda: (Yanit(yedek), None, "[1 yedek]"))
|
||
self.assertEqual(self.sent(), yedek) # the refused text never reaches the client
|
||
self.assertNotIn(b"yapamam", self.h.wfile.getvalue())
|
||
self.assertTrue(any("GECMEDI" in s for s in self.logs))
|
||
self.h.send_response.assert_called_once() # headers once, although two upstream answers
|
||
|
||
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.h._denetimli_akis(Yanit(body), "[1]", None, None, "yap", {}, lambda: self.fail("fallback opened"))
|
||
self.assertEqual(self.sent(), body)
|
||
self.assertEqual(self.asked, [])
|
||
|
||
def test_unreachable_jev_or_fallback_keeps_the_first_answer(self):
|
||
self.puan = None
|
||
body = sse("Bunu yapamam.")
|
||
self.h._denetimli_akis(Yanit(body), "[1]", None, None, "yap", {}, lambda: self.fail("fallback opened"))
|
||
self.assertEqual(self.sent(), body)
|
||
self.h.wfile = io.BytesIO()
|
||
self.puan = scores(ret=0.9)
|
||
|
||
def bozuk():
|
||
raise OSError("yedek acilamadi")
|
||
self.h._denetimli_akis(Yanit(body), "[1]", None, None, "yap", {}, bozuk)
|
||
self.assertEqual(self.sent(), body)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|