yilsemandClaude Opus 5.5
23e8aa8191
bridge.py is now a small splitter with no features: a request for one of Codex's own models goes to chatgpt.com byte for byte, a request for an added model goes to kopru.py, which runs as its own process. Codex's own models no longer pass through the feature code at all, and restarting or losing the bridge process does not touch them. Measured with real Codex on spare ports: own model answers; forced compaction on the own model completes (twice) while the bridge process is restarted; added model answers; no own-model request reaches the bridge process. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
91 lines
5.4 KiB
Python
91 lines
5.4 KiB
Python
"""Provider-switch history regression tests, without startup or network access."""
|
|
import ast
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
|
|
tree = ast.parse(Path(__file__).with_name("kopru.py").read_text(encoding="utf-8"))
|
|
names = {"flatten_request", "fix_tool_ids", "drop_foreign_items", "SseRewriter", "phase_damgala",
|
|
"asistan_mesaji_mi"}
|
|
nodes = [n for n in tree.body if isinstance(n, (ast.FunctionDef, ast.ClassDef)) and n.name in names]
|
|
env = {"json": json, "re": __import__("re"), "log": lambda message: None, "pencere": __import__("pencere")}
|
|
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"} for t in node.targets):
|
|
nodes.insert(0, node)
|
|
exec(compile(ast.Module(body=nodes, type_ignores=[]), "bridge.py", "exec"), env)
|
|
|
|
|
|
class ModelSwitch(unittest.TestCase):
|
|
def test_external_to_chatgpt_preserves_messages_and_paired_tool_calls(self):
|
|
req = {"input": [
|
|
{"type": "reasoning", "id": "rs_native", "encrypted_content": "native-reasoning"},
|
|
{"type": "reasoning", "id": "rs_chatcmpl-foreign"},
|
|
{"type": "item_reference", "id": "chatcmpl-foreign"},
|
|
{"type": "message", "id": "msg_chatcmpl-foreign", "role": "assistant", "content": "remember-4827"},
|
|
{"type": "custom_tool_call", "id": "fc_chatcmpl-call", "call_id": "call_4827", "name": "probe", "input": "hello"},
|
|
{"type": "custom_tool_call_output", "id": "fc_chatcmpl-output", "call_id": "call_4827", "output": "result-4827"},
|
|
]}
|
|
self.assertEqual(env["fix_tool_ids"](req), 2)
|
|
self.assertEqual(env["drop_foreign_items"](req), 2)
|
|
self.assertEqual(req["input"][0]["encrypted_content"], "native-reasoning")
|
|
self.assertEqual(req["input"][1]["content"], "remember-4827")
|
|
self.assertEqual(req["input"][2]["id"], "ctc_chatcmpl-call")
|
|
self.assertEqual(req["input"][3]["id"], "ctco_chatcmpl-output")
|
|
self.assertEqual(req["input"][2]["call_id"], req["input"][3]["call_id"])
|
|
self.assertEqual(req["input"][3]["output"], "result-4827")
|
|
|
|
def test_chatgpt_to_external_preserves_custom_tool_history(self):
|
|
original = {"access_programs": {"cyber": "standard"}, "input": [
|
|
{"type": "custom_tool_call", "id": "ctc_4827", "call_id": "call_4827", "name": "probe", "input": "remember-4827"},
|
|
{"type": "custom_tool_call_output", "id": "ctco_4827", "call_id": "call_4827", "output": "result-4827"},
|
|
], "tools": [{"type": "namespace", "name": "functions", "tools": [{"type": "custom", "name": "probe"}]}]}
|
|
req = copy.deepcopy(original)
|
|
self.assertEqual(env["flatten_request"](req), {"probe"})
|
|
env["fix_tool_ids"](req)
|
|
self.assertEqual(json.loads(req["input"][0]["arguments"]), {"input": "remember-4827"})
|
|
self.assertEqual(req["input"][1]["output"], "result-4827")
|
|
self.assertEqual(req["input"][0]["call_id"], req["input"][1]["call_id"])
|
|
self.assertEqual(req["input"][0]["type"], "function_call")
|
|
self.assertEqual(req["input"][1]["type"], "function_call_output")
|
|
self.assertIn("access_programs", original)
|
|
|
|
def test_streamed_external_custom_tool_can_return_to_codex(self):
|
|
event = {"type": "response.output_item.done", "item": {
|
|
"type": "function_call", "id": "fc_chatcmpl-probe", "call_id": "call_4827",
|
|
"name": "probe", "arguments": json.dumps({"input": "remember-4827"})}}
|
|
raw = ("event: response.output_item.done\ndata: " + json.dumps(event) + "\n\n").encode()
|
|
adapter = env["SseRewriter"]({"probe"})
|
|
out = adapter.feed(raw[:25]) + adapter.feed(raw[25:]) + adapter.flush()
|
|
rewritten = json.loads(next(line[6:] for line in out.decode().splitlines() if line.startswith("data: ")))
|
|
self.assertEqual(rewritten["item"]["type"], "custom_tool_call")
|
|
self.assertEqual(rewritten["item"]["input"], "remember-4827")
|
|
self.assertEqual(rewritten["item"]["call_id"], "call_4827")
|
|
self.assertEqual(rewritten["item"]["id"], "ctc_chatcmpl-probe")
|
|
|
|
def test_custom_stream_ids_are_valid_before_history_is_saved(self):
|
|
# Real failure after disabling the bridge and returning to OpenAI:
|
|
# input[32].id fc_call_... was saved as a custom_tool_call by Codex.
|
|
item = {"type": "function_call", "id": "fc_call_3f427ecc539d4c779c1252077701216f",
|
|
"call_id": "call_original", "name": "probe",
|
|
"arguments": json.dumps({"input": "keep this exact tool input"})}
|
|
adapter = env["SseRewriter"]({"probe"})
|
|
converted = []
|
|
for kind in ("response.output_item.added", "response.output_item.done"):
|
|
converted.append(adapter._fix({"type": kind, "item": dict(item)})["item"])
|
|
final = adapter._fix({"type": "response.completed", "response": {"output": [dict(item)]}})
|
|
converted.append(final["response"]["output"][0])
|
|
for result in converted:
|
|
self.assertEqual(result["type"], "custom_tool_call")
|
|
self.assertEqual(result["id"], "ctc_call_3f427ecc539d4c779c1252077701216f")
|
|
self.assertEqual(result["call_id"], "call_original")
|
|
self.assertEqual(result["input"], "keep this exact tool input")
|
|
history = {"input": converted}
|
|
self.assertEqual(env["fix_tool_ids"](history), 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|