claude2openai.py

""""
  Copyright [2006] [Naresh Mehta] https://www.naresh.se

  Licensed under the Apache License, Version 2.0 (the "License");
  you may not use this file except in compliance with the License.
  You may obtain a copy of the License at

      http://www.apache.org/licenses/LICENSE-2.0

  Unless required by applicable law or agreed to in writing, software
  distributed under the License is distributed on an "AS IS" BASIS,
  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  See the License for the specific language governing permissions and
  limitations under the License.


Claude-to-OpenAI request/response translation server.

This server accepts Claude-style requests on `/v1/messages`, converts them to
OpenAI-compatible chat completions, forwards them to a configured upstream,
then converts the response back to Claude format.
"""

import json
import logging
import os
import sys
import time
import tomllib
import urllib.error
import urllib.parse
import urllib.request
import uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Any, Iterable

LOGGER = logging.getLogger("claude2openai")


def _configure_logging(config: dict[str, Any]) -> None:
    """Configure structured logging based on the server config."""
    server_config = config.get("server", {})
    level_name = str(server_config.get("log_level", "INFO")).upper()
    level = getattr(logging, level_name, logging.INFO)
    logging.basicConfig(
        level=level,
        format="%(asctime)s %(levelname)s %(name)s %(message)s",
        datefmt="%Y-%m-%dT%H:%M:%S%z",
    )


def _read_json_body(handler: BaseHTTPRequestHandler) -> dict[str, Any]:
    """Decode the request body JSON into a dictionary."""
    content_length = int(handler.headers.get("Content-Length", "0"))
    raw_body = handler.rfile.read(content_length)
    if not raw_body:
        return {}
    return json.loads(raw_body.decode("utf-8"))


def _log_preview(label: str, payload: dict[str, Any], enabled: bool, limit: int) -> None:
    """Log a truncated JSON preview for tracing translations."""
    if not enabled:
        return
    try:
        rendered = json.dumps(payload, ensure_ascii=False)
    except TypeError:
        rendered = json.dumps({"_unserializable": str(payload)}, ensure_ascii=False)
    preview = rendered[:limit]
    LOGGER.info("%s: %s", label, preview)


def _write_json_response(
    handler: BaseHTTPRequestHandler, status: int, payload: dict[str, Any]
) -> None:
    """Send a JSON response with a status code."""
    data = json.dumps(payload).encode("utf-8")
    handler.send_response(status)
    handler.send_header("Content-Type", "application/json")
    handler.send_header("Content-Length", str(len(data)))
    handler.end_headers()
    handler.wfile.write(data)


def _write_sse_event(handler: BaseHTTPRequestHandler, payload: dict[str, Any]) -> None:
    """Write a single Server-Sent Event frame."""
    data = json.dumps(payload).encode("utf-8")
    event_type = payload.get("type")
    if event_type:
        handler.wfile.write(b"event: ")
        handler.wfile.write(str(event_type).encode("utf-8"))
        handler.wfile.write(b"\n")
    handler.wfile.write(b"data: ")
    handler.wfile.write(data)
    handler.wfile.write(b"\n\n")
    handler.wfile.flush()


def _log_sse_event(payload: dict[str, Any], enabled: bool, limit: int) -> None:
    """Log SSE events for debugging client formatting."""
    if not enabled:
        return
    try:
        rendered = json.dumps(payload, ensure_ascii=False)
    except TypeError:
        rendered = json.dumps({"_unserializable": str(payload)}, ensure_ascii=False)
    LOGGER.info("SSE event: %s", rendered[:limit])


def _looks_like_tool_json(text: str) -> bool:
    """Heuristic to detect tool-call JSON rendered as text."""
    snippet = text.strip()
    if snippet.startswith("{"):
        unescaped = snippet.replace('\\"', '"')
    else:
        unescaped = snippet.replace('\\"', '"').lstrip()
    if not unescaped.startswith("{"):
        return False
    has_type = '"type"' in unescaped and '"function"' in unescaped
    has_name = '"name"' in unescaped
    has_params = '"parameters"' in unescaped
    has_top_level_function = '"type"' in unescaped and '"function"' in unescaped and has_name and has_params
    has_tool_calls = '"tool_calls"' in unescaped and '"function"' in unescaped
    has_question_tool = '"AskUserQuestion"' in unescaped
    return (
        (has_type and (has_name or has_params))
        or has_top_level_function
        or has_tool_calls
        or has_question_tool
    )


def _looks_like_topic_json(text: str) -> bool:
    """Heuristic to detect topic-detection JSON rendered as text."""
    snippet = text.strip().replace('\\"', '"')
    return snippet.startswith("{") and '"isNewTopic"' in snippet and '"title"' in snippet


def _looks_like_cli_question_json(text: str) -> bool:
    """Heuristic for Claude CLI question payloads rendered as JSON text."""
    snippet = text.strip().replace('\\"', '"')
    has_header = '"header"' in snippet
    has_options = '"options"' in snippet
    has_multi = '"multiSelect"' in snippet or '"multi_select"' in snippet
    return has_header and has_options and has_multi


def _looks_like_cli_question_fragment(text: str) -> bool:
    """Detect partial CLI question JSON fragments."""
    snippet = text.replace('\\"', '"')
    return '"header"' in snippet or '"options"' in snippet or '"multiSelect"' in snippet

def _extract_text(content: Any) -> str:
    """Extract plain text from Claude-style content blocks."""
    if isinstance(content, str):
        return content
    if isinstance(content, dict):
        if content.get("type") == "text":
            return str(content.get("text", ""))
        return ""
    if isinstance(content, list):
        parts = []
        for block in content:
            if isinstance(block, str):
                parts.append(block)
                continue
            if isinstance(block, dict) and block.get("type") == "text":
                parts.append(str(block.get("text", "")))
        return "".join(parts)
    return ""


def _stringify_tool_input(value: Any) -> str:
    """Serialize tool input into a JSON string for OpenAI tool calls."""
    if value is None:
        return "{}"
    if isinstance(value, str):
        return value
    try:
        return json.dumps(value, separators=(",", ":"))
    except TypeError:
        return json.dumps({"_raw": str(value)}, separators=(",", ":"))


def _stringify_tool_result_content(content: Any) -> str:
    """Normalize tool result content into a string."""
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts: list[str] = []
        for block in content:
            if isinstance(block, str):
                parts.append(block)
                continue
            if not isinstance(block, dict):
                continue
            block_type = block.get("type")
            if block_type == "text":
                parts.append(str(block.get("text", "")))
            elif block_type == "json":
                if "json" in block:
                    parts.append(json.dumps(block.get("json"), separators=(",", ":")))
                elif "text" in block:
                    parts.append(str(block.get("text", "")))
        return "".join(parts)
    try:
        return json.dumps(content, separators=(",", ":"))
    except TypeError:
        return str(content)


def _normalize_openai_tools(tools: Any) -> list[dict[str, Any]] | None:
    """Convert Claude-style tools to OpenAI tool definitions."""
    if not isinstance(tools, list):
        return None
    normalized: list[dict[str, Any]] = []
    for tool in tools:
        if not isinstance(tool, dict):
            continue
        if tool.get("type") == "function" and "function" in tool:
            normalized.append(tool)
            continue
        name = tool.get("name")
        description = tool.get("description")
        parameters = tool.get("input_schema") or tool.get("parameters") or {"type": "object"}
        if name:
            normalized.append(
                {
                    "type": "function",
                    "function": {
                        "name": name,
                        "description": description,
                        "parameters": parameters,
                    },
                }
            )
    return normalized or None


def _normalize_tool_choice(tool_choice: Any) -> Any:
    """Convert Claude tool_choice into OpenAI-compatible tool_choice."""
    if not isinstance(tool_choice, dict):
        return tool_choice
    choice_type = tool_choice.get("type")
    if choice_type == "tool" and tool_choice.get("name"):
        return {"type": "function", "function": {"name": tool_choice["name"]}}
    return tool_choice


def _apply_tool_overrides(
    openai_payload: dict[str, Any],
    disable_tools: bool,
    tool_choice_override: str | None,
) -> None:
    """Apply tool configuration overrides to the OpenAI payload."""
    if disable_tools:
        openai_payload.pop("tools", None)
        openai_payload.pop("tool_choice", None)
        return

    if not tool_choice_override:
        return

    override = tool_choice_override.strip()
    if override in {"none", "auto", "required"}:
        openai_payload["tool_choice"] = override
        return

    openai_payload["tool_choice"] = {"type": "function", "function": {"name": override}}


def _as_bool(value: Any) -> bool:
    """Parse common truthy/falsey values from config."""
    if isinstance(value, bool):
        return value
    if value is None:
        return False
    if isinstance(value, (int, float)):
        return value != 0
    if isinstance(value, str):
        normalized = value.strip().lower()
        if normalized in {"true", "1", "yes", "y", "on"}:
            return True
        if normalized in {"false", "0", "no", "n", "off", "none", "null", ""}:
            return False
    return False


def _resolve_openai_model(
    payload_model: str | None,
    default_model: str | None,
    model_map: dict[str, str] | None,
) -> str | None:
    """Select an OpenAI model from a Claude model and optional mapping."""
    if payload_model and model_map:
        mapped = model_map.get(payload_model)
        if mapped:
            return mapped
        for pattern, target in model_map.items():
            if pattern.endswith("*") and payload_model.startswith(pattern[:-1]):
                return target
            if pattern.startswith("*") and payload_model.endswith(pattern[1:]):
                return target
    return default_model or payload_model


def _claude_to_openai(
    payload: dict[str, Any],
    default_model: str | None,
    model_map: dict[str, str] | None,
    disable_tools: bool,
    tool_choice_override: str | None,
    response_format_override: str | None,
    system_prompt_suffix: str | None,
    system_prompt_mode: str | None,
) -> dict[str, Any]:
    """Translate a Claude request payload into OpenAI chat completions format."""
    messages: list[dict[str, Any]] = []
    system_content = _extract_text(payload.get("system"))
    mode = (system_prompt_mode or "append").strip().lower()
    suffix = (system_prompt_suffix or "").strip()
    if mode == "replace":
        system_content = suffix
    elif mode == "strip":
        system_content = ""
    else:
        if suffix:
            if system_content:
                system_content = f"{system_content}\n\n{suffix}"
            else:
                system_content = suffix
    if system_content:
        messages.append({"role": "system", "content": system_content})

    for message in payload.get("messages", []):
        role = message.get("role", "user")
        content = message.get("content")
        if isinstance(content, list):
            text_parts: list[str] = []
            tool_calls: list[dict[str, Any]] = []
            tool_results: list[dict[str, Any]] = []
            for block in content:
                if isinstance(block, str):
                    text_parts.append(block)
                    continue
                if not isinstance(block, dict):
                    continue
                block_type = block.get("type")
                if block_type == "text":
                    text_parts.append(str(block.get("text", "")))
                elif block_type == "tool_use":
                    tool_calls.append(block)
                elif block_type == "tool_result":
                    tool_results.append(block)

            if text_parts or not tool_calls:
                messages.append({"role": role, "content": "".join(text_parts)})

            if tool_calls:
                assistant_tools = []
                for call in tool_calls:
                    call_id = call.get("id") or f"tool_{uuid.uuid4().hex}"
                    assistant_tools.append(
                        {
                            "id": call_id,
                            "type": "function",
                            "function": {
                                "name": call.get("name"),
                                "arguments": _stringify_tool_input(call.get("input")),
                            },
                        }
                    )
                messages.append({"role": "assistant", "tool_calls": assistant_tools})

            for result in tool_results:
                tool_call_id = result.get("tool_use_id") or result.get("id")
                if not tool_call_id:
                    tool_call_id = f"tool_{uuid.uuid4().hex}"
                    LOGGER.warning(
                        "tool_result missing tool_use_id; generated %s",
                        tool_call_id,
                    )
                result_content = _stringify_tool_result_content(result.get("content"))
                messages.append(
                    {
                        "role": "tool",
                        "tool_call_id": tool_call_id,
                        "content": result_content,
                    }
                )
            continue

        content_text = _extract_text(content)
        if role == "assistant":
            stripped = content_text.strip()
            if not stripped or stripped in {"{", "}", "{}"}:
                LOGGER.warning("Skipping empty assistant message content.")
                continue
            if (
                stripped.startswith("{")
                and (
                    _looks_like_tool_json(stripped)
                    or _looks_like_topic_json(stripped)
                    or _looks_like_cli_question_json(stripped)
                )
            ):
                LOGGER.warning("Skipping assistant JSON-like content block.")
                continue
        messages.append({"role": role, "content": content_text})

    openai_payload: dict[str, Any] = {
        "model": _resolve_openai_model(payload.get("model"), default_model, model_map),
        "messages": messages,
    }

    max_tokens = payload.get("max_tokens", payload.get("max_tokens_to_sample"))
    if max_tokens is not None:
        openai_payload["max_tokens"] = max_tokens

    for key in (
        "temperature",
        "top_p",
        "presence_penalty",
        "frequency_penalty",
        "n",
        "user",
        "seed",
        "logprobs",
        "top_logprobs",
    ):
        if key in payload:
            openai_payload[key] = payload[key]

    if "response_format" in payload:
        openai_payload["response_format"] = payload["response_format"]

    tools = _normalize_openai_tools(payload.get("tools"))
    if tools:
        openai_payload["tools"] = tools

    if "tool_choice" in payload:
        openai_payload["tool_choice"] = _normalize_tool_choice(payload.get("tool_choice"))

    _apply_tool_overrides(openai_payload, disable_tools, tool_choice_override)

    if response_format_override:
        openai_payload["response_format"] = {"type": response_format_override}

    stop_sequences = payload.get("stop_sequences")
    if stop_sequences:
        openai_payload["stop"] = stop_sequences

    if payload.get("stream") is True:
        openai_payload["stream"] = True

    return openai_payload


def _openai_to_claude(payload: dict[str, Any]) -> dict[str, Any]:
    """Translate an OpenAI chat completions response into Claude format."""
    choices = payload.get("choices", [])
    first_choice = choices[0] if choices else {}
    message = first_choice.get("message", {})
    content_text = message.get("content") or ""
    tool_calls = message.get("tool_calls") or []
    legacy_function_call = message.get("function_call")
    finish_reason = first_choice.get("finish_reason")
    usage = payload.get("usage", {})

    stop_reason = finish_reason
    if finish_reason == "stop":
        stop_reason = "end_turn"
    elif finish_reason == "length":
        stop_reason = "max_tokens"
    elif finish_reason == "tool_calls":
        stop_reason = "tool_use"

    content_blocks: list[dict[str, Any]] = []
    if content_text:
        content_blocks.append({"type": "text", "text": content_text})

    if legacy_function_call:
        tool_calls = [
            {
                "id": legacy_function_call.get("id") or "tool_call_0",
                "function": {
                    "name": legacy_function_call.get("name"),
                    "arguments": legacy_function_call.get("arguments"),
                },
            }
        ]

    for call in tool_calls:
        function = call.get("function", {})
        raw_args = function.get("arguments") or "{}"
        try:
            parsed_args = json.loads(raw_args)
        except json.JSONDecodeError:
            parsed_args = {"_raw": raw_args}
        content_blocks.append(
            {
                "type": "tool_use",
                "id": call.get("id") or f"tool_{uuid.uuid4().hex}",
                "name": function.get("name"),
                "input": parsed_args,
            }
        )

    return {
        "id": payload.get("id"),
        "type": "message",
        "role": "assistant",
        "model": payload.get("model"),
        "content": content_blocks,
        "stop_reason": stop_reason,
        "stop_sequence": None,
        "usage": {
            "input_tokens": usage.get("prompt_tokens"),
            "output_tokens": usage.get("completion_tokens"),
        },
    }


def _emit_sse_from_claude_message(
    handler: BaseHTTPRequestHandler, message: dict[str, Any], merge_text_blocks: bool
) -> None:
    """Emit a minimal Claude SSE sequence from a full message."""
    message_id = message.get("id") or f"msg_{uuid.uuid4().hex}"
    model = message.get("model")
    _write_sse_event(
        handler,
        {
            "type": "message_start",
            "message": {
                "id": message_id,
                "type": "message",
                "role": "assistant",
                "model": model,
                "content": [],
                "stop_reason": None,
                "stop_sequence": None,
                "usage": {"input_tokens": None, "output_tokens": None},
            },
        },
    )

    content_blocks = message.get("content") or []
    if merge_text_blocks:
        merged: list[dict[str, Any]] = []
        buffer: list[str] = []
        for block in content_blocks:
            if block.get("type") == "text":
                buffer.append(block.get("text", ""))
                continue
            if buffer:
                merged.append({"type": "text", "text": "".join(buffer)})
                buffer = []
            merged.append(block)
        if buffer:
            merged.append({"type": "text", "text": "".join(buffer)})
        content_blocks = merged
    for index, block in enumerate(content_blocks):
        block_type = block.get("type")
        _write_sse_event(
            handler,
            {
                "type": "content_block_start",
                "index": index,
                "content_block": {"type": block_type, **{k: v for k, v in block.items() if k != "type"}},
            },
        )
        if block_type == "text":
            text = block.get("text", "")
            if text:
                _write_sse_event(
                    handler,
                    {
                        "type": "content_block_delta",
                        "index": index,
                        "delta": {"type": "text_delta", "text": text},
                    },
                )
        elif block_type == "tool_use":
            _write_sse_event(
                handler, {"type": "content_block_delta", "index": index, "delta": {}}
            )
        _write_sse_event(handler, {"type": "content_block_stop", "index": index})

    usage = message.get("usage") or {}
    _write_sse_event(
        handler,
        {
            "type": "message_delta",
            "delta": {
                "stop_reason": message.get("stop_reason"),
                "stop_sequence": message.get("stop_sequence"),
            },
            "usage": {
                "input_tokens": usage.get("input_tokens"),
                "output_tokens": usage.get("output_tokens"),
            },
        },
    )
    _write_sse_event(handler, {"type": "message_stop"})


class UpstreamHTTPError(Exception):
    """HTTP error wrapper for upstream requests."""
    def __init__(self, status_code: int, detail: str) -> None:
        super().__init__(detail)
        self.status_code = status_code
        self.detail = detail


class UpstreamConnectionError(Exception):
    """Connection error wrapper for upstream requests."""
    def __init__(self, detail: str) -> None:
        super().__init__(detail)
        self.detail = detail


def _should_retry_http_error(status_code: int) -> bool:
    """Return True for retriable upstream HTTP errors."""
    return 500 <= status_code < 600


def _read_with_retries(
    request: urllib.request.Request,
    timeout: float,
    retries: int,
    backoff_seconds: float,
) -> urllib.response.addinfourl:
    """Read an upstream response with retries for transient failures."""
    attempt = 0
    while True:
        try:
            return urllib.request.urlopen(request, timeout=timeout)
        except urllib.error.HTTPError as exc:
            if _should_retry_http_error(exc.code) and attempt < retries:
                time.sleep(backoff_seconds * (2**attempt))
                attempt += 1
                continue
            try:
                error_body = exc.read().decode("utf-8")
            except Exception:
                error_body = exc.reason
            raise UpstreamHTTPError(exc.code, str(error_body)) from exc
        except urllib.error.URLError as exc:
            if attempt < retries:
                time.sleep(backoff_seconds * (2**attempt))
                attempt += 1
                continue
            raise UpstreamConnectionError(str(exc)) from exc


def _openai_request_json(
    openai_url: str,
    api_key: str,
    payload: dict[str, Any],
    timeout: float,
    retries: int,
    backoff_seconds: float,
    log_responses: bool,
    log_limit: int,
) -> dict[str, Any]:
    """Send a non-streaming OpenAI request and parse JSON."""
    request_data = json.dumps(payload).encode("utf-8")
    request = urllib.request.Request(openai_url, data=request_data, method="POST")
    request.add_header("Content-Type", "application/json")
    request.add_header("Authorization", f"Bearer {api_key}")

    with _read_with_retries(request, timeout, retries, backoff_seconds) as response:
        response_body = response.read().decode("utf-8")
        if log_responses:
            preview = response_body[:log_limit]
            LOGGER.info("Upstream response body: %s", preview)
        return json.loads(response_body)


def _iter_openai_stream_lines(response: urllib.response.addinfourl) -> Iterable[str]:
    """Yield non-empty lines from an OpenAI stream response."""
    while True:
        line = response.readline()
        if not line:
            break
        decoded = line.decode("utf-8").strip()
        if decoded:
            yield decoded


def _stream_openai_to_claude(
    handler: BaseHTTPRequestHandler,
    openai_url: str,
    api_key: str,
    payload: dict[str, Any],
    timeout: float,
    retries: int,
    backoff_seconds: float,
    log_responses: bool,
    log_limit: int,
    merge_text_blocks: bool,
    strict_stream_text_blocks: bool,
    drop_empty_text_deltas: bool,
    stream_sentence_buffer: bool,
    log_translations: bool,
    log_translation_limit: int,
    log_sse_events: bool,
    log_sse_limit: int,
    filter_tool_json_text: bool,
    filter_topic_json_text: bool,
    empty_stream_fallback: str,
    empty_stream_fallback_text: str | None,
) -> None:
    """Relay OpenAI streaming responses as Claude SSE events."""
    request_data = json.dumps(payload).encode("utf-8")
    request = urllib.request.Request(openai_url, data=request_data, method="POST")
    request.add_header("Content-Type", "application/json")
    request.add_header("Authorization", f"Bearer {api_key}")

    message_id = f"msg_{uuid.uuid4().hex}"
    active_message_id: str | None = None
    model = payload.get("model")
    started = False
    finished = False
    stop_reason = None
    output_tokens = None
    usage_output_tokens = None
    usage_input_tokens = None
    last_usage: tuple[int | None, int | None] | None = None
    text_block_index: int | None = None
    text_buffer: list[str] = []
    translated_text_parts: list[str] = []
    suppress_cli_json = False
    suppressed_text_parts: list[str] = []
    tool_block_indices: dict[int, int] = {}
    next_block_index = 0

    def _ensure_started() -> None:
        nonlocal started
        if started:
            return
        _write_sse_event(
            handler,
            {
                "type": "message_start",
                "message": {
                    "id": message_id,
                    "type": "message",
                    "role": "assistant",
                    "model": model,
                    "content": [],
                    "stop_reason": None,
                    "stop_sequence": None,
                    "usage": {"input_tokens": None, "output_tokens": None},
                },
            },
        )
        _log_sse_event(
            {
                "type": "message_start",
                "message": {
                    "id": message_id,
                    "type": "message",
                    "role": "assistant",
                    "model": model,
                    "content": [],
                    "stop_reason": None,
                    "stop_sequence": None,
                    "usage": {"input_tokens": None, "output_tokens": None},
                },
            },
            log_sse_events,
            log_sse_limit,
        )
        started = True

    with _read_with_retries(request, timeout, retries, backoff_seconds) as response:
        content_type = response.headers.get("Content-Type", "")
        if "text/event-stream" not in content_type:
            response_body = response.read().decode("utf-8")
            if log_responses:
                preview = response_body[:log_limit]
                LOGGER.info("Upstream response body: %s", preview)
            openai_response = json.loads(response_body)
            claude_response = _openai_to_claude(openai_response)
            _log_preview(
                "Translated OpenAI -> Claude",
                claude_response,
                log_translations,
                log_translation_limit,
            )
            _emit_sse_from_claude_message(handler, claude_response, merge_text_blocks)
            return
        for line in _iter_openai_stream_lines(response):
            if finished:
                break
            if not line.startswith("data:"):
                continue
            data = line[len("data:") :].strip()
            if data == "[DONE]":
                break
            if log_responses:
                preview = data[:log_limit]
                LOGGER.info("Upstream stream chunk: %s", preview)
            try:
                chunk = json.loads(data)
            except json.JSONDecodeError:
                continue

            usage = chunk.get("usage") or {}
            if usage:
                usage_output_tokens = usage.get("completion_tokens", usage_output_tokens)
                usage_input_tokens = usage.get("prompt_tokens", usage_input_tokens)

            chunk_id = chunk.get("id")
            if active_message_id is None and chunk_id:
                active_message_id = chunk_id
            if active_message_id and chunk_id and chunk_id != active_message_id and started:
                continue
            if not started:
                message_id = chunk_id or message_id
                model = chunk.get("model", model)

            choices = chunk.get("choices", [])
            if not choices:
                if usage:
                    current_usage = (usage_input_tokens, usage_output_tokens)
                    if current_usage != last_usage:
                        last_usage = current_usage
                        _write_sse_event(
                            handler,
                            {
                                "type": "message_delta",
                                "delta": {"stop_reason": None, "stop_sequence": None},
                                "usage": {
                                    "input_tokens": usage_input_tokens,
                                    "output_tokens": usage_output_tokens,
                                },
                            },
                        )
                        _log_sse_event(
                            {
                                "type": "message_delta",
                                "delta": {"stop_reason": None, "stop_sequence": None},
                                "usage": {
                                    "input_tokens": usage_input_tokens,
                                    "output_tokens": usage_output_tokens,
                                },
                            },
                            log_sse_events,
                            log_sse_limit,
                        )
                continue
            choice = choices[0]
            delta = choice.get("delta", {})
            if "content" in delta and delta["content"] is not None:
                if drop_empty_text_deltas and delta["content"] == "":
                    continue
                if stream_sentence_buffer:
                    text_buffer.append(delta["content"])
                    buffered = "".join(text_buffer)
                    if filter_tool_json_text and _looks_like_cli_question_fragment(buffered):
                        suppress_cli_json = True
                    if suppress_cli_json:
                        suppressed_text_parts.append(buffered)
                        continue
                    if any(token in buffered for token in (".", "!", "?", "\n")):
                        text_buffer = []
                        if filter_tool_json_text and (
                            _looks_like_tool_json(buffered)
                            or _looks_like_topic_json(buffered)
                            or _looks_like_cli_question_json(buffered)
                        ):
                            suppress_cli_json = True
                            suppressed_text_parts.append(buffered)
                            continue
                        if text_block_index is None:
                            text_block_index = next_block_index
                            next_block_index += 1
                            _ensure_started()
                            _write_sse_event(
                                handler,
                                {
                                    "type": "content_block_start",
                                    "index": text_block_index,
                                    "content_block": {"type": "text", "text": ""},
                                },
                            )
                            _log_sse_event(
                                {
                                    "type": "content_block_start",
                                    "index": text_block_index,
                                    "content_block": {"type": "text", "text": ""},
                                },
                                log_sse_events,
                                log_sse_limit,
                            )
                        output_tokens = (output_tokens or 0) + 1
                        _write_sse_event(
                            handler,
                            {
                                "type": "content_block_delta",
                                "index": text_block_index,
                                "delta": {"type": "text_delta", "text": buffered},
                            },
                        )
                        _log_sse_event(
                            {
                                "type": "content_block_delta",
                                "index": text_block_index,
                                "delta": {"type": "text_delta", "text": buffered},
                            },
                            log_sse_events,
                            log_sse_limit,
                        )
                        translated_text_parts.append(buffered)
                else:
                    if filter_tool_json_text and (
                        _looks_like_tool_json(delta["content"])
                        or _looks_like_topic_json(delta["content"])
                        or _looks_like_cli_question_json(delta["content"])
                    ):
                        suppress_cli_json = True
                        suppressed_text_parts.append(delta["content"])
                        continue
                    if filter_tool_json_text and _looks_like_cli_question_fragment(
                        delta["content"]
                    ):
                        suppress_cli_json = True
                        suppressed_text_parts.append(delta["content"])
                        continue
                    if text_block_index is None:
                        text_block_index = next_block_index
                        next_block_index += 1
                        _ensure_started()
                        _ensure_started()
                        _write_sse_event(
                            handler,
                            {
                                "type": "content_block_start",
                                "index": text_block_index,
                                "content_block": {"type": "text", "text": ""},
                            },
                        )
                        _log_sse_event(
                            {
                                "type": "content_block_start",
                                "index": text_block_index,
                                "content_block": {"type": "text", "text": ""},
                            },
                            log_sse_events,
                            log_sse_limit,
                        )
                    output_tokens = (output_tokens or 0) + 1
                    _write_sse_event(
                        handler,
                        {
                            "type": "content_block_delta",
                            "index": text_block_index,
                            "delta": {"type": "text_delta", "text": delta["content"]},
                        },
                    )
                    _log_sse_event(
                        {
                            "type": "content_block_delta",
                            "index": text_block_index,
                            "delta": {"type": "text_delta", "text": delta["content"]},
                        },
                        log_sse_events,
                        log_sse_limit,
                    )
                    translated_text_parts.append(delta["content"])
            if not strict_stream_text_blocks and "tool_calls" in delta and delta["tool_calls"]:
                for tool_delta in delta["tool_calls"]:
                    tool_index = tool_delta.get("index", 0)
                    if tool_index not in tool_block_indices:
                        tool_block_indices[tool_index] = next_block_index
                        next_block_index += 1
                        function = tool_delta.get("function", {})
                        _write_sse_event(
                            handler,
                            {
                                "type": "content_block_start",
                                "index": tool_block_indices[tool_index],
                                "content_block": {
                                    "type": "tool_use",
                                    "id": tool_delta.get("id")
                                    or f"tool_{uuid.uuid4().hex}",
                                    "name": function.get("name"),
                                    "input": {},
                                },
                            },
                        )
                    function = tool_delta.get("function", {})
                    if "arguments" in function and function["arguments"] is not None:
                        _write_sse_event(
                            handler,
                            {
                                "type": "content_block_delta",
                                "index": tool_block_indices[tool_index],
                                "delta": {
                                    "type": "input_json_delta",
                                    "partial_json": function["arguments"],
                                },
                            },
                        )
                    _log_sse_event(
                        {
                            "type": "content_block_delta",
                            "index": tool_block_indices[tool_index],
                            "delta": {
                                "type": "input_json_delta",
                                "partial_json": function["arguments"],
                            },
                        },
                        log_sse_events,
                        log_sse_limit,
                    )
            if choice.get("finish_reason") is not None:
                stop_reason = choice.get("finish_reason")

            if usage:
                current_usage = (usage_input_tokens, usage_output_tokens)
                if current_usage != last_usage:
                    last_usage = current_usage
                    _write_sse_event(
                        handler,
                        {
                            "type": "message_delta",
                            "delta": {"stop_reason": None, "stop_sequence": None},
                            "usage": {
                                "input_tokens": usage_input_tokens,
                                "output_tokens": usage_output_tokens,
                            },
                        },
                    )
                    _log_sse_event(
                        {
                            "type": "message_delta",
                            "delta": {"stop_reason": None, "stop_sequence": None},
                            "usage": {
                                "input_tokens": usage_input_tokens,
                                "output_tokens": usage_output_tokens,
                            },
                        },
                        log_sse_events,
                        log_sse_limit,
                    )

    if text_buffer:
        buffered = "".join(text_buffer)
        text_buffer = []
        if suppress_cli_json:
            suppressed_text_parts.append(buffered)
            buffered = ""
        if filter_tool_json_text and (
            _looks_like_tool_json(buffered)
            or _looks_like_topic_json(buffered)
            or _looks_like_cli_question_json(buffered)
        ):
            suppressed_text_parts.append(buffered)
            buffered = ""
        if buffered:
            if text_block_index is None:
                text_block_index = next_block_index
                next_block_index += 1
                _ensure_started()
                _write_sse_event(
                    handler,
                    {
                        "type": "content_block_start",
                        "index": text_block_index,
                        "content_block": {"type": "text", "text": ""},
                    },
                )
                _log_sse_event(
                    {
                        "type": "content_block_start",
                        "index": text_block_index,
                        "content_block": {"type": "text", "text": ""},
                    },
                    log_sse_events,
                    log_sse_limit,
                )
            output_tokens = (output_tokens or 0) + 1
            _write_sse_event(
                handler,
                {
                    "type": "content_block_delta",
                    "index": text_block_index,
                    "delta": {"type": "text_delta", "text": buffered},
                },
            )
            _log_sse_event(
                {
                    "type": "content_block_delta",
                    "index": text_block_index,
                    "delta": {"type": "text_delta", "text": buffered},
                },
                log_sse_events,
                log_sse_limit,
            )
            translated_text_parts.append(buffered)

    if not started and suppressed_text_parts:
        fallback_mode = empty_stream_fallback.strip().lower()
        if fallback_mode not in {"pass_through", "passthrough", "placeholder", "none"}:
            fallback_mode = "placeholder"
        if fallback_mode in {"pass_through", "passthrough"}:
            fallback_text = "".join(suppressed_text_parts)
        elif fallback_mode == "placeholder":
            fallback_text = empty_stream_fallback_text or ""
        else:
            fallback_text = ""

        if fallback_text:
            _write_sse_event(
                handler,
                {
                    "type": "message_start",
                    "message": {
                        "id": message_id,
                        "type": "message",
                        "role": "assistant",
                        "model": model,
                        "content": [],
                        "stop_reason": None,
                        "stop_sequence": None,
                        "usage": {"input_tokens": None, "output_tokens": None},
                    },
                },
            )
            _log_sse_event(
                {
                    "type": "message_start",
                    "message": {
                        "id": message_id,
                        "type": "message",
                        "role": "assistant",
                        "model": model,
                        "content": [],
                        "stop_reason": None,
                        "stop_sequence": None,
                        "usage": {"input_tokens": None, "output_tokens": None},
                    },
                },
                log_sse_events,
                log_sse_limit,
            )
            _write_sse_event(
                handler,
                {
                    "type": "content_block_start",
                    "index": 0,
                    "content_block": {"type": "text", "text": ""},
                },
            )
            _log_sse_event(
                {
                    "type": "content_block_start",
                    "index": 0,
                    "content_block": {"type": "text", "text": ""},
                },
                log_sse_events,
                log_sse_limit,
            )
            _write_sse_event(
                handler,
                {
                    "type": "content_block_delta",
                    "index": 0,
                    "delta": {"type": "text_delta", "text": fallback_text},
                },
            )
            _log_sse_event(
                {
                    "type": "content_block_delta",
                    "index": 0,
                    "delta": {"type": "text_delta", "text": fallback_text},
                },
                log_sse_events,
                log_sse_limit,
            )
            _write_sse_event(handler, {"type": "content_block_stop", "index": 0})
            _log_sse_event(
                {"type": "content_block_stop", "index": 0},
                log_sse_events,
                log_sse_limit,
            )
            _write_sse_event(
                handler,
                {
                    "type": "message_delta",
                    "delta": {"stop_reason": "end_turn", "stop_sequence": None},
                    "usage": {
                        "input_tokens": usage_input_tokens,
                        "output_tokens": usage_output_tokens,
                    },
                },
            )
            _log_sse_event(
                {
                    "type": "message_delta",
                    "delta": {"stop_reason": "end_turn", "stop_sequence": None},
                    "usage": {
                        "input_tokens": usage_input_tokens,
                        "output_tokens": usage_output_tokens,
                    },
                },
                log_sse_events,
                log_sse_limit,
            )
            _write_sse_event(handler, {"type": "message_stop"})
            _log_sse_event({"type": "message_stop"}, log_sse_events, log_sse_limit)
            return

    end_indices = list(tool_block_indices.values())
    if text_block_index is not None:
        end_indices.append(text_block_index)
    if started:
        for index in sorted(end_indices):
            _write_sse_event(handler, {"type": "content_block_stop", "index": index})
            _log_sse_event(
                {"type": "content_block_stop", "index": index},
                log_sse_events,
                log_sse_limit,
            )
    if stop_reason == "stop":
        stop_reason = "end_turn"
    elif stop_reason == "length":
        stop_reason = "max_tokens"
    elif stop_reason == "tool_calls":
        stop_reason = "tool_use"
    if strict_stream_text_blocks and stop_reason == "tool_use":
        stop_reason = "end_turn"
    if stop_reason is None and translated_text_parts:
        stop_reason = "end_turn"
    usage_output_tokens = usage_output_tokens if usage_output_tokens is not None else output_tokens
    if log_translations:
        _log_preview(
            "Translated OpenAI -> Claude",
            {
                "type": "message",
                "role": "assistant",
                "content": [{"type": "text", "text": "".join(translated_text_parts)}],
                "stop_reason": stop_reason,
                "usage": {
                    "input_tokens": usage_input_tokens,
                    "output_tokens": usage_output_tokens,
                },
            },
            True,
            log_translation_limit,
        )
    if started:
        _write_sse_event(
            handler,
            {
                "type": "message_delta",
                "delta": {"stop_reason": stop_reason, "stop_sequence": None},
                "usage": {
                    "input_tokens": usage_input_tokens,
                    "output_tokens": usage_output_tokens,
                },
            },
        )
        _log_sse_event(
            {
                "type": "message_delta",
                "delta": {"stop_reason": stop_reason, "stop_sequence": None},
                "usage": {
                    "input_tokens": usage_input_tokens,
                    "output_tokens": usage_output_tokens,
                },
            },
            log_sse_events,
            log_sse_limit,
        )
        _write_sse_event(handler, {"type": "message_stop"})
        _log_sse_event({"type": "message_stop"}, log_sse_events, log_sse_limit)
        finished = True


def _build_openai_url(base_url: str) -> str:
    """Normalize base URL and append the chat completions path."""
    normalized = base_url.rstrip("/")
    if normalized.endswith("/v1"):
        normalized = normalized[: -len("/v1")]
    return f"{normalized}/v1/chat/completions"


class Claude2OpenAIServer(BaseHTTPRequestHandler):
    """HTTP handler for Claude-compatible endpoints."""
    config: dict[str, Any] = {}

    def do_GET(self) -> None:
        """Serve a health check endpoint."""
        path = urllib.parse.urlparse(self.path).path.rstrip("/")
        if path == "/health":
            _write_json_response(self, 200, {"status": "ok"})
            return
        _write_json_response(self, 404, {"error": "not found"})

    def do_POST(self) -> None:
        """Handle Claude messages and forward them upstream."""
        path = urllib.parse.urlparse(self.path).path.rstrip("/")
        if not (path.endswith("/v1/messages") or path.endswith("/v1/v1/messages")):
            _write_json_response(self, 404, {"error": "not found"})
            return

        try:
            payload = _read_json_body(self)
        except json.JSONDecodeError:
            _write_json_response(self, 400, {"error": "invalid json"})
            return

        openai_config = self.config.get("openai", {})
        default_model = openai_config.get("model")
        model_map = self.config.get("model_map")
        disable_tools = _as_bool(openai_config.get("disable_tools", False))
        tool_choice_override = openai_config.get("tool_choice_override")
        response_format_override = openai_config.get("response_format_override")
        system_prompt_suffix = openai_config.get("system_prompt_suffix")
        system_prompt_mode = openai_config.get("system_prompt_mode")
        log_translations = _as_bool(openai_config.get("log_translated_messages", False))
        log_translation_limit = int(
            openai_config.get("log_translated_messages_limit", 2000)
        )
        openai_payload = _claude_to_openai(
            payload,
            default_model,
            model_map,
            disable_tools,
            tool_choice_override,
            response_format_override,
            system_prompt_suffix,
            system_prompt_mode,
        )
        _log_preview(
            "Translated Claude -> OpenAI",
            openai_payload,
            log_translations,
            log_translation_limit,
        )
        if not openai_payload.get("model"):
            _write_json_response(self, 400, {"error": "missing model"})
            return

        base_url = openai_config.get("base_url", "").rstrip("/")
        api_key = openai_config.get("api_key")
        if not base_url or not api_key:
            _write_json_response(self, 500, {"error": "openai config missing"})
            return

        openai_url = _build_openai_url(base_url)
        timeout = float(openai_config.get("timeout_seconds", 60))
        retries = int(openai_config.get("retries", 0))
        backoff_seconds = float(openai_config.get("retry_backoff_seconds", 0.5))
        log_responses = bool(openai_config.get("log_upstream_response", False))
        log_limit = int(openai_config.get("log_upstream_response_limit", 2000))
        merge_text_blocks = _as_bool(openai_config.get("merge_text_blocks", True))
        strict_stream_text_blocks = _as_bool(
            openai_config.get("strict_stream_text_blocks", False)
        )
        drop_empty_text_deltas = _as_bool(
            openai_config.get("drop_empty_text_deltas", False)
        )
        stream_sentence_buffer = _as_bool(
            openai_config.get("stream_sentence_buffer", False)
        )
        log_sse_events = _as_bool(openai_config.get("log_sse_events", False))
        log_sse_limit = int(openai_config.get("log_sse_events_limit", 2000))
        filter_tool_json_text = _as_bool(
            openai_config.get("filter_tool_json_text", False)
        )
        filter_topic_json_text = _as_bool(
            openai_config.get("filter_topic_json_text", False)
        )
        empty_stream_fallback = str(
            openai_config.get("empty_stream_fallback", "none")
        )
        empty_stream_fallback_text = openai_config.get("empty_stream_fallback_text")

        if openai_payload.get("stream") is True:
            self.send_response(200)
            self.send_header("Content-Type", "text/event-stream")
            self.send_header("Cache-Control", "no-cache")
            self.send_header("Connection", "close")
            self.end_headers()
            try:
                _stream_openai_to_claude(
                    self,
                    openai_url,
                    api_key,
                    openai_payload,
                    timeout,
                    retries,
                    backoff_seconds,
                    log_responses,
                    log_limit,
                    merge_text_blocks,
                    strict_stream_text_blocks,
                    drop_empty_text_deltas,
                    stream_sentence_buffer,
                    log_translations,
                    log_translation_limit,
                    log_sse_events,
                    log_sse_limit,
                    filter_tool_json_text,
                    filter_topic_json_text,
                    empty_stream_fallback,
                    empty_stream_fallback_text,
                )
            except UpstreamHTTPError as exc:
                LOGGER.error("Upstream error (stream): %s", exc.detail)
                _write_sse_event(
                    self,
                    {"type": "error", "error": "upstream error", "detail": exc.detail},
                )
            except UpstreamConnectionError as exc:
                LOGGER.error("Upstream connection failed (stream): %s", exc.detail)
                _write_sse_event(
                    self,
                    {
                        "type": "error",
                        "error": "upstream connection failed",
                        "detail": exc.detail,
                    },
                )
            self.close_connection = True
            return

        try:
            openai_response = _openai_request_json(
                openai_url,
                api_key,
                openai_payload,
                timeout,
                retries,
                backoff_seconds,
                log_responses,
                log_limit,
            )
        except UpstreamHTTPError as exc:
            LOGGER.error("Upstream error: %s", exc.detail)
            _write_json_response(
                self, 502, {"error": "upstream error", "detail": exc.detail}
            )
            return
        except UpstreamConnectionError as exc:
            LOGGER.error("Upstream connection failed: %s", exc.detail)
            _write_json_response(
                self,
                502,
                {"error": "upstream connection failed", "detail": exc.detail},
            )
            return

        claude_response = _openai_to_claude(openai_response)
        _log_preview(
            "Translated OpenAI -> Claude",
            claude_response,
            log_translations,
            log_translation_limit,
        )
        _write_json_response(self, 200, claude_response)


def _load_config() -> dict[str, Any]:
    """Load config.toml from C2O_CONFIG or default path."""
    config_path = os.environ.get("C2O_CONFIG", "config.toml")
    try:
        with open(config_path, "rb") as handle:
            return tomllib.load(handle)
    except FileNotFoundError:
        print(f"Config file not found: {config_path}", file=sys.stderr)
        sys.exit(1)
    except tomllib.TOMLDecodeError as exc:
        print(f"Invalid config.toml: {exc}", file=sys.stderr)
        sys.exit(1)


def main() -> None:
    """Entry point for the Claude2OpenAI server."""
    config = _load_config()
    _configure_logging(config)
    server_config = config.get("server", {})
    host = server_config.get("host", "127.0.0.1")
    port = int(server_config.get("port", 8080))

    Claude2OpenAIServer.config = config
    httpd = ThreadingHTTPServer((host, port), Claude2OpenAIServer)
    LOGGER.info("Claude2OpenAI listening on http://%s:%s", host, port)
    httpd.serve_forever()


if __name__ == "__main__":
    main()