"""Focused regression tests for administrator model configuration helpers."""

from __future__ import annotations

import errno
from pathlib import Path
from types import SimpleNamespace

import pytest
import yaml
from fastapi import HTTPException
from langchain_core.messages import HumanMessage
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from starlette.routing import Match

from app.gateway import services
from app.gateway.routers import model_management as router
from app.gateway.routers import models as public_models
from deerflow.config.app_config import AppConfig
from deerflow.config.model_config import ModelConfig
from deerflow.models.factory import create_chat_model
from deerflow.persistence.base import Base
from deerflow.persistence.model_capabilities import (
    MemoryModelCapabilityStore,
    ModelCapabilityRepository,
    ModelCapabilityRow,
    import_legacy_reasoning_effort_capabilities,
)


def test_config_atomic_replace_retries_transient_busy_filesystem_error(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path = tmp_path / "config.yaml"
    original = "models:\n  - name: old\n"
    updated = "models:\n  - name: new\n"
    config_path.write_text(original, encoding="utf-8")
    real_replace = router.os.replace
    attempts = 0

    def replace_with_transient_busy(source: str, target: Path) -> None:
        nonlocal attempts
        attempts += 1
        if attempts < 3:
            raise OSError(errno.EBUSY, "resource busy", str(target))
        real_replace(source, target)

    monkeypatch.setattr(router.os, "replace", replace_with_transient_busy)
    monkeypatch.setattr(router, "sleep", lambda _delay: None)

    backup = router._backup_and_write_config(config_path, original, updated)

    assert attempts == 3
    assert config_path.read_text(encoding="utf-8") == updated
    assert backup.read_text(encoding="utf-8") == original


def test_config_atomic_replace_does_not_retry_permanent_error(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path = tmp_path / "config.yaml"
    original = "models:\n  - name: old\n"
    config_path.write_text(original, encoding="utf-8")
    attempts = 0

    def fail_replace(_source: str, _target: Path) -> None:
        nonlocal attempts
        attempts += 1
        raise OSError(errno.EACCES, "permission denied")

    monkeypatch.setattr(router.os, "replace", fail_replace)

    with pytest.raises(OSError, match="permission denied"):
        router._backup_and_write_config(config_path, original, "models:\n  - name: new\n")

    assert attempts == 1
    assert config_path.read_text(encoding="utf-8") == original
    assert not list(tmp_path.glob(".config.yaml.*.tmp"))


def test_config_single_file_mount_is_updated_without_atomic_rename(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path = tmp_path / "config.yaml"
    original = "models:\n  - name: old\n"
    updated = "models:\n  - name: mounted-new\n"
    config_path.write_text(original, encoding="utf-8")
    monkeypatch.setattr(router, "_is_file_mountpoint", lambda _path: True)

    def unexpected_replace(_source: str, _target: Path) -> None:
        pytest.fail("single-file bind mount must not be replaced by rename")

    monkeypatch.setattr(router.os, "replace", unexpected_replace)

    backup = router._backup_and_write_config(config_path, original, updated)

    assert config_path.read_text(encoding="utf-8") == updated
    assert backup.read_text(encoding="utf-8") == original
    assert not list(tmp_path.glob(".config.yaml.*.tmp"))


def test_replace_models_block_keeps_other_sections_and_comments() -> None:
    raw = """config_version: 1
models:
  # Existing provider note
  - name: old-model
    use: langchain_openai:ChatOpenAI
    model: old-model
# Disabled provider example remains useful to operators.
# - name: disabled-model
#   model: disabled-model
agents:
  default: lead_agent
"""

    replaced = router._replace_models_block(
        raw,
        [
            {
                "name": "new-model",
                "use": "langchain_openai:ChatOpenAI",
                "model": "new-model",
            }
        ],
    )

    parsed = yaml.safe_load(replaced)
    assert parsed["models"] == [
        {
            "name": "new-model",
            "use": "langchain_openai:ChatOpenAI",
            "model": "new-model",
        }
    ]
    assert parsed["agents"] == {"default": "lead_agent"}
    assert "# Existing provider note" in replaced
    assert "# Disabled provider example remains useful to operators." in replaced


def test_connection_error_redaction_hides_literal_and_resolved_keys() -> None:
    message = "provider rejected raw-secret and resolved-secret"
    assert router._redact_text(message, ["raw-secret", "resolved-secret"]) == "provider rejected *** and ***"


def test_legacy_model_edit_preserves_write_file_transport() -> None:
    existing = {
        "name": "bridge-model",
        "use": "langchain_openai:ChatOpenAI",
        "model": "bridge-model",
        "write_file_transport": "text_bridge",
    }

    merged = router._request_to_raw_model(
        router.ModelUpsertRequest(
            name="bridge-model",
            use="langchain_openai:ChatOpenAI",
            model="bridge-model-v2",
        ),
        existing,
    )

    assert merged["model"] == "bridge-model-v2"
    assert merged["write_file_transport"] == "text_bridge"

    changed = router._request_to_raw_model(
        router.ModelUpsertRequest(
            name="bridge-model",
            use="langchain_openai:ChatOpenAI",
            model="bridge-model-v2",
            write_file_transport="native",
        ),
        merged,
    )
    assert changed["write_file_transport"] == "native"
    assert type(changed["write_file_transport"]) is str


def test_model_yaml_render_strips_database_owned_capability_from_every_row() -> None:
    assert router._strip_database_owned_fields(
        [
            {"name": "first", "supports_reasoning_effort": True},
            {"name": "second", "supports_reasoning_effort": False},
        ]
    ) == [{"name": "first"}, {"name": "second"}]


def test_model_mutation_routes_accept_names_with_slashes() -> None:
    for method in ("PUT", "DELETE"):
        route = next(item for item in router.router.routes if getattr(item, "methods", set()) == {method} and getattr(item, "path", "") == "/api/admin/models/{model_name:path}")
        match, child_scope = route.matches(
            {
                "type": "http",
                "method": method,
                "path": "/api/admin/models/zai-org/GLM-5-FP8",
            }
        )
        assert match is Match.FULL
        assert child_scope["path_params"] == {"model_name": "zai-org/GLM-5-FP8"}


def test_reasoning_capability_route_precedes_generic_model_update_and_accepts_slashes() -> None:
    scope = {
        "type": "http",
        "method": "PUT",
        "path": "/api/admin/models/capabilities/reasoning-effort/zai-org/GLM-5-FP8",
    }
    matching_paths = [route.path for route in router.router.routes if route.matches(scope)[0] is Match.FULL]

    assert matching_paths[:2] == [
        "/api/admin/models/capabilities/reasoning-effort/{model_name:path}",
        "/api/admin/models/{model_name:path}",
    ]


def test_write_file_transport_route_precedes_generic_model_update() -> None:
    scope = {
        "type": "http",
        "method": "PUT",
        "path": "/api/admin/models/write-file-transport/zai-org/GLM-5-FP8",
    }
    matching_paths = [route.path for route in router.router.routes if route.matches(scope)[0] is Match.FULL]

    assert matching_paths[:2] == [
        "/api/admin/models/write-file-transport/{model_name:path}",
        "/api/admin/models/{model_name:path}",
    ]


def _admin_request():
    return SimpleNamespace(
        state=SimpleNamespace(user=SimpleNamespace(system_role="admin")),
        app=SimpleNamespace(state=SimpleNamespace()),
    )


@pytest.mark.asyncio
async def test_write_file_transport_toggle_persists_native_and_buffered_gateway(
    tmp_path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        """models:
  - name: flash
    use: langchain_openai:ChatOpenAI
    model: flash
""",
        encoding="utf-8",
    )
    saved: dict[str, object] = {}

    async def save_models(_request, raw_before: str, models: list[dict[str, object]]) -> list[dict[str, object]]:
        saved["raw_before"] = raw_before
        saved["models"] = models
        return models

    monkeypatch.setattr(router, "_resolve_config_path", lambda: config_path)
    monkeypatch.setattr(router, "_save_models", save_models)

    enabled = await router.update_write_file_transport(
        "flash",
        router.WriteFileTransportToggleRequest(write_file_transport="buffered_gateway"),
        _admin_request(),
    )

    assert enabled.write_file_transport == "buffered_gateway"
    assert saved["models"] == [
        {
            "name": "flash",
            "use": "langchain_openai:ChatOpenAI",
            "model": "flash",
            "write_file_transport": "buffered_gateway",
        }
    ]

    disabled = await router.update_write_file_transport(
        "flash",
        router.WriteFileTransportToggleRequest(write_file_transport="native"),
        _admin_request(),
    )

    assert disabled.write_file_transport == "native"
    assert saved["models"][0]["write_file_transport"] == "native"


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("use", "use_responses_api"),
    [
        ("langchain_anthropic:ChatAnthropic", None),
        ("langchain_openai:ChatOpenAI", True),
    ],
)
async def test_write_file_transport_rejects_incompatible_models(
    use: str,
    use_responses_api: bool | None,
    tmp_path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path = tmp_path / "config.yaml"
    original = f"""models:
  - name: incompatible
    use: {use}
    model: incompatible
"""
    if use_responses_api:
        original = original.replace(
            "    model: incompatible\n",
            "    model: incompatible\n    use_responses_api: true\n",
        )
    config_path.write_text(original, encoding="utf-8")
    monkeypatch.setattr(router, "_resolve_config_path", lambda: config_path)

    async def fail_save_models(_request, _raw_before, _models):
        raise AssertionError("incompatible models must not be persisted")

    monkeypatch.setattr(router, "_save_models", fail_save_models)

    with pytest.raises(HTTPException) as exc_info:
        await router.update_write_file_transport(
            "incompatible",
            router.WriteFileTransportToggleRequest(write_file_transport="buffered_gateway"),
            _admin_request(),
        )

    assert exc_info.value.status_code == 422
    assert config_path.read_text(encoding="utf-8") == original


def test_strict_stream_integrity_route_precedes_generic_model_update() -> None:
    scope = {
        "type": "http",
        "method": "PUT",
        "path": "/api/admin/models/strict-stream-integrity/zai-org/GLM-5-FP8",
    }
    matching_paths = [route.path for route in router.router.routes if route.matches(scope)[0] is Match.FULL]

    assert matching_paths[:2] == [
        "/api/admin/models/strict-stream-integrity/{model_name:path}",
        "/api/admin/models/{model_name:path}",
    ]


@pytest.mark.asyncio
async def test_strict_stream_integrity_toggle_persists_explicit_bool(
    tmp_path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        """models:
  - name: flash
    use: langchain_openai:ChatOpenAI
    model: flash
""",
        encoding="utf-8",
    )
    saved: dict[str, object] = {}

    async def save_models(_request, raw_before: str, models: list[dict[str, object]]) -> list[dict[str, object]]:
        saved["models"] = models
        return models

    monkeypatch.setattr(router, "_resolve_config_path", lambda: config_path)
    monkeypatch.setattr(router, "_save_models", save_models)

    enabled = await router.update_strict_stream_integrity(
        "flash",
        router.StrictStreamIntegrityToggleRequest(strict_stream_integrity=True),
        _admin_request(),
    )

    assert enabled.strict_stream_integrity is True
    assert saved["models"] == [
        {
            "name": "flash",
            "use": "langchain_openai:ChatOpenAI",
            "model": "flash",
            "strict_stream_integrity": True,
        }
    ]

    disabled = await router.update_strict_stream_integrity(
        "flash",
        router.StrictStreamIntegrityToggleRequest(strict_stream_integrity=False),
        _admin_request(),
    )

    assert disabled.strict_stream_integrity is False
    assert saved["models"][0]["strict_stream_integrity"] is False


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("use", "use_responses_api"),
    [
        ("langchain_anthropic:ChatAnthropic", None),
        ("langchain_openai:ChatOpenAI", True),
    ],
)
async def test_strict_stream_integrity_rejects_incompatible_models(
    use: str,
    use_responses_api: bool | None,
    tmp_path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path = tmp_path / "config.yaml"
    original = f"""models:
  - name: incompatible
    use: {use}
    model: incompatible
"""
    if use_responses_api:
        original = original.replace(
            "    model: incompatible\n",
            "    model: incompatible\n    use_responses_api: true\n",
        )
    config_path.write_text(original, encoding="utf-8")
    monkeypatch.setattr(router, "_resolve_config_path", lambda: config_path)

    async def fail_save_models(_request, _raw_before, _models):
        raise AssertionError("incompatible models must not be persisted")

    monkeypatch.setattr(router, "_save_models", fail_save_models)

    with pytest.raises(HTTPException) as exc_info:
        await router.update_strict_stream_integrity(
            "incompatible",
            router.StrictStreamIntegrityToggleRequest(strict_stream_integrity=True),
            _admin_request(),
        )

    assert exc_info.value.status_code == 422
    assert config_path.read_text(encoding="utf-8") == original


def test_legacy_model_edit_preserves_strict_stream_integrity() -> None:
    existing = {
        "name": "relay-model",
        "use": "langchain_openai:ChatOpenAI",
        "model": "relay-model",
        "strict_stream_integrity": True,
    }

    merged = router._request_to_raw_model(
        router.ModelUpsertRequest(
            name="relay-model",
            use="langchain_openai:ChatOpenAI",
            model="relay-model-v2",
        ),
        existing,
    )

    assert merged["model"] == "relay-model-v2"
    assert merged["strict_stream_integrity"] is True

    changed = router._request_to_raw_model(
        router.ModelUpsertRequest(
            name="relay-model",
            use="langchain_openai:ChatOpenAI",
            model="relay-model-v2",
            strict_stream_integrity=False,
        ),
        merged,
    )
    assert changed["strict_stream_integrity"] is False


def _reasoning_request(store, *, config: AppConfig | None = None):
    return SimpleNamespace(
        state=SimpleNamespace(
            user=SimpleNamespace(id="admin-1", system_role="admin"),
        ),
        app=SimpleNamespace(
            state=SimpleNamespace(
                model_capability_store=store,
                config=config,
            )
        ),
    )


def _qwen_model(
    *,
    name: str = "zai-org/GLM-5-FP8",
    supports_thinking: bool = True,
    legacy_reasoning_effort: bool = False,
) -> ModelConfig:
    return ModelConfig(
        name=name,
        use="langchain_openai:ChatOpenAI",
        model="Qwen/Qwen3.8-Flash-Next",
        api_key="test-key",
        base_url="http://models.invalid/v1",
        supports_thinking=supports_thinking,
        # This legacy parser field is deliberately the opposite of the DB row.
        supports_reasoning_effort=legacy_reasoning_effort,
    )


@pytest.mark.asyncio
async def test_legacy_true_capability_is_seeded_when_database_row_is_missing() -> None:
    store = MemoryModelCapabilityStore()

    imported = await import_legacy_reasoning_effort_capabilities(
        store,
        [_qwen_model(name="legacy", legacy_reasoning_effort=True)],
    )

    assert imported == ["legacy"]
    assert (await store.get_capability("legacy"))["supports_reasoning_effort"] is True


@pytest.mark.asyncio
@pytest.mark.parametrize("persisted_value", [False, True])
async def test_legacy_true_never_overwrites_existing_database_capability(
    persisted_value: bool,
) -> None:
    store = MemoryModelCapabilityStore()
    await store.set_reasoning_effort_support(
        "legacy",
        supports_reasoning_effort=persisted_value,
        updated_by="admin-1",
    )

    imported = await import_legacy_reasoning_effort_capabilities(
        store,
        [_qwen_model(name="legacy", legacy_reasoning_effort=True)],
    )
    persisted = await store.get_capability("legacy")

    assert imported == []
    assert persisted["supports_reasoning_effort"] is persisted_value
    assert persisted["updated_by"] == "admin-1"


@pytest.mark.asyncio
async def test_legacy_false_config_does_not_create_database_capability() -> None:
    store = MemoryModelCapabilityStore()

    imported = await import_legacy_reasoning_effort_capabilities(
        store,
        [_qwen_model(name="legacy", legacy_reasoning_effort=False)],
    )

    assert imported == []
    assert await store.get_capability("legacy") is None


@pytest.mark.asyncio
async def test_sql_model_capability_store_persists_across_repository_instances(tmp_path) -> None:
    engine = create_async_engine(
        f"sqlite+aiosqlite:///{tmp_path / 'model-capabilities.db'}",
    )
    async with engine.begin() as connection:
        await connection.run_sync(
            lambda sync_connection: Base.metadata.create_all(
                sync_connection,
                tables=[ModelCapabilityRow.__table__],
            )
        )
    session_factory = async_sessionmaker(engine, expire_on_commit=False)

    first = ModelCapabilityRepository(session_factory)
    await first.set_reasoning_effort_support(
        "zai-org/GLM-5-FP8",
        supports_reasoning_effort=True,
        updated_by="admin-1",
    )
    second = ModelCapabilityRepository(session_factory)

    persisted = await second.get_capability("zai-org/GLM-5-FP8")
    assert persisted is not None
    assert persisted["model_name"] == "zai-org/GLM-5-FP8"
    assert persisted["supports_reasoning_effort"] is True
    assert persisted["updated_by"] == "admin-1"
    assert persisted["updated_at"]
    await engine.dispose()


@pytest.mark.asyncio
async def test_reasoning_capability_endpoint_and_model_apis_use_database_not_yaml(
    tmp_path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path = tmp_path / "config.yaml"
    original = """models:
  - name: zai-org/GLM-5-FP8
    use: langchain_openai:ChatOpenAI
    model: Qwen/Qwen3.8-Flash-Next
    supports_thinking: true
    supports_reasoning_effort: false
"""
    config_path.write_text(original, encoding="utf-8")
    monkeypatch.setattr(router, "_resolve_config_path", lambda: config_path)
    store = MemoryModelCapabilityStore()
    config = AppConfig.model_construct(models=[_qwen_model()])
    request = _reasoning_request(store, config=config)

    updated = await router.update_reasoning_effort_capability(
        "zai-org/GLM-5-FP8",
        router.ReasoningEffortCapabilityRequest(
            supports_reasoning_effort=True,
            default_reasoning_effort="low",
        ),
        request,
    )
    admin_list = await router.list_model_configurations(request)
    public_list = await public_models.list_models(request, config)

    assert updated.supports_reasoning_effort is True
    assert updated.default_reasoning_effort == "low"
    assert admin_list.models[0].supports_reasoning_effort is True
    assert public_list.models[0].supports_reasoning_effort is True
    assert public_list.models[0].default_reasoning_effort == "low"
    assert (await store.get_capability("zai-org/GLM-5-FP8"))["default_reasoning_effort"] == "low"
    assert config_path.read_text(encoding="utf-8") == original


@pytest.mark.asyncio
async def test_reasoning_capability_can_only_be_enabled_for_thinking_models(
    tmp_path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        """models:
  - name: flash
    use: langchain_openai:ChatOpenAI
    model: flash
    supports_thinking: false
""",
        encoding="utf-8",
    )
    monkeypatch.setattr(router, "_resolve_config_path", lambda: config_path)
    store = MemoryModelCapabilityStore()

    with pytest.raises(HTTPException) as exc_info:
        await router.update_reasoning_effort_capability(
            "flash",
            router.ReasoningEffortCapabilityRequest(supports_reasoning_effort=True),
            _reasoning_request(store),
        )

    assert exc_info.value.status_code == 422
    assert await store.get_capability("flash") is None


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("supports_thinking", "expected_capability"),
    [(True, True), (False, False)],
)
async def test_model_update_preserves_capability_unless_thinking_is_disabled(
    supports_thinking: bool,
    expected_capability: bool,
    tmp_path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        """models:
  - name: qwen
    use: langchain_openai:ChatOpenAI
    model: Qwen/Qwen3.8-Flash-Next
    supports_thinking: true
""",
        encoding="utf-8",
    )
    monkeypatch.setattr(router, "_resolve_config_path", lambda: config_path)

    async def _save_models(_request, _raw_before, rows):
        return rows

    monkeypatch.setattr(router, "_save_models", _save_models)
    store = MemoryModelCapabilityStore()
    await store.set_reasoning_effort_support(
        "qwen",
        supports_reasoning_effort=True,
    )
    body = router.ModelUpsertRequest(
        name="qwen",
        use="langchain_openai:ChatOpenAI",
        model="Qwen/Qwen3.8-Flash-Next",
        supports_thinking=supports_thinking,
    )
    assert "supports_reasoning_effort" not in body.model_fields_set

    result = await router.update_model_configuration(
        "qwen",
        body,
        _reasoning_request(store),
    )

    assert result.supports_reasoning_effort is expected_capability
    assert (await store.get_capability("qwen"))["supports_reasoning_effort"] is expected_capability


@pytest.mark.asyncio
async def test_delete_model_configuration_removes_matching_model_and_activates_config(
    tmp_path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        """models:
  - name: zai-org/GLM-5-FP8
    use: langchain_openai:ChatOpenAI
    model: GLM-5-FP8
  - name: keep-me
    use: langchain_openai:ChatOpenAI
    model: keep-me
""",
        encoding="utf-8",
    )
    saved: dict[str, object] = {}

    async def save_models(_request, raw_before: str, models: list[dict[str, object]]) -> list[dict[str, object]]:
        saved["raw_before"] = raw_before
        saved["models"] = models
        return models

    monkeypatch.setattr(router, "_resolve_config_path", lambda: config_path)
    monkeypatch.setattr(router, "_save_models", save_models)
    request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(system_role="admin")))

    await router.delete_model_configuration("zai-org/GLM-5-FP8", request)

    assert "zai-org/GLM-5-FP8" in str(saved["raw_before"])
    assert saved["models"] == [
        {
            "name": "keep-me",
            "use": "langchain_openai:ChatOpenAI",
            "model": "keep-me",
        }
    ]


@pytest.mark.asyncio
async def test_reorder_model_configurations_persists_a_complete_permutation(
    tmp_path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        """models:
  - name: first
    use: langchain_openai:ChatOpenAI
    model: first
  - name: second
    use: langchain_openai:ChatOpenAI
    model: second
""",
        encoding="utf-8",
    )
    saved: dict[str, object] = {}

    async def save_models(_request, raw_before: str, models: list[dict[str, object]]) -> list[dict[str, object]]:
        saved["raw_before"] = raw_before
        saved["models"] = models
        return models

    monkeypatch.setattr(router, "_resolve_config_path", lambda: config_path)
    monkeypatch.setattr(router, "_save_models", save_models)
    request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(system_role="admin")))

    result = await router.reorder_model_configurations(
        router.ModelOrderRequest(names=["second", "first"]),
        request,
    )

    assert [item.name for item in result] == ["second", "first"]
    assert [item["name"] for item in saved["models"]] == ["second", "first"]


@pytest.mark.asyncio
async def test_reorder_model_configurations_rejects_stale_or_partial_orders(
    tmp_path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        """models:
  - name: first
    use: langchain_openai:ChatOpenAI
    model: first
  - name: second
    use: langchain_openai:ChatOpenAI
    model: second
""",
        encoding="utf-8",
    )
    monkeypatch.setattr(router, "_resolve_config_path", lambda: config_path)
    request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(system_role="admin")))

    with pytest.raises(Exception) as exc_info:
        await router.reorder_model_configurations(router.ModelOrderRequest(names=["first"]), request)

    assert getattr(exc_info.value, "status_code", None) == 409


@pytest.mark.asyncio
async def test_connection_test_uses_one_minimal_prompt_without_writing_config(
    tmp_path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path = tmp_path / "config.yaml"
    original = "models: []\nagent: lead_agent\n"
    config_path.write_text(original, encoding="utf-8")
    messages_seen = []

    class FakeChatModel:
        async def ainvoke(self, messages):
            messages_seen.extend(messages)
            return SimpleNamespace(content="OK")

    monkeypatch.setattr(router, "_resolve_config_path", lambda: config_path)
    monkeypatch.setattr(router, "create_chat_model", lambda *_args, **_kwargs: FakeChatModel())
    request = SimpleNamespace(
        state=SimpleNamespace(user=SimpleNamespace(system_role="admin")),
        app=SimpleNamespace(state=SimpleNamespace(config=SimpleNamespace(model_copy=lambda **_kwargs: SimpleNamespace()))),
    )
    body = router.ModelTestRequest(
        model=router.ModelUpsertRequest(
            name="test-model",
            use="langchain_openai:ChatOpenAI",
            model="test-model",
            api_key="raw-secret",
        )
    )

    result = await router.test_model_connection(body, request)

    assert result.success is True
    assert result.preview == "OK"
    assert len(messages_seen) == 1
    assert messages_seen[0].content == "Reply with exactly: OK"
    assert config_path.read_text(encoding="utf-8") == original


def test_force_disable_thinking_skips_generic_openai_models_without_thinking_support() -> None:
    """LiteLLM proxies reject vendor-specific thinking payloads for ordinary chat models."""
    config = AppConfig.model_construct(
        models=[
            ModelConfig(
                name="litellm-qwen",
                use="langchain_openai:ChatOpenAI",
                model="qwen3.6",
                api_key="test-key",
                base_url="http://litellm.example/v1",
                supports_thinking=False,
            )
        ]
    )

    chat_model = create_chat_model(
        "litellm-qwen",
        app_config=config,
        force_disable_thinking=True,
    )

    assert chat_model.extra_body is None


def test_openai_compatible_thinking_sends_enable_flag_by_default() -> None:
    config = AppConfig.model_construct(
        models=[
            ModelConfig(
                name="qwen-thinking",
                use="langchain_openai:ChatOpenAI",
                model="Qwen3.8-27B",
                api_key="test-key",
                base_url="http://vllm.example/v1",
                supports_thinking=True,
            )
        ]
    )

    chat_model = create_chat_model(
        "qwen-thinking",
        app_config=config,
        thinking_enabled=True,
    )

    assert chat_model.extra_body == {
        "enable_thinking": True,
        "chat_template_kwargs": {"enable_thinking": True},
    }


@pytest.mark.parametrize(
    ("requested", "expected"),
    [
        ("minimal", "low"),
        ("low", "low"),
        ("medium", "medium"),
        ("high", "xhigh"),
        ("xhigh", "xhigh"),
    ],
)
def test_qwen_reasoning_effort_is_top_level_only(requested: str, expected: str) -> None:
    config = AppConfig.model_construct(models=[_qwen_model(name="qwen")])

    chat_model = create_chat_model(
        "qwen",
        app_config=config,
        thinking_enabled=True,
        reasoning_effort=requested,
    )
    payload = chat_model._get_request_payload([HumanMessage(content="hello")])

    assert payload["reasoning_effort"] == expected
    assert "reasoning_effort" not in payload["extra_body"]["chat_template_kwargs"]
    assert payload["extra_body"]["chat_template_kwargs"]["enable_thinking"] is True


def test_final_reasoning_policy_overrides_both_preseeded_runtime_containers() -> None:
    config = {
        "configurable": {
            "thinking_enabled": True,
            "reasoning_effort": "high",
        },
        "context": {
            "reasoning_effort": "minimal",
        },
    }
    # The public body context merges with setdefault. The capability gate must
    # therefore run afterwards and assign its normalized value to both views.
    services.merge_run_context_overrides(
        config,
        {
            "thinking_enabled": True,
            "reasoning_effort": "medium",
        },
    )

    normalized = services.enforce_reasoning_effort_run_config(
        config,
        capability_enabled=True,
        model_supports_thinking=True,
        default_reasoning_effort="low",
    )

    assert normalized == "low"
    assert config["configurable"]["reasoning_effort"] == "low"
    assert config["context"]["reasoning_effort"] == "low"


def test_reasoning_policy_defaults_supported_thinking_model_to_xhigh() -> None:
    config = {
        "configurable": {"thinking_enabled": True},
        "context": {},
    }

    normalized = services.enforce_reasoning_effort_run_config(
        config,
        capability_enabled=True,
        model_supports_thinking=True,
    )

    assert normalized == "xhigh"
    assert config["configurable"]["reasoning_effort"] == "xhigh"
    assert config["context"]["reasoning_effort"] == "xhigh"


@pytest.mark.parametrize(
    ("capability_enabled", "model_supports_thinking", "thinking_enabled"),
    [
        (False, True, True),
        (True, False, True),
        (True, True, False),
    ],
)
def test_reasoning_policy_strips_effort_when_selection_is_not_available(
    capability_enabled: bool,
    model_supports_thinking: bool,
    thinking_enabled: bool,
) -> None:
    config = {
        "configurable": {"reasoning_effort": "low"},
        "context": {
            "thinking_enabled": thinking_enabled,
            "reasoning_effort": "medium",
        },
    }

    normalized = services.enforce_reasoning_effort_run_config(
        config,
        capability_enabled=capability_enabled,
        model_supports_thinking=model_supports_thinking,
    )

    assert normalized is None
    assert "reasoning_effort" not in config["configurable"]
    assert "reasoning_effort" not in config["context"]


def test_reasoning_policy_uses_global_default_instead_of_stale_thread_value() -> None:
    config = {
        "configurable": {},
        "context": {
            "thinking_enabled": True,
            "reasoning_effort": "extreme",
        },
    }

    normalized = services.enforce_reasoning_effort_run_config(
        config,
        capability_enabled=True,
        model_supports_thinking=True,
        default_reasoning_effort="low",
    )
    assert normalized == "low"
    assert config["configurable"]["reasoning_effort"] == "low"
    assert config["context"]["reasoning_effort"] == "low"


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("request_config", "body_context", "expected_model", "expected_effort"),
    [
        (
            {"context": {}},
            {
                "model_name": "qwen-a",
                "thinking_enabled": True,
                "reasoning_effort": "minimal",
            },
            "qwen-a",
            "medium",
        ),
        (
            {
                "context": {
                    "model_name": "qwen-b",
                    "thinking_enabled": True,
                    "reasoning_effort": "high",
                }
            },
            {
                "model_name": "qwen-a",
                "thinking_enabled": True,
                "reasoning_effort": "xhigh",
            },
            "qwen-b",
            None,
        ),
    ],
)
async def test_start_run_applies_database_reasoning_policy_after_final_context_merge(
    request_config: dict,
    body_context: dict,
    expected_model: str,
    expected_effort: str | None,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import analysis_projects
    from app.gateway.llmwiki_rag import LlmWikiRagContext

    app_config = AppConfig.model_construct(
        models=[
            _qwen_model(name="qwen-a"),
            _qwen_model(name="qwen-b"),
        ]
    )
    store = MemoryModelCapabilityStore()
    await store.set_reasoning_effort_support(
        "qwen-a",
        supports_reasoning_effort=True,
        default_reasoning_effort="medium",
    )

    class ThreadStore:
        async def get(self, _thread_id):
            return None

        async def create(self, *_args, **_kwargs):
            return None

        async def update_status(self, *_args, **_kwargs):
            return None

        async def update_metadata(self, *_args, **_kwargs):
            return None

        async def update_assistant_id(self, *_args, **_kwargs):
            return None

    record = SimpleNamespace(
        run_id="run-1",
        metadata={},
        task=None,
    )

    class RunManager:
        async def create_or_reject(self, *_args, **kwargs):
            record.metadata = dict(kwargs["metadata"])
            return record

        async def set_status(self, *_args, **_kwargs):
            return None

        def watch_completion(self, _record):
            return None

    run_context = SimpleNamespace(
        app_config=app_config,
        thread_store=ThreadStore(),
    )
    captured: dict[str, object] = {}

    async def prepare_project_run(body, _thread_id, _request):
        return body, None

    async def no_fixed_answer(**_kwargs):
        return None

    async def empty_rag_context(**_kwargs):
        return LlmWikiRagContext([], [], 0, "", [], "")

    async def capture_run(**kwargs):
        captured.update(kwargs["run_agent_kwargs"])

    monkeypatch.setattr(analysis_projects, "prepare_project_run", prepare_project_run)
    monkeypatch.setattr(services, "get_stream_bridge", lambda _request: object())
    monkeypatch.setattr(services, "get_run_manager", lambda _request: RunManager())
    monkeypatch.setattr(services, "get_run_context", lambda _request: run_context)
    monkeypatch.setattr(services, "get_concurrency_gate", lambda _request: None)
    monkeypatch.setattr(services, "get_effective_user_id", lambda: "user-1")
    monkeypatch.setattr(services, "_find_fixed_answer", no_fixed_answer)
    monkeypatch.setattr(services, "_get_model_concurrency_limit", lambda _name: None)
    monkeypatch.setattr(services, "_get_user_model_concurrency_limit", lambda _name: None)
    monkeypatch.setattr(services, "resolve_agent_factory", lambda _assistant_id: object())
    monkeypatch.setattr(services, "build_llmwiki_rag_context", empty_rag_context)
    monkeypatch.setattr(services, "apply_knowledge_space_strict_scope", lambda *_args, **_kwargs: False)
    monkeypatch.setattr(services, "_run_agent_with_concurrency_lease", capture_run)

    body = SimpleNamespace(
        input={"messages": [{"role": "user", "content": "hello"}]},
        config=request_config,
        context=body_context,
        metadata={},
        assistant_id=None,
        on_disconnect="continue",
        multitask_strategy="reject",
        stream_mode=["values"],
        stream_subgraphs=False,
        interrupt_before=None,
        interrupt_after=None,
        command=None,
    )
    request = SimpleNamespace(
        app=SimpleNamespace(
            state=SimpleNamespace(
                config=app_config,
                model_capability_store=store,
            )
        ),
        client=None,
        headers={},
    )

    result = await services.start_run(
        body,
        "thread-1",
        request,
        enforce_access=False,
    )
    await result.task

    final_config = captured["config"]
    assert result.metadata["model_name"] == expected_model
    if expected_effort is None:
        assert "reasoning_effort" not in final_config["configurable"]
        assert "reasoning_effort" not in final_config["context"]
        assert "reasoning_effort" not in result.metadata
    else:
        assert final_config["configurable"]["reasoning_effort"] == expected_effort
        assert final_config["context"]["reasoning_effort"] == expected_effort
        assert result.metadata["reasoning_effort"] == expected_effort