test_openclaude_compat.py

"""Tests for the notebook-independent OpenClaude adapter."""

from __future__ import annotations

import unittest

from openclaude_compat import (
    TOOL_PROTOCOL_MARKER,
    TOOL_RECAP_CHARACTERS,
    add_system_instruction,
    has_tool_protocol,
    normalize_openclaude_messages,
)
from openai_compat import tool_protocol_instruction


TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "WebFetch",
            "description": "Fetch a page.",
            "parameters": {
                "type": "object",
                "properties": {
                    "url": {"type": "string"},
                    "prompt": {"type": "string"},
                },
                "required": ["url", "prompt"],
            },
        },
    }
]


class OpenClaudeCompatibilityTests(unittest.TestCase):
    def test_parallel_results_are_mapped_by_id_and_stay_contiguous(self) -> None:
        normalized = normalize_openclaude_messages(
            [
                {"role": "user", "content": "Faça."},
                {
                    "role": "assistant",
                    "content": "",
                    "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": "1→source",
                },
            ]
        )
        self.assertEqual(
            [message["role"] for message in normalized],
            ["user", "assistant", "tool", "tool", "user"],
        )
        self.assertEqual(normalized[2]["name"], "Bash")
        self.assertEqual(normalized[3]["name"], "Read")
        self.assertIn("Bash result:\n/root", normalized[4]["content"])
        self.assertIn("source", normalized[4]["content"])
        self.assertNotIn("1→", normalized[4]["content"])

    def test_read_recap_is_bounded_and_preserves_head_and_tail(self) -> None:
        content = "\n".join(
            f"{index}→line-{index}" for index in range(3000)
        )
        normalized = normalize_openclaude_messages(
            [
                {"role": "user", "content": "Leia."},
                {
                    "role": "assistant",
                    "content": "",
                    "tool_calls": [
                        {
                            "id": "read_id",
                            "type": "function",
                            "function": {
                                "name": "Read",
                                "arguments": '{"file_path":"/tmp/large.txt"}',
                            },
                        }
                    ],
                },
                {
                    "role": "tool",
                    "tool_call_id": "read_id",
                    "content": content,
                },
            ]
        )
        recap = normalized[-1]["content"]
        self.assertLess(len(recap), TOOL_RECAP_CHARACTERS + 100)
        self.assertIn("line-0", recap)
        self.assertIn("line-2999", recap)
        self.assertIn("characters omitted", recap)

    def test_unknown_tool_call_id_is_client_error(self) -> None:
        with self.assertRaisesRegex(ValueError, "unknown tool_call_id"):
            normalize_openclaude_messages(
                [
                    {
                        "role": "tool",
                        "tool_call_id": "missing",
                        "content": "result",
                    }
                ]
            )

    def test_continuation_nudge_and_system_reminder_are_removed(self) -> None:
        normalized = normalize_openclaude_messages(
            [
                {"role": "user", "content": "Faça."},
                {
                    "role": "user",
                    "content": (
                        "<system-reminder>internal</system-reminder>"
                        "Continue with the task. If you were interrupted, "
                        "resume your thought."
                    ),
                },
            ]
        )
        self.assertEqual(normalized, [{"role": "user", "content": "Faça."}])

    def test_protocol_keeps_webfetch_constraint_without_schema_duplication(self) -> None:
        instruction = tool_protocol_instruction(TOOLS)
        self.assertIn(TOOL_PROTOCOL_MARKER, instruction)
        self.assertIn("WebFetch requires both url and prompt", instruction)
        self.assertIn("Available tool names:", instruction)
        self.assertNotIn('"parameters":', instruction)

    def test_protocol_does_not_call_unlisted_toolsearch(self) -> None:
        instruction = tool_protocol_instruction(
            [
                {
                    "type": "function",
                    "function": {
                        "name": "Bash",
                        "description": "Run a command.",
                        "parameters": {"type": "object"},
                    },
                }
            ]
        )
        self.assertIn("Deferred tools are unavailable", instruction)
        self.assertNotIn("ToolSearch", instruction)

    def test_instruction_is_inserted_before_latest_user(self) -> None:
        prepared = add_system_instruction(
            [
                {"role": "system", "content": "base"},
                {"role": "user", "content": "first"},
                {"role": "assistant", "content": "reply"},
                {"role": "user", "content": "latest"},
            ],
            "policy",
        )
        self.assertEqual(prepared[-2], {"role": "system", "content": "policy"})
        self.assertEqual(prepared[-1]["content"], "latest")

    def test_existing_protocol_is_detected(self) -> None:
        self.assertTrue(
            has_tool_protocol(
                [{"role": "system", "content": TOOL_PROTOCOL_MARKER}]
            )
        )
        self.assertFalse(has_tool_protocol([{"role": "user", "content": "oi"}]))


if __name__ == "__main__":
    unittest.main()