Spaces:
Running on Zero
Running on Zero
| """End-to-end pure-Python regression for the OpenClaude tool wire contract.""" | |
| from __future__ import annotations | |
| import json | |
| import unittest | |
| from openai_compat import ( | |
| analyze_tool_flow, | |
| indexed_tool_calls, | |
| resolve_tool_choice, | |
| select_tools, | |
| tool_names, | |
| ) | |
| from openclaude_compat import normalize_openclaude_messages | |
| from tool_calls import extract_tool_calls | |
| GLOB = { | |
| "type": "function", | |
| "function": { | |
| "name": "Glob", | |
| "description": "Find files by glob pattern.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "pattern": {"type": "string"}, | |
| "path": {"type": "string"}, | |
| }, | |
| "required": ["pattern"], | |
| }, | |
| }, | |
| } | |
| READ = { | |
| "type": "function", | |
| "function": { | |
| "name": "Read", | |
| "description": "Read a file.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": {"file_path": {"type": "string"}}, | |
| "required": ["file_path"], | |
| }, | |
| }, | |
| } | |
| class OpenClaudeToolContractTests(unittest.TestCase): | |
| def test_repository_summary_becomes_structured_openai_tool_call(self) -> None: | |
| # This is the public failure shape reported with Qwen coder models: | |
| # OpenClaude asks for a repository summary with tool_choice=auto. | |
| messages = [ | |
| {"role": "user", "content": "Summarize this repository structure."} | |
| ] | |
| tools = [GLOB, READ] | |
| state = analyze_tool_flow(messages, tools) | |
| choice = resolve_tool_choice("auto", state) | |
| selected, mode = select_tools(tools, choice) | |
| self.assertEqual(mode, "forced") | |
| self.assertEqual([t["function"]["name"] for t in selected], ["Glob"]) | |
| # Exact Qwen2.5-Coder native function-call syntax. | |
| model_text = ( | |
| '<tool_call>{"name":"Glob","arguments":{"pattern":"**/*"}}' | |
| "</tool_call>" | |
| ) | |
| calls, visible = extract_tool_calls(model_text, tool_names(selected)) | |
| self.assertEqual(visible, "") | |
| self.assertEqual(len(calls), 1) | |
| self.assertEqual(calls[0]["type"], "function") | |
| self.assertEqual(calls[0]["function"]["name"], "Glob") | |
| self.assertEqual( | |
| json.loads(calls[0]["function"]["arguments"]), | |
| {"pattern": "**/*"}, | |
| ) | |
| # OpenClaude's streaming converter requires a stable `index`, while the | |
| # non-streaming converter consumes the same id/name/arguments payload. | |
| streamed = indexed_tool_calls(calls) | |
| self.assertEqual(streamed[0]["index"], 0) | |
| self.assertTrue(streamed[0]["id"].startswith("call_")) | |
| def test_tool_result_round_trip_preserves_call_id_and_arguments_mapping(self) -> None: | |
| history = [ | |
| {"role": "user", "content": "Summarize this repository structure."}, | |
| { | |
| "role": "assistant", | |
| "content": None, | |
| "tool_calls": [ | |
| { | |
| "id": "call_contract_1", | |
| "type": "function", | |
| "function": { | |
| "name": "Glob", | |
| "arguments": '{"pattern":"**/*"}', | |
| }, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "call_contract_1", | |
| "name": "Glob", | |
| "content": "app.py\nopenai_compat.py\ntool_calls.py", | |
| }, | |
| ] | |
| normalized = normalize_openclaude_messages(history) | |
| assistant = next(m for m in normalized if m["role"] == "assistant") | |
| result = next(m for m in normalized if m["role"] == "tool") | |
| call = assistant["tool_calls"][0] | |
| self.assertEqual(call["id"], "call_contract_1") | |
| self.assertEqual(call["function"]["arguments"], {"pattern": "**/*"}) | |
| self.assertEqual(result["tool_call_id"], "call_contract_1") | |
| self.assertEqual(result["name"], "Glob") | |
| followup_state = analyze_tool_flow(history, [GLOB, READ]) | |
| # After evidence exists, auto stays available instead of being poisoned | |
| # by the first turn. The model may summarize or request another tool. | |
| self.assertEqual(resolve_tool_choice("auto", followup_state), "auto") | |
| if __name__ == "__main__": | |
| unittest.main() | |