from __future__ import annotations

import json
import os
import sys
from typing import Any, Iterator
from urllib import error, request

from .agent_types import (
    AssistantTurn,
    ModelConfig,
    OutputSchemaConfig,
    StreamEvent,
    ToolCall,
    UsageStats,
)


class OpenAICompatError(RuntimeError):
    """Raised when the local OpenAI-compatible backend returns an invalid response."""


PROVIDER_MIN_TEMPERATURES = {
    # 实测这些 provider 会拒绝 temperature=0；这里做最小值兜底，
    # 保持默认确定性取向的同时避免模型切换后请求直接 400。
    'minimax': 0.01,
    'wenxin': 0.1,
}

ANTHROPIC_MESSAGES_MODEL_PREFIXES = (
    'ppio/pa/claude-',
)

ANTHROPIC_VERSION = '2023-06-01'
ANTHROPIC_MAX_TOKENS = 16384
DEFAULT_MODEL_IDLE_TIMEOUT_SECONDS = 60.0


def _join_url(base_url: str, suffix: str) -> str:
    base = base_url.rstrip('/')
    return f'{base}/{suffix.lstrip("/")}'


def _normalize_content(content: Any) -> str:
    if content is None:
        return ''
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts: list[str] = []
        for item in content:
            if isinstance(item, str):
                parts.append(item)
                continue
            if not isinstance(item, dict):
                parts.append(str(item))
                continue
            if item.get('type') == 'text' and isinstance(item.get('text'), str):
                parts.append(item['text'])
                continue
            if isinstance(item.get('text'), str):
                parts.append(item['text'])
                continue
            parts.append(json.dumps(item, ensure_ascii=True))
        return ''.join(parts)
    return str(content)


def _parse_tool_arguments(raw_arguments: Any) -> dict[str, Any]:
    if raw_arguments is None:
        return {}
    if isinstance(raw_arguments, dict):
        return raw_arguments
    if isinstance(raw_arguments, str):
        raw_arguments = raw_arguments.strip()
        if not raw_arguments:
            return {}
        try:
            parsed = json.loads(raw_arguments)
        except json.JSONDecodeError as exc:
            preview = raw_arguments[:240].replace('\n', '\\n')
            raise OpenAICompatError(
                'Invalid tool arguments returned by model: '
                f'line {exc.lineno} column {exc.colno}: {exc.msg}; '
                f'preview={preview!r}'
            ) from exc
        if not isinstance(parsed, dict):
            raise OpenAICompatError(
                f'Tool arguments must decode to an object, got {type(parsed).__name__}'
            )
        return parsed
    raise OpenAICompatError(
        f'Unsupported tool arguments payload: {type(raw_arguments).__name__}'
    )


def _tool_call_ids(message: dict[str, Any]) -> list[str]:
    raw_tool_calls = message.get('tool_calls')
    if not isinstance(raw_tool_calls, list):
        return []
    ids: list[str] = []
    for index, raw_call in enumerate(raw_tool_calls):
        if not isinstance(raw_call, dict):
            continue
        call_id = raw_call.get('id')
        ids.append(call_id if isinstance(call_id, str) and call_id else f'call_{index}')
    return ids


def _contiguous_tool_result_ids(
    messages: list[dict[str, Any]],
    start_index: int,
) -> set[str]:
    ids: set[str] = set()
    index = start_index
    while index < len(messages) and messages[index].get('role') == 'tool':
        tool_call_id = messages[index].get('tool_call_id')
        if isinstance(tool_call_id, str) and tool_call_id:
            ids.add(tool_call_id)
        index += 1
    return ids


def _filter_tool_calls_by_ids(
    message: dict[str, Any],
    valid_ids: set[str],
) -> dict[str, Any]:
    raw_tool_calls = message.get('tool_calls')
    if not isinstance(raw_tool_calls, list):
        return message
    filtered: list[dict[str, Any]] = []
    for index, raw_call in enumerate(raw_tool_calls):
        if not isinstance(raw_call, dict):
            continue
        call_id = raw_call.get('id')
        normalized_id = (
            call_id if isinstance(call_id, str) and call_id else f'call_{index}'
        )
        if normalized_id in valid_ids:
            filtered.append(raw_call)
    if len(filtered) == len(raw_tool_calls):
        return message
    next_message = dict(message)
    if filtered:
        next_message['tool_calls'] = filtered
    else:
        next_message.pop('tool_calls', None)
    return next_message


def _sanitize_anthropic_tool_messages(
    messages: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    """清理 Anthropic/Bedrock 严格要求的 tool_use/tool_result 邻接关系。

    OpenAI 风格历史在异常中断、裁剪或前端恢复时可能留下 assistant tool_calls，
    但下一条消息不再紧跟对应 tool 结果。Anthropic Messages API 会直接拒绝
    这种历史。这里在出网关前删除缺失结果的 tool_call，并跳过没有前置
    tool_use 的孤儿 tool 结果，保留普通文本上下文继续对话。
    """
    sanitized: list[dict[str, Any]] = []
    pending_tool_ids: set[str] = set()

    for index, message in enumerate(messages):
        role = message.get('role')
        if role == 'assistant':
            tool_ids = _tool_call_ids(message)
            if tool_ids:
                result_ids = _contiguous_tool_result_ids(messages, index + 1)
                valid_ids = set(tool_ids).intersection(result_ids)
                message = _filter_tool_calls_by_ids(message, valid_ids)
                pending_tool_ids = valid_ids
            else:
                pending_tool_ids = set()
            sanitized.append(message)
            continue

        if role == 'tool':
            tool_call_id = message.get('tool_call_id')
            if isinstance(tool_call_id, str) and tool_call_id in pending_tool_ids:
                sanitized.append(message)
            continue

        pending_tool_ids = set()
        sanitized.append(message)

    return sanitized


def _optional_int(value: Any) -> int:
    if isinstance(value, bool):
        return 0
    if isinstance(value, int):
        return value
    if isinstance(value, float):
        return int(value)
    if isinstance(value, str):
        try:
            return int(value)
        except ValueError:
            return 0
    return 0


def _temperature_for_model(model: str, configured: float) -> float:
    provider = model.split('/', 1)[0].strip().lower() if '/' in model else ''
    minimum = PROVIDER_MIN_TEMPERATURES.get(provider)
    if minimum is None:
        return configured
    return max(configured, minimum)


def _uses_anthropic_messages_api(model: str, base_url: str) -> bool:
    normalized_model = model.strip().lower()
    normalized_base = base_url.rstrip('/').lower()
    return normalized_base.endswith('/anthropic') or any(
        normalized_model.startswith(prefix)
        for prefix in ANTHROPIC_MESSAGES_MODEL_PREFIXES
    )


_UNKNOWN_EVENT_LOG_LIMIT = 8
_unknown_event_seen: dict[str, int] = {}


def _log_unknown_anthropic_event(kind: str, payload: dict[str, Any]) -> None:
    """Emit a one-line stderr warning the first few times we see an unknown event.

    The streaming handler used to silently drop any block/delta type it didn't
    recognise. When the upstream proxy started emitting frames we didn't decode,
    the model's output budget was burned with no observable effect. This log
    surfaces the situation without flooding logs on every turn.
    """
    seen = _unknown_event_seen.get(kind, 0)
    if seen >= _UNKNOWN_EVENT_LOG_LIMIT:
        return
    _unknown_event_seen[kind] = seen + 1
    try:
        snippet = json.dumps(payload, ensure_ascii=False)[:400]
    except Exception:  # pragma: no cover - defensive
        snippet = repr(payload)[:400]
    print(
        f'[openai_compat] unknown anthropic stream event {kind!r}: {snippet}',
        file=sys.stderr,
        flush=True,
    )


def _mark_last_message_cache_breakpoint(messages: list[dict[str, Any]]) -> None:
    """Attach cache_control to the last content block of the last message.

    Anthropic prompt cache requires the breakpoint at the *end* of the cacheable
    prefix. The next turn appends a new message → the previous last becomes the
    second-to-last with its breakpoint still in the prefix, so everything up to
    that point is reused on the next call.
    """
    if not messages:
        return
    last = messages[-1]
    content = last.get('content')
    if not isinstance(content, list) or not content:
        return
    last_block = content[-1]
    if not isinstance(last_block, dict):
        return
    new_block = dict(last_block)
    new_block['cache_control'] = {'type': 'ephemeral'}
    new_content = list(content)
    new_content[-1] = new_block
    last['content'] = new_content


def _anthropic_base_url(base_url: str) -> str:
    base = base_url.rstrip('/')
    if base.lower().endswith('/anthropic'):
        return base
    if base.lower().endswith('/v1'):
        return f'{base[:-3]}/anthropic'
    return f'{base}/anthropic'


def _parse_usage(payload: Any) -> UsageStats:
    if not isinstance(payload, dict):
        return UsageStats()
    completion_details = payload.get('completion_tokens_details')
    if not isinstance(completion_details, dict):
        completion_details = {}
    return UsageStats(
        input_tokens=(
            _optional_int(payload.get('input_tokens'))
            or _optional_int(payload.get('prompt_tokens'))
            or _optional_int(payload.get('prompt_eval_count'))
        ),
        output_tokens=(
            _optional_int(payload.get('output_tokens'))
            or _optional_int(payload.get('completion_tokens'))
            or _optional_int(payload.get('eval_count'))
        ),
        cache_creation_input_tokens=_optional_int(
            payload.get('cache_creation_input_tokens')
        ),
        cache_read_input_tokens=_optional_int(payload.get('cache_read_input_tokens')),
        reasoning_tokens=(
            _optional_int(payload.get('reasoning_tokens'))
            or _optional_int(completion_details.get('reasoning_tokens'))
        ),
    )


def _build_response_format(
    schema: OutputSchemaConfig | None,
) -> dict[str, Any] | None:
    if schema is None:
        return None
    return {
        'type': 'json_schema',
        'json_schema': {
            'name': schema.name,
            'schema': schema.schema,
            'strict': schema.strict,
        },
    }


class OpenAICompatClient:
    """Minimal OpenAI-compatible chat client for local model servers."""

    def __init__(self, config: ModelConfig) -> None:
        self.config = config

    def _request_timeout_seconds(self) -> float:
        """Return per-socket model timeout.

        `ModelConfig.timeout_seconds` is the whole request budget used by the
        product, but urllib applies it as an idle socket timeout. Keeping it at
        one hour makes a bad upstream stream hold a session lock for an hour.
        Use a shorter idle timeout by default; any normally streaming response
        keeps extending this naturally because bytes continue to arrive.
        """
        raw = os.environ.get('CLAW_MODEL_IDLE_TIMEOUT_SECONDS', '').strip()
        try:
            configured = float(raw) if raw else DEFAULT_MODEL_IDLE_TIMEOUT_SECONDS
        except ValueError:
            configured = DEFAULT_MODEL_IDLE_TIMEOUT_SECONDS
        configured = max(5.0, configured)
        return min(float(self.config.timeout_seconds), configured)

    def complete(
        self,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]],
        *,
        output_schema: OutputSchemaConfig | None = None,
    ) -> AssistantTurn:
        if self._uses_anthropic_messages_api():
            return self._complete_anthropic_messages(
                messages=messages,
                tools=tools,
                output_schema=output_schema,
            )
        payload = self._request_json(
            self._build_payload(
                messages=messages,
                tools=tools,
                stream=False,
                output_schema=output_schema,
            )
        )
        choices = payload.get('choices')
        if not isinstance(choices, list) or not choices:
            raise OpenAICompatError('Local model backend returned no choices')
        first_choice = choices[0]
        if not isinstance(first_choice, dict):
            raise OpenAICompatError('Local model backend returned malformed choice data')

        message = first_choice.get('message')
        if not isinstance(message, dict):
            raise OpenAICompatError('Local model backend returned no assistant message')

        content = _normalize_content(message.get('content'))
        tool_calls = self._parse_tool_calls_from_message(message)

        finish_reason = first_choice.get('finish_reason')
        if finish_reason is not None and not isinstance(finish_reason, str):
            finish_reason = str(finish_reason)

        return AssistantTurn(
            content=content,
            tool_calls=tuple(tool_calls),
            finish_reason=finish_reason,
            raw_message=message,
            usage=_parse_usage(payload.get('usage')),
        )

    def stream(
        self,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]],
        *,
        output_schema: OutputSchemaConfig | None = None,
    ) -> Iterator[StreamEvent]:
        if self._uses_anthropic_messages_api():
            yield from self._stream_anthropic_messages(
                messages=messages,
                tools=tools,
                output_schema=output_schema,
            )
            return
        payload = self._build_payload(
            messages=messages,
            tools=tools,
            stream=True,
            output_schema=output_schema,
        )
        req = request.Request(
            _join_url(self.config.base_url, '/chat/completions'),
            data=json.dumps(payload).encode('utf-8'),
            headers={
                'Authorization': f'Bearer {self.config.api_key}',
                'Content-Type': 'application/json',
            },
            method='POST',
        )
        try:
            with request.urlopen(req, timeout=self._request_timeout_seconds()) as response:
                yield StreamEvent(type='message_start')
                for event_payload in self._iter_sse_payloads(response):
                    yield from self._parse_stream_payload(event_payload)
        except error.HTTPError as exc:
            detail = exc.read().decode('utf-8', errors='replace')
            raise OpenAICompatError(
                f'HTTP {exc.code} from local model backend: {detail}'
            ) from exc
        except error.URLError as exc:
            raise OpenAICompatError(
                f'Unable to reach local model backend at {self.config.base_url}: {exc.reason}'
            ) from exc

    def _uses_anthropic_messages_api(self) -> bool:
        return _uses_anthropic_messages_api(self.config.model, self.config.base_url)

    def _request_json(self, payload: dict[str, Any]) -> dict[str, Any]:
        body = json.dumps(payload).encode('utf-8')
        req = request.Request(
            _join_url(self.config.base_url, '/chat/completions'),
            data=body,
            headers={
                'Authorization': f'Bearer {self.config.api_key}',
                'Content-Type': 'application/json',
            },
            method='POST',
        )
        try:
            with request.urlopen(req, timeout=self._request_timeout_seconds()) as response:
                raw = response.read()
        except error.HTTPError as exc:
            detail = exc.read().decode('utf-8', errors='replace')
            raise OpenAICompatError(
                f'HTTP {exc.code} from local model backend: {detail}'
            ) from exc
        except error.URLError as exc:
            raise OpenAICompatError(
                f'Unable to reach local model backend at {self.config.base_url}: {exc.reason}'
            ) from exc

        try:
            payload = json.loads(raw.decode('utf-8'))
        except json.JSONDecodeError as exc:
            raise OpenAICompatError('Local model backend returned invalid JSON') from exc
        if not isinstance(payload, dict):
            raise OpenAICompatError('Local model backend returned malformed JSON payload')
        return payload

    def _build_payload(
        self,
        *,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]],
        stream: bool,
        output_schema: OutputSchemaConfig | None,
    ) -> dict[str, Any]:
        payload: dict[str, Any] = {
            'model': self.config.model,
            'messages': messages,
            'temperature': _temperature_for_model(self.config.model, self.config.temperature),
            'stream': stream,
        }
        if tools:
            payload['tools'] = tools
            payload['tool_choice'] = 'auto'
        if stream:
            payload['stream_options'] = {'include_usage': True}
        response_format = _build_response_format(output_schema)
        if response_format is not None:
            payload['response_format'] = response_format
        return payload

    def _build_anthropic_payload(
        self,
        *,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]],
        stream: bool,
        output_schema: OutputSchemaConfig | None,
    ) -> dict[str, Any]:
        system_parts: list[str] = []
        anthropic_messages: list[dict[str, Any]] = []
        for message in _sanitize_anthropic_tool_messages(messages):
            role = message.get('role')
            if role == 'system':
                content = _normalize_content(message.get('content')).strip()
                if content:
                    system_parts.append(content)
                continue
            if role == 'assistant':
                blocks = self._anthropic_assistant_blocks(message)
                if blocks:
                    self._append_anthropic_message(anthropic_messages, 'assistant', blocks)
                continue
            if role == 'tool':
                tool_call_id = message.get('tool_call_id')
                if not isinstance(tool_call_id, str) or not tool_call_id:
                    tool_call_id = 'toolu_unknown'
                self._append_anthropic_message(
                    anthropic_messages,
                    'user',
                    [
                        {
                            'type': 'tool_result',
                            'tool_use_id': tool_call_id,
                            'content': _normalize_content(message.get('content')),
                        }
                    ],
                )
                continue
            if role == 'user':
                blocks = self._anthropic_text_blocks(message.get('content'))
                if blocks:
                    self._append_anthropic_message(anthropic_messages, 'user', blocks)

        system_text = '\n\n'.join(system_parts) if system_parts else ''
        if output_schema is not None:
            schema_hint = (
                f'请严格输出符合 JSON Schema `{output_schema.name}` 的 JSON，'
                '不要输出额外解释。'
            )
            system_text = f'{system_text}\n\n{schema_hint}'.strip() if system_text else schema_hint

        cache_enabled = (
            os.environ.get('CLAW_DISABLE_PROMPT_CACHE', '').strip().lower()
            not in {'1', 'true', 'yes', 'on'}
        )

        payload: dict[str, Any] = {
            'model': self.config.model,
            'messages': anthropic_messages,
            'max_tokens': ANTHROPIC_MAX_TOKENS,
            'stream': stream,
        }
        if system_text:
            if cache_enabled:
                payload['system'] = [
                    {
                        'type': 'text',
                        'text': system_text,
                        'cache_control': {'type': 'ephemeral'},
                    }
                ]
            else:
                payload['system'] = system_text
        converted_tools = self._anthropic_tools(tools)
        if converted_tools:
            if cache_enabled:
                converted_tools = list(converted_tools)
                last_tool = dict(converted_tools[-1])
                last_tool['cache_control'] = {'type': 'ephemeral'}
                converted_tools[-1] = last_tool
            payload['tools'] = converted_tools
        if cache_enabled and anthropic_messages:
            _mark_last_message_cache_breakpoint(anthropic_messages)
        return payload

    def _anthropic_headers(self) -> dict[str, str]:
        return {
            'Authorization': f'Bearer {self.config.api_key}',
            'Content-Type': 'application/json',
            'anthropic-version': ANTHROPIC_VERSION,
            'api-key': self.config.api_key,
            'x-api-key': self.config.api_key,
        }

    def _anthropic_messages_url(self) -> str:
        return _join_url(_anthropic_base_url(self.config.base_url), '/v1/messages')

    def _complete_anthropic_messages(
        self,
        *,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]],
        output_schema: OutputSchemaConfig | None,
    ) -> AssistantTurn:
        payload = self._build_anthropic_payload(
            messages=messages,
            tools=tools,
            stream=False,
            output_schema=output_schema,
        )
        body = json.dumps(payload).encode('utf-8')
        req = request.Request(
            self._anthropic_messages_url(),
            data=body,
            headers=self._anthropic_headers(),
            method='POST',
        )
        try:
            with request.urlopen(req, timeout=self._request_timeout_seconds()) as response:
                raw = response.read()
        except error.HTTPError as exc:
            detail = exc.read().decode('utf-8', errors='replace')
            raise OpenAICompatError(
                f'HTTP {exc.code} from local model backend: {detail}'
            ) from exc
        except error.URLError as exc:
            raise OpenAICompatError(
                f'Unable to reach local model backend at {self.config.base_url}: {exc.reason}'
            ) from exc

        try:
            response_payload = json.loads(raw.decode('utf-8'))
        except json.JSONDecodeError as exc:
            raise OpenAICompatError('Local model backend returned invalid JSON') from exc
        if not isinstance(response_payload, dict):
            raise OpenAICompatError('Local model backend returned malformed JSON payload')
        return self._parse_anthropic_message_response(response_payload)

    def _stream_anthropic_messages(
        self,
        *,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]],
        output_schema: OutputSchemaConfig | None,
    ) -> Iterator[StreamEvent]:
        payload = self._build_anthropic_payload(
            messages=messages,
            tools=tools,
            stream=True,
            output_schema=output_schema,
        )
        req = request.Request(
            self._anthropic_messages_url(),
            data=json.dumps(payload).encode('utf-8'),
            headers=self._anthropic_headers(),
            method='POST',
        )
        try:
            with request.urlopen(req, timeout=self._request_timeout_seconds()) as response:
                yield StreamEvent(type='message_start')
                tool_block_indexes: dict[int, int] = {}
                next_tool_index = 0
                finish_reason: str | None = None
                for event_payload in self._iter_sse_payloads(response):
                    event_type = event_payload.get('type')
                    if event_type == 'message_start':
                        message = event_payload.get('message')
                        usage = _parse_usage(
                            message.get('usage') if isinstance(message, dict) else None
                        )
                        if usage.total_tokens:
                            yield StreamEvent(
                                type='usage',
                                usage=usage,
                                raw_event=event_payload,
                            )
                        continue
                    if event_type == 'content_block_start':
                        block_index = event_payload.get('index')
                        content_block = event_payload.get('content_block')
                        if not isinstance(block_index, int) or not isinstance(content_block, dict):
                            continue
                        block_type = content_block.get('type')
                        if block_type != 'tool_use':
                            if block_type not in ('text', 'thinking', 'redacted_thinking'):
                                _log_unknown_anthropic_event(
                                    'content_block_start', event_payload
                                )
                            continue
                        tool_index = next_tool_index
                        next_tool_index += 1
                        tool_block_indexes[block_index] = tool_index
                        tool_input = content_block.get('input')
                        arguments = (
                            json.dumps(tool_input, ensure_ascii=True)
                            if isinstance(tool_input, dict) and tool_input
                            else ''
                        )
                        yield StreamEvent(
                            type='tool_call_delta',
                            tool_call_index=tool_index,
                            tool_call_id=(
                                content_block.get('id')
                                if isinstance(content_block.get('id'), str)
                                else None
                            ),
                            tool_name=(
                                content_block.get('name')
                                if isinstance(content_block.get('name'), str)
                                else None
                            ),
                            arguments_delta=arguments,
                            raw_event=event_payload,
                        )
                        continue
                    if event_type == 'content_block_delta':
                        delta = event_payload.get('delta')
                        if not isinstance(delta, dict):
                            continue
                        delta_type = delta.get('type')
                        if delta_type == 'text_delta':
                            text = delta.get('text')
                            if isinstance(text, str) and text:
                                yield StreamEvent(
                                    type='content_delta',
                                    delta=text,
                                    raw_event=event_payload,
                                )
                            continue
                        if delta_type == 'input_json_delta':
                            block_index = event_payload.get('index')
                            partial_json = delta.get('partial_json')
                            if (
                                isinstance(block_index, int)
                                and block_index in tool_block_indexes
                                and isinstance(partial_json, str)
                                and partial_json
                            ):
                                yield StreamEvent(
                                    type='tool_call_delta',
                                    tool_call_index=tool_block_indexes[block_index],
                                    arguments_delta=partial_json,
                                    raw_event=event_payload,
                                )
                            continue
                        if delta_type in ('thinking_delta', 'signature_delta'):
                            # Extended-thinking blocks: we don't surface them as
                            # content (the runtime has nowhere to put them), but
                            # they previously vanished silently and burned the
                            # whole 4k output budget before tool_use could start.
                            # Logging once is enough to confirm they're flowing.
                            continue
                        _log_unknown_anthropic_event(
                            f'content_block_delta:{delta_type}', event_payload
                        )
                        continue
                    if event_type == 'message_delta':
                        delta = event_payload.get('delta')
                        if isinstance(delta, dict) and isinstance(delta.get('stop_reason'), str):
                            finish_reason = delta['stop_reason']
                        usage = _parse_usage(event_payload.get('usage'))
                        if usage.total_tokens:
                            yield StreamEvent(
                                type='usage',
                                usage=usage,
                                raw_event=event_payload,
                            )
                        continue
                    if event_type == 'message_stop':
                        yield StreamEvent(
                            type='message_stop',
                            finish_reason=finish_reason,
                            raw_event=event_payload,
                        )
        except error.HTTPError as exc:
            detail = exc.read().decode('utf-8', errors='replace')
            raise OpenAICompatError(
                f'HTTP {exc.code} from local model backend: {detail}'
            ) from exc
        except error.URLError as exc:
            raise OpenAICompatError(
                f'Unable to reach local model backend at {self.config.base_url}: {exc.reason}'
            ) from exc

    def _parse_anthropic_message_response(self, payload: dict[str, Any]) -> AssistantTurn:
        raw_content = payload.get('content')
        if not isinstance(raw_content, list):
            raise OpenAICompatError('Anthropic backend returned no content blocks')
        text_parts: list[str] = []
        tool_calls: list[ToolCall] = []
        for index, block in enumerate(raw_content):
            if not isinstance(block, dict):
                continue
            if block.get('type') == 'text':
                text = block.get('text')
                if isinstance(text, str):
                    text_parts.append(text)
                continue
            if block.get('type') == 'tool_use':
                name = block.get('name')
                if not isinstance(name, str) or not name:
                    raise OpenAICompatError('Tool call missing function name')
                call_id = block.get('id')
                if not isinstance(call_id, str) or not call_id:
                    call_id = f'call_{index}'
                arguments = block.get('input')
                if not isinstance(arguments, dict):
                    arguments = {}
                tool_calls.append(ToolCall(id=call_id, name=name, arguments=arguments))
        finish_reason = payload.get('stop_reason')
        if finish_reason is not None and not isinstance(finish_reason, str):
            finish_reason = str(finish_reason)
        return AssistantTurn(
            content=''.join(text_parts),
            tool_calls=tuple(tool_calls),
            finish_reason=finish_reason,
            raw_message=payload,
            usage=_parse_usage(payload.get('usage')),
        )

    def _anthropic_assistant_blocks(self, message: dict[str, Any]) -> list[dict[str, Any]]:
        blocks: list[dict[str, Any]] = []
        content = _normalize_content(message.get('content'))
        if content:
            blocks.append({'type': 'text', 'text': content})
        raw_tool_calls = message.get('tool_calls')
        if isinstance(raw_tool_calls, list):
            for index, raw_call in enumerate(raw_tool_calls):
                if not isinstance(raw_call, dict):
                    continue
                function_block = raw_call.get('function')
                if not isinstance(function_block, dict):
                    continue
                name = function_block.get('name')
                if not isinstance(name, str) or not name:
                    continue
                call_id = raw_call.get('id')
                if not isinstance(call_id, str) or not call_id:
                    call_id = f'call_{index}'
                try:
                    arguments = _parse_tool_arguments(function_block.get('arguments'))
                except OpenAICompatError:
                    arguments = {}
                blocks.append(
                    {
                        'type': 'tool_use',
                        'id': call_id,
                        'name': name,
                        'input': arguments,
                    }
                )
        return blocks

    def _anthropic_text_blocks(self, content: Any) -> list[dict[str, Any]]:
        normalized = _normalize_content(content)
        if not normalized:
            return []
        return [{'type': 'text', 'text': normalized}]

    def _anthropic_tools(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
        converted: list[dict[str, Any]] = []
        for tool in tools:
            if not isinstance(tool, dict):
                continue
            function_block = tool.get('function')
            if not isinstance(function_block, dict):
                continue
            name = function_block.get('name')
            if not isinstance(name, str) or not name:
                continue
            entry: dict[str, Any] = {
                'name': name,
                'input_schema': function_block.get('parameters') or {'type': 'object'},
            }
            description = function_block.get('description')
            if isinstance(description, str) and description:
                entry['description'] = description
            converted.append(entry)
        return converted

    def _append_anthropic_message(
        self,
        messages: list[dict[str, Any]],
        role: str,
        blocks: list[dict[str, Any]],
    ) -> None:
        if not blocks:
            return
        if messages and messages[-1].get('role') == role:
            previous = messages[-1].get('content')
            if isinstance(previous, list):
                previous.extend(blocks)
                return
        messages.append({'role': role, 'content': blocks})

    def _parse_tool_calls_from_message(self, message: dict[str, Any]) -> list[ToolCall]:
        tool_calls: list[ToolCall] = []
        raw_tool_calls = message.get('tool_calls')
        if isinstance(raw_tool_calls, list):
            for idx, raw_call in enumerate(raw_tool_calls):
                if not isinstance(raw_call, dict):
                    raise OpenAICompatError('Malformed tool call payload from model')
                function_block = raw_call.get('function') or {}
                if not isinstance(function_block, dict):
                    raise OpenAICompatError('Malformed tool call function payload from model')
                name = function_block.get('name')
                if not isinstance(name, str) or not name:
                    raise OpenAICompatError('Tool call missing function name')
                call_id = raw_call.get('id')
                if not isinstance(call_id, str) or not call_id:
                    call_id = f'call_{idx}'
                arguments = _parse_tool_arguments(function_block.get('arguments'))
                tool_calls.append(ToolCall(id=call_id, name=name, arguments=arguments))
        elif isinstance(message.get('function_call'), dict):
            function_call = message['function_call']
            name = function_call.get('name')
            if not isinstance(name, str) or not name:
                raise OpenAICompatError('Function call missing name')
            arguments = _parse_tool_arguments(function_call.get('arguments'))
            tool_calls.append(ToolCall(id='call_0', name=name, arguments=arguments))
        return tool_calls

    def _iter_sse_payloads(self, response: Any) -> Iterator[dict[str, Any]]:
        buffer: list[str] = []
        while True:
            line = response.readline()
            if not line:
                break
            if isinstance(line, bytes):
                text = line.decode('utf-8', errors='replace')
            else:
                text = str(line)
            stripped = text.strip()
            if not stripped:
                if not buffer:
                    continue
                joined = '\n'.join(buffer)
                buffer.clear()
                if joined == '[DONE]':
                    break
                try:
                    payload = json.loads(joined)
                except json.JSONDecodeError as exc:
                    raise OpenAICompatError(
                        f'Invalid JSON in streaming response: {joined!r}'
                    ) from exc
                if not isinstance(payload, dict):
                    raise OpenAICompatError('Malformed SSE payload from model backend')
                yield payload
                continue
            if stripped.startswith('data:'):
                buffer.append(stripped[5:].strip())

        if buffer:
            joined = '\n'.join(buffer)
            if joined != '[DONE]':
                try:
                    payload = json.loads(joined)
                except json.JSONDecodeError as exc:
                    raise OpenAICompatError(
                        f'Invalid trailing JSON in streaming response: {joined!r}'
                    ) from exc
                if not isinstance(payload, dict):
                    raise OpenAICompatError('Malformed trailing SSE payload from model backend')
                yield payload

    def _parse_stream_payload(
        self,
        payload: dict[str, Any],
    ) -> Iterator[StreamEvent]:
        usage = _parse_usage(payload.get('usage'))
        if usage.total_tokens:
            yield StreamEvent(
                type='usage',
                usage=usage,
                raw_event=payload,
            )

        choices = payload.get('choices')
        if not isinstance(choices, list):
            return

        for choice in choices:
            if not isinstance(choice, dict):
                continue
            delta = choice.get('delta')
            if not isinstance(delta, dict):
                delta = {}
            content = delta.get('content')
            if isinstance(content, str) and content:
                yield StreamEvent(
                    type='content_delta',
                    delta=content,
                    raw_event=choice,
                )
            tool_calls = delta.get('tool_calls')
            if isinstance(tool_calls, list):
                for raw_tool_call in tool_calls:
                    if not isinstance(raw_tool_call, dict):
                        continue
                    function_block = raw_tool_call.get('function')
                    if not isinstance(function_block, dict):
                        function_block = {}
                    yield StreamEvent(
                        type='tool_call_delta',
                        tool_call_index=(
                            raw_tool_call.get('index')
                            if isinstance(raw_tool_call.get('index'), int)
                            else 0
                        ),
                        tool_call_id=(
                            raw_tool_call.get('id')
                            if isinstance(raw_tool_call.get('id'), str)
                            else None
                        ),
                        tool_name=(
                            function_block.get('name')
                            if isinstance(function_block.get('name'), str)
                            else None
                        ),
                        arguments_delta=(
                            function_block.get('arguments')
                            if isinstance(function_block.get('arguments'), str)
                            else ''
                        ),
                        raw_event=raw_tool_call,
                    )
            finish_reason = choice.get('finish_reason')
            if finish_reason is not None:
                if not isinstance(finish_reason, str):
                    finish_reason = str(finish_reason)
                yield StreamEvent(
                    type='message_stop',
                    finish_reason=finish_reason,
                    raw_event=choice,
                )