yilsemandClaude Opus 5.5
69daf809e0
Claude rejects a turn whose latest assistant thinking blocks are not exactly
as produced ("cannot be modified" / "Invalid signature in thinking block").
Following Hermes (_manage_thinking_signatures: a dead-signature turn is
replayed without thinking), the bridge reads the first upstream piece and,
on exactly these messages, resends that one request once with reasoning
effort none before anything reaches Codex. Other errors are not resent.
The rejected request is kept as log/dusunme-400-*.json for diagnosis.
The overflow guard and layer filter still used a flat 2 bytes/token; they
now use the measured ratio. Ship IDE beta.48 (bridge version 27).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XeNgWNzPjzUKQuS32RUYXx
88 lines
5.0 KiB
Python
88 lines
5.0 KiB
Python
"""Regression coverage for Codex -> external API access_programs compatibility."""
|
|
import ast
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import unittest
|
|
from unittest.mock import Mock
|
|
import urllib.request
|
|
|
|
|
|
# Load only the actual adapter and routing method: no catalog/network/thread
|
|
# startup or credential loading is needed to exercise the production code.
|
|
source = ast.parse(Path(__file__).with_name("kopru.py").read_text(encoding="utf-8"))
|
|
flatten = next(n for n in source.body if isinstance(n, ast.FunctionDef) and n.name == "flatten_request")
|
|
claude_sonu = next(n for n in source.body if isinstance(n, ast.FunctionDef) and n.name == "claude_sonu_duzelt")
|
|
handler = next(n for n in source.body if isinstance(n, ast.ClassDef) and n.name == "H")
|
|
responses = next(n for n in handler.body if isinstance(n, ast.FunctionDef) and n.name == "_responses")
|
|
ozgun = next(n for n in handler.body if isinstance(n, ast.FunctionDef) and n.name == "_ozgun")
|
|
taraf = [n for n in source.body if isinstance(n, ast.FunctionDef) and n.name in ("ek_model_mu", "yonlendirilen")]
|
|
|
|
|
|
class AccessPrograms(unittest.TestCase):
|
|
def setUp(self):
|
|
self.upstream = Mock(return_value=Mock(status=200))
|
|
self.env = {
|
|
"WATCH_OWNER": None, "WatchResponse": lambda response, accept: response, "json": json,
|
|
"urllib": Mock(request=Mock(Request=urllib.request.Request, urlopen=self.upstream)),
|
|
"decompress": lambda body, encoding: body,
|
|
"MODEL_MAP": {},
|
|
"CFG": {"nine_router_catalog": {"enabled": True, "route": "9router", "prefix": "9r/"},
|
|
"resim_sadelestir": False, "pano_kanca_user": False},
|
|
"ROUTES": {"9router": {"url": "https://example.invalid/v1/responses", "key": "test-only", "flatten_tools": True}},
|
|
"fix_tool_ids": lambda req: 0,
|
|
"drop_foreign_items": lambda req: 0,
|
|
"SseRewriter": Mock(),
|
|
"log": lambda message: None, "pencere": __import__("pencere"), "gecis_context": __import__("gecis_context"), "dusunme_imzasi_bozuk": __import__("kopru").dusunme_imzasi_bozuk, "dusunme_ornegi_sakla": __import__("kopru").dusunme_ornegi_sakla, "OnekliYanit": __import__("kopru").OnekliYanit, "IDE_BILGI": {}, "yarim": __import__("yarim"), "jev_denetim": __import__("jev_denetim"), "internet": __import__("internet"), "is_hatti": __import__("is_hatti"), "GORULEN_ISTEMCI": set(), "arac_semalarini_duzelt": lambda req: 0, "nokta_mesaji_mi": lambda req: False, "EXTRA_MODELS": [],
|
|
"DEBUG": False, "UA": "test", "RETRIES": 1, "UPSTREAM_IDLE_TIMEOUT": 3900,
|
|
}
|
|
exec(compile(ast.Module(body=taraf + [flatten, claude_sonu, responses, ozgun], type_ignores=[]), "bridge.py", "exec"), self.env)
|
|
|
|
def request(self, model):
|
|
return {"model": model, "access_programs": {"cyber": "standard"},
|
|
"metadata": {"access_programs": "preserve nested data"},
|
|
"input": [{"role": "user", "content": [{"type": "input_text", "text": "hello"}]}],
|
|
"tools": [{"type": "function", "name": "echo", "parameters": {"type": "object", "properties": {}}}]}
|
|
|
|
def test_adapter_removes_only_top_level_field(self):
|
|
req = self.request("meta/muse-spark-1.3")
|
|
original = copy.deepcopy(req)
|
|
self.env["flatten_request"](req)
|
|
self.assertNotIn("access_programs", req)
|
|
self.assertEqual(req["metadata"], original["metadata"])
|
|
self.assertEqual(req["input"], original["input"])
|
|
self.assertEqual(req["tools"][0]["name"], "echo")
|
|
|
|
def test_absent_null_and_empty_values(self):
|
|
for value in (None, {}, {"cyber": "standard"}):
|
|
with self.subTest(value=value):
|
|
req = {"input": [], "access_programs": value}
|
|
self.env["flatten_request"](req)
|
|
self.assertNotIn("access_programs", req)
|
|
self.env["flatten_request"]({"input": []})
|
|
|
|
def test_model_and_combo_external_request_strips_field(self):
|
|
for model in ("9r/meta/muse-spark-1.3-contributor", "9r/Fusion-Hizli"):
|
|
with self.subTest(model=model):
|
|
body = json.dumps(self.request(model)).encode()
|
|
fake = Mock(headers={})
|
|
self.env["_responses"](fake, 1, "/responses", body)
|
|
sent = json.loads(self.upstream.call_args.args[0].data)
|
|
self.assertNotIn("access_programs", sent)
|
|
self.assertEqual(sent["model"], model.removeprefix("9r/"))
|
|
self.assertIn("access_programs", json.loads(body))
|
|
fake._stream.assert_called_once()
|
|
|
|
def test_chatgpt_passthrough_preserves_original_body(self):
|
|
body = json.dumps(self.request("gpt-6-astra")).encode()
|
|
fake = Mock(headers={})
|
|
fake._ozgun = lambda *a: self.env["_ozgun"](fake, *a)
|
|
self.env["_responses"](fake, 1, "/responses", body)
|
|
self.assertEqual(fake._passthrough.call_args.args[2], body)
|
|
self.assertFalse(fake._passthrough.call_args.kwargs["rewritten"])
|
|
self.upstream.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|