aiturk-hermes-ide/apps/desktop/assets/codex-bridge/test_model_switch.py

"""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", "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()