feat(codex-bridge): ship the local Codex bridge with the IDE
Adds the IDE's models and combos to Codex's model list and keeps conversation summaries readable across providers, so switching models in Codex no longer drops the conversation. Standard-library Python, run on the agent venv. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HRDGW1LiKcDUZYUdKBoWYW
This commit is contained in:
1 parent
df63432c92
commit
1ce5ab82cf
13 files changed
+2071
No files matched your search
@@ -0,0 +1,68 @@
|
||||
"""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")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user