"""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": (
"\nBash\n"
"\n"
"Create code and run tests.\n"
"ola\nsnip_id=x"
),
}
]
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": (
"\nWebSearch\n"
"\n"
"Use this skill to create code "
"and run tests.\n"
"oi\n"
"snip_id=abc"
),
}
],
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": (
"The required parameter `prompt` "
"is missing"
),
},
],
[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()