yilsemandClaude Opus 5.5
e2793fa38f
From ten independent read-only audits (11 Oct 2026), verified against the code: - a permanent gateway error (no tools, 401/402/403, bare upstream 400) ends the turn with a visible assistant message instead of a silent retry and an empty turn; - "modeli ... yapma", "X modeli için test yap" and general "hangi model ..." questions are no longer commands; a name whose version does not fit, or that several models share, is asked back; - the stream is forwarded as it arrives (read1) and an upstream drop is no longer logged as a client stop nor saved as an interrupted answer; - a lone "." joining a running turn is removed without the hook messages after it; - the interrupted-answer note finds the user message behind hook messages and detects completion by event type; a per-minute rate limit is not learned as a context window. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
115 lines
6.8 KiB
Python
115 lines
6.8 KiB
Python
"""Bridge 7: tool-less models are learned, flat gateway errors are translated, instructions are never trimmed."""
|
||
import ast
|
||
import json
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
import pencere
|
||
|
||
tree = ast.parse(Path(__file__).with_name("bridge.py").read_text(encoding="utf-8"))
|
||
names = {"fix_tool_ids", "SseRewriter", "phase_damgala", "asistan_mesaji_mi", "nokta_mesaji_mi", "hazir_yaniti",
|
||
"sema_duzelt", "arac_semalarini_duzelt"}
|
||
nodes = [n for n in tree.body if isinstance(n, (ast.FunctionDef, ast.ClassDef)) and n.name in names]
|
||
for node in tree.body:
|
||
if isinstance(node, ast.Assign) and any(isinstance(t, ast.Name) and t.id in
|
||
{"ID_PREFIX", "FOREIGN_ID_MARK", "REFERENCE_TYPES", "INLINE_TYPES", "ARAC_OGELERI", "NOKTA_ISARETLERI"} for t in node.targets):
|
||
nodes.insert(0, node)
|
||
env = {"json": json, "re": __import__("re"), "time": __import__("time"), "log": lambda message: None, "pencere": pencere}
|
||
exec(compile(ast.Module(body=nodes, type_ignores=[]), "bridge.py", "exec"), env)
|
||
|
||
|
||
def feed(event, **kw):
|
||
adapter = env["SseRewriter"](set(), **kw)
|
||
raw = b"data: " + json.dumps(event).encode() + b"\n\n"
|
||
return [json.loads(line[6:]) for line in (adapter.feed(raw) + adapter.flush()).decode().splitlines()
|
||
if line.startswith("data: ")]
|
||
|
||
|
||
class Ek(unittest.TestCase):
|
||
def test_model_without_tool_calls_is_reported_and_remembered(self):
|
||
seen = []
|
||
flat = {"type": "error", "code": "bad_request", "status_code": 400, "message":
|
||
'[400]: {"error":{"message":"This model does not support tool calls.","code":"tool_choice_unsupported"}}'}
|
||
out = feed(flat, hedef="m", aracsiz=seen.append)
|
||
# A permanent error ends the turn with a visible answer instead of a silent retry.
|
||
self.assertEqual([e["type"] for e in out], ["response.output_item.added", "response.output_text.delta",
|
||
"response.output_item.done", "response.completed"])
|
||
self.assertIn("araç çağrısını desteklemiyor", out[1]["delta"])
|
||
self.assertEqual(out[2]["item"]["phase"], "final_answer")
|
||
self.assertEqual(seen, ["m"])
|
||
home = tempfile.mkdtemp()
|
||
self.assertTrue(pencere.ogren_aracsiz(home, "m"))
|
||
self.assertFalse(pencere.ogren_aracsiz(home, "m"))
|
||
self.assertFalse(pencere.coz("m", "model", "cc/claude-opus-5-5", aracsiz=pencere.aracsizlar(home))["arac"])
|
||
self.assertTrue(pencere.coz("n", "model", "cc/claude-opus-5-5", aracsiz=pencere.aracsizlar(home))["arac"])
|
||
|
||
def test_flat_overflow_error_is_translated_and_learned(self):
|
||
seen = []
|
||
flat = {"type": "error", "code": "bad_request", "message":
|
||
"Your prompt exceeds the model's context length. Max context tokens: 65536."}
|
||
out = feed(flat, hedef="m", tasma=lambda hedef, metin: seen.append(pencere.sinir_oku(metin)))
|
||
self.assertEqual(out[0]["response"]["error"]["code"], "context_length_exceeded")
|
||
self.assertEqual(seen, [65536])
|
||
self.assertEqual(feed({"type": "error", "code": "server_error", "message": "upstream 503"})[0]["type"], "error")
|
||
|
||
def test_guard_never_trims_instruction_messages(self):
|
||
big = "KURAL " * 40000
|
||
req = {"input": [
|
||
{"type": "message", "role": "developer", "content": [{"type": "input_text", "text": big}]},
|
||
{"type": "message", "role": "user", "content": [{"type": "input_text", "text": "eski " * 40000}]},
|
||
{"type": "message", "role": "user", "content": [{"type": "input_text", "text": "soru"}]}]}
|
||
pencere.sigdir(req, 65536)
|
||
self.assertEqual(req["input"][0]["content"][0]["text"], big)
|
||
self.assertIn("çıkarıldı (köprü)", req["input"][1]["content"][0]["text"])
|
||
|
||
|
||
class Sema(unittest.TestCase):
|
||
def test_top_level_union_becomes_one_object_schema(self):
|
||
req = {"tools": [
|
||
{"type": "function", "name": "a", "parameters": {"oneOf": [
|
||
{"type": "object", "properties": {"x": {"type": "string"}, "k": {"type": "string"}}, "required": ["x", "k"]},
|
||
{"type": "object", "properties": {"y": {"type": "integer"}, "k": {"type": "string"}}, "required": ["y", "k"]}]}},
|
||
{"type": "function", "name": "b", "parameters": {"type": "object", "properties": {"z": {"anyOf": [{"type": "string"}, {"type": "null"}]}}}},
|
||
{"type": "function", "name": "c", "parameters": {"allOf": [
|
||
{"type": "object", "properties": {"p": {"type": "string"}}, "required": ["p"]},
|
||
{"type": "object", "properties": {"q": {"type": "string"}}, "required": ["q"]}]}}]}
|
||
self.assertEqual(env["arac_semalarini_duzelt"](req), 2)
|
||
a, b, c = (t["parameters"] for t in req["tools"])
|
||
self.assertEqual((a["type"], sorted(a["properties"]), a["required"]), ("object", ["k", "x", "y"], ["k"]))
|
||
self.assertNotIn("oneOf", a)
|
||
self.assertIn("anyOf", b["properties"]["z"]) # nested unions are left alone
|
||
self.assertEqual((sorted(c["properties"]), c["required"]), (["p", "q"], ["p", "q"]))
|
||
self.assertEqual(env["arac_semalarini_duzelt"](req), 0)
|
||
|
||
|
||
class Nokta(unittest.TestCase):
|
||
def user(self, content):
|
||
return {"type": "message", "role": "user", "content": content}
|
||
|
||
def test_only_a_lone_dot_from_the_user_is_answered_locally(self):
|
||
is_dot = env["nokta_mesaji_mi"]
|
||
self.assertTrue(is_dot({"input": [self.user([{"type": "input_text", "text": " . "}])]}))
|
||
self.assertTrue(is_dot({"input": [self.user("x"), self.user(".")]}))
|
||
for other in ("..", "?", "Selam", "", "1."):
|
||
self.assertFalse(is_dot({"input": [self.user([{"type": "input_text", "text": other}])]}), other)
|
||
self.assertFalse(is_dot({"input": [self.user("."), {"type": "function_call_output", "call_id": "c", "output": "x"}]}))
|
||
self.assertFalse(is_dot({"input": [self.user([{"type": "input_text", "text": "."}, {"type": "input_image", "image_url": "x"}])]}))
|
||
self.assertFalse(is_dot({"input": [{"type": "message", "role": "assistant", "content": "."}]}))
|
||
self.assertFalse(is_dot({"input": []}))
|
||
|
||
def test_local_answer_is_a_complete_response_with_a_final_answer(self):
|
||
raw = env["hazir_yaniti"]("ide/x", "cc/opus-5-5 hazır.").decode()
|
||
events = [json.loads(line[6:]) for line in raw.splitlines() if line.startswith("data: ")]
|
||
self.assertEqual([e["type"] for e in events], ["response.created", "response.output_item.added",
|
||
"response.output_text.delta", "response.output_item.done", "response.completed"])
|
||
done = events[-1]["response"]
|
||
self.assertEqual(done["status"], "completed")
|
||
self.assertEqual(done["output"][0]["phase"], "final_answer")
|
||
self.assertEqual(done["output"][0]["content"][0]["text"], "cc/opus-5-5 hazır.")
|
||
self.assertEqual(done["usage"]["total_tokens"], 0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|