yilsemandClaude Opus 5.5
adf73eeedd
Model switch lost the conversation: on 10 Oct a 1 MB thread was compacted by the newly selected small model into 2201 characters. The compaction summary for a bridge model is now written by a subscription model (ozet_kopru.ozet_modeli, default gpt-6.1-sol); if that request is refused before it starts, the target model is used as before. A summary far shorter than the conversation warrants is not accepted: Codex gets response.failed and keeps the history. The line under each added model no longer repeats its name. Teams show their members, fallback chains their order, single models their provider and the traits readable from the name; the text fits one phone row (30 characters). Also normalizes tool call ids in the streamed response so Codex persists ids the subscription backend accepts (fc_ -> ctc_), and bumps the bridge protocol to 6. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
90 lines
5.3 KiB
Python
90 lines
5.3 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("bridge.py").read_text(encoding="utf-8"))
|
|
names = {"flatten_request", "fix_tool_ids", "drop_foreign_items", "SseRewriter"}
|
|
nodes = [n for n in tree.body if isinstance(n, (ast.FunctionDef, ast.ClassDef)) and n.name in names]
|
|
env = {"json": json, "log": lambda message: None}
|
|
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"} 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()
|