odysseus/tests/test_foreground_model_routing.py

"""Regression coverage for strict foreground model selection."""

import asyncio
import json
from types import SimpleNamespace

import httpx
import pytest
from fastapi import HTTPException
import core.database as database
import src.agent_loop as agent_loop
import src.endpoint_resolver as endpoint_resolver
import src.foreground_model_routing as foreground_model_routing
import src.llm_core as llm_core
import routes.chat_routes as chat_routes
import routes.chat_helpers as chat_helpers
import routes.prefs_routes as prefs_routes
from src.request_models import ChatRequest
from src.tool_approvals import document_content_digest
from src.foreground_model_routing import (
    FOREGROUND_AVAILABILITY_STATUSES,
    MAX_FOREGROUND_FALLBACKS,
    ForegroundModelPolicy,
    build_foreground_model_candidates,
    resolve_foreground_model_policy,
)


def _collect(gen):
    async def _run():
        return [chunk async for chunk in gen]

    return asyncio.run(_run())


class _EmptyQuery:
    def filter(self, *args, **kwargs):
        return self

    def order_by(self, *args, **kwargs):
        return self

    def first(self):
        return None


class _EmptyDb:
    def query(self, *args, **kwargs):
        return _EmptyQuery()

    def close(self):
        return None


class _RouteRequest:
    def __init__(self, mode, privileges=None):
        self.headers = {}
        auth_manager = None
        if privileges is not None:
            auth_manager = SimpleNamespace(get_privileges=lambda user: privileges)
        self.app = SimpleNamespace(state=SimpleNamespace(auth_manager=auth_manager))
        self.state = SimpleNamespace(current_user="alice")
        self._form = {
            "message": "hello",
            "session": "session-1",
            "mode": mode,
            "compare_mode": "true",
        }

    async def form(self):
        return self._form


@pytest.mark.parametrize(
    ("status", "expected"),
    [
        (429, 429),
        ("503", 503),
        (429.9, None),
        (True, None),
        ("429.9", None),
    ],
)
def test_stream_failure_status_uses_exact_http_statuses(status, expected):
    chunk = f'event: error\ndata: {json.dumps({"status": status})}\n\n'
    assert chat_routes._stream_failure_status(chunk) == expected


def _chat_stream_endpoint(
    monkeypatch,
    mode,
    captured,
    *,
    agent_chunks=None,
    chat_chunks=None,
    capture_completion=False,
    capture_context=False,
    endpoint_url="https://selected.example/v1",
):
    def add_message(message):
        captured.setdefault("added_messages", []).append(message)

    session = SimpleNamespace(
        endpoint_url=endpoint_url,
        model="selected-model",
        headers={"Authorization": "Bearer selected"},
        name="test",
        history=[],
        add_message=add_message,
    )
    session_manager = SimpleNamespace(
        get_session=lambda session_id: session,
        save_sessions=lambda: None,
    )
    context = SimpleNamespace(
        user="alice",
        messages=[{"role": "user", "content": "hello"}],
        route_messages=[
            {"role": "user", "content": "old one"},
            {"role": "assistant", "content": "old answer"},
            {"role": "user", "content": "hello"},
        ],
        preprocessed=SimpleNamespace(attachment_meta=[]),
        auto_opened_docs=[],
        rag_sources=[],
        web_sources=[],
        used_memories=[],
        uploaded_files=[],
        uprefs={},
        was_compacted=False,
        context_trimmed=False,
        context_length=4096,
        context_messages_before_trim=1,
        context_messages_after_trim=1,
        context_tokens_before_trim=10,
        context_tokens_after_trim=10,
        preset=SimpleNamespace(temperature=0.2, max_tokens=128, character_name=None),
    )

    async def fake_build_context(*args, **kwargs):
        if capture_context:
            captured["build_context"] = kwargs
        return context

    async def fake_chat_stream(candidates, messages, **kwargs):
        captured["chat"] = candidates
        if chat_chunks is not None:
            for chunk in chat_chunks:
                if isinstance(chunk, BaseException):
                    raise chunk
                yield chunk
            return
        yield f'data: {json.dumps({"delta": "done"})}\n\n'
        yield "data: [DONE]\n\n"

    async def fake_agent_stream(endpoint_url, model, messages, **kwargs):
        captured["agent"] = {
            "primary": (endpoint_url, model, kwargs.get("headers")),
            "fallbacks": kwargs.get("fallbacks"),
        }
        if kwargs.get("external_untrusted_context_seen"):
            captured["agent_external_untrusted_context_seen"] = True
        if kwargs.get("exact_approval") is not None:
            captured["exact_approval"] = kwargs["exact_approval"]
            captured["approval_disabled_tools"] = set(
                kwargs.get("disabled_tools") or ()
            )
        if agent_chunks is not None:
            for chunk in agent_chunks:
                if isinstance(chunk, BaseException):
                    raise chunk
                yield chunk
            return
        yield f'data: {json.dumps({"delta": "done"})}\n\n'
        yield "data: [DONE]\n\n"

    monkeypatch.setattr(chat_routes, "coerce_message_and_session", lambda *args, **kwargs: ("hello", "session-1"))
    monkeypatch.setattr(chat_routes, "_verify_session_owner", lambda *args, **kwargs: None)
    monkeypatch.setattr(chat_routes, "effective_user", lambda request: "alice")
    monkeypatch.setattr(chat_routes, "_clear_orphaned_session_endpoint", lambda *args, **kwargs: False)
    monkeypatch.setattr(chat_routes, "_recover_empty_session_model", lambda *args, **kwargs: False)
    monkeypatch.setattr(chat_routes, "_enforce_chat_privileges", lambda *args, **kwargs: None)
    monkeypatch.setattr(chat_routes, "resolve_session_auth", lambda *args, **kwargs: None)
    monkeypatch.setattr(chat_routes, "get_session_mode", lambda session_id: "chat")
    monkeypatch.setattr(chat_routes, "set_session_mode", lambda *args, **kwargs: None)
    monkeypatch.setattr(chat_routes, "build_chat_context", fake_build_context)
    monkeypatch.setattr(chat_routes, "SessionLocal", _EmptyDb)
    monkeypatch.setattr(chat_routes, "_is_image_generation_session", lambda *args, **kwargs: False)
    monkeypatch.setattr(chat_routes, "stream_llm_with_fallback", fake_chat_stream)
    monkeypatch.setattr(chat_routes, "stream_agent_loop", fake_agent_stream)
    if capture_completion:
        monkeypatch.setattr(
            chat_routes,
            "save_assistant_response",
            lambda *args, **kwargs: captured.setdefault("saved", []).append((args, kwargs)),
        )
        monkeypatch.setattr(
            chat_routes,
            "run_post_response_tasks",
            lambda *args, **kwargs: captured.setdefault("post_processed", []).append((args, kwargs)),
        )
    else:
        monkeypatch.setattr(chat_routes, "save_assistant_response", lambda *args, **kwargs: None)
        monkeypatch.setattr(chat_routes, "run_post_response_tasks", lambda *args, **kwargs: None)
    monkeypatch.setattr(chat_routes, "estimate_tokens", lambda messages: 10)
    monkeypatch.setattr(
        chat_routes,
        "accumulate_token_usage",
        lambda *args, **kwargs: captured.setdefault("accumulated_usage", []).append((args, kwargs)),
    )
    monkeypatch.setattr(foreground_model_routing, "_load_policy_preferences", lambda owner=None: {
        "default_model_fallbacks": [
            {"endpoint_id": "legacy", "model": "legacy-model"},
        ],
    })

    import src.settings as settings

    monkeypatch.setattr(
        settings,
        "get_setting",
        lambda key, default=None: default,
    )
    monkeypatch.setattr(
        settings,
        "get_user_setting",
        lambda key, owner="", default=None: (
            [{"endpoint_id": "legacy", "model": "legacy-model"}]
            if key == "default_model_fallbacks"
            else default
        ),
    )

    router = chat_routes.setup_chat_routes(
        session_manager,
        SimpleNamespace(),
        SimpleNamespace(),
        SimpleNamespace(),
        SimpleNamespace(),
        SimpleNamespace(),
    )
    return next(route.endpoint for route in router.routes if route.path == "/api/chat_stream")


@pytest.mark.asyncio
@pytest.mark.parametrize("mode", ["chat", "agent"])
async def test_chat_stream_route_keeps_selected_model_strict_with_legacy_data(monkeypatch, mode):
    captured = {}
    endpoint = _chat_stream_endpoint(monkeypatch, mode, captured)

    response = await endpoint(_RouteRequest(mode))
    async for _ in response.body_iterator:
        pass

    selected = (
        "https://selected.example/v1",
        "selected-model",
        {"Authorization": "Bearer selected"},
    )
    if mode == "chat":
        assert captured == {"chat": [selected]}
    else:
        assert captured == {"agent": {"primary": selected, "fallbacks": []}}


@pytest.mark.asyncio
async def test_chat_stream_consumes_exact_tool_approval_for_own_session(monkeypatch):
    from src.tool_capabilities import capabilities_for_action

    captured = {}
    endpoint = _chat_stream_endpoint(monkeypatch, "agent", captured)
    tool_content = '{"content":"replacement"}'
    pending = chat_routes.tool_approval_store.create(
        owner="alice",
        session_id="session-1",
        origin_run_id="run-1",
        tool_name="update_document",
        content=tool_content,
        workspace=None,
        document_id="document-7",
        document_version=4,
        document_digest=document_content_digest("original"),
        external_untrusted_context_seen=True,
        capabilities=capabilities_for_action("update_document", tool_content),
    )
    request = _RouteRequest("agent")
    request._form.update(
        {
            "tool_approval_id": pending.approval_id,
            "tool_approval_decision": "approve",
            "active_doc_id": "document-changed-in-browser",
            "compare_mode": "false",
        }
    )

    response = await endpoint(request)
    async for _ in response.body_iterator:
        pass

    grant = captured["exact_approval"]
    assert grant.pending == pending
    assert chat_routes.tool_approval_store.peek(pending.approval_id) is None
    assert grant.matches(
        owner="alice",
        session_id="session-1",
        tool_name="update_document",
        content=tool_content,
        workspace=None,
    )
    assert "update_document" not in captured["approval_disabled_tools"]


@pytest.mark.asyncio
async def test_chat_stream_approval_restores_exact_shell_turn_toggle(monkeypatch):
    from src.tool_capabilities import capabilities_for_action

    captured = {}
    endpoint = _chat_stream_endpoint(monkeypatch, "agent", captured)
    pending = chat_routes.tool_approval_store.create(
        owner="alice",
        session_id="session-1",
        origin_run_id="run-1",
        tool_name="bash",
        content="printf exact",
        workspace=None,
        external_untrusted_context_seen=True,
        capabilities=capabilities_for_action("bash", "printf exact"),
    )
    request = _RouteRequest("chat")
    request._form.update(
        {
            "allow_bash": "false",
            "tool_approval_id": pending.approval_id,
            "tool_approval_decision": "approve",
        }
    )

    response = await endpoint(request)
    async for _ in response.body_iterator:
        pass

    assert captured["exact_approval"].pending == pending
    assert "bash" not in captured["approval_disabled_tools"]


@pytest.mark.asyncio
async def test_chat_stream_denial_keeps_originating_run_tainted(monkeypatch):
    from src.tool_capabilities import capabilities_for_action

    captured = {}
    endpoint = _chat_stream_endpoint(monkeypatch, "agent", captured)
    pending = chat_routes.tool_approval_store.create(
        owner="alice",
        session_id="session-1",
        origin_run_id="run-1",
        tool_name="bash",
        content="printf retry",
        workspace=None,
        external_untrusted_context_seen=True,
        capabilities=capabilities_for_action("bash", "printf retry"),
    )
    request = _RouteRequest("agent")
    request._form.update(
        {
            "tool_approval_id": pending.approval_id,
            "tool_approval_decision": "deny",
        }
    )

    response = await endpoint(request)
    async for _ in response.body_iterator:
        pass

    assert "exact_approval" not in captured
    assert captured["agent_external_untrusted_context_seen"] is True
    assert chat_routes.tool_approval_store.peek(pending.approval_id) is None


@pytest.mark.asyncio
async def test_chat_stream_normal_reply_retires_pending_action_but_keeps_taint(
    monkeypatch,
):
    from src.tool_capabilities import capabilities_for_action

    captured = {}
    endpoint = _chat_stream_endpoint(monkeypatch, "agent", captured)
    pending = chat_routes.tool_approval_store.create(
        owner="alice",
        session_id="session-1",
        origin_run_id="run-1",
        tool_name="bash",
        content="printf retry",
        workspace=None,
        external_untrusted_context_seen=True,
        capabilities=capabilities_for_action("bash", "printf retry"),
    )

    response = await endpoint(_RouteRequest("agent"))
    async for _ in response.body_iterator:
        pass

    assert "exact_approval" not in captured
    assert captured["agent_external_untrusted_context_seen"] is True
    assert chat_routes.tool_approval_store.peek(pending.approval_id) is None


@pytest.mark.asyncio
async def test_chat_stream_approval_ignores_research_and_new_attachments(monkeypatch):
    from src.tool_capabilities import capabilities_for_action

    captured = {}
    endpoint = _chat_stream_endpoint(
        monkeypatch,
        "agent",
        captured,
        capture_context=True,
    )
    monkeypatch.setattr(chat_routes, "get_session_mode", lambda _session_id: "research_pending")
    pending = chat_routes.tool_approval_store.create(
        owner="alice",
        session_id="session-1",
        origin_run_id="run-1",
        tool_name="bash",
        content="printf exact",
        workspace=None,
        external_untrusted_context_seen=True,
        capabilities=capabilities_for_action("bash", "printf exact"),
    )
    request = _RouteRequest("agent")
    request._form.update(
        {
            "attachments": '["unrelated-upload"]',
            "use_research": "true",
            "tool_approval_id": pending.approval_id,
            "tool_approval_decision": "approve",
        }
    )

    response = await endpoint(request)
    async for _ in response.body_iterator:
        pass

    assert captured["exact_approval"].pending == pending
    assert captured["build_context"]["att_ids"] == []
    assert "agent" in captured
    assert "chat" not in captured


@pytest.mark.asyncio
@pytest.mark.parametrize("mode", ["chat", "agent"])
@pytest.mark.parametrize("endpoint_url", ["", None])
async def test_chat_stream_rejects_missing_selected_endpoint_before_fallback(
    monkeypatch,
    mode,
    endpoint_url,
):
    captured = {}
    endpoint = _chat_stream_endpoint(
        monkeypatch,
        mode,
        captured,
        endpoint_url=endpoint_url,
    )

    with pytest.raises(HTTPException) as exc:
        await endpoint(_RouteRequest(mode))

    assert exc.value.status_code == 400
    assert "not configured" in str(exc.value.detail)
    assert captured == {}


@pytest.mark.asyncio
@pytest.mark.parametrize("mode", ["chat", "agent"])
async def test_chat_stream_route_uses_only_new_explicit_fallback_policy(monkeypatch, mode):
    captured = {}
    endpoint = _chat_stream_endpoint(monkeypatch, mode, captured)
    monkeypatch.setattr(
        foreground_model_routing,
        "_load_policy_preferences",
        lambda owner=None: {
            "foreground_fallback_enabled": True,
            "foreground_model_fallbacks": [
                {"endpoint_id": "backup", "model": "backup-model"},
            ],
            "default_model_fallbacks": [
                {"endpoint_id": "legacy", "model": "legacy-model"},
            ],
        },
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda entries, owner=None, require_exact_model=False: [
            ("https://backup.example/v1", "backup-model", {"Authorization": "Bearer backup"}),
        ],
    )

    response = await endpoint(_RouteRequest(mode))
    async for _ in response.body_iterator:
        pass

    selected = (
        "https://selected.example/v1",
        "selected-model",
        {"Authorization": "Bearer selected"},
    )
    backup = (
        "https://backup.example/v1",
        "backup-model",
        {"Authorization": "Bearer backup"},
    )
    if mode == "chat":
        assert captured == {"chat": [selected, backup]}
    else:
        assert captured == {"agent": {"primary": selected, "fallbacks": [backup]}}


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("primary_context", "backup_context", "expected_counts"),
    [
        (100, 1000, (1, 3)),
        (1000, 100, (3, 1)),
    ],
)
async def test_streaming_chat_shapes_each_candidate_from_route_neutral_history(
    monkeypatch,
    primary_context,
    backup_context,
    expected_counts,
):
    captured = {}
    endpoint = _chat_stream_endpoint(monkeypatch, "chat", captured)
    monkeypatch.setattr(foreground_model_routing, "_load_policy_preferences", lambda owner=None: {
        "foreground_fallback_enabled": True,
        "foreground_model_fallbacks": [
            {"endpoint_id": "backup", "model": "backup-model"},
        ],
    })
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda *args, **kwargs: [("https://backup.example/v1", "backup-model", {})],
    )
    async def fake_compact(
        session, url, model, messages, headers=None, owner=None, **kwargs
    ):
        return (
            list(messages),
            backup_context if model == "backup-model" else primary_context,
            False,
        )

    monkeypatch.setattr(chat_routes, "maybe_compact", fake_compact)
    monkeypatch.setattr(
        chat_routes,
        "trim_for_context",
        lambda messages, budget: list(messages) if budget >= 1000 else list(messages[-1:]),
    )

    async def fake_stream(candidates, messages, **kwargs):
        factory = kwargs["candidate_request_factory"]
        requests = [
            await factory(index, *candidate)
            for index, candidate in enumerate(candidates)
        ]
        captured["request_counts"] = tuple(
            len(request["messages"]) for request in requests
        )
        yield 'data: {"type": "fallback", "candidate_index": 1, "selected_model": "selected-model", "answered_by": "backup-model"}\n\n'
        yield 'data: {"delta": "backup"}\n\n'
        yield "data: [DONE]\n\n"

    monkeypatch.setattr(chat_routes, "stream_llm_with_fallback", fake_stream)

    response = await endpoint(_RouteRequest("chat"))
    async for _chunk in response.body_iterator:
        pass

    assert captured["request_counts"] == expected_counts


@pytest.mark.asyncio
async def test_streaming_chat_persists_only_answering_route_compaction(monkeypatch):
    captured = {}
    applied = []
    endpoint = _chat_stream_endpoint(monkeypatch, "chat", captured)
    monkeypatch.setattr(foreground_model_routing, "_load_policy_preferences", lambda owner=None: {
        "foreground_fallback_enabled": True,
        "foreground_model_fallbacks": [
            {"endpoint_id": "backup", "model": "backup-model"},
        ],
    })
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda *args, **kwargs: [("https://backup.example/v1", "backup-model", {})],
    )

    async def fake_compact(
        session, url, model, messages, headers=None, owner=None,
        *, persist=True, compaction_state=None,
    ):
        assert persist is False
        compaction_state.update({"route": model, "applied": False})
        return ([{"role": "system", "content": f"summary for {model}"}, *messages], 1000, True)

    def fake_apply(session, state):
        if not state or state.get("applied"):
            return False
        state["applied"] = True
        applied.append(state["route"])
        return True

    async def fake_stream(candidates, messages, **kwargs):
        factory = kwargs["candidate_request_factory"]
        for index, candidate in enumerate(candidates):
            await factory(index, *candidate)
        yield 'data: {"type": "fallback", "candidate_index": 1, "selected_model": "selected-model", "answered_by": "backup-model"}\n\n'
        yield 'data: {"delta": "backup"}\n\n'
        yield "data: [DONE]\n\n"

    monkeypatch.setattr(chat_routes, "maybe_compact", fake_compact)
    monkeypatch.setattr(chat_routes, "apply_compaction_state", fake_apply)
    monkeypatch.setattr(chat_routes, "stream_llm_with_fallback", fake_stream)

    response = await endpoint(_RouteRequest("chat"))
    chunks = [chunk async for chunk in response.body_iterator]

    assert applied == ["backup-model"]
    assert any('"type": "compacted"' in chunk for chunk in chunks)


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("selected_url", "selected_cost_tracked", "backup_url", "expected_cost_tracked"),
    [
        ("http://localhost:11434/v1", False, "https://backup.example/v1", True),
        ("https://selected.example/v1", True, "http://localhost:11434/v1", False),
    ],
)
async def test_streaming_chat_cost_uses_answering_route_classification(
    monkeypatch,
    selected_url,
    selected_cost_tracked,
    backup_url,
    expected_cost_tracked,
):
    captured = {}
    chunks = [
        'data: {"type": "fallback", "candidate_index": 1, "selected_model": "selected-model", "answered_by": "backup-model"}\n\n',
        'data: {"type": "usage", "data": {"model": "backup-model", "input_tokens": 20, "output_tokens": 5}}\n\n',
        'data: {"delta": "backup answer"}\n\n',
        "data: [DONE]\n\n",
    ]
    endpoint = _chat_stream_endpoint(
        monkeypatch,
        "chat",
        captured,
        chat_chunks=chunks,
        capture_completion=True,
        endpoint_url=selected_url,
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "_load_policy_preferences",
        lambda owner=None: {
            "foreground_fallback_enabled": True,
            "foreground_model_fallbacks": [
                {"endpoint_id": "backup", "model": "backup-model"},
            ],
        },
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda *args, **kwargs: [
            (backup_url, "backup-model", {}),
        ],
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_route_descriptor",
        lambda *args, **kwargs: {
            "endpoint_id": "selected",
            "endpoint_label": "Selected local endpoint",
            "endpoint_cost_tracked": selected_cost_tracked,
        },
    )

    response = await endpoint(_RouteRequest("chat"))
    emitted = [chunk async for chunk in response.body_iterator]

    metrics = json.loads(next(
        chunk for chunk in emitted if '"type": "metrics"' in chunk
    )[6:])["data"]
    assert metrics["endpoint_id"] == "backup"
    assert metrics["endpoint_cost_tracked"] is expected_cost_tracked
    saved_args, _saved_kwargs = captured["saved"][0]
    assert saved_args[4]["endpoint_cost_tracked"] is expected_cost_tracked


@pytest.mark.asyncio
@pytest.mark.parametrize("selected_cost_tracked", [False, True])
async def test_streaming_chat_persists_selected_route_cost_classification(
    monkeypatch,
    selected_cost_tracked,
):
    captured = {}
    endpoint = _chat_stream_endpoint(
        monkeypatch,
        "chat",
        captured,
        chat_chunks=[
            'data: {"type": "usage", "data": {"model": "selected-model", "input_tokens": 20, "output_tokens": 5}}\n\n',
            'data: {"delta": "selected answer"}\n\n',
            "data: [DONE]\n\n",
        ],
        capture_completion=True,
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_route_descriptor",
        lambda *args, **kwargs: {
            "endpoint_id": "selected",
            "endpoint_label": "Selected endpoint",
            "endpoint_cost_tracked": selected_cost_tracked,
        },
    )

    response = await endpoint(_RouteRequest("chat"))
    emitted = [chunk async for chunk in response.body_iterator]

    metrics = json.loads(next(
        chunk for chunk in emitted if '"type": "metrics"' in chunk
    )[6:])["data"]
    assert metrics["endpoint_cost_tracked"] is selected_cost_tracked
    saved_args, _saved_kwargs = captured["saved"][0]
    assert saved_args[4]["endpoint_cost_tracked"] is selected_cost_tracked


@pytest.mark.asyncio
async def test_chat_stream_route_does_not_save_or_postprocess_terminal_agent_error(monkeypatch):
    captured = {}
    error_chunk = 'event: error\ndata: {"status": 401, "error": "invalid key"}\n\n'
    endpoint = _chat_stream_endpoint(
        monkeypatch,
        "agent",
        captured,
        agent_chunks=[error_chunk],
        capture_completion=True,
    )

    response = await endpoint(_RouteRequest("agent"))
    chunks = [chunk async for chunk in response.body_iterator]

    assert error_chunk in chunks
    assert "saved" not in captured
    assert "post_processed" not in captured


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("provider_status", "expected_status", "expected_message"),
    [
        (401, 401, "Model request failed (HTTP 401)"),
        (429.9, None, "Model request failed"),
    ],
)
async def test_chat_stream_persists_completed_tools_before_later_terminal_error(
    monkeypatch,
    provider_status,
    expected_status,
    expected_message,
):
    captured = {}
    terminal_metadata = {
        "failed": True,
        "failure": {
            "status": provider_status,
            "message": "credential-shaped provider detail",
        },
        "model": "backup-model",
        "requested_model": "selected-model",
        "endpoint_id": "backup-endpoint",
        "endpoint_label": "Backup endpoint",
        "tool_events": [
            {"round": 1, "tool": "bash", "output": "created", "exit_code": 0},
        ],
        "round_texts": ["partial answer"],
        "round_models": ["backup-model"],
        "round_endpoint_ids": ["backup-endpoint"],
        "round_endpoint_labels": ["Backup endpoint"],
        "input_tokens": 75,
        "output_tokens": 15,
        "usage_source": "real",
        "endpoint_cost_tracked": True,
    }
    chunks = [
        'data: {"delta": "partial answer"}\n\n',
        f'data: {json.dumps({"type": "agent_terminal", "data": terminal_metadata})}\n\n',
        f'event: error\ndata: {json.dumps({"status": provider_status, "error": "invalid key"})}\n\n',
    ]
    endpoint = _chat_stream_endpoint(
        monkeypatch,
        "agent",
        captured,
        agent_chunks=chunks,
        capture_completion=True,
    )

    response = await endpoint(_RouteRequest("agent"))
    emitted = [chunk async for chunk in response.body_iterator]

    assert any(chunk.startswith("event: error") for chunk in emitted)
    assert not any(chunk == "data: [DONE]\n\n" for chunk in emitted)
    assert len(captured["saved"]) == 1
    saved_args, _saved_kwargs = captured["saved"][0]
    assert "partial answer" in saved_args[3]
    assert f"Agent stopped: {expected_message}" in saved_args[3]
    assert "credential-shaped provider detail" not in saved_args[3]
    assert saved_args[4]["failure"] == {
        "status": expected_status,
        "message": expected_message,
    }
    assert saved_args[4]["failed"] is True
    assert saved_args[4]["tool_events"][0]["output"] == "created"
    assert captured["accumulated_usage"][0][0][1] == saved_args[4]
    assert saved_args[4]["input_tokens"] == 75
    assert saved_args[4]["output_tokens"] == 15
    assert "post_processed" not in captured
    assert all(chunk != "data: [DONE]\n\n" for chunk in chunks)


@pytest.mark.asyncio
async def test_chat_stream_persists_partial_terminal_error_with_route_provenance(monkeypatch):
    captured = {}
    chunks = [
        'data: {"type": "fallback", "candidate_index": 1, "selected_model": "selected-model", "answered_by": "backup-model", "answered_by_endpoint_id": "backup", "answered_by_endpoint_label": "Backup endpoint"}\n\n',
        'data: {"delta": "visible partial"}\n\n',
        'event: error\ndata: {"status": 503, "error": "credential-shaped provider detail"}\n\n',
        "data: [DONE]\n\n",
    ]
    endpoint = _chat_stream_endpoint(
        monkeypatch,
        "chat",
        captured,
        chat_chunks=chunks,
        capture_completion=True,
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "_load_policy_preferences",
        lambda owner=None: {
            "foreground_fallback_enabled": True,
            "foreground_model_fallbacks": [
                {"endpoint_id": "backup", "model": "backup-model"},
            ],
        },
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda *args, **kwargs: [
            ("https://backup.example/v1", "backup-model", {"Authorization": "Bearer backup"}),
        ],
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_route_descriptor",
        lambda *args, **kwargs: {
            "endpoint_id": "selected",
            "endpoint_label": "Selected endpoint",
        },
    )

    response = await endpoint(_RouteRequest("chat"))
    emitted = [chunk async for chunk in response.body_iterator]

    assert any(chunk.startswith("event: error") for chunk in emitted)
    assert not any(chunk == "data: [DONE]\n\n" for chunk in emitted)
    assert len(captured["saved"]) == 1
    saved_args, _saved_kwargs = captured["saved"][0]
    assert saved_args[3] == (
        "visible partial\n\n"
        "[Response stopped: Model request failed (HTTP 503)]"
    )
    assert "credential-shaped provider detail" not in str(saved_args)
    assert saved_args[4]["failure"] == {
        "status": 503,
        "message": "Model request failed (HTTP 503)",
    }
    assert saved_args[4]["model"] == "backup-model"
    assert saved_args[4]["requested_model"] == "selected-model"
    assert saved_args[4]["endpoint_id"] == "backup"
    assert saved_args[4]["endpoint_label"] == "backup"
    assert saved_args[4]["requested_endpoint_id"] == "selected"
    assert saved_args[4]["requested_endpoint_label"] == "Selected endpoint"
    assert saved_args[4]["endpoint_cost_tracked"] is True
    assert saved_args[4]["input_tokens"] == 10
    assert saved_args[4]["output_tokens"] == len("visible partial") // 4
    assert saved_args[4]["usage_source"] == "estimated"
    assert captured["accumulated_usage"][0][0][1] == saved_args[4]
    chat_terminal = json.loads(next(
        chunk for chunk in emitted if '"type": "chat_terminal"' in chunk
    )[6:])["data"]
    assert chat_terminal == saved_args[4]
    assert "post_processed" not in captured


@pytest.mark.asyncio
async def test_chat_terminal_preserves_real_usage_and_accumulates_once(monkeypatch):
    captured = {}
    endpoint = _chat_stream_endpoint(
        monkeypatch,
        "chat",
        captured,
        chat_chunks=[
            'data: {"type": "usage", "data": {"model": "selected-model", "input_tokens": 123, "output_tokens": 17, "usage_source": "real"}}\n\n',
            'data: {"delta": "visible partial"}\n\n',
            'event: error\ndata: {"status": 503, "error": "provider detail"}\n\n',
        ],
        capture_completion=True,
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_route_descriptor",
        lambda *args, **kwargs: {
            "endpoint_id": "selected",
            "endpoint_label": "Selected paid endpoint",
            "endpoint_cost_tracked": True,
        },
    )

    response = await endpoint(_RouteRequest("chat"))
    emitted = [chunk async for chunk in response.body_iterator]

    saved_metrics = captured["saved"][0][0][4]
    assert saved_metrics["input_tokens"] == 123
    assert saved_metrics["output_tokens"] == 17
    assert saved_metrics["usage_source"] == "real"
    assert saved_metrics["endpoint_cost_tracked"] is True
    assert saved_metrics["failed"] is True
    assert len(captured["accumulated_usage"]) == 1
    assert captured["accumulated_usage"][0][0][1] == saved_metrics
    assert len([
        chunk for chunk in emitted if '"type": "metrics"' in chunk
    ]) == 1
    assert len([
        chunk for chunk in emitted if '"type": "chat_terminal"' in chunk
    ]) == 1
    assert "post_processed" not in captured


@pytest.mark.asyncio
async def test_cancelled_agent_fallback_saves_endpoint_and_round_provenance(monkeypatch):
    captured = {}
    chunks = [
        'data: {"type": "model_actual", "round": 1, "model": "selected-provider-alias"}\n\n',
        'data: {"type": "agent_step", "round": 2}\n\n',
        'data: {"type": "fallback", "round": 2, "selected_model": "selected-model", "answered_by": "backup-model", "answered_by_endpoint_id": "account-two", "answered_by_endpoint_label": "Account two"}\n\n',
        'data: {"delta": "partial answer"}\n\n',
        asyncio.CancelledError(),
    ]
    endpoint = _chat_stream_endpoint(
        monkeypatch,
        "agent",
        captured,
        agent_chunks=chunks,
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "_load_policy_preferences",
        lambda owner=None: {
            "foreground_fallback_enabled": True,
            "foreground_model_fallbacks": [
                {"endpoint_id": "account-two", "model": "selected-model"},
            ],
        },
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda *args, **kwargs: [
            ("https://backup.example/v1", "selected-model", {"Authorization": "Bearer two"}),
        ],
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_route_descriptor",
        lambda *args, **kwargs: {
            "endpoint_id": "account-one",
            "endpoint_label": "Account one",
        },
    )

    response = await endpoint(_RouteRequest("agent"))
    with pytest.raises(asyncio.CancelledError):
        async for _chunk in response.body_iterator:
            pass

    saved = captured["added_messages"][-1]
    assert saved.metadata["requested_endpoint_id"] == "account-one"
    assert saved.metadata["endpoint_id"] == "account-two"
    assert saved.metadata["model"] == "backup-model"
    assert saved.metadata["round_models"] == ["selected-provider-alias", "backup-model"]
    assert saved.metadata["round_endpoint_ids"] == ["account-one", "account-two"]
    assert saved.metadata["round_endpoint_labels"] == ["Account one", "Account two"]


@pytest.mark.asyncio
async def test_cancelled_chat_fallback_saves_same_model_endpoint_provenance(monkeypatch):
    captured = {}
    chunks = [
        'data: {"type": "fallback", "candidate_index": 1, "selected_model": "selected-model", "answered_by": "selected-model", "answered_by_endpoint_id": "account-two", "answered_by_endpoint_label": "Account two"}\n\n',
        'data: {"delta": "partial answer"}\n\n',
        asyncio.CancelledError(),
    ]
    endpoint = _chat_stream_endpoint(
        monkeypatch,
        "chat",
        captured,
        chat_chunks=chunks,
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "_load_policy_preferences",
        lambda owner=None: {
            "foreground_fallback_enabled": True,
            "foreground_model_fallbacks": [
                {"endpoint_id": "account-two", "model": "selected-model"},
            ],
        },
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda *args, **kwargs: [
            ("https://backup.example/v1", "selected-model", {"Authorization": "Bearer two"}),
        ],
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_route_descriptor",
        lambda *args, **kwargs: {
            "endpoint_id": "account-one",
            "endpoint_label": "Account one",
        },
    )

    response = await endpoint(_RouteRequest("chat"))
    with pytest.raises(asyncio.CancelledError):
        async for _chunk in response.body_iterator:
            pass

    saved = captured["added_messages"][-1]
    assert saved.metadata["requested_endpoint_id"] == "account-one"
    assert saved.metadata["endpoint_id"] == "account-two"
    assert saved.metadata["requested_endpoint_label"] == "Account one"
    assert saved.metadata["endpoint_label"] == "account-two"


@pytest.mark.asyncio
async def test_chat_stream_route_excludes_fallback_outside_non_admin_allowlist(monkeypatch):
    captured = {}
    endpoint = _chat_stream_endpoint(monkeypatch, "chat", captured)
    monkeypatch.setattr(
        foreground_model_routing,
        "_load_policy_preferences",
        lambda owner=None: {
            "foreground_fallback_enabled": True,
            "foreground_model_fallbacks": [
                {"endpoint_id": "backup", "model": "blocked-model"},
            ],
        },
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda *args, **kwargs: pytest.fail("unauthorized fallback reached endpoint resolution"),
    )

    response = await endpoint(_RouteRequest("chat", privileges={
        "allowed_models": ["selected-model"],
        "allowed_models_restricted": True,
        "max_messages_per_day": 0,
    }))
    async for _ in response.body_iterator:
        pass

    assert captured == {"chat": [(
        "https://selected.example/v1",
        "selected-model",
        {"Authorization": "Bearer selected"},
    )]}


class _NonStreamChatHandler:
    async def handle_memory_command(self, sess, message):
        return None


def _chat_endpoint(
    monkeypatch,
    *,
    owner="alice",
    endpoint_url="https://selected.example/v1",
):
    saved = []
    session = SimpleNamespace(
        endpoint_url=endpoint_url,
        model="selected-model",
        headers={"Authorization": "Bearer selected"},
        history=[],
        add_message=saved.append,
    )
    session_manager = SimpleNamespace(
        get_session=lambda session_id: session,
        save_sessions=lambda: None,
    )
    context = SimpleNamespace(
        user=owner,
        messages=[{"role": "user", "content": "hello"}],
        route_messages=[
            {"role": "user", "content": "old one"},
            {"role": "assistant", "content": "old answer"},
            {"role": "user", "content": "hello"},
        ],
        context_length=100,
        uprefs={},
        preset=SimpleNamespace(
            temperature=0.2,
            max_tokens=128,
            character_name=None,
        ),
    )

    async def fake_build_context(*args, **kwargs):
        return context

    monkeypatch.setattr(chat_routes, "_verify_session_owner", lambda *args, **kwargs: None)
    monkeypatch.setattr(chat_routes, "effective_user", lambda request: owner)
    monkeypatch.setattr(chat_routes, "_clear_orphaned_session_endpoint", lambda *args, **kwargs: False)
    monkeypatch.setattr(chat_routes, "_recover_empty_session_model", lambda *args, **kwargs: False)
    monkeypatch.setattr(chat_routes, "_enforce_chat_privileges", lambda *args, **kwargs: None)
    monkeypatch.setattr(chat_routes, "build_chat_context", fake_build_context)
    monkeypatch.setattr(chat_routes, "clean_thinking_for_save", lambda reply, metadata: (reply, metadata))
    monkeypatch.setattr(chat_routes, "run_post_response_tasks", lambda *args, **kwargs: None)

    import core.database as database

    monkeypatch.setattr(database, "update_session_last_accessed", lambda session_id: None)

    router = chat_routes.setup_chat_routes(
        session_manager,
        _NonStreamChatHandler(),
        SimpleNamespace(),
        SimpleNamespace(),
        SimpleNamespace(),
        SimpleNamespace(),
    )
    endpoint = next(route.endpoint for route in router.routes if route.path == "/api/chat")
    return endpoint, saved


@pytest.mark.asyncio
async def test_nonstream_chat_is_strict_by_default_and_reports_selected_route(monkeypatch):
    calls = []
    monkeypatch.setattr(foreground_model_routing, "_load_policy_preferences", lambda owner: {})
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda *args, **kwargs: pytest.fail("strict non-stream Chat resolved fallback entries"),
    )

    async def fake_call(url, model, messages, **kwargs):
        calls.append((url, model, kwargs.get("headers")))
        return "selected answer"

    monkeypatch.setattr(llm_core, "llm_call_async", fake_call)
    endpoint, saved = _chat_endpoint(monkeypatch)

    response = await endpoint(
        _RouteRequest("chat"),
        ChatRequest(message="hello", session="session-1"),
    )

    assert calls == [(
        "https://selected.example/v1",
        "selected-model",
        {"Authorization": "Bearer selected"},
    )]
    assert response == {
        "response": "selected answer",
        "requested_model": "selected-model",
        "model": "selected-model",
        "requested_endpoint_id": None,
        "requested_endpoint_label": "Selected route",
        "endpoint_id": None,
        "endpoint_label": "Selected route",
    }
    assert saved[-1].metadata == {
        "model": "selected-model",
        "requested_model": "selected-model",
        "endpoint_id": None,
        "endpoint_label": "Selected route",
        "requested_endpoint_id": None,
        "requested_endpoint_label": "Selected route",
        "context_length": 100,
        "context_trimmed": False,
    }


@pytest.mark.asyncio
@pytest.mark.parametrize("endpoint_url", ["", None])
async def test_nonstream_chat_rejects_missing_selected_endpoint_before_fallback(
    monkeypatch,
    endpoint_url,
):
    endpoint, saved = _chat_endpoint(monkeypatch, endpoint_url=endpoint_url)

    with pytest.raises(HTTPException) as exc:
        await endpoint(
            _RouteRequest("chat"),
            ChatRequest(message="hello", session="session-1"),
        )

    assert exc.value.status_code == 400
    assert "not configured" in str(exc.value.detail)
    assert saved == []


@pytest.mark.asyncio
async def test_nonstream_chat_opt_in_advances_only_on_eligible_failure(monkeypatch):
    calls = []
    monkeypatch.setattr(foreground_model_routing, "_load_policy_preferences", lambda owner: {
        "foreground_fallback_enabled": True,
        "foreground_model_fallbacks": [
            {"endpoint_id": "backup", "model": "backup-model"},
        ],
    })
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda entries, owner=None, require_exact_model=False: [
            ("https://backup.example/v1", "backup-model", {"Authorization": "Bearer backup"}),
        ],
    )

    async def fake_call(url, model, messages, **kwargs):
        calls.append((url, model, kwargs.get("headers")))
        if model == "selected-model":
            raise HTTPException(503, "selected unavailable")
        return "backup answer"

    monkeypatch.setattr(llm_core, "llm_call_async", fake_call)
    endpoint, saved = _chat_endpoint(monkeypatch)

    response = await endpoint(
        _RouteRequest("chat"),
        ChatRequest(message="hello", session="session-1"),
    )

    assert [call[1] for call in calls] == ["selected-model", "backup-model"]
    assert response == {
        "response": "backup answer",
        "requested_model": "selected-model",
        "model": "backup-model",
        "requested_endpoint_id": None,
        "requested_endpoint_label": "Selected route",
        "endpoint_id": "backup",
        "endpoint_label": "backup",
    }
    assert saved[-1].metadata == {
        "model": "backup-model",
        "requested_model": "selected-model",
        "endpoint_id": "backup",
        "endpoint_label": "backup",
        "requested_endpoint_id": None,
        "requested_endpoint_label": "Selected route",
        "context_length": 128000,
        "context_trimmed": False,
    }


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("primary_context", "backup_context", "expected_counts"),
    [
        (100, 1000, (1, 3)),
        (1000, 100, (3, 1)),
    ],
)
async def test_nonstream_chat_shapes_each_candidate_from_route_neutral_history(
    monkeypatch,
    primary_context,
    backup_context,
    expected_counts,
):
    calls = []
    monkeypatch.setattr(foreground_model_routing, "_load_policy_preferences", lambda owner: {
        "foreground_fallback_enabled": True,
        "foreground_model_fallbacks": [
            {"endpoint_id": "backup", "model": "backup-model"},
        ],
    })
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda *args, **kwargs: [("https://backup.example/v1", "backup-model", {})],
    )
    async def fake_compact(
        session, url, model, messages, headers=None, owner=None, **kwargs
    ):
        return (
            list(messages),
            backup_context if model == "backup-model" else primary_context,
            False,
        )

    monkeypatch.setattr(chat_routes, "maybe_compact", fake_compact)
    monkeypatch.setattr(
        chat_routes,
        "trim_for_context",
        lambda messages, budget: list(messages) if budget >= 1000 else list(messages[-1:]),
    )

    async def fake_call(url, model, messages, **kwargs):
        calls.append((model, list(messages)))
        if model == "selected-model":
            raise HTTPException(503, "selected unavailable")
        return "backup answer"

    monkeypatch.setattr(llm_core, "llm_call_async", fake_call)
    endpoint, _saved = _chat_endpoint(monkeypatch)

    response = await endpoint(
        _RouteRequest("chat"),
        ChatRequest(message="hello", session="session-1"),
    )

    assert tuple(len(messages) for _model, messages in calls) == expected_counts
    assert response["model"] == "backup-model"


@pytest.mark.asyncio
async def test_nonstream_same_model_fallback_persists_endpoint_identity(monkeypatch):
    monkeypatch.setattr(foreground_model_routing, "_load_policy_preferences", lambda owner: {
        "foreground_fallback_enabled": True,
        "foreground_model_fallbacks": [
            {"endpoint_id": "account-two", "model": "selected-model"},
        ],
    })
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda *args, **kwargs: [(
            "https://selected.example/v1",
            "selected-model",
            {"Authorization": "Bearer account-two"},
        )],
    )

    async def fake_call(url, model, messages, **kwargs):
        if kwargs.get("headers", {}).get("Authorization") == "Bearer selected":
            raise HTTPException(429, "rate limited")
        return "second account answer"

    monkeypatch.setattr(llm_core, "llm_call_async", fake_call)
    endpoint, saved = _chat_endpoint(monkeypatch)

    response = await endpoint(
        _RouteRequest("chat"),
        ChatRequest(message="hello", session="session-1"),
    )

    assert response["requested_model"] == response["model"] == "selected-model"
    assert response["endpoint_id"] == "account-two"
    assert saved[-1].metadata["endpoint_id"] == "account-two"


@pytest.mark.asyncio
async def test_nonstream_chat_does_not_fallback_on_ineligible_failure(monkeypatch):
    calls = []
    monkeypatch.setattr(foreground_model_routing, "_load_policy_preferences", lambda owner: {
        "foreground_fallback_enabled": True,
        "foreground_model_fallbacks": [
            {"endpoint_id": "backup", "model": "backup-model"},
        ],
    })
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda entries, owner=None, require_exact_model=False: [
            ("https://backup.example/v1", "backup-model", {}),
        ],
    )

    async def fake_call(url, model, messages, **kwargs):
        calls.append(model)
        raise HTTPException(401, "invalid key")

    monkeypatch.setattr(llm_core, "llm_call_async", fake_call)
    endpoint, _saved = _chat_endpoint(monkeypatch)

    with pytest.raises(HTTPException) as exc:
        await endpoint(
            _RouteRequest("chat"),
            ChatRequest(message="hello", session="session-1"),
        )

    assert exc.value.status_code == 401
    assert calls == ["selected-model"]


@pytest.mark.asyncio
async def test_nonstream_chat_does_not_fallback_on_endpoint_configuration_error(monkeypatch):
    calls = []
    monkeypatch.setattr(foreground_model_routing, "_load_policy_preferences", lambda owner: {
        "foreground_fallback_enabled": True,
        "foreground_model_fallbacks": [
            {"endpoint_id": "backup", "model": "backup-model"},
        ],
    })
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda *args, **kwargs: [("https://backup.example/v1", "backup-model", {})],
    )

    async def fake_post(client, url, headers, **kwargs):
        calls.append(url)
        raise httpx.UnsupportedProtocol("unsupported protocol")

    monkeypatch.setattr(llm_core, "httpx_post_kimi_aware_async", fake_post)
    monkeypatch.setattr(llm_core, "_is_host_dead", lambda url: False)
    monkeypatch.setattr(llm_core, "note_model_activity", lambda *args, **kwargs: None)
    monkeypatch.setattr(llm_core, "_get_cached_response", lambda key: None)
    endpoint, saved = _chat_endpoint(monkeypatch, endpoint_url="ftp://selected.example")

    with pytest.raises(HTTPException) as exc:
        await endpoint(
            _RouteRequest("chat"),
            ChatRequest(message="hello", session="session-1"),
        )

    assert exc.value.status_code == 502
    assert len(calls) == 1
    assert saved == []


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("primary_body", "expected_status"),
    [
        ({"error": {"type": "invalid_request_error", "message": "unsupported model"}}, 400),
        ({"unexpected": "successful but malformed provider body"}, 502),
    ],
)
async def test_nonstream_chat_real_parser_never_falls_back_on_provider_or_schema_error(
    monkeypatch,
    primary_body,
    expected_status,
):
    calls = []
    monkeypatch.setattr(foreground_model_routing, "_load_policy_preferences", lambda owner: {
        "foreground_fallback_enabled": True,
        "foreground_model_fallbacks": [
            {"endpoint_id": "backup", "model": "backup-model"},
        ],
    })
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda entries, owner=None, require_exact_model=False: [
            ("https://backup.example/v1", "backup-model", {}),
        ],
    )

    class _Response:
        is_success = True
        status_code = 200
        text = ""

        def __init__(self, body):
            self._body = body

        def json(self):
            return self._body

    async def fake_post(_client, target_url, _headers, **kwargs):
        calls.append(target_url)
        if "selected.example" in target_url:
            return _Response(primary_body)
        return _Response({"choices": [{"message": {"content": "backup answer"}}]})

    monkeypatch.setattr(llm_core, "_get_http_client", lambda: object())
    monkeypatch.setattr(llm_core, "httpx_post_kimi_aware_async", fake_post)
    monkeypatch.setattr(llm_core, "_is_host_dead", lambda url: False)
    monkeypatch.setattr(llm_core, "note_model_activity", lambda *args, **kwargs: None)
    monkeypatch.setattr(llm_core, "_get_cached_response", lambda key: None)
    monkeypatch.setattr(llm_core, "_set_cached_response", lambda *args, **kwargs: None)
    endpoint, saved = _chat_endpoint(monkeypatch)

    with pytest.raises(HTTPException) as exc:
        await endpoint(
            _RouteRequest("chat"),
            ChatRequest(message="hello", session="session-1"),
        )

    assert exc.value.status_code == expected_status
    assert len(calls) == 1
    assert "selected.example" in calls[0]
    assert saved == []


@pytest.mark.asyncio
@pytest.mark.parametrize("use_fallback", [False, True])
async def test_nonstream_chat_real_parser_persists_provider_model_alias(
    monkeypatch,
    use_fallback,
):
    calls = []
    if use_fallback:
        monkeypatch.setattr(foreground_model_routing, "_load_policy_preferences", lambda owner: {
            "foreground_fallback_enabled": True,
            "foreground_model_fallbacks": [
                {"endpoint_id": "backup", "model": "backup-model"},
            ],
        })
        monkeypatch.setattr(
            foreground_model_routing,
            "resolve_fallback_entries",
            lambda *args, **kwargs: [
                ("https://backup.example/v1", "backup-model", {}),
            ],
        )
    else:
        monkeypatch.setattr(
            foreground_model_routing,
            "_load_policy_preferences",
            lambda owner: {},
        )

    class _Response:
        is_success = True
        status_code = 200
        text = ""

        def __init__(self, body):
            self._body = body

        def json(self):
            return self._body

    async def fake_post(_client, target_url, _headers, **kwargs):
        calls.append(target_url)
        if use_fallback and "selected.example" in target_url:
            return _Response({
                "error": {
                    "status": 503,
                    "message": "selected unavailable",
                },
            })
        return _Response({
            "model": "provider-backup-alias" if use_fallback else "provider-selected-alias",
            "choices": [{"message": {"content": "provider answer"}}],
        })

    monkeypatch.setattr(llm_core, "_get_http_client", lambda: object())
    monkeypatch.setattr(llm_core, "httpx_post_kimi_aware_async", fake_post)
    monkeypatch.setattr(llm_core, "_is_host_dead", lambda url: False)
    monkeypatch.setattr(llm_core, "note_model_activity", lambda *args, **kwargs: None)
    monkeypatch.setattr(llm_core, "_get_cached_response", lambda key: None)
    monkeypatch.setattr(llm_core, "_set_cached_response", lambda *args, **kwargs: None)
    endpoint, saved = _chat_endpoint(monkeypatch)

    response = await endpoint(
        _RouteRequest("chat"),
        ChatRequest(message="hello", session="session-1"),
    )

    expected_model = (
        "provider-backup-alias" if use_fallback else "provider-selected-alias"
    )
    assert response["response"] == "provider answer"
    assert response["model"] == expected_model
    assert saved[-1].metadata["model"] == expected_model
    assert response["endpoint_id"] == ("backup" if use_fallback else None)
    assert len(calls) == (2 if use_fallback else 1)


@pytest.mark.asyncio
async def test_nonstream_chat_does_not_treat_empty_response_as_unavailability(monkeypatch):
    calls = []
    monkeypatch.setattr(foreground_model_routing, "_load_policy_preferences", lambda owner: {
        "foreground_fallback_enabled": True,
        "foreground_model_fallbacks": [
            {"endpoint_id": "backup", "model": "backup-model"},
        ],
    })
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda entries, owner=None, require_exact_model=False: [
            ("https://backup.example/v1", "backup-model", {}),
        ],
    )

    async def fake_call(url, model, messages, **kwargs):
        calls.append(model)
        return ""

    monkeypatch.setattr(llm_core, "llm_call_async", fake_call)
    endpoint, _saved = _chat_endpoint(monkeypatch)

    response = await endpoint(
        _RouteRequest("chat"),
        ChatRequest(message="hello", session="session-1"),
    )

    assert calls == ["selected-model"]
    assert response["model"] == "selected-model"


@pytest.mark.asyncio
async def test_nonstream_named_owner_does_not_inherit_flat_opt_in(monkeypatch):
    calls = []
    monkeypatch.setattr(prefs_routes, "_load", lambda: {
        "foreground_fallback_enabled": True,
        "foreground_model_fallbacks": [
            {"endpoint_id": "shared", "model": "shared-model"},
        ],
    })
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda *args, **kwargs: pytest.fail("flat opt-in resolved a candidate for bob"),
    )

    async def fake_call(url, model, messages, **kwargs):
        calls.append(model)
        raise HTTPException(503, "selected unavailable")

    monkeypatch.setattr(llm_core, "llm_call_async", fake_call)
    endpoint, _saved = _chat_endpoint(monkeypatch, owner="bob")

    with pytest.raises(HTTPException) as exc:
        await endpoint(
            _RouteRequest("chat"),
            ChatRequest(message="hello", session="session-1"),
        )

    assert exc.value.status_code == 503
    assert calls == ["selected-model"]


def test_candidate_builder_appends_only_policy_authorized_fallbacks():
    """Chat and Agent share the same candidate-building policy boundary."""

    authorized = [("https://opt-in.example/v1", "opt-in-model", {})]
    policy = ForegroundModelPolicy(enabled=True, fallback_candidates=tuple(authorized))

    assert build_foreground_model_candidates(
        "https://selected.example/v1",
        "selected-model",
        {"Authorization": "Bearer selected"},
        owner="alice",
        policy=policy,
    ) == [
        ("https://selected.example/v1", "selected-model", {"Authorization": "Bearer selected"}),
        *authorized,
    ]


def test_selected_endpoint_id_wins_over_route_equality_scan(monkeypatch):
    selected = {
        "endpoint_id": "account-two",
        "endpoint_label": "Account two",
        "endpoint_cost_tracked": True,
    }
    seen = []

    def exact_id_descriptor(endpoint_id, url, model, headers, owner=None):
        seen.append((endpoint_id, url, model, headers, owner))
        return selected

    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_route_descriptor_by_id",
        exact_id_descriptor,
        raising=False,
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_route_descriptor",
        lambda *args, **kwargs: {
            "endpoint_id": "account-one",
            "endpoint_label": "Account one",
            "endpoint_cost_tracked": True,
        },
    )

    descriptors = foreground_model_routing.build_foreground_route_descriptors(
        "https://provider.example/v1/chat/completions",
        "same-model",
        {"Authorization": "Bearer shared-key"},
        owner="alice",
        policy=ForegroundModelPolicy(),
        selected_endpoint_id="account-two",
    )

    assert descriptors == [selected]
    assert seen == [(
        "account-two",
        "https://provider.example/v1/chat/completions",
        "same-model",
        {"Authorization": "Bearer shared-key"},
        "alice",
    )]


@pytest.mark.asyncio
async def test_chat_stream_threads_form_endpoint_id_to_descriptor_builder(monkeypatch):
    captured = {}
    endpoint = _chat_stream_endpoint(monkeypatch, "chat", captured)
    seen = []

    def fake_descriptors(*args, selected_endpoint_id=None, **kwargs):
        seen.append(selected_endpoint_id)
        return [{
            "endpoint_id": selected_endpoint_id,
            "endpoint_label": "Selected endpoint",
            "endpoint_cost_tracked": True,
        }]

    monkeypatch.setattr(
        chat_routes,
        "build_foreground_route_descriptors",
        fake_descriptors,
    )
    request = _RouteRequest("chat")
    request._form["selected_endpoint_id"] = "account-two"

    response = await endpoint(request)
    async for _chunk in response.body_iterator:
        pass

    assert seen == ["account-two"]


@pytest.mark.asyncio
async def test_nonstream_chat_threads_request_endpoint_id_to_descriptor_builder(
    monkeypatch,
):
    seen = []

    def fake_descriptors(*args, selected_endpoint_id=None, **kwargs):
        seen.append(selected_endpoint_id)
        return [{
            "endpoint_id": selected_endpoint_id,
            "endpoint_label": "Selected endpoint",
            "endpoint_cost_tracked": True,
        }]

    async def fake_call(url, model, messages, **kwargs):
        return "selected answer"

    monkeypatch.setattr(
        foreground_model_routing,
        "_load_policy_preferences",
        lambda owner: {},
    )
    monkeypatch.setattr(
        chat_routes,
        "build_foreground_route_descriptors",
        fake_descriptors,
    )
    monkeypatch.setattr(llm_core, "llm_call_async", fake_call)
    endpoint, _saved = _chat_endpoint(monkeypatch)

    await endpoint(
        _RouteRequest("chat"),
        ChatRequest(
            message="hello",
            session="session-1",
            selected_endpoint_id="account-two",
        ),
    )

    assert seen == ["account-two"]


def test_strict_policy_builds_only_the_selected_chat_candidate():
    candidates = build_foreground_model_candidates(
        "https://selected.example/v1",
        "selected-model",
        {"Authorization": "Bearer selected"},
        owner="alice",
        policy=ForegroundModelPolicy(),
    )

    assert candidates == [
        ("https://selected.example/v1", "selected-model", {"Authorization": "Bearer selected"})
    ]


def test_legacy_chat_resolver_is_disconnected():
    assert not hasattr(endpoint_resolver, "resolve_chat_fallback_candidates")


def test_retired_silent_endpoint_switcher_is_not_callable():
    assert not hasattr(chat_helpers, "try_fallback_endpoint")


@pytest.mark.parametrize(
    "prefs",
    [
        {},
        {"foreground_fallback_enabled": False, "foreground_model_fallbacks": [{"endpoint_id": "b", "model": "m"}]},
        {"foreground_fallback_enabled": "true", "foreground_model_fallbacks": [{"endpoint_id": "b", "model": "m"}]},
        {"foreground_fallback_enabled": True, "foreground_model_fallbacks": []},
        {"default_model_fallbacks": [{"endpoint_id": "legacy", "model": "legacy"}]},
    ],
)
def test_foreground_policy_fails_closed_without_explicit_complete_opt_in(monkeypatch, prefs):
    monkeypatch.setattr(foreground_model_routing, "_load_policy_preferences", lambda owner=None: prefs)
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda entries, owner=None: pytest.fail("disabled policy resolved fallback entries"),
    )

    assert resolve_foreground_model_policy("alice") == ForegroundModelPolicy()


def test_foreground_policy_resolves_ordered_owner_scoped_entries(monkeypatch):
    entries = [
        {"endpoint_id": f"ep-{i}", "model": f"model-{i}"}
        for i in range(MAX_FOREGROUND_FALLBACKS + 2)
    ]
    seen = {}
    monkeypatch.setattr(
        foreground_model_routing,
        "_load_policy_preferences",
        lambda owner=None: {
            "foreground_fallback_enabled": True,
            "foreground_model_fallbacks": entries,
        },
    )

    def fake_resolve(resolved_entries, owner=None, *, require_exact_model=False):
        seen["entries"] = resolved_entries
        seen["owner"] = owner
        seen["require_exact_model"] = require_exact_model
        return [("https://backup.example/v1", "backup", {"Authorization": "secret"})]

    monkeypatch.setattr(foreground_model_routing, "resolve_fallback_entries", fake_resolve)

    policy = resolve_foreground_model_policy("alice")

    assert policy.enabled is True
    assert policy.fallback_candidates == (
        ("https://backup.example/v1", "backup", {"Authorization": "secret"}),
    )
    assert policy.eligible_statuses == FOREGROUND_AVAILABILITY_STATUSES
    assert policy.fallback_on_empty is False
    assert seen == {
        "entries": entries[:MAX_FOREGROUND_FALLBACKS],
        "owner": "alice",
        "require_exact_model": True,
    }


def test_foreground_policy_filters_allowed_models_before_maximum_slice(monkeypatch):
    disallowed = [
        {"endpoint_id": f"blocked-{i}", "model": f"blocked-model-{i}"}
        for i in range(MAX_FOREGROUND_FALLBACKS)
    ]
    allowed_entry = {"endpoint_id": "allowed", "model": "allowed-model"}
    seen = {}
    monkeypatch.setattr(
        foreground_model_routing,
        "_load_policy_preferences",
        lambda owner=None: {
            "foreground_fallback_enabled": True,
            "foreground_model_fallbacks": [*disallowed, allowed_entry],
        },
    )

    def fake_resolve(entries, owner=None, *, require_exact_model=False):
        seen["entries"] = entries
        return [("https://allowed.example/v1", "allowed-model", {})]

    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        fake_resolve,
    )

    policy = resolve_foreground_model_policy(
        "alice",
        allowed_models={"allowed-model"},
    )

    assert policy.enabled is True
    assert policy.fallback_candidates == (
        ("https://allowed.example/v1", "allowed-model", {}),
    )
    assert seen["entries"] == [allowed_entry]


def test_compatibility_resolver_keeps_descriptor_aligned_when_an_entry_is_skipped(
    monkeypatch,
):
    entries = [
        {"endpoint_id": "missing", "model": "missing-model"},
        {"endpoint_id": "backup", "model": "backup-model"},
    ]
    monkeypatch.setattr(
        foreground_model_routing,
        "_load_policy_preferences",
        lambda owner=None: {
            "foreground_fallback_enabled": True,
            "foreground_model_fallbacks": entries,
        },
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda *args, **kwargs: [
            ("https://backup.example/v1", "backup-model", {})
        ],
    )

    policy = resolve_foreground_model_policy("alice")

    assert policy.fallback_candidates == (
        ("https://backup.example/v1", "backup-model", {}),
    )
    assert policy.fallback_descriptors[0]["endpoint_id"] == "backup"


def test_foreground_policy_loads_only_the_requested_users_preferences(monkeypatch):
    by_owner = {
        "alice": {
            "foreground_fallback_enabled": True,
            "foreground_model_fallbacks": [{"endpoint_id": "alice-backup", "model": "alice-model"}],
        },
        "bob": {
            "foreground_fallback_enabled": False,
            "foreground_model_fallbacks": [{"endpoint_id": "bob-backup", "model": "bob-model"}],
        },
    }
    seen = []
    monkeypatch.setattr(foreground_model_routing, "_load_policy_preferences", lambda owner=None: by_owner[owner])

    def fake_resolve(entries, owner=None, *, require_exact_model=False):
        seen.append((entries, owner))
        assert require_exact_model is True
        return [(f"https://{owner}.example/v1", f"{owner}-model", {})]

    monkeypatch.setattr(foreground_model_routing, "resolve_fallback_entries", fake_resolve)

    assert resolve_foreground_model_policy("alice").enabled is True
    assert resolve_foreground_model_policy("bob") == ForegroundModelPolicy()
    assert seen == [([{"endpoint_id": "alice-backup", "model": "alice-model"}], "alice")]


def test_named_owner_does_not_inherit_flat_single_user_fallback_consent(monkeypatch):
    monkeypatch.setattr(prefs_routes, "_load", lambda: {
        "foreground_fallback_enabled": True,
        "foreground_model_fallbacks": [
            {"endpoint_id": "shared-backup", "model": "shared-model"},
        ],
    })
    monkeypatch.setattr(
        prefs_routes,
        "_load_for_user",
        lambda owner=None: pytest.fail("named foreground policy used flat compatibility loader"),
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda *args, **kwargs: pytest.fail("flat consent resolved candidate endpoints for bob"),
    )

    assert resolve_foreground_model_policy("bob") == ForegroundModelPolicy()


def test_named_owner_unrelated_save_does_not_import_flat_fallback_consent(
    monkeypatch,
    tmp_path,
):
    prefs_file = tmp_path / "user_prefs.json"
    prefs_file.write_text(json.dumps({
        "foreground_fallback_enabled": True,
        "foreground_model_fallbacks": [
            {"endpoint_id": "single-user", "model": "single-model"},
        ],
        "default_model_fallbacks": [
            {"endpoint_id": "legacy", "model": "legacy-model"},
        ],
    }), encoding="utf-8")
    monkeypatch.setattr(prefs_routes, "PREFS_FILE", str(prefs_file))
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda *args, **kwargs: pytest.fail("Bob inherited flat fallback consent"),
    )

    assert resolve_foreground_model_policy("bob") == ForegroundModelPolicy()
    bob = prefs_routes._load_for_user("bob")
    bob["theme"] = "dark"
    prefs_routes._save_for_user("bob", bob)

    assert resolve_foreground_model_policy("bob") == ForegroundModelPolicy()
    raw = prefs_routes._load()
    assert raw["_users"]["bob"] == {"theme": "dark"}
    assert raw["default_model_fallbacks"][0]["endpoint_id"] == "legacy"


def test_startup_pref_migration_does_not_transfer_flat_fallback_consent(
    monkeypatch,
    tmp_path,
):
    database_file = tmp_path / "app.db"
    database_file.touch()
    (tmp_path / "auth.json").write_text(json.dumps({
        "users": {
            "alice": {"is_admin": True},
        },
    }), encoding="utf-8")
    prefs_file = tmp_path / "user_prefs.json"
    flat_fallbacks = [
        {"endpoint_id": "single-user", "model": "single-model"},
    ]
    legacy_fallbacks = [
        {"endpoint_id": "legacy", "model": "legacy-model"},
    ]
    prefs_file.write_text(json.dumps({
        "theme": "dark",
        "foreground_fallback_enabled": True,
        "foreground_model_fallbacks": flat_fallbacks,
        "default_model_fallbacks": legacy_fallbacks,
    }), encoding="utf-8")
    monkeypatch.setattr(database, "DATABASE_URL", f"sqlite:///{database_file}")
    monkeypatch.setattr(database, "AUTH_FILE", str(tmp_path / "auth.json"))
    monkeypatch.setattr(database, "MEMORY_FILE", str(tmp_path / "memory.json"))
    monkeypatch.setattr(database, "USER_PREFS_FILE", str(prefs_file))
    monkeypatch.setattr(prefs_routes, "PREFS_FILE", str(prefs_file))
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda *args, **kwargs: pytest.fail(
            "startup migration transferred flat fallback consent"
        ),
    )

    database._migrate_assign_legacy_owner()

    raw = json.loads(prefs_file.read_text(encoding="utf-8"))
    assert raw["foreground_fallback_enabled"] is True
    assert raw["foreground_model_fallbacks"] == flat_fallbacks
    assert raw["_users"]["alice"] == {
        "theme": "dark",
        "default_model_fallbacks": legacy_fallbacks,
    }
    assert resolve_foreground_model_policy("alice") == ForegroundModelPolicy()


def test_auth_disabled_write_in_multiuser_store_does_not_grant_named_consent(
    monkeypatch,
    tmp_path,
):
    prefs_file = tmp_path / "user_prefs.json"
    prefs_file.write_text(json.dumps({
        "_users": {
            "alice": {"theme": "dark"},
            "bob": {"theme": "light"},
        },
    }), encoding="utf-8")
    monkeypatch.setattr(prefs_routes, "PREFS_FILE", str(prefs_file))
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda *args, **kwargs: pytest.fail(
            "auth-disabled consent was written into a named owner"
        ),
    )

    ownerless = prefs_routes._load_for_user(None)
    ownerless["foreground_fallback_enabled"] = True
    ownerless["foreground_model_fallbacks"] = [
        {"endpoint_id": "single-user", "model": "single-model"},
    ]
    prefs_routes._save_for_user(None, ownerless)

    raw = prefs_routes._load()
    assert raw["foreground_fallback_enabled"] is True
    assert raw["foreground_model_fallbacks"][0]["endpoint_id"] == "single-user"
    assert raw["_users"]["alice"] == {"theme": "dark"}
    assert raw["_users"]["bob"] == {"theme": "light"}
    assert resolve_foreground_model_policy("alice") == ForegroundModelPolicy()
    assert resolve_foreground_model_policy("bob") == ForegroundModelPolicy()


def test_named_owner_resolves_only_an_actual_scoped_preferences_dict(monkeypatch):
    entry = {"endpoint_id": "alice-backup", "model": "alice-model"}
    monkeypatch.setattr(prefs_routes, "_load", lambda: {
        "_users": {
            "alice": {
                "foreground_fallback_enabled": True,
                "foreground_model_fallbacks": [entry],
            },
            "bob": ["not", "a", "preferences", "dict"],
        },
    })
    seen = []
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda entries, owner=None, require_exact_model=False: (
            seen.append((entries, owner, require_exact_model))
            or [("https://alice.example/v1", "alice-model", {})]
        ),
    )

    assert resolve_foreground_model_policy("alice").enabled is True
    assert resolve_foreground_model_policy("bob") == ForegroundModelPolicy()
    assert seen == [([entry], "alice", True)]


def test_auth_disabled_owner_none_preserves_flat_single_user_policy(monkeypatch):
    prefs = {
        "foreground_fallback_enabled": True,
        "foreground_model_fallbacks": [
            {"endpoint_id": "single-user-backup", "model": "backup-model"},
        ],
    }
    seen = []
    monkeypatch.setattr(prefs_routes, "_load_for_user", lambda owner=None: prefs)
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda entries, owner=None, require_exact_model=False: (
            seen.append((entries, owner, require_exact_model))
            or [("https://backup.example/v1", "backup-model", {})]
        ),
    )

    policy = resolve_foreground_model_policy(None)

    assert policy.enabled is True
    assert policy.fallback_candidates == (("https://backup.example/v1", "backup-model", {}),)
    assert seen == [(
        [{"endpoint_id": "single-user-backup", "model": "backup-model"}],
        None,
        True,
    )]


def test_foreground_policy_filters_models_outside_the_callers_allowlist(monkeypatch):
    seen = []
    monkeypatch.setattr(
        foreground_model_routing,
        "_load_policy_preferences",
        lambda owner=None: {
            "foreground_fallback_enabled": True,
            "foreground_model_fallbacks": [
                {"endpoint_id": "allowed", "model": "allowed-model"},
                {"endpoint_id": "blocked", "model": "blocked-model"},
            ],
        },
    )

    def fake_resolve(entries, owner=None, *, require_exact_model=False):
        seen.extend(entries)
        return [("https://allowed.example/v1", "allowed-model", {})]

    monkeypatch.setattr(foreground_model_routing, "resolve_fallback_entries", fake_resolve)

    policy = resolve_foreground_model_policy("alice", allowed_models={"allowed-model"})

    assert policy.enabled is True
    assert seen == [{"endpoint_id": "allowed", "model": "allowed-model"}]


def test_foreground_policy_is_strict_when_all_fallbacks_are_disallowed(monkeypatch):
    monkeypatch.setattr(
        foreground_model_routing,
        "_load_policy_preferences",
        lambda owner=None: {
            "foreground_fallback_enabled": True,
            "foreground_model_fallbacks": [
                {"endpoint_id": "blocked", "model": "blocked-model"},
            ],
        },
    )
    monkeypatch.setattr(
        foreground_model_routing,
        "resolve_fallback_entries",
        lambda *args, **kwargs: pytest.fail("disallowed entry reached credential resolution"),
    )

    assert resolve_foreground_model_policy(
        "alice",
        allowed_models={"selected-model"},
    ) == ForegroundModelPolicy()


def test_utility_resolver_does_not_inherit_legacy_chat_fallbacks(monkeypatch):
    seen_keys = []

    def fake_resolve(setting_key, owner=None):
        seen_keys.append((setting_key, owner))
        return [("https://utility.example/v1", "utility-model", {})]

    monkeypatch.setattr(endpoint_resolver, "_resolve_fallback_candidates", fake_resolve)

    assert endpoint_resolver.resolve_utility_fallback_candidates(owner="alice") == [
        ("https://utility.example/v1", "utility-model", {})
    ]
    assert seen_keys == [("utility_model_fallbacks", "alice")]


def test_multi_round_agent_uses_only_selected_model(monkeypatch):
    """Every Agent round receives only the selected foreground candidate."""

    seen_candidates = []
    round_number = 0

    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(agent_loop, "estimate_tokens", lambda *args, **kwargs: 10)
    monkeypatch.setattr(agent_loop, "blocked_tools_for_owner", lambda owner: set())
    async def fake_stream(candidates, messages, **kwargs):
        nonlocal round_number
        round_number += 1
        seen_candidates.append([(url, model) for url, model, _headers in candidates])
        if round_number == 1:
            call = {"name": "bash", "arguments": json.dumps({"command": "printf ok"})}
            yield f'data: {json.dumps({"type": "tool_calls", "calls": [call]})}\n\n'
        else:
            yield f'data: {json.dumps({"delta": "done"})}\n\n'
        yield "data: [DONE]\n\n"

    async def fake_execute(block, *args, **kwargs):
        return "bash", {"output": "ok", "exit_code": 0}

    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)
    monkeypatch.setattr(agent_loop, "execute_tool_block", fake_execute)

    chunks = _collect(
        agent_loop.stream_agent_loop(
            "https://selected.example/v1",
            "selected-model",
            [{"role": "user", "content": "Run one tool and report back."}],
            max_rounds=3,
            relevant_tools={"bash"},
            fallbacks=[],
            _is_teacher_run=True,
        )
    )

    assert seen_candidates == [
        [("https://selected.example/v1", "selected-model")],
        [("https://selected.example/v1", "selected-model")],
    ]
    assert any('"delta": "done"' in chunk for chunk in chunks)


def test_multi_round_agent_pins_answering_fallback_for_the_run(monkeypatch):
    """A tool round must not silently switch back to the selected model."""

    seen_candidates = []
    round_number = 0
    primary = ("https://selected.example/v1", "selected-model", {})
    backup = ("https://backup.example/v1", "backup-model", {"Authorization": "backup"})
    route_modes = []

    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(agent_loop, "estimate_tokens", lambda *args, **kwargs: 10)
    monkeypatch.setattr(
        agent_loop,
        "_agent_route_tool_mode",
        lambda url, model, owner=None, headers=None: route_modes.append((url, model, owner, headers)) or (model == "selected-model", False, False),
    )

    async def fake_stream(candidates, messages, **kwargs):
        nonlocal round_number
        round_number += 1
        seen_candidates.append(candidates)
        assert kwargs["fallback_statuses"] == FOREGROUND_AVAILABILITY_STATUSES
        assert kwargs["fallback_on_empty"] is False
        if round_number == 1:
            yield f'data: {json.dumps({"type": "fallback", "selected_model": "selected-model", "answered_by": "backup-model", "candidate_index": 1})}\n\n'
            call = {"name": "bash", "arguments": json.dumps({"command": "printf ok"})}
            yield f'data: {json.dumps({"type": "tool_calls", "calls": [call]})}\n\n'
        else:
            yield f'data: {json.dumps({"delta": "done"})}\n\n'
        yield "data: [DONE]\n\n"

    async def fake_execute(block, *args, **kwargs):
        return "bash", {"output": "ok", "exit_code": 0}

    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)
    monkeypatch.setattr(agent_loop, "execute_tool_block", fake_execute)

    chunks = _collect(
        agent_loop.stream_agent_loop(
            primary[0],
            primary[1],
            [{"role": "user", "content": "Run one tool and report back."}],
            max_rounds=3,
            relevant_tools={"bash"},
            headers={},
            fallbacks=[backup],
            fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
            fallback_on_empty=False,
            _is_teacher_run=True,
        )
    )

    assert seen_candidates == [[primary, backup], [backup]]
    fallback_event = next(chunk for chunk in chunks if '"type": "fallback"' in chunk)
    fallback_data = json.loads(fallback_event[6:])
    assert fallback_data["pinned_for_run"] is True
    assert fallback_data["round"] == 1
    metrics_event = next(chunk for chunk in chunks if '"type": "metrics"' in chunk)
    metrics = json.loads(metrics_event[6:])["data"]
    assert metrics["requested_model"] == "selected-model"
    assert metrics["model"] == "backup-model"
    assert metrics["round_models"] == ["backup-model", "backup-model"]
    assert metrics["tool_events"][0]["model"] == "backup-model"
    assert route_modes == [
        ("https://selected.example/v1", "selected-model", None, {}),
        ("https://backup.example/v1", "backup-model", None, {"Authorization": "backup"}),
    ]


def test_late_agent_fallback_records_each_round_and_stays_pinned(monkeypatch):
    seen_candidates = []
    round_number = 0
    primary = ("https://selected.example/v1", "selected-model", {})
    backup = ("https://backup.example/v1", "backup-model", {})

    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(agent_loop, "estimate_tokens", lambda *args, **kwargs: 10)
    monkeypatch.setattr(agent_loop, "_agent_route_tool_mode", lambda url, model, owner=None, headers=None: (True, False, False))

    async def fake_stream(candidates, messages, **kwargs):
        nonlocal round_number
        round_number += 1
        seen_candidates.append(candidates)
        if round_number == 1:
            yield 'data: {"delta": "primary round"}\n\n'
            call = {"name": "bash", "arguments": json.dumps({"command": "printf one"})}
            yield f'data: {json.dumps({"type": "tool_calls", "calls": [call]})}\n\n'
        elif round_number == 2:
            yield f'data: {json.dumps({"type": "fallback", "selected_model": "selected-model", "answered_by": "backup-model", "candidate_index": 1})}\n\n'
            yield f'data: {json.dumps({"type": "model_actual", "requested_model": "backup-model", "model": "provider-backup-alias"})}\n\n'
            yield 'data: {"delta": "backup round"}\n\n'
            call = {"name": "bash", "arguments": json.dumps({"command": "printf two"})}
            yield f'data: {json.dumps({"type": "tool_calls", "calls": [call]})}\n\n'
        else:
            yield 'data: {"delta": "backup final"}\n\n'
        yield "data: [DONE]\n\n"

    async def fake_execute(block, *args, **kwargs):
        # Keep this routing-only test untainted with a content-free fixture.
        # Any model-visible shell error is workspace-derived and correctly
        # reaches the exact-approval boundary on the next action.
        return "bash", {"exit_code": 1}

    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)
    monkeypatch.setattr(agent_loop, "execute_tool_block", fake_execute)

    chunks = _collect(
        agent_loop.stream_agent_loop(
            primary[0],
            primary[1],
            [{"role": "user", "content": "Run two tools and report back."}],
            headers=primary[2],
            max_rounds=4,
            relevant_tools={"bash"},
            fallbacks=[backup],
            fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
            fallback_on_empty=False,
            _is_teacher_run=True,
        )
    )

    assert seen_candidates == [[primary, backup], [primary, backup], [backup]]
    fallback_data = json.loads(next(chunk for chunk in chunks if '"type": "fallback"' in chunk)[6:])
    assert fallback_data["round"] == 2
    model_actual = json.loads(next(chunk for chunk in chunks if '"type": "model_actual"' in chunk)[6:])
    assert model_actual["round"] == 2
    assert model_actual["requested_model"] == "selected-model"
    assert model_actual["model"] == "provider-backup-alias"
    metrics = json.loads(next(chunk for chunk in chunks if '"type": "metrics"' in chunk)[6:])["data"]
    assert metrics["round_models"] == ["selected-model", "provider-backup-alias", "backup-model"]
    assert [event["model"] for event in metrics["tool_events"]] == [
        "selected-model",
        "provider-backup-alias",
    ]


@pytest.mark.parametrize("status", [400, 401, 404])
def test_agent_terminal_first_round_error_has_no_success_completion(monkeypatch, status):
    calls = 0
    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(agent_loop, "estimate_tokens", lambda *args, **kwargs: 10)

    async def fake_stream(candidates, messages, **kwargs):
        nonlocal calls
        calls += 1
        yield f'event: error\ndata: {json.dumps({"status": status, "error": "provider rejected request"})}\n\n'

    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)

    chunks = _collect(
        agent_loop.stream_agent_loop(
            "https://selected.example/v1",
            "selected-model",
            [{"role": "user", "content": "hello"}],
            max_rounds=3,
            relevant_tools=set(),
            fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
            fallback_on_empty=False,
            _is_teacher_run=True,
        )
    )

    assert calls == 1
    assert any(chunk.startswith("event: error") for chunk in chunks)
    assert not any('"type": "metrics"' in chunk for chunk in chunks), chunks
    assert "data: [DONE]\n\n" not in chunks
    assert not any("empty response" in chunk.lower() for chunk in chunks)


@pytest.mark.parametrize(
    ("provider_status", "expected_status", "expected_message"),
    [
        (400, 400, "Model request failed (HTTP 400)"),
        (429.9, None, "Model request failed"),
    ],
)
def test_agent_terminal_later_round_error_stops_after_completed_tool(
    monkeypatch,
    provider_status,
    expected_status,
    expected_message,
):
    calls = 0
    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(agent_loop, "estimate_tokens", lambda *args, **kwargs: 10)
    monkeypatch.setattr(agent_loop, "blocked_tools_for_owner", lambda owner: set())
    monkeypatch.setattr(
        agent_loop,
        "_agent_route_tool_mode",
        lambda *args, **kwargs: (True, False, False),
    )

    async def fake_stream(candidates, messages, **kwargs):
        nonlocal calls
        calls += 1
        if calls == 1:
            tool_call = {
                "name": "bash",
                "arguments": json.dumps({"command": "printf one"}),
            }
            yield f'data: {json.dumps({"type": "tool_calls", "calls": [tool_call]})}\n\n'
            yield "data: [DONE]\n\n"
            return
        yield 'data: {"delta": "partial second-round prose"}\n\n'
        yield f'event: error\ndata: {json.dumps({"status": provider_status, "error": "unsupported model"})}\n\n'

    async def fake_execute(block, *args, **kwargs):
        return "bash", {"output": "ok", "exit_code": 0}

    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)
    monkeypatch.setattr(agent_loop, "execute_tool_block", fake_execute)

    chunks = _collect(
        agent_loop.stream_agent_loop(
            "https://selected.example/v1",
            "selected-model",
            [{"role": "user", "content": "Run one tool."}],
            max_rounds=3,
            relevant_tools={"bash"},
            fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
            fallback_on_empty=False,
            _is_teacher_run=True,
        )
    )

    assert calls == 2
    assert sum('"type": "agent_step"' in chunk for chunk in chunks) == 1
    terminal = json.loads(next(
        chunk for chunk in chunks if '"type": "agent_terminal"' in chunk
    )[6:])["data"]
    assert terminal["failed"] is True
    assert terminal["failure"]["status"] == expected_status
    assert terminal["tool_events"][0]["output"] == "ok"
    assert terminal["failure"] == {
        "status": expected_status,
        "message": expected_message,
    }
    assert terminal["round_models"] == ["selected-model", "selected-model"]
    assert terminal["round_texts"][-1] == (
        "partial second-round prose\n\n"
        f"[Agent stopped: {expected_message}]"
    )
    assert any(chunk.startswith("event: error") for chunk in chunks)
    assert not any('"type": "metrics"' in chunk for chunk in chunks)
    assert "data: [DONE]\n\n" not in chunks
    assert not any("empty response" in chunk.lower() for chunk in chunks)


@pytest.mark.parametrize(
    ("provider_status", "expected_status", "expected_message"),
    [
        (503, 503, "Model request failed (HTTP 503)"),
        (429.9, None, "Model request failed"),
    ],
)
def test_direct_low_signal_partial_error_emits_terminal_history(
    monkeypatch,
    provider_status,
    expected_status,
    expected_message,
):
    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(agent_loop, "estimate_tokens", lambda *args, **kwargs: 10)
    monkeypatch.setattr(
        agent_loop,
        "_classify_agent_request",
        lambda messages, latest: {
            "low_signal": True,
            "continuation": False,
            "domains": [],
            "retrieval_query": latest,
        },
    )
    monkeypatch.setattr(agent_loop, "_is_casual_low_signal", lambda latest: True)

    async def fake_stream(candidates, messages, **kwargs):
        yield 'data: {"delta": "visible direct partial"}\n\n'
        yield f'event: error\ndata: {json.dumps({"status": provider_status, "error": "provider detail"})}\n\n'

    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)

    chunks = _collect(agent_loop.stream_agent_loop(
        "https://selected.example/v1",
        "selected-model",
        [{"role": "user", "content": "hello"}],
        relevant_tools=set(),
        route_descriptors=[{
            "endpoint_id": "selected",
            "endpoint_label": "Selected",
            "endpoint_cost_tracked": True,
        }],
        fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
        fallback_on_empty=False,
        _is_teacher_run=True,
    ))

    terminal = json.loads(next(
        chunk for chunk in chunks if '"type": "agent_terminal"' in chunk
    )[6:])["data"]
    assert terminal["round_texts"] == [
        "visible direct partial\n\n"
        f"[Agent stopped: {expected_message}]"
    ]
    assert terminal["failure"] == {
        "status": expected_status,
        "message": expected_message,
    }
    assert terminal["endpoint_cost_tracked"] is True
    assert terminal["usage_buckets"][0]["endpoint_cost_tracked"] is True
    assert any(chunk.startswith("event: error") for chunk in chunks)
    assert "data: [DONE]\n\n" not in chunks


@pytest.mark.parametrize("terminal_error", [False, True])
def test_direct_low_signal_fallback_estimates_winning_route_prompt(
    monkeypatch,
    terminal_error,
):
    primary = ("https://selected.example/v1", "generic-model", {})
    backup = ("https://backup.example/v1", "odysseus-qwen-backup", {})
    candidate_requests = []

    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(
        agent_loop,
        "_classify_agent_request",
        lambda messages, latest: {
            "low_signal": True,
            "continuation": False,
            "domains": [],
            "retrieval_query": latest,
        },
    )
    monkeypatch.setattr(agent_loop, "_is_casual_low_signal", lambda latest: True)
    monkeypatch.setattr(
        agent_loop,
        "_is_odysseus_qwen_model",
        lambda candidate_model: candidate_model == backup[1],
    )
    monkeypatch.setattr(
        agent_loop,
        "_minimal_odysseus_general_messages",
        lambda messages, include_memory=True: [
            {"role": "system", "content": "larger backup route prompt"},
            *list(messages),
        ],
    )
    monkeypatch.setattr(
        agent_loop,
        "estimate_tokens",
        lambda request_messages: (
            99
            if any(
                message.get("content") == "larger backup route prompt"
                for message in request_messages
            )
            else 7
        ),
    )

    async def fake_stream(candidates, messages, **kwargs):
        assert messages == [{"role": "user", "content": "hello"}]
        request = kwargs["candidate_request_factory"](1, *backup)
        candidate_requests.append(request["messages"])
        fallback_event = {
            "type": "fallback",
            "selected_model": primary[1],
            "answered_by": backup[1],
            "candidate_index": 1,
            "selected_endpoint_id": "selected",
            "selected_endpoint_label": "Selected",
            "selected_endpoint_cost_tracked": False,
            "answered_by_endpoint_id": "backup",
            "answered_by_endpoint_label": "Backup",
            "answered_by_endpoint_cost_tracked": True,
        }
        yield "data: " + json.dumps(fallback_event) + "\n\n"
        yield 'data: {"type": "usage", "data": {"input_tokens": null, "output_tokens": 1}}\n\n'
        yield 'data: {"delta": "backup response"}\n\n'
        if terminal_error:
            yield 'event: error\ndata: {"status": 503, "error": "unavailable"}\n\n'
        else:
            yield "data: [DONE]\n\n"

    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)

    chunks = _collect(agent_loop.stream_agent_loop(
        primary[0],
        primary[1],
        [{"role": "user", "content": "hello"}],
        relevant_tools=set(),
        fallbacks=[backup],
        route_descriptors=[
            {
                "endpoint_id": "selected",
                "endpoint_label": "Selected",
                "endpoint_cost_tracked": False,
            },
            {
                "endpoint_id": "backup",
                "endpoint_label": "Backup",
                "endpoint_cost_tracked": True,
            },
        ],
        fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
        fallback_on_empty=False,
        _is_teacher_run=True,
    ))

    assert candidate_requests[0][0] == {
        "role": "system",
        "content": "larger backup route prompt",
    }
    event_type = "agent_terminal" if terminal_error else "metrics"
    accounted = json.loads(next(
        chunk for chunk in chunks if f'"type": "{event_type}"' in chunk
    )[6:])["data"]
    assert accounted["model"] == backup[1]
    assert accounted["endpoint_id"] == "backup"
    assert accounted["input_tokens"] == 99
    assert accounted["usage_buckets"] == [{
        "round": 1,
        "model": backup[1],
        "endpoint_id": "backup",
        "endpoint_label": "Backup",
        "input_tokens": 99,
        "output_tokens": len("backup response") // 4,
        "usage_source": "estimated",
        "endpoint_cost_tracked": True,
    }]


def test_direct_low_signal_configuration_error_surfaces_without_fake_success(monkeypatch):
    monkeypatch.setattr(
        agent_loop,
        "get_setting",
        lambda key, default=None: (
            "not-an-int" if key == "agent_stream_timeout_seconds" else default
        ),
    )
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(
        agent_loop,
        "_classify_agent_request",
        lambda messages, latest: {
            "low_signal": True,
            "continuation": False,
            "domains": [],
            "retrieval_query": latest,
        },
    )
    monkeypatch.setattr(agent_loop, "_is_casual_low_signal", lambda latest: True)

    chunks = _collect(agent_loop.stream_agent_loop(
        "https://selected.example/v1",
        "selected-model",
        [{"role": "user", "content": "hello"}],
        relevant_tools=set(),
        fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
        fallback_on_empty=False,
        _is_teacher_run=True,
    ))

    assert len(chunks) == 1
    assert chunks[0].startswith("event: error")
    payload = json.loads(chunks[0].split("data: ", 1)[1])
    assert payload == {
        "error": "Model request failed",
        "status": 500,
        "fallback_eligible": False,
    }
    assert not any('"delta": "Hey."' in chunk for chunk in chunks)
    assert not any('"type": "metrics"' in chunk for chunk in chunks)
    assert "data: [DONE]\n\n" not in chunks


def test_direct_low_signal_empty_completion_surfaces_without_fake_success(monkeypatch):
    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(
        agent_loop,
        "_classify_agent_request",
        lambda messages, latest: {
            "low_signal": True,
            "continuation": False,
            "domains": [],
            "retrieval_query": latest,
        },
    )
    monkeypatch.setattr(agent_loop, "_is_casual_low_signal", lambda latest: True)

    async def empty_stream(*args, **kwargs):
        yield "data: [DONE]\n\n"

    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", empty_stream)
    chunks = _collect(agent_loop.stream_agent_loop(
        "https://selected.example/v1",
        "selected-model",
        [{"role": "user", "content": "hello"}],
        relevant_tools=set(),
        fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
        fallback_on_empty=False,
        _is_teacher_run=True,
    ))

    assert len(chunks) == 1
    payload = json.loads(chunks[0].split("data: ", 1)[1])
    assert payload["error"] == "Model returned an empty response"
    assert payload["fallback_eligible"] is False
    assert not any('"delta": "Hey."' in chunk for chunk in chunks)
    assert not any('"type": "metrics"' in chunk for chunk in chunks)


def test_reasoning_only_agent_error_emits_terminal_history(monkeypatch):
    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(agent_loop, "estimate_tokens", lambda *args, **kwargs: 10)

    async def fake_stream(candidates, messages, **kwargs):
        yield 'data: {"delta": "private reasoning partial", "thinking": true}\n\n'
        yield 'event: error\ndata: {"status": 504, "error": "provider detail"}\n\n'

    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)

    chunks = _collect(agent_loop.stream_agent_loop(
        "https://selected.example/v1",
        "selected-model",
        [{"role": "user", "content": "Investigate this failure."}],
        relevant_tools={"bash"},
        fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
        fallback_on_empty=False,
        _is_teacher_run=True,
    ))

    terminal = json.loads(next(
        chunk for chunk in chunks if '"type": "agent_terminal"' in chunk
    )[6:])["data"]
    assert terminal["thinking"] == "private reasoning partial"
    assert terminal["round_texts"] == [
        "[Agent stopped: Model request failed (HTTP 504)]"
    ]
    assert any(chunk.startswith("event: error") for chunk in chunks)
    assert "data: [DONE]\n\n" not in chunks


def test_toolless_multi_round_agent_persists_round_route_provenance(monkeypatch):
    calls = 0
    primary = ("https://selected.example/v1", "selected-model", {})
    backup = ("https://backup.example/v1", "backup-model", {})
    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(agent_loop, "estimate_tokens", lambda *args, **kwargs: 10)

    async def fake_stream(candidates, messages, **kwargs):
        nonlocal calls
        calls += 1
        if calls == 1:
            yield 'data: {"delta": "Let me check that now"}\n\n'
        else:
            yield 'data: {"type": "fallback", "selected_model": "selected-model", "answered_by": "backup-model", "candidate_index": 1, "selected_endpoint_id": "selected-ep", "selected_endpoint_label": "Selected endpoint", "selected_endpoint_cost_tracked": false, "answered_by_endpoint_id": "backup-ep", "answered_by_endpoint_label": "Backup endpoint", "answered_by_endpoint_cost_tracked": true}\n\n'
            yield 'data: {"delta": "final answer"}\n\n'
        yield "data: [DONE]\n\n"

    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)

    chunks = _collect(agent_loop.stream_agent_loop(
        primary[0],
        primary[1],
        [{"role": "user", "content": "Please investigate."}],
        headers=primary[2],
        max_rounds=3,
        relevant_tools=set(),
        fallbacks=[backup],
        route_descriptors=[
            {"endpoint_id": "selected-ep", "endpoint_label": "Selected endpoint", "endpoint_cost_tracked": False},
            {"endpoint_id": "backup-ep", "endpoint_label": "Backup endpoint", "endpoint_cost_tracked": True},
        ],
        fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
        fallback_on_empty=False,
        _is_teacher_run=True,
    ))

    metrics = json.loads(next(
        chunk for chunk in chunks if '"type": "metrics"' in chunk
    )[6:])["data"]
    assert metrics["round_texts"] == ["Let me check that now", "final answer"]
    assert metrics["round_models"] == ["selected-model", "backup-model"]
    assert metrics["round_endpoint_ids"] == ["selected-ep", "backup-ep"]
    assert metrics["endpoint_id"] == "backup-ep"
    assert metrics["requested_endpoint_id"] == "selected-ep"
    assert metrics["endpoint_cost_tracked"] is True
    assert "tool_events" not in metrics


def test_agent_metrics_attribute_usage_to_each_answering_route(monkeypatch):
    calls = 0
    primary = ("https://paid.example/v1", "selected-model", {})
    backup = ("http://localhost:11434/v1", "backup-model", {})
    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(agent_loop, "estimate_tokens", lambda *args, **kwargs: 10)
    monkeypatch.setattr(
        agent_loop,
        "_agent_route_tool_mode",
        lambda *args, **kwargs: (True, False, False),
    )

    async def fake_stream(candidates, messages, **kwargs):
        nonlocal calls
        calls += 1
        if calls == 1:
            yield 'data: {"type": "model_actual", "model": "selected-alias"}\n\n'
            yield 'data: {"type": "usage", "data": {"model": "selected-alias", "input_tokens": 100, "output_tokens": 10}}\n\n'
            tool_call = {
                "name": "bash",
                "arguments": json.dumps({"command": "printf one"}),
            }
            yield f'data: {json.dumps({"type": "tool_calls", "calls": [tool_call]})}\n\n'
        else:
            yield 'data: {"type": "fallback", "selected_model": "selected-model", "answered_by": "backup-model", "candidate_index": 1, "selected_endpoint_id": "paid", "selected_endpoint_label": "Paid", "selected_endpoint_cost_tracked": true, "answered_by_endpoint_id": "local", "answered_by_endpoint_label": "Local", "answered_by_endpoint_cost_tracked": false}\n\n'
            yield 'data: {"type": "model_actual", "model": "backup-alias"}\n\n'
            yield 'data: {"type": "usage", "data": {"model": "backup-alias", "input_tokens": 200, "output_tokens": 20}}\n\n'
            yield 'data: {"delta": "done"}\n\n'
        yield "data: [DONE]\n\n"

    async def fake_execute(block, *args, **kwargs):
        return "bash", {"output": "ok", "exit_code": 0}

    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)
    monkeypatch.setattr(agent_loop, "execute_tool_block", fake_execute)

    chunks = _collect(agent_loop.stream_agent_loop(
        primary[0],
        primary[1],
        [{"role": "user", "content": "Run one tool."}],
        headers=primary[2],
        max_rounds=3,
        relevant_tools={"bash"},
        fallbacks=[backup],
        route_descriptors=[
            {"endpoint_id": "paid", "endpoint_label": "Paid", "endpoint_cost_tracked": True},
            {"endpoint_id": "local", "endpoint_label": "Local", "endpoint_cost_tracked": False},
        ],
        fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
        fallback_on_empty=False,
        _is_teacher_run=True,
    ))

    metrics = json.loads(next(
        chunk for chunk in chunks if '"type": "metrics"' in chunk
    )[6:])["data"]
    assert metrics["input_tokens"] == 300
    assert metrics["output_tokens"] == 30
    assert metrics["usage_source"] == "real"
    assert metrics["usage_buckets"] == [
        {
            "round": 1,
            "model": "selected-alias",
            "endpoint_id": "paid",
            "endpoint_label": "Paid",
            "input_tokens": 100,
            "output_tokens": 10,
            "usage_source": "real",
            "endpoint_cost_tracked": True,
        },
        {
            "round": 2,
            "model": "backup-alias",
            "endpoint_id": "local",
            "endpoint_label": "Local",
            "input_tokens": 200,
            "output_tokens": 20,
            "usage_source": "real",
            "endpoint_cost_tracked": False,
        },
    ]


@pytest.mark.parametrize("malformed_input", [None, "bad", 10**1000])
def test_agent_round_ignores_malformed_usage_and_uses_estimate(
    monkeypatch,
    malformed_input,
):
    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(agent_loop, "estimate_tokens", lambda *args, **kwargs: 10)
    monkeypatch.setattr(
        agent_loop,
        "_agent_route_tool_mode",
        lambda *args, **kwargs: (True, False, False),
    )

    async def fake_stream(candidates, messages, **kwargs):
        usage_event = {
            "type": "usage",
            "data": {
                "input_tokens": malformed_input,
                "output_tokens": 1,
            },
        }
        yield "data: " + json.dumps(usage_event) + "\n\n"
        yield 'data: {"delta": "valid answer"}\n\n'
        yield "data: [DONE]\n\n"

    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)

    chunks = _collect(agent_loop.stream_agent_loop(
        "https://selected.example/v1",
        "selected-model",
        [{"role": "user", "content": "Run a detailed investigation."}],
        max_rounds=1,
        relevant_tools={"bash"},
        route_descriptors=[{
            "endpoint_id": "selected",
            "endpoint_label": "Selected",
            "endpoint_cost_tracked": True,
        }],
        fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
        fallback_on_empty=False,
        _is_teacher_run=True,
    ))

    metrics = json.loads(next(
        chunk for chunk in chunks if '"type": "metrics"' in chunk
    )[6:])["data"]
    assert metrics["usage_source"] == "estimated"
    assert metrics["input_tokens"] == 10
    assert metrics["usage_buckets"][0]["input_tokens"] == 10
    assert metrics["usage_buckets"][0]["output_tokens"] == len("valid answer") // 4


@pytest.mark.parametrize(
    ("synthesis_result", "expected_answer"),
    [
        ("Recovered final answer.", "Recovered final answer."),
        (
            "",
            "I gathered some search results but couldn't pull a clean answer together. "
            "Want me to try a more specific question, or summarize what I did find?",
        ),
    ],
)
def test_force_answer_recovery_persists_and_bills_pinned_fallback_route(
    monkeypatch,
    synthesis_result,
    expected_answer,
):
    primary = ("https://selected.example/v1", "selected-model", {})
    backup = (
        "https://backup.example/v1",
        "backup-model",
        {"Authorization": "Bearer backup"},
    )
    requests_by_round = []
    synthesis_calls = []

    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(agent_loop, "estimate_tokens", lambda *args, **kwargs: 10)
    monkeypatch.setattr(agent_loop, "blocked_tools_for_owner", lambda owner: set())
    monkeypatch.setattr(
        agent_loop,
        "_agent_route_tool_mode",
        lambda *args, **kwargs: (True, False, False),
    )

    async def fake_compact(
        session, url, model, messages, headers=None, owner=None,
        *, persist=True, compaction_state=None,
    ):
        return (list(messages), 4096, False)

    async def fake_stream(candidates, messages, **kwargs):
        round_index = len(requests_by_round)
        requests_by_round.append([(url, model) for url, model, _ in candidates])
        factory = kwargs["candidate_request_factory"]
        for index, candidate in enumerate(candidates):
            await factory(index, *candidate)
        if round_index == 0:
            fallback_event = {
                "type": "fallback",
                "selected_model": primary[1],
                "answered_by": backup[1],
                "candidate_index": 1,
                "selected_endpoint_id": "selected-ep",
                "selected_endpoint_label": "Selected",
                "selected_endpoint_cost_tracked": False,
                "answered_by_endpoint_id": "backup-ep",
                "answered_by_endpoint_label": "Backup",
                "answered_by_endpoint_cost_tracked": True,
            }
            yield "data: " + json.dumps(fallback_event) + "\n\n"
        tool_call = {
            "name": "bash",
            "arguments": json.dumps({"command": "printf repeated"}),
        }
        yield f'data: {json.dumps({"type": "tool_calls", "calls": [tool_call]})}\n\n'
        yield "data: [DONE]\n\n"

    async def fake_execute(block, *args, **kwargs):
        # The repeated-call recovery is the subject here, not provenance. Use
        # a content-free failure; model-visible shell errors correctly arm the
        # exact-approval gate.
        return "bash", {"exit_code": 1}

    async def fake_synthesis(**kwargs):
        synthesis_calls.append(kwargs)
        return synthesis_result

    monkeypatch.setattr(agent_loop, "maybe_compact", fake_compact)
    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)
    monkeypatch.setattr(agent_loop, "execute_tool_block", fake_execute)
    monkeypatch.setattr(llm_core, "llm_call_async", fake_synthesis)

    chunks = _collect(agent_loop.stream_agent_loop(
        primary[0],
        primary[1],
        [{"role": "user", "content": "Keep checking until you can answer."}],
        headers=primary[2],
        max_rounds=6,
        relevant_tools={"bash"},
        fallbacks=[backup],
        route_descriptors=[
            {
                "endpoint_id": "selected-ep",
                "endpoint_label": "Selected",
                "endpoint_cost_tracked": False,
            },
            {
                "endpoint_id": "backup-ep",
                "endpoint_label": "Backup",
                "endpoint_cost_tracked": True,
            },
        ],
        fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
        fallback_on_empty=False,
        _is_teacher_run=True,
    ))

    assert requests_by_round[0] == [
        (primary[0], primary[1]),
        (backup[0], backup[1]),
    ]
    assert requests_by_round[1:] == [[(backup[0], backup[1])]] * 5
    assert len(synthesis_calls) == 1
    assert synthesis_calls[0]["url"] == backup[0]
    assert synthesis_calls[0]["model"] == backup[1]
    assert synthesis_calls[0]["headers"] == backup[2]

    metrics = json.loads(next(
        chunk for chunk in chunks if '"type": "metrics"' in chunk
    )[6:])["data"]
    assert metrics["round_texts"][-1] == expected_answer
    assert metrics["round_models"][-1] == backup[1]
    assert metrics["round_endpoint_ids"][-1] == "backup-ep"
    assert metrics["usage_buckets"][-1] == {
        "round": 6,
        "model": backup[1],
        "endpoint_id": "backup-ep",
        "endpoint_label": "Backup",
        "input_tokens": 10,
        "output_tokens": len(synthesis_result) // 4,
        "usage_source": "estimated",
        "endpoint_cost_tracked": True,
    }
    assert len(metrics["usage_buckets"]) == 7


def test_agent_terminal_retains_completed_paid_fallback_usage(monkeypatch):
    calls = 0
    primary = ("http://localhost:11434/v1", "selected-model", {})
    backup = ("https://paid.example/v1", "backup-model", {})
    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(agent_loop, "estimate_tokens", lambda *args, **kwargs: 10)
    monkeypatch.setattr(
        agent_loop,
        "_agent_route_tool_mode",
        lambda *args, **kwargs: (True, False, False),
    )

    async def fake_stream(candidates, messages, **kwargs):
        nonlocal calls
        calls += 1
        if calls == 1:
            yield 'data: {"type": "fallback", "selected_model": "selected-model", "answered_by": "backup-model", "candidate_index": 1, "selected_endpoint_id": "local", "selected_endpoint_label": "Local", "selected_endpoint_cost_tracked": false, "answered_by_endpoint_id": "paid", "answered_by_endpoint_label": "Paid", "answered_by_endpoint_cost_tracked": true}\n\n'
            yield 'data: {"type": "usage", "data": {"model": "backup-model", "input_tokens": 125, "output_tokens": 25}}\n\n'
            tool_call = {
                "name": "bash",
                "arguments": json.dumps({"command": "printf one"}),
            }
            yield f'data: {json.dumps({"type": "tool_calls", "calls": [tool_call]})}\n\n'
            yield "data: [DONE]\n\n"
            return
        yield 'event: error\ndata: {"status": 400, "error": "unsupported model"}\n\n'

    async def fake_execute(block, *args, **kwargs):
        return "bash", {"output": "ok", "exit_code": 0}

    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)
    monkeypatch.setattr(agent_loop, "execute_tool_block", fake_execute)

    chunks = _collect(agent_loop.stream_agent_loop(
        primary[0],
        primary[1],
        [{"role": "user", "content": "Run one tool."}],
        headers=primary[2],
        max_rounds=3,
        relevant_tools={"bash"},
        fallbacks=[backup],
        route_descriptors=[
            {"endpoint_id": "local", "endpoint_label": "Local", "endpoint_cost_tracked": False},
            {"endpoint_id": "paid", "endpoint_label": "Paid", "endpoint_cost_tracked": True},
        ],
        fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
        fallback_on_empty=False,
        _is_teacher_run=True,
    ))

    terminal = json.loads(next(
        chunk for chunk in chunks if '"type": "agent_terminal"' in chunk
    )[6:])["data"]
    assert terminal["input_tokens"] == 125
    assert terminal["output_tokens"] == 25
    assert terminal["usage_source"] == "real"
    assert terminal["usage_buckets"] == [{
        "round": 1,
        "model": "backup-model",
        "endpoint_id": "paid",
        "endpoint_label": "Paid",
        "input_tokens": 125,
        "output_tokens": 25,
        "usage_source": "real",
        "endpoint_cost_tracked": True,
    }]
    assert not any('"type": "metrics"' in chunk for chunk in chunks)


def test_agent_builds_backup_prompt_and_tool_transport_before_attempt(monkeypatch):
    requests = []
    primary = ("https://selected.example/v1", "selected-model", {})
    backup = ("http://localhost:11434/api/chat", "backup-model", {})

    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(agent_loop, "estimate_tokens", lambda *args, **kwargs: 10)
    monkeypatch.setattr(agent_loop, "blocked_tools_for_owner", lambda owner: set())
    monkeypatch.setattr(
        agent_loop,
        "_agent_route_tool_mode",
        lambda url, model, owner=None, headers=None: (model == "selected-model", model == "backup-model", False),
    )

    def fake_build(messages, model, *args, **kwargs):
        return (
            list(messages) + [{
                "role": "system",
                "content": f"route prompt for {model}",
                "_agent_injected": "prompt",
            }],
            [],
        )

    monkeypatch.setattr(agent_loop, "_build_system_prompt", fake_build)

    async def fake_stream(candidates, messages, **kwargs):
        factory = kwargs["candidate_request_factory"]
        requests.extend([
            await factory(index, *candidate)
            for index, candidate in enumerate(candidates)
        ])
        yield f'data: {json.dumps({"type": "fallback", "selected_model": primary[1], "answered_by": backup[1], "candidate_index": 1})}\n\n'
        yield 'data: {"delta": "backup answer"}\n\n'
        yield "data: [DONE]\n\n"

    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)

    chunks = _collect(
        agent_loop.stream_agent_loop(
            primary[0],
            primary[1],
            [{"role": "user", "content": "Use bash if needed."}],
            headers=primary[2],
            max_rounds=1,
            relevant_tools={"bash"},
            fallbacks=[backup],
            fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
            fallback_on_empty=False,
            _is_teacher_run=True,
        )
    )

    assert requests[0]["kwargs"]["tools"]
    assert requests[1]["kwargs"]["tools"] is None
    backup_contents = [message.get("content") for message in requests[1]["messages"]]
    assert "route prompt for backup-model" in backup_contents
    assert "route prompt for selected-model" not in backup_contents
    assert any('"delta": "backup answer"' in chunk for chunk in chunks)


@pytest.mark.parametrize(
    ("primary_context", "backup_context", "expected_fallback_message_count"),
    [
        (1000, 100, 2),
        (100, 1000, 22),
    ],
)
def test_agent_fallback_request_uses_candidate_context_budget(
    monkeypatch,
    primary_context,
    backup_context,
    expected_fallback_message_count,
):
    requests_by_round = []
    context_lookups = []
    trim_budgets = []
    round_number = 0
    primary = ("https://selected.example/v1", "selected-model", {})
    backup = ("https://backup.example/v1", "backup-model", {})
    latest_user = "LATEST USER TURN MUST SURVIVE"
    history = [
        {"role": "user" if index % 2 == 0 else "assistant", "content": f"history-{index}"}
        for index in range(20)
    ] + [{"role": "user", "content": latest_user}]

    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(agent_loop, "estimate_tokens", lambda messages: len(messages) * 10)
    monkeypatch.setattr(agent_loop, "blocked_tools_for_owner", lambda owner: set())
    monkeypatch.setattr(
        agent_loop,
        "_agent_route_tool_mode",
        lambda *args, **kwargs: (True, False, False),
    )

    def fake_build(messages, model, *args, **kwargs):
        return ([{
            "role": "system",
            "content": f"route prompt for {model}",
            "_agent_injected": "prompt",
        }] + list(messages), [])

    monkeypatch.setattr(agent_loop, "_build_system_prompt", fake_build)

    import src.context_budget as context_budget
    import src.context_compactor as context_compactor
    import src.model_context as model_context

    def fake_context(candidate_url, candidate_model, fallback=0):
        context_lookups.append((candidate_url, candidate_model, fallback))
        return backup_context if candidate_model == "backup-model" else primary_context

    def fake_compute(soft_budget, candidate_context, explicit, hard_max=None):
        return candidate_context

    def fake_trim(messages, effective_budget, reserve_tokens=0):
        trim_budgets.append(effective_budget)
        if effective_budget != 100:
            return list(messages)
        route_prompt = next(
            message for message in messages
            if message.get("_agent_injected") == "prompt"
        )
        current_user = next(
            message for message in reversed(messages)
            if message.get("role") == "user"
        )
        return [route_prompt, current_user]

    monkeypatch.setattr(model_context, "budget_context_for_model", fake_context)
    monkeypatch.setattr(context_budget, "compute_input_token_budget", fake_compute)
    monkeypatch.setattr(context_budget, "budget_is_explicit", lambda value: False)
    monkeypatch.setattr(context_compactor, "trim_for_context", fake_trim)

    async def fake_stream(candidates, messages, **kwargs):
        nonlocal round_number
        round_number += 1
        factory = kwargs["candidate_request_factory"]
        requests = [
            await factory(index, *candidate)
            for index, candidate in enumerate(candidates)
        ]
        requests_by_round.append(requests)
        if round_number == 1:
            yield f'data: {json.dumps({"type": "fallback", "selected_model": primary[1], "answered_by": backup[1], "candidate_index": 1})}\n\n'
            tool_call = {
                "name": "bash",
                "arguments": json.dumps({"command": "printf one"}),
            }
            yield f'data: {json.dumps({"type": "tool_calls", "calls": [tool_call]})}\n\n'
        else:
            yield 'data: {"delta": "pinned backup answer"}\n\n'
        yield "data: [DONE]\n\n"

    async def fake_execute(block, *args, **kwargs):
        return "bash", {"output": "ok", "exit_code": 0}

    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)
    monkeypatch.setattr(agent_loop, "execute_tool_block", fake_execute)

    chunks = _collect(
        agent_loop.stream_agent_loop(
            primary[0],
            primary[1],
            history,
            headers=primary[2],
            max_rounds=2,
            relevant_tools={"bash"},
            fallbacks=[backup],
            fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
            fallback_on_empty=False,
            context_length=4096,
            _is_teacher_run=True,
        )
    )

    assert [(url, model) for url, model, _fallback in context_lookups] == [
        (primary[0], primary[1]),
        (backup[0], backup[1]),
        (backup[0], backup[1]),
    ]
    assert trim_budgets == [primary_context, backup_context, backup_context]
    fallback_messages = requests_by_round[0][1]["messages"]
    assert len(fallback_messages) == expected_fallback_message_count
    assert fallback_messages[0]["content"] == "route prompt for backup-model"
    assert any(
        message == {"role": "user", "content": latest_user}
        for message in fallback_messages
    )
    assert all("selected-model" not in str(message) for message in fallback_messages)
    pinned_messages = requests_by_round[1][0]["messages"]
    assert any(
        message == {"role": "user", "content": latest_user}
        for message in pinned_messages
    )
    assert pinned_messages[0]["content"] == "route prompt for backup-model"
    metrics = json.loads(next(
        chunk for chunk in chunks if '"type": "metrics"' in chunk
    )[6:])["data"]
    assert metrics["context_length"] == backup_context


def test_agent_persists_only_answering_route_compaction(monkeypatch):
    primary = ("https://selected.example/v1", "selected-model", {})
    backup = ("https://backup.example/v1", "backup-model", {})
    applied = []

    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(agent_loop, "estimate_tokens", lambda *args, **kwargs: 10)
    monkeypatch.setattr(agent_loop, "blocked_tools_for_owner", lambda owner: set())

    async def fake_compact(
        session, url, model, messages, headers=None, owner=None,
        *, persist=True, compaction_state=None,
    ):
        assert persist is False
        compaction_state.update({"route": model, "applied": False})
        return (list(messages), 1000, True)

    def fake_apply(session, state):
        if not state or state.get("applied"):
            return False
        state["applied"] = True
        applied.append(state["route"])
        return True

    async def fake_stream(candidates, messages, **kwargs):
        factory = kwargs["candidate_request_factory"]
        for index, candidate in enumerate(candidates):
            await factory(index, *candidate)
        yield f'data: {json.dumps({"type": "fallback", "selected_model": primary[1], "answered_by": backup[1], "candidate_index": 1})}\n\n'
        yield 'data: {"delta": "backup answer"}\n\n'
        yield "data: [DONE]\n\n"

    monkeypatch.setattr(agent_loop, "maybe_compact", fake_compact)
    monkeypatch.setattr(agent_loop, "apply_compaction_state", fake_apply)
    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)

    _collect(
        agent_loop.stream_agent_loop(
            primary[0],
            primary[1],
            [{"role": "user", "content": "Run a command after checking the route."}],
            headers=primary[2],
            history_session=object(),
            max_rounds=1,
            relevant_tools={"bash"},
            fallbacks=[backup],
            fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
            fallback_on_empty=False,
            _is_teacher_run=True,
        )
    )

    assert applied == ["backup-model"]


def test_agent_deferred_compaction_survives_duplicate_primary_fallback(monkeypatch):
    primary = ("https://selected.example/v1", "selected-model", {})
    compacted_routes = []

    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(agent_loop, "estimate_tokens", lambda *args, **kwargs: 10)
    monkeypatch.setattr(agent_loop, "blocked_tools_for_owner", lambda owner: set())

    async def fake_compact(
        session, url, model, messages, headers=None, owner=None,
        *, persist=True, compaction_state=None,
    ):
        assert persist is False
        compacted_routes.append((url, model))
        return (list(messages), 1000, False)

    async def fake_stream(candidates, messages, **kwargs):
        assert candidates == [primary]
        request = await kwargs["candidate_request_factory"](0, *primary)
        assert request["messages"]
        yield 'data: {"delta": "answer"}\n\n'
        yield "data: [DONE]\n\n"

    monkeypatch.setattr(agent_loop, "maybe_compact", fake_compact)
    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)

    chunks = _collect(agent_loop.stream_agent_loop(
        primary[0],
        primary[1],
        [{"role": "user", "content": "Investigate this."}],
        headers=primary[2],
        relevant_tools={"bash"},
        fallbacks=[primary],
        defer_context_shaping=True,
        max_rounds=1,
        fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
        fallback_on_empty=False,
        _is_teacher_run=True,
    ))

    assert compacted_routes == [(primary[0], primary[1])]
    assert any('"delta": "answer"' in chunk for chunk in chunks)


def test_skill_activation_reaches_later_fallback_request_and_pinned_round(monkeypatch):
    requests_by_round = []
    round_number = 0
    primary = ("https://selected.example/v1", "selected-model", {})
    backup = ("https://backup.example/v1", "odysseus-qwen-backup", {})

    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
    monkeypatch.setattr(agent_loop, "estimate_tokens", lambda *args, **kwargs: 10)
    monkeypatch.setattr(agent_loop, "blocked_tools_for_owner", lambda owner: set())
    monkeypatch.setattr(
        agent_loop,
        "_is_odysseus_qwen_model",
        lambda model: model == backup[1],
    )
    monkeypatch.setattr(
        agent_loop,
        "_agent_route_tool_mode",
        lambda url, model, owner=None, headers=None: (
            model == "selected-model",
            False,
            False,
        ),
    )

    def fake_build(messages, model, *args, **kwargs):
        route_tools = sorted(kwargs.get("relevant_tools") or [])
        return (
            list(messages) + [{
                "role": "system",
                "content": f"route={model}; tools={','.join(route_tools)}",
                "_agent_injected": "prompt",
            }],
            [],
        )

    monkeypatch.setattr(agent_loop, "_build_system_prompt", fake_build)

    import services.memory.skills as skills_module
    import src.tool_policy as tool_policy

    class FakeSkillsManager:
        def __init__(self, data_dir):
            pass

        def load(self, owner=None):
            return [{
                "name": "runtime-skill",
                "requires_toolsets": ["grep"],
            }]

        def get_relevant_skills(self, *args, **kwargs):
            return []

    monkeypatch.setattr(skills_module, "SkillsManager", FakeSkillsManager)
    monkeypatch.setattr(tool_policy, "known_tool_names", lambda: {"manage_skills", "grep"})

    async def fake_stream(candidates, messages, **kwargs):
        nonlocal round_number
        round_number += 1
        factory = kwargs["candidate_request_factory"]
        requests = [
            await factory(index, *candidate)
            for index, candidate in enumerate(candidates)
        ]
        requests_by_round.append((list(candidates), requests))

        if round_number == 1:
            call = {
                "name": "manage_skills",
                "arguments": json.dumps({"action": "view", "name": "runtime-skill"}),
            }
            yield f'data: {json.dumps({"type": "tool_calls", "calls": [call]})}\n\n'
        elif round_number == 2:
            yield f'data: {json.dumps({"type": "fallback", "selected_model": primary[1], "answered_by": backup[1], "candidate_index": 1})}\n\n'
            call = {
                "name": "grep",
                "arguments": json.dumps({"pattern": "needle", "path": "."}),
            }
            yield f'data: {json.dumps({"type": "tool_calls", "calls": [call]})}\n\n'
        else:
            yield 'data: {"delta": "pinned backup answer"}\n\n'
        yield "data: [DONE]\n\n"

    async def fake_execute(block, *args, **kwargs):
        return block.tool_type, {"output": "ok", "exit_code": 0}

    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)
    monkeypatch.setattr(agent_loop, "execute_tool_block", fake_execute)

    chunks = _collect(
        agent_loop.stream_agent_loop(
            primary[0],
            primary[1],
            [{"role": "user", "content": "Load runtime-skill, then use it."}],
            headers=primary[2],
            max_rounds=3,
            relevant_tools={"manage_skills"},
            fallbacks=[backup],
            fallback_statuses=FOREGROUND_AVAILABILITY_STATUSES,
            fallback_on_empty=False,
            _is_teacher_run=True,
        )
    )

    round_two_candidates, round_two_requests = requests_by_round[1]
    assert round_two_candidates == [primary, backup]
    primary_schema_names = {
        schema["function"]["name"]
        for schema in round_two_requests[0]["kwargs"]["tools"]
    }
    assert "grep" in primary_schema_names
    assert round_two_requests[1]["kwargs"]["tools"] is None
    assert any(
        "route=odysseus-qwen-backup; tools=grep,manage_skills" in (message.get("content") or "")
        for message in round_two_requests[1]["messages"]
    )

    round_three_candidates, round_three_requests = requests_by_round[2]
    assert round_three_candidates == [backup]
    assert any(
        "route=odysseus-qwen-backup; tools=grep,manage_skills" in (message.get("content") or "")
        for message in round_three_requests[0]["messages"]
    )
    assert any('"delta": "pinned backup answer"' in chunk for chunk in chunks)