Spaces:
Running on Zero
Running on Zero
| """Contract tests for OpenAI request normalization.""" | |
| from __future__ import annotations | |
| import unittest | |
| from openai_compat import ( | |
| MAX_SCHEMA_DESCRIPTION_CHARS, | |
| MAX_TOOL_DESCRIPTION_CHARS, | |
| _tool_result_events, | |
| analyze_tool_flow, | |
| indexed_tool_calls, | |
| is_simple_greeting, | |
| normalize_messages, | |
| normalize_tools, | |
| resolve_tool_choice, | |
| select_tools, | |
| tool_choice_instruction, | |
| ) | |
| TOOLS = [ | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "Read", | |
| "description": "Read a file", | |
| "parameters": { | |
| "type": "object", | |
| "properties": {"file_path": {"type": "string"}}, | |
| "required": ["file_path"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "Bash", | |
| "parameters": {"type": "object", "properties": {}}, | |
| }, | |
| }, | |
| ] | |
| EDIT_TOOL = { | |
| "type": "function", | |
| "function": { | |
| "name": "Edit", | |
| "parameters": {"type": "object", "properties": {}}, | |
| }, | |
| } | |
| GLOB_TOOL = { | |
| "type": "function", | |
| "function": { | |
| "name": "Glob", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "pattern": {"type": "string"}, | |
| "path": {"type": "string"}, | |
| }, | |
| "required": ["pattern"], | |
| }, | |
| }, | |
| } | |
| WEB_TOOLS = [ | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "WebSearch", | |
| "parameters": { | |
| "type": "object", | |
| "properties": {"query": {"type": "string"}}, | |
| "required": ["query"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "WebFetch", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "url": {"type": "string"}, | |
| "prompt": {"type": "string"}, | |
| }, | |
| "required": ["url", "prompt"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "ToolSearch", | |
| "parameters": {"type": "object", "properties": {}}, | |
| }, | |
| }, | |
| ] | |
| class OpenAICompatibilityTests(unittest.TestCase): | |
| def test_verbose_tool_metadata_is_compacted_without_losing_schema(self) -> None: | |
| tools = normalize_tools( | |
| [ | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "Bash", | |
| "description": "manual " * 2_000, | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "command": { | |
| "type": "string", | |
| "description": "command help " * 1_000, | |
| } | |
| }, | |
| "required": ["command"], | |
| "additionalProperties": False, | |
| }, | |
| }, | |
| } | |
| ] | |
| ) | |
| function = tools[0]["function"] | |
| parameters = function["parameters"] | |
| self.assertLessEqual( | |
| len(function["description"]), MAX_TOOL_DESCRIPTION_CHARS | |
| ) | |
| self.assertTrue(function["description"].endswith("…")) | |
| self.assertLessEqual( | |
| len(parameters["properties"]["command"]["description"]), | |
| MAX_SCHEMA_DESCRIPTION_CHARS, | |
| ) | |
| self.assertEqual(parameters["required"], ["command"]) | |
| self.assertFalse(parameters["additionalProperties"]) | |
| def test_invalid_parameter_schema_is_replaced(self) -> None: | |
| tools = normalize_tools( | |
| [{"type": "function", "function": {"name": "Read", "parameters": "bad"}}] | |
| ) | |
| self.assertEqual( | |
| tools[0]["function"]["parameters"], | |
| {"type": "object", "properties": {}}, | |
| ) | |
| def test_duplicate_tool_names_are_deduplicated_case_insensitively(self) -> None: | |
| duplicate = { | |
| "type": "function", | |
| "function": { | |
| "name": "read", | |
| "description": "duplicate alias", | |
| "parameters": {"type": "object", "properties": {}}, | |
| }, | |
| } | |
| normalized = normalize_tools([TOOLS[0], duplicate]) | |
| self.assertEqual(len(normalized), 1) | |
| self.assertEqual(normalized[0]["function"]["name"], "Read") | |
| def test_input_schema_alias_and_bare_function_are_supported(self) -> None: | |
| tools = normalize_tools( | |
| [{"name": "Search", "input_schema": {"type": "object"}}] | |
| ) | |
| self.assertEqual(tools[0]["function"]["name"], "Search") | |
| self.assertEqual( | |
| tools[0]["function"]["parameters"], {"type": "object"} | |
| ) | |
| def test_tool_choice_none_hides_all_tools(self) -> None: | |
| tools, mode = select_tools(TOOLS, "none") | |
| self.assertEqual(tools, []) | |
| self.assertEqual(mode, "none") | |
| def test_explicit_no_tools_instruction_resolves_auto_to_none(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| { | |
| "role": "system", | |
| "content": "Não use ferramentas nesta verificação.", | |
| }, | |
| {"role": "user", "content": "Responda apenas OK."}, | |
| ], | |
| TOOLS, | |
| ) | |
| self.assertTrue(state.can_finalize) | |
| self.assertEqual(resolve_tool_choice("auto", state), "none") | |
| def test_other_tools_prohibition_preserves_forced_read(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| { | |
| "role": "system", | |
| "content": ( | |
| "Use somente Read quando necessário. " | |
| "Não use outras ferramentas." | |
| ), | |
| }, | |
| {"role": "user", "content": "Leia README.md."}, | |
| ], | |
| TOOLS, | |
| ) | |
| self.assertTrue(state.requires_tool) | |
| self.assertEqual( | |
| resolve_tool_choice("auto", state)["function"]["name"], | |
| "Read", | |
| ) | |
| def test_explicit_bash_request_forces_bash(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| { | |
| "role": "user", | |
| "content": ( | |
| "Usar a ferramenta Bash para executar o comando pwd " | |
| "e informar o diretório retornado." | |
| ), | |
| } | |
| ], | |
| TOOLS, | |
| ) | |
| self.assertTrue(state.requires_tool) | |
| self.assertEqual( | |
| resolve_tool_choice("auto", state), | |
| {"type": "function", "function": {"name": "Bash"}}, | |
| ) | |
| def test_required_choice_is_restricted_to_explicit_read(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| { | |
| "role": "user", | |
| "content": ( | |
| "Use obrigatoriamente a ferramenta Read para ler " | |
| "/tmp/continuar.txt." | |
| ), | |
| } | |
| ], | |
| TOOLS, | |
| ) | |
| self.assertEqual( | |
| resolve_tool_choice("required", state), | |
| {"type": "function", "function": {"name": "Read"}}, | |
| ) | |
| def test_required_choice_remains_required_for_plain_greeting(self) -> None: | |
| state = analyze_tool_flow( | |
| [{"role": "user", "content": "oi"}], | |
| TOOLS, | |
| ) | |
| self.assertFalse(state.active) | |
| self.assertIsNone(resolve_tool_choice(None, state)) | |
| self.assertEqual(resolve_tool_choice("auto", state), "auto") | |
| self.assertEqual(resolve_tool_choice("required", state), "required") | |
| def test_openclaude_greeting_metadata_is_fast_path_safe(self) -> None: | |
| messages = [ | |
| { | |
| "role": "user", | |
| "content": ( | |
| "<available-deferred-tools>\nBash\n" | |
| "</available-deferred-tools>\n" | |
| "<system-reminder>Create code and run tests.</system-reminder>\n" | |
| "ola\n<system-reminder>snip_id=x</system-reminder>" | |
| ), | |
| } | |
| ] | |
| self.assertTrue(is_simple_greeting(messages)) | |
| self.assertFalse(is_simple_greeting([{"role": "user", "content": "ola, leia app.py"}])) | |
| def test_openclaude_metadata_does_not_become_user_intent(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| { | |
| "role": "user", | |
| "content": ( | |
| "<available-deferred-tools>\nWebSearch\n" | |
| "</available-deferred-tools>\n" | |
| "<system-reminder>Use this skill to create code " | |
| "and run tests.</system-reminder>\n" | |
| "oi\n" | |
| "<system-reminder>snip_id=abc</system-reminder>" | |
| ), | |
| } | |
| ], | |
| TOOLS, | |
| ) | |
| self.assertFalse(state.active) | |
| self.assertFalse(state.requires_tool) | |
| self.assertEqual(resolve_tool_choice("required", state), "required") | |
| def test_forced_tool_choice_is_case_insensitive_and_restrictive(self) -> None: | |
| tools, mode = select_tools( | |
| TOOLS, | |
| {"type": "function", "function": {"name": "read"}}, | |
| ) | |
| self.assertEqual([tool["function"]["name"] for tool in tools], ["Read"]) | |
| self.assertEqual(mode, "forced") | |
| self.assertIn("Read", tool_choice_instruction(mode, tools)) | |
| def test_unknown_forced_tool_is_rejected(self) -> None: | |
| with self.assertRaisesRegex(ValueError, "not defined"): | |
| select_tools( | |
| TOOLS, | |
| {"type": "function", "function": {"name": "DeleteEverything"}}, | |
| ) | |
| def test_required_without_tools_is_rejected(self) -> None: | |
| with self.assertRaisesRegex(ValueError, "at least one tool"): | |
| select_tools([], "required") | |
| def test_tool_history_arguments_become_mappings(self) -> None: | |
| messages = normalize_messages( | |
| [ | |
| { | |
| "role": "assistant", | |
| "content": None, | |
| "tool_calls": [ | |
| { | |
| "id": "call_1", | |
| "type": "function", | |
| "function": { | |
| "name": "Read", | |
| "arguments": '{"file_path":"/tmp/a.txt"}', | |
| }, | |
| } | |
| ], | |
| }, | |
| {"role": "tool", "tool_call_id": "call_1", "content": "ok"}, | |
| ] | |
| ) | |
| self.assertEqual( | |
| messages[0]["tool_calls"][0]["function"]["arguments"], | |
| {"file_path": "/tmp/a.txt"}, | |
| ) | |
| self.assertEqual(messages[1]["tool_call_id"], "call_1") | |
| def test_extra_instruction_merges_with_initial_system_message(self) -> None: | |
| messages = normalize_messages( | |
| [{"role": "system", "content": "Base"}], | |
| "Must call Read.", | |
| ) | |
| self.assertEqual(len(messages), 1) | |
| self.assertIn("Base", messages[0]["content"]) | |
| self.assertIn("Must call Read.", messages[0]["content"]) | |
| def test_streamed_tool_calls_receive_stable_indices(self) -> None: | |
| calls = [ | |
| {"id": "call_a", "type": "function", "function": {"name": "Read"}}, | |
| {"id": "call_b", "type": "function", "function": {"name": "Bash"}}, | |
| ] | |
| indexed = indexed_tool_calls(calls) | |
| self.assertEqual([call["index"] for call in indexed], [0, 1]) | |
| self.assertNotIn("index", calls[0]) | |
| def test_parallel_results_are_resolved_by_id_even_when_reordered(self) -> None: | |
| events = _tool_result_events( | |
| [ | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "read_id", | |
| "type": "function", | |
| "function": { | |
| "name": "Read", | |
| "arguments": '{"file_path":"/tmp/a"}', | |
| }, | |
| }, | |
| { | |
| "id": "bash_id", | |
| "type": "function", | |
| "function": { | |
| "name": "Bash", | |
| "arguments": '{"command":"pwd"}', | |
| }, | |
| }, | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "bash_id", | |
| "content": "/root", | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "read_id", | |
| "content": "source", | |
| }, | |
| ] | |
| ) | |
| self.assertEqual([event.name for event in events], ["Bash", "Read"]) | |
| self.assertEqual(events[0].arguments, {"command": "pwd"}) | |
| self.assertEqual(events[1].arguments, {"file_path": "/tmp/a"}) | |
| def test_agentic_read_requires_another_tool(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "read_id", | |
| "type": "function", | |
| "function": { | |
| "name": "Read", | |
| "arguments": '{"file_path":"/tmp/a"}', | |
| }, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "read_id", | |
| "content": "source", | |
| }, | |
| ], | |
| [*TOOLS, EDIT_TOOL], | |
| ) | |
| self.assertTrue(state.requires_tool) | |
| self.assertEqual(resolve_tool_choice(None, state), "required") | |
| self.assertEqual(resolve_tool_choice("auto", state), "required") | |
| self.assertEqual(resolve_tool_choice("none", state), "none") | |
| def test_read_only_flow_can_answer_normally(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "read_id", | |
| "type": "function", | |
| "function": {"name": "Read", "arguments": "{}"}, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "read_id", | |
| "content": "source", | |
| }, | |
| ], | |
| [TOOLS[0]], | |
| ) | |
| self.assertTrue(state.active) | |
| self.assertTrue(state.can_finalize) | |
| self.assertIsNone(resolve_tool_choice(None, state)) | |
| self.assertEqual(resolve_tool_choice("auto", state), "auto") | |
| def test_read_result_preserves_explicit_required_choice(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| { | |
| "role": "user", | |
| "content": "Use a ferramenta Read para ler README.md.", | |
| }, | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "read_id", | |
| "type": "function", | |
| "function": { | |
| "name": "Read", | |
| "arguments": '{"file_path":"README.md"}', | |
| }, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "read_id", | |
| "content": "conteúdo lido", | |
| }, | |
| ], | |
| [TOOLS[0]], | |
| ) | |
| self.assertTrue(state.can_finalize) | |
| self.assertEqual(resolve_tool_choice("required", state), "required") | |
| def test_edit_then_passing_test_allows_final_response(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "edit_id", | |
| "type": "function", | |
| "function": {"name": "Edit", "arguments": "{}"}, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "edit_id", | |
| "content": "updated", | |
| }, | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "test_id", | |
| "type": "function", | |
| "function": { | |
| "name": "Bash", | |
| "arguments": { | |
| "command": "python3 -m unittest -v" | |
| }, | |
| }, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "test_id", | |
| "content": "Ran 3 tests in 0.1s\n\nOK", | |
| }, | |
| ], | |
| [*TOOLS, EDIT_TOOL], | |
| ) | |
| self.assertTrue(state.can_finalize) | |
| self.assertFalse(state.requires_tool) | |
| self.assertIsNone(resolve_tool_choice(None, state)) | |
| self.assertEqual(resolve_tool_choice("auto", state), "auto") | |
| def test_edit_after_passing_test_requires_fresh_verification(self) -> None: | |
| messages = [ | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "edit_1", | |
| "type": "function", | |
| "function": {"name": "Edit", "arguments": "{}"}, | |
| } | |
| ], | |
| }, | |
| {"role": "tool", "tool_call_id": "edit_1", "content": "updated"}, | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "test_id", | |
| "type": "function", | |
| "function": { | |
| "name": "Bash", | |
| "arguments": { | |
| "command": "python3 -m unittest -v" | |
| }, | |
| }, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "test_id", | |
| "content": "Ran 3 tests\n\nOK", | |
| }, | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "edit_2", | |
| "type": "function", | |
| "function": {"name": "Edit", "arguments": "{}"}, | |
| } | |
| ], | |
| }, | |
| {"role": "tool", "tool_call_id": "edit_2", "content": "updated again"}, | |
| ] | |
| state = analyze_tool_flow(messages, [*TOOLS, EDIT_TOOL]) | |
| self.assertTrue(state.requires_tool) | |
| self.assertFalse(state.can_finalize) | |
| def test_successful_web_search_forces_synthesis_without_more_tools(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "search_id", | |
| "type": "function", | |
| "function": { | |
| "name": "WebSearch", | |
| "arguments": '{"query":"noticias RJ"}', | |
| }, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "search_id", | |
| "content": "Notícia atual — https://example.test/rj", | |
| }, | |
| ], | |
| [*TOOLS, EDIT_TOOL, *WEB_TOOLS], | |
| ) | |
| self.assertTrue(state.can_finalize) | |
| self.assertIsNone(resolve_tool_choice(None, state)) | |
| self.assertEqual(resolve_tool_choice("auto", state), "auto") | |
| self.assertIn("Do not repeat WebFetch", state.instruction or "") | |
| def test_webfetch_schema_error_requires_tool_search_without_evidence(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "fetch_id", | |
| "type": "function", | |
| "function": { | |
| "name": "WebFetch", | |
| "arguments": '{"url":"https://example.test"}', | |
| }, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "fetch_id", | |
| "content": ( | |
| "<tool_use_error>The required parameter `prompt` " | |
| "is missing</tool_use_error>" | |
| ), | |
| }, | |
| ], | |
| [WEB_TOOLS[0], WEB_TOOLS[2]], | |
| ) | |
| self.assertTrue(state.requires_tool) | |
| self.assertIn("select:WebFetch", state.instruction or "") | |
| self.assertEqual( | |
| resolve_tool_choice(None, state), | |
| { | |
| "type": "function", | |
| "function": {"name": "ToolSearch"}, | |
| }, | |
| ) | |
| def test_new_real_user_message_resets_completed_flow(self) -> None: | |
| history = [ | |
| { | |
| "role": "user", | |
| "content": "Implemente a solução.", | |
| }, | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "edit_old", | |
| "type": "function", | |
| "function": {"name": "Edit", "arguments": "{}"}, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "edit_old", | |
| "content": "updated", | |
| }, | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "test_old", | |
| "type": "function", | |
| "function": { | |
| "name": "Bash", | |
| "arguments": { | |
| "command": "python3 -m unittest -v" | |
| }, | |
| }, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "test_old", | |
| "content": "Ran 2 tests\n\nOK", | |
| }, | |
| {"role": "user", "content": "Agora implemente outra funcionalidade."}, | |
| ] | |
| state = analyze_tool_flow(history, [*TOOLS, EDIT_TOOL]) | |
| self.assertTrue(state.requires_tool) | |
| self.assertFalse(state.can_finalize) | |
| self.assertEqual(resolve_tool_choice(None, state), "required") | |
| def test_synthetic_continuation_does_not_reset_dirty_flow(self) -> None: | |
| history = [ | |
| {"role": "user", "content": "Implemente a solução."}, | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "read_id", | |
| "type": "function", | |
| "function": {"name": "Read", "arguments": "{}"}, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "read_id", | |
| "content": "source", | |
| }, | |
| { | |
| "role": "user", | |
| "content": ( | |
| "Continue with the task. If you were interrupted, " | |
| "resume your thought." | |
| ), | |
| }, | |
| ] | |
| self.assertTrue( | |
| analyze_tool_flow(history, [*TOOLS, EDIT_TOOL]).requires_tool | |
| ) | |
| def test_zero_failures_and_status_200_are_not_errors(self) -> None: | |
| history = [ | |
| {"role": "user", "content": "Implemente e teste."}, | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "edit_id", | |
| "type": "function", | |
| "function": {"name": "Edit", "arguments": "{}"}, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "edit_id", | |
| "content": "updated", | |
| }, | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "test_id", | |
| "type": "function", | |
| "function": { | |
| "name": "Bash", | |
| "arguments": {"command": "pytest -q"}, | |
| }, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "test_id", | |
| "content": "5 passed, 0 failed, 0 errors; status code 200", | |
| }, | |
| ] | |
| self.assertTrue( | |
| analyze_tool_flow(history, [*TOOLS, EDIT_TOOL]).can_finalize | |
| ) | |
| def test_first_bash_inspection_requires_continuation(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| {"role": "user", "content": "Implemente a solução."}, | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "ls_id", | |
| "type": "function", | |
| "function": { | |
| "name": "Bash", | |
| "arguments": {"command": "ls -la"}, | |
| }, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "ls_id", | |
| "content": "solution.py\ntest_solution.py", | |
| }, | |
| ], | |
| [*TOOLS, EDIT_TOOL], | |
| ) | |
| self.assertTrue(state.requires_tool) | |
| def test_parallel_edit_and_test_do_not_count_as_causal_verification(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| {"role": "user", "content": "Implemente e teste."}, | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "edit_parallel", | |
| "type": "function", | |
| "function": {"name": "Edit", "arguments": "{}"}, | |
| }, | |
| { | |
| "id": "test_parallel", | |
| "type": "function", | |
| "function": { | |
| "name": "Bash", | |
| "arguments": {"command": "pytest -q"}, | |
| }, | |
| }, | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "edit_parallel", | |
| "content": "updated", | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "test_parallel", | |
| "content": "5 passed", | |
| }, | |
| ], | |
| [*TOOLS, EDIT_TOOL], | |
| ) | |
| self.assertTrue(state.requires_tool) | |
| self.assertFalse(state.can_finalize) | |
| def test_silent_test_script_is_positive_evidence(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| {"role": "user", "content": "Implemente e teste."}, | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "edit_id", | |
| "type": "function", | |
| "function": {"name": "Edit", "arguments": "{}"}, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "edit_id", | |
| "content": "updated", | |
| }, | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "script_id", | |
| "type": "function", | |
| "function": { | |
| "name": "Bash", | |
| "arguments": { | |
| "command": "bash test_solution.sh" | |
| }, | |
| }, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "script_id", | |
| "content": "Bash completed without textual output", | |
| }, | |
| ], | |
| [*TOOLS, EDIT_TOOL], | |
| ) | |
| self.assertTrue(state.can_finalize) | |
| def test_toolsearch_success_keeps_webfetch_recovery_pending(self) -> None: | |
| history = [ | |
| {"role": "user", "content": "Use WebFetch."}, | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "fetch_bad", | |
| "type": "function", | |
| "function": { | |
| "name": "WebFetch", | |
| "arguments": { | |
| "url": "https://example.test" | |
| }, | |
| }, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "fetch_bad", | |
| "content": "Invalid tool parameters: prompt is missing", | |
| }, | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "search_tool", | |
| "type": "function", | |
| "function": { | |
| "name": "ToolSearch", | |
| "arguments": { | |
| "query": "select:WebFetch" | |
| }, | |
| }, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "search_tool", | |
| "content": "WebFetch schema loaded", | |
| }, | |
| ] | |
| state = analyze_tool_flow(history, WEB_TOOLS) | |
| self.assertTrue(state.requires_tool) | |
| self.assertEqual(state.forced_tool, "WebFetch") | |
| self.assertEqual( | |
| resolve_tool_choice(None, state)["function"]["name"], | |
| "WebFetch", | |
| ) | |
| def test_read_only_error_does_not_activate_agentic_gate(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| {"role": "user", "content": "Leia o arquivo."}, | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "read_bad", | |
| "type": "function", | |
| "function": { | |
| "name": "Read", | |
| "arguments": {"file_path": "/missing"}, | |
| }, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "read_bad", | |
| "content": "No such file", | |
| }, | |
| ], | |
| [TOOLS[0]], | |
| ) | |
| self.assertFalse(state.active) | |
| def test_initial_local_memory_inspection_forces_bash(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| { | |
| "role": "user", | |
| "content": "Verifique a memória RAM do notebook.", | |
| } | |
| ], | |
| TOOLS, | |
| ) | |
| self.assertTrue(state.requires_tool) | |
| self.assertEqual(state.forced_tool, "Bash") | |
| self.assertEqual( | |
| resolve_tool_choice(None, state)["function"]["name"], | |
| "Bash", | |
| ) | |
| def test_local_cat_inspection_can_finish_with_edit_tools_available(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| { | |
| "role": "user", | |
| "content": "Verifique a memória RAM do notebook.", | |
| }, | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "memory_id", | |
| "type": "function", | |
| "function": { | |
| "name": "Bash", | |
| "arguments": { | |
| "command": "cat /proc/meminfo | head" | |
| }, | |
| }, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "memory_id", | |
| "content": "MemTotal: 4023456 kB", | |
| }, | |
| ], | |
| [*TOOLS, EDIT_TOOL], | |
| ) | |
| self.assertFalse(state.requires_tool) | |
| self.assertFalse(state.can_finalize) | |
| def test_read_only_request_can_finish_with_edit_tools_available(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| {"role": "user", "content": "Leia o arquivo README.md."}, | |
| { | |
| "role": "assistant", | |
| "tool_calls": [ | |
| { | |
| "id": "read_only_id", | |
| "type": "function", | |
| "function": { | |
| "name": "Read", | |
| "arguments": {"file_path": "README.md"}, | |
| }, | |
| } | |
| ], | |
| }, | |
| { | |
| "role": "tool", | |
| "tool_call_id": "read_only_id", | |
| "content": "Documentação do projeto.", | |
| }, | |
| ], | |
| [*TOOLS, EDIT_TOOL], | |
| ) | |
| self.assertFalse(state.requires_tool) | |
| self.assertTrue(state.can_finalize) | |
| def test_initial_current_news_request_forces_websearch(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| { | |
| "role": "user", | |
| "content": "Pesquise na web as últimas notícias do RJ.", | |
| } | |
| ], | |
| WEB_TOOLS, | |
| ) | |
| self.assertTrue(state.requires_tool) | |
| self.assertEqual(state.forced_tool, "WebSearch") | |
| def test_initial_programming_request_requires_a_tool(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| { | |
| "role": "user", | |
| "content": "Corrija o código e rode os testes.", | |
| } | |
| ], | |
| [*TOOLS, EDIT_TOOL], | |
| ) | |
| self.assertTrue(state.requires_tool) | |
| self.assertIsNone(state.forced_tool) | |
| self.assertEqual(resolve_tool_choice(None, state), "required") | |
| def test_do_it_now_followup_requires_a_tool(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| {"role": "user", "content": "Mostre como verificar a RAM."}, | |
| { | |
| "role": "assistant", | |
| "content": "Você pode executar free -h.", | |
| }, | |
| {"role": "user", "content": "Faça isso agora."}, | |
| ], | |
| TOOLS, | |
| ) | |
| self.assertTrue(state.requires_tool) | |
| self.assertEqual( | |
| resolve_tool_choice("auto", state)["function"]["name"], | |
| "Bash", | |
| ) | |
| def test_openclaude_auto_keeps_tools_visible_for_unclassified_task(self) -> None: | |
| state = analyze_tool_flow( | |
| [{"role": "user", "content": "Compare these two design options."}], | |
| [*TOOLS, EDIT_TOOL], | |
| ) | |
| self.assertFalse(state.requires_tool) | |
| self.assertIsNone(resolve_tool_choice(None, state)) | |
| self.assertEqual(resolve_tool_choice("auto", state), "auto") | |
| tools, mode = select_tools(TOOLS, resolve_tool_choice("auto", state)) | |
| self.assertEqual(mode, "auto") | |
| self.assertEqual(len(tools), len(TOOLS)) | |
| def test_repository_summary_requires_real_inspection(self) -> None: | |
| state = analyze_tool_flow( | |
| [{"role": "user", "content": "Summarize this repository structure."}], | |
| [*TOOLS, EDIT_TOOL], | |
| ) | |
| self.assertTrue(state.active) | |
| self.assertTrue(state.requires_tool) | |
| self.assertEqual(resolve_tool_choice("auto", state), "required") | |
| tools, mode = select_tools(TOOLS, resolve_tool_choice("auto", state)) | |
| self.assertEqual(mode, "required") | |
| self.assertTrue(tools) | |
| def test_repository_summary_prefers_glob_when_openclaude_advertises_it(self) -> None: | |
| all_tools = [*TOOLS, GLOB_TOOL, EDIT_TOOL] | |
| state = analyze_tool_flow( | |
| [{"role": "user", "content": "Summarize this repository structure."}], | |
| all_tools, | |
| ) | |
| choice = resolve_tool_choice("auto", state) | |
| self.assertTrue(state.requires_tool) | |
| self.assertEqual(state.forced_tool, "Glob") | |
| self.assertEqual(choice["function"]["name"], "Glob") | |
| selected, mode = select_tools(all_tools, choice) | |
| self.assertEqual(mode, "forced") | |
| self.assertEqual([tool["function"]["name"] for tool in selected], ["Glob"]) | |
| def test_old_user_no_tools_instruction_does_not_poison_future_turn(self) -> None: | |
| state = analyze_tool_flow( | |
| [ | |
| {"role": "user", "content": "Não use ferramentas; explique só em texto."}, | |
| {"role": "assistant", "content": "Certo."}, | |
| {"role": "user", "content": "Agora analise este repositório."}, | |
| ], | |
| [*TOOLS, EDIT_TOOL], | |
| ) | |
| self.assertTrue(state.requires_tool) | |
| self.assertEqual(resolve_tool_choice("auto", state), "required") | |
| def test_current_user_no_tools_instruction_still_disables_tools(self) -> None: | |
| state = analyze_tool_flow( | |
| [{"role": "user", "content": "Analise este repositório sem ferramentas."}], | |
| [*TOOLS, EDIT_TOOL], | |
| ) | |
| self.assertTrue(state.can_finalize) | |
| self.assertEqual(resolve_tool_choice("auto", state), "none") | |
| def test_initial_gate_does_not_force_tools_for_plain_conversation(self) -> None: | |
| for prompt in ("Olá, tudo bem?", "Escreva um poema curto."): | |
| with self.subTest(prompt=prompt): | |
| state = analyze_tool_flow( | |
| [{"role": "user", "content": prompt}], | |
| [*TOOLS, EDIT_TOOL, *WEB_TOOLS], | |
| ) | |
| self.assertFalse(state.active) | |
| self.assertIsNone(resolve_tool_choice(None, state)) | |
| self.assertEqual(resolve_tool_choice("auto", state), "auto") | |
| class ResearchPersistenceFlowTests(unittest.TestCase): | |
| def _tools(self): | |
| return [ | |
| { | |
| "type": "function", | |
| "function": {"name": "WebSearch", "description": "Search web", "parameters": {"type": "object"}}, | |
| }, | |
| { | |
| "type": "function", | |
| "function": {"name": "Write", "description": "Write file", "parameters": {"type": "object"}}, | |
| }, | |
| ] | |
| def test_web_evidence_for_save_request_forces_write(self): | |
| messages = [ | |
| {"role": "user", "content": "pesquise ultimas noticias do rj e salve como txt"}, | |
| { | |
| "role": "assistant", | |
| "content": None, | |
| "tool_calls": [{ | |
| "id": "call_search", | |
| "type": "function", | |
| "function": {"name": "WebSearch", "arguments": '{"query":"ultimas noticias RJ"}'}, | |
| }], | |
| }, | |
| {"role": "tool", "tool_call_id": "call_search", "name": "WebSearch", "content": "Noticia A\nNoticia B"}, | |
| ] | |
| state = analyze_tool_flow(messages, self._tools()) | |
| self.assertTrue(state.requires_tool) | |
| self.assertEqual(state.forced_tool, "Write") | |
| choice = resolve_tool_choice("auto", state) | |
| self.assertEqual(choice["function"]["name"], "Write") | |
| def test_successful_write_finishes_research_save_request(self): | |
| messages = [ | |
| {"role": "user", "content": "pesquise ultimas noticias do rj e salve como txt"}, | |
| { | |
| "role": "assistant", | |
| "content": None, | |
| "tool_calls": [{ | |
| "id": "call_search", | |
| "type": "function", | |
| "function": {"name": "WebSearch", "arguments": '{"query":"ultimas noticias RJ"}'}, | |
| }], | |
| }, | |
| {"role": "tool", "tool_call_id": "call_search", "name": "WebSearch", "content": "Noticia A"}, | |
| { | |
| "role": "assistant", | |
| "content": None, | |
| "tool_calls": [{ | |
| "id": "call_write", | |
| "type": "function", | |
| "function": {"name": "Write", "arguments": '{"file_path":"noticias_rj.txt","content":"Noticia A"}'}, | |
| }], | |
| }, | |
| {"role": "tool", "tool_call_id": "call_write", "name": "Write", "content": "Wrote noticias_rj.txt"}, | |
| ] | |
| state = analyze_tool_flow(messages, self._tools()) | |
| self.assertTrue(state.can_finalize) | |
| self.assertFalse(state.requires_tool) | |
| self.assertIn("saved", state.reason) | |
| if __name__ == "__main__": | |
| unittest.main() | |