openclaude_compat.py

"""OpenClaude-specific prompting and message normalization for the Space.

The Space owns this adapter so notebook clients can connect directly to its
OpenAI-compatible endpoint.  No conversation state is stored in the process;
all decisions are reconstructed from the request history.
"""

from __future__ import annotations

import os
import re
from collections.abc import Mapping
from typing import Any

from tool_calls import normalize_openai_tool_arguments


TOOL_PROTOCOL_MARKER = "OPENAI TOOL CALL FORMAT — MANDATORY"
TOOL_RECAP_CHARACTERS = int(os.getenv("TOOL_RECAP_CHARACTERS", "6000"))
SYSTEM_REMINDER_RE = re.compile(
    r"<system-reminder\b[^>]*>.*?</system-reminder>",
    re.DOTALL | re.IGNORECASE,
)


def _content_text(content: Any) -> str:
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        return "\n".join(
            str(block.get("text", ""))
            for block in content
            if isinstance(block, Mapping)
            and block.get("type") in {"text", "input_text"}
        )
    return "" if content is None else str(content)


def _tool_name(call: Mapping[str, Any]) -> str | None:
    function = call.get("function")
    if not isinstance(function, Mapping):
        return None
    name = function.get("name")
    return name if isinstance(name, str) and name else None


def _is_continuation_nudge(text: str) -> bool:
    folded = text.casefold()
    return (
        "<system-reminder>" in folded
        or (
            "continue with the task" in folded
            and "resume your thought" in folded
        )
    )


def _strip_system_reminders(text: str) -> str:
    cleaned = SYSTEM_REMINDER_RE.sub("", str(text))
    return re.sub(r"\n{3,}", "\n\n", cleaned).strip()


def _bound_recap(text: str) -> str:
    """Keep evidence recaps bounded so one tool result cannot dominate context."""
    limit = max(256, TOOL_RECAP_CHARACTERS)
    if len(text) <= limit:
        return text
    head = limit * 2 // 3
    tail = limit - head
    return (
        text[:head]
        + f"\n...[{len(text) - limit} characters omitted]...\n"
        + text[-tail:]
    )


def _read_recap(content: str) -> str:
    lines: list[str] = []
    for raw_line in _strip_system_reminders(content).splitlines():
        line = raw_line.strip()
        if not line or line.startswith("<system-reminder"):
            continue
        match = re.match(r"^\d+→\s*(.*)$", line)
        if match:
            line = match.group(1).strip()
        if line:
            lines.append(line)
    return _bound_recap("\n".join(lines).strip())


def _tool_recap(tool_name: str, content: str) -> str:
    cleaned = _strip_system_reminders(content)
    if not cleaned:
        return f"{tool_name} completed without textual output."
    return f"{tool_name} result:\n{_bound_recap(cleaned)}"


def normalize_openclaude_messages(messages: object) -> list[dict[str, Any]]:
    """Preserve native tool history and add bounded evidence recaps.

    OpenClaude may return parallel results in a different order from the calls.
    Results are therefore matched by ``tool_call_id`` rather than by position.
    The recap is emitted only after the whole result batch, so parallel tool
    messages remain contiguous for Qwen's chat template.
    """
    if not isinstance(messages, list):
        raise ValueError("messages must be a list")

    normalized: list[dict[str, Any]] = []
    pending_by_id: dict[str, str] = {}
    pending_order: list[str] = []
    pending_recaps: list[str] = []
    generated_call_number = 0

    def flush_recaps() -> None:
        if not pending_recaps:
            return
        normalized.append(
            {
                "role": "user",
                "content": "[Tool results received]\n"
                + "\n\n".join(pending_recaps),
            }
        )
        pending_recaps.clear()

    for raw_message in messages:
        if not isinstance(raw_message, Mapping):
            raise ValueError("each message must be an object")
        message = dict(raw_message)
        raw_role = str(message.get("role", "user")).casefold()
        content = _content_text(message.get("content"))

        if raw_role != "tool":
            flush_recaps()

        if raw_role in {"system", "developer"}:
            normalized.append({"role": "system", "content": content})
            continue

        if raw_role == "assistant":
            calls: list[dict[str, Any]] = []
            raw_calls = message.get("tool_calls")
            if not isinstance(raw_calls, list):
                raw_calls = []
            for raw_call in raw_calls:
                if not isinstance(raw_call, Mapping):
                    continue
                name = _tool_name(raw_call)
                if not name:
                    continue
                generated_call_number += 1
                call_id = raw_call.get("id")
                if not isinstance(call_id, str) or not call_id:
                    call_id = f"call_normalized_{generated_call_number}"
                if call_id in pending_by_id:
                    raise ValueError(f"duplicate tool_call id: {call_id}")
                function = raw_call.get("function")
                arguments = (
                    function.get("arguments", {})
                    if isinstance(function, Mapping)
                    else {}
                )
                calls.append(
                    {
                        "id": call_id,
                        "type": "function",
                        "function": {
                            "name": name,
                            "arguments": normalize_openai_tool_arguments(
                                arguments
                            ),
                        },
                    }
                )
                pending_by_id[call_id] = name
                pending_order.append(call_id)

            if content and (
                "[tool results received]" in content.casefold()
                or _is_continuation_nudge(content)
            ):
                continue
            normalized.append(
                {
                    "role": "assistant",
                    "content": content if content else None,
                    **({"tool_calls": calls} if calls else {}),
                }
            )
            continue

        if raw_role == "tool":
            call_id = message.get("tool_call_id")
            tool_name: str | None = None
            if isinstance(call_id, str) and call_id:
                tool_name = pending_by_id.pop(call_id, None)
                if tool_name is None:
                    explicit_name = message.get("name")
                    if isinstance(explicit_name, str) and explicit_name:
                        tool_name = explicit_name
                    else:
                        raise ValueError(
                            "tool result references unknown tool_call_id: "
                            f"{call_id}"
                        )
                if call_id in pending_order:
                    pending_order.remove(call_id)
            elif pending_order:
                call_id = pending_order.pop(0)
                tool_name = pending_by_id.pop(call_id)
            else:
                explicit_name = message.get("name")
                if not isinstance(explicit_name, str) or not explicit_name:
                    raise ValueError("tool result is missing tool_call_id")
                tool_name = explicit_name
                call_id = None

            entry: dict[str, Any] = {
                "role": "tool",
                "name": tool_name,
                "content": content,
            }
            if isinstance(call_id, str) and call_id:
                entry["tool_call_id"] = call_id
            normalized.append(entry)
            recap = (
                _read_recap(content)
                if tool_name.casefold() == "read"
                else _tool_recap(tool_name, content)
            )
            if recap:
                pending_recaps.append(recap)
            continue

        original_content = content
        content = _strip_system_reminders(content)
        if original_content and not content:
            continue
        if _is_continuation_nudge(content):
            continue
        normalized.append({"role": "user", "content": content})

    flush_recaps()
    return normalized


def has_tool_protocol(messages: object) -> bool:
    if not isinstance(messages, list):
        return False
    return any(
        isinstance(message, Mapping)
        and TOOL_PROTOCOL_MARKER in _content_text(message.get("content"))
        for message in messages
    )


def add_system_instruction(
    messages: list[dict[str, Any]], instruction: str | None
) -> list[dict[str, Any]]:
    """Insert request-local instructions near the current user turn."""
    if not instruction:
        return messages
    prepared = list(messages)
    insert_at = 0
    for index in range(len(prepared) - 1, -1, -1):
        if prepared[index].get("role") == "user":
            insert_at = index
            break
    prepared.insert(insert_at, {"role": "system", "content": instruction})
    return prepared