core-python/tests/test_space_chat.py

"""Tests for the shared public chat runtime."""

from __future__ import annotations

import pytest

from maris_core.space_chat import (
    DEFAULT_CHAT_MODEL,
    SpaceChatRequest,
    build_space_chat_messages,
    generate_space_chat_reply,
    resolve_space_chat_models,
)
from maris_core.utils.hf_integration import HFIntegration


def test_space_chat_request_accepts_external_hf_model_ids() -> None:
    request = SpaceChatRequest(message="Sveiki", model="Qwen/Qwen3-Coder-480B-A35B-Instruct")

    assert request.model == "Qwen/Qwen3-Coder-480B-A35B-Instruct"


class _StopIterationChoices:
    def __bool__(self) -> bool:
        return True

    def __getitem__(self, index: int) -> object:
        raise StopIteration(index)

    def __iter__(self):
        return iter(())


class _MalformedChatClient:
    def __init__(self, token: str | None = None) -> None:
        self.token = token
        self.generation_models: list[str] = []

    def chat_completion(self, **_: object) -> dict[str, object]:
        return {"choices": _StopIterationChoices()}

    def text_generation(self, **kwargs: object) -> str:
        self.generation_models.append(str(kwargs["model"]))
        return "Fallback response from text_generation."


class _StopIterationChatClient:
    def __init__(self, token: str | None = None) -> None:
        self.token = token
        self.generation_models: list[str] = []

    def chat_completion(self, **_: object) -> dict[str, object]:
        return next(iter(()))

    def text_generation(self, **kwargs: object) -> str:
        self.generation_models.append(str(kwargs["model"]))
        return "Fallback response after StopIteration."


@pytest.mark.asyncio
async def test_generate_space_chat_reply_uses_hf_inference_space_runtime_config(
    monkeypatch,
) -> None:
    captured_kwargs: dict[str, object] = {}

    async def fake_save_conversation(
        self,
        prompt: str,
        response: str,
        metadata: dict[str, object] | None = None,
    ) -> None:
        del self, prompt, response, metadata

    class _ConfiguredChatClient:
        def __init__(self, **kwargs: object) -> None:
            captured_kwargs.update(kwargs)

        def chat_completion(self, **kwargs: object) -> dict[str, object]:
            return {
                "choices": [
                    {
                        "message": {
                            "content": f"Atbilde no {kwargs['model']}",
                        }
                    }
                ]
            }

    monkeypatch.setattr(HFIntegration, "save_conversation", fake_save_conversation)
    monkeypatch.setenv("HF_INFERENCE_API_KEY", "hf_inference_secret")

    response = await generate_space_chat_reply(
        SpaceChatRequest(message="Palīdzi ar manu jautājumu"),
        client_factory=_ConfiguredChatClient,
    )

    assert response.model == DEFAULT_CHAT_MODEL
    assert captured_kwargs == {
        "provider": "hf-inference",
        "base_url": "https://api-inference.huggingface.co",
        "token": "hf_inference_secret",
    }


@pytest.mark.asyncio
async def test_generate_space_chat_reply_falls_back_on_malformed_chat_completion(
    monkeypatch,
) -> None:
    async def fake_save_conversation(
        self,
        prompt: str,
        response: str,
        metadata: dict[str, object] | None = None,
    ) -> None:
        del self, prompt, response, metadata

    monkeypatch.setattr(HFIntegration, "save_conversation", fake_save_conversation)

    response = await generate_space_chat_reply(
        SpaceChatRequest(message="Palīdzi ar manu jautājumu"),
        client_factory=_MalformedChatClient,
    )

    assert response.response == "Fallback response from text_generation."
    assert response.model == DEFAULT_CHAT_MODEL


@pytest.mark.asyncio
async def test_generate_space_chat_reply_falls_back_when_chat_completion_raises_stop_iteration(
    monkeypatch,
) -> None:
    async def fake_save_conversation(
        self,
        prompt: str,
        response: str,
        metadata: dict[str, object] | None = None,
    ) -> None:
        del self, prompt, response, metadata

    monkeypatch.setattr(HFIntegration, "save_conversation", fake_save_conversation)

    response = await generate_space_chat_reply(
        SpaceChatRequest(message="Palīdzi ar manu jautājumu"),
        client_factory=_StopIterationChatClient,
    )

    assert response.response == "Fallback response after StopIteration."
    assert response.model == DEFAULT_CHAT_MODEL


def test_resolve_space_chat_models_appends_hidden_fallbacks(monkeypatch) -> None:
    monkeypatch.setenv("MARIS_CHAT_MODEL", "MarisUK/maris-ai-text")
    monkeypatch.setenv(
        "MARIS_CHAT_FALLBACK_MODELS",
        "MarisUK/maris-assistant-runtime-fallback,Qwen/Qwen3-Coder-480B-A35B-Instruct",
    )

    models = resolve_space_chat_models("MarisUK/offline-model")

    assert models == (
        "MarisUK/offline-model",
        "MarisUK/maris-ai-text",
        "MarisUK/maris-assistant-runtime-fallback",
        "Qwen/Qwen3-Coder-480B-A35B-Instruct",
    )


def test_build_space_chat_messages_drops_pending_duplicate_user_turn() -> None:
    messages = build_space_chat_messages(
        SpaceChatRequest(
            message="Palīdzi ar plānu",
            history=[
                {"role": "user", "content": "Iepriekšējais jautājums"},
                {"role": "assistant", "content": "Iepriekšējā atbilde"},
                {"role": "user", "content": "Palīdzi ar plānu"},
            ],
        )
    )

    assert messages[-3:] == [
        {"role": "user", "content": "Iepriekšējais jautājums"},
        {"role": "assistant", "content": "Iepriekšējā atbilde"},
        {"role": "user", "content": "Palīdzi ar plānu"},
    ]


class _RecoveringChatClient:
    def __init__(self, token: str | None = None) -> None:
        self.token = token
        self.chat_models: list[str] = []

    def chat_completion(self, **kwargs: object) -> dict[str, object]:
        model = str(kwargs["model"])
        self.chat_models.append(model)
        if model == "MarisUK/offline-model":
            raise OSError("primary model unavailable")
        return {
            "choices": [
                {
                    "message": {
                        "content": f"Atbilde no {model}",
                    }
                }
            ]
        }


@pytest.mark.asyncio
async def test_generate_space_chat_reply_tries_fallback_models_when_primary_fails(
    monkeypatch,
) -> None:
    async def fake_save_conversation(
        self,
        prompt: str,
        response: str,
        metadata: dict[str, object] | None = None,
    ) -> None:
        del self, prompt, response, metadata

    monkeypatch.setattr(HFIntegration, "save_conversation", fake_save_conversation)
    monkeypatch.setenv("MARIS_CHAT_MODEL", "MarisUK/offline-model")
    monkeypatch.setenv("MARIS_CHAT_MODELS", "")
    monkeypatch.setenv("MARIS_CHAT_FALLBACK_MODELS", "Qwen/Qwen3-Coder-480B-A35B-Instruct")

    response = await generate_space_chat_reply(
        SpaceChatRequest(message="Sveiki"),
        client_factory=_RecoveringChatClient,
    )

    assert response.response == "Atbilde no Qwen/Qwen3-Coder-480B-A35B-Instruct"
    assert response.model == "Qwen/Qwen3-Coder-480B-A35B-Instruct"


class _UnavailableChatClient:
    def __init__(self, token: str | None = None) -> None:
        self.token = token

    def chat_completion(self, **kwargs: object) -> dict[str, object]:
        del kwargs
        raise RuntimeError("provider rejected model")

    def text_generation(self, **kwargs: object) -> str:
        del kwargs
        raise RuntimeError("generation unavailable")


@pytest.mark.asyncio
async def test_generate_space_chat_reply_returns_emergency_fallback_when_all_models_fail(
    monkeypatch,
) -> None:
    async def fake_save_conversation(
        self,
        prompt: str,
        response: str,
        metadata: dict[str, object] | None = None,
    ) -> None:
        del self, prompt, metadata
        assert "drošo fallback režīmu" in response

    monkeypatch.setattr(HFIntegration, "save_conversation", fake_save_conversation)

    response = await generate_space_chat_reply(
        SpaceChatRequest(message="Sveiki", model="deepseek-ai/DeepSeek-V3.2"),
        client_factory=_UnavailableChatClient,
    )

    assert "drošo fallback režīmu" in response.response
    assert response.model == DEFAULT_CHAT_MODEL


@pytest.mark.asyncio
async def test_generate_space_chat_reply_saves_rich_metadata(monkeypatch) -> None:
    captured_metadata: dict[str, object] = {}

    async def fake_save_conversation(
        self,
        prompt: str,
        response: str,
        metadata: dict[str, object] | None = None,
    ) -> None:
        del self
        assert prompt == "Sveiki"
        assert response == "Atbilde no MarisUK/maris-ai-text"
        captured_metadata.update(metadata or {})

    class _HealthyChatClient:
        def __init__(self, token: str | None = None) -> None:
            self.token = token

        def chat_completion(self, **kwargs: object) -> dict[str, object]:
            return {
                "choices": [
                    {
                        "message": {
                            "content": f"Atbilde no {kwargs['model']}",
                        }
                    }
                ]
            }

    monkeypatch.setattr(HFIntegration, "save_conversation", fake_save_conversation)

    response = await generate_space_chat_reply(
        SpaceChatRequest(
            message="Sveiki",
            history=[{"role": "user", "content": "Sveiki"}],
            persona_id="strategist",
            session_id="space-session-1",
        ),
        client_factory=_HealthyChatClient,
    )

    assert response.model == DEFAULT_CHAT_MODEL
    assert captured_metadata == {
        "session_id": "space-session-1",
        "persona_id": "strategist",
        "requested_model": DEFAULT_CHAT_MODEL,
        "resolved_model": DEFAULT_CHAT_MODEL,
        "history_messages": 1,
        "detected_emotion": "neutral",
        "emotion_confidence": 0.45,
        "response_style": "clear_grounded",
        "space_repo": "MarisUK/maris.ai.chat",
        "public_space_chat": True,
    }