"""Invocation-local text and completion checks for compatible chat streams."""

from __future__ import annotations

from collections.abc import Mapping
from dataclasses import dataclass, field
from typing import Any


def needs_qwen_flash_next_integrity(*model_names: str | None) -> bool:
    """Match the affected model id or its local alias, without broad Qwen rules."""
    return any(
        name.strip().lower().replace("_", "-").rsplit("/", 1)[-1] == "qwen3.8-flash-next"
        for name in model_names
        if isinstance(name, str)
    )


class IncompleteChatStreamError(RuntimeError):
    """A provider stream ended before a complete response could be committed.

    Only a fixed reason is exposed. Provider text, URLs and error bodies must
    never appear in an exception that the gateway may send to a browser.
    """

    def __init__(self, reason: str) -> None:
        self.reason = reason
        super().__init__(f"Incomplete OpenAI-compatible chat stream: {reason}")


@dataclass
class ChatStreamState:
    """Track one request, never a model instance or a persisted conversation."""

    observed_chat_frame: bool = False
    terminal_received: bool = False
    content: dict[int, str] = field(default_factory=dict)
    reasoning: dict[int, str] = field(default_factory=dict)

    def observe(self, choices: list[dict[str, Any]]) -> None:
        self.observed_chat_frame = True
        # LangChain consumes the first choice, so another choice's finish
        # reason must not make an unfinished first choice look complete.
        if choices and choices[0].get("finish_reason"):
            self.terminal_received = True

    def ensure_finished(self) -> None:
        # Responses API and adapter-generated chunks do not pass through the
        # chat-completions converter and have their own completion semantics.
        if self.observed_chat_frame and not self.terminal_received:
            raise IncompleteChatStreamError("missing_terminal_frame")

    def normalize_text(
        self,
        *,
        index: int,
        delta: dict[str, Any],
        original_delta: Mapping[str, Any] | None,
        message: Mapping[str, Any] | None,
    ) -> None:
        """Convert explicit full-message snapshots to suffixes.

        A real delta always wins and is appended verbatim, including repeated
        words. Only text supplied through ``message`` is treated as a full
        snapshot; prefix guessing on ordinary deltas would lose real tokens.
        """
        for fields, accumulated in (
            (("content",), self.content),
            (("reasoning_content", "reasoning"), self.reasoning),
        ):
            source = original_delta or {}
            current = accumulated.get(index, "")
            incoming = _first_text(source, fields)
            if incoming is not None:
                accumulated[index] = current + incoming[1]
                continue
            # Structured content and vendor reasoning objects are opaque.
            # Preserve them instead of replacing them with a text snapshot.
            if any(source.get(name) is not None and not isinstance(source[name], str) for name in fields):
                continue

            snapshot = _first_text(message or {}, fields)
            if snapshot is None:
                continue
            name, text = snapshot
            if text.startswith(current):
                suffix = text[len(current):]
                accumulated[index] = text
            elif current.startswith(text):
                # A duplicate/stale snapshot adds nothing to the live text.
                suffix = ""
            else:
                # A delta cannot retract already-yielded bytes. Retry before
                # committing contradictory text into the conversation.
                raise IncompleteChatStreamError("inconsistent_message_snapshot")
            for alias in fields:
                delta.pop(alias, None)
            delta[name] = suffix


def _first_text(source: Mapping[str, Any], fields: tuple[str, ...]) -> tuple[str, str] | None:
    for name in fields:
        value = source.get(name)
        if isinstance(value, str) and value:
            return name, value
    return None