"""GET /sessions/{id}/info — the session-info dialog's single source.
The model identity is not on the session row: it lives in the route engine,
which knows the active entry (model, target, host, window) and the health of
each entry behind it. These tests pin the join, and the window precedence the
session head now shares with the agent.
"""
from datetime import datetime, timezone
import pytest
from fastapi import HTTPException
import navi.api.routes.sessions as sessions_mod
from navi.auth import User
from navi.core.registry import ProfileRegistry
from navi.core.session import Session
from navi.llm.config import LLMConfig, Route, RouteEntry, Target
from navi.profiles.base import AgentProfile
_FAKE_USER = User(
id="test-user", email="test@example.com", display_name="Test", role="admin", permissions=[]
)
class _Health:
def __init__(self, targets=None, missing=None):
self._targets = targets or {"cloud": {"state": "ok"}, "home": {"state": "ok"}}
self._missing = missing or []
def snapshot(self):
return {"targets": self._targets, "missing_models": self._missing}
class _Backend:
"""Enough of FallbackOllamaBackend for the endpoint."""
def __init__(self, config, active_route="default", health=None):
self.config = config
self.health = health or _Health()
self._active_route = active_route
self._active = {
"default": {"model": "glm-5.3-flash:cloud", "target": "cloud",
"primary": True, "primary_model": "glm-5.3-flash:cloud"}
}
def active_report(self):
return self._active if self._active_route else {}
def active_window(self, route_id):
route = self.config.route(route_id)
return route.entries[0].context_window if route is not None else None
class _Registry:
def __init__(self, backend):
self._backend = backend
def get(self, key):
if self._backend is None:
raise KeyError(key)
return self._backend
class _Store:
def __init__(self, session):
self.session = session
async def get_meta(self, session_id):
return self.session if session_id == self.session.id else None
async def get(self, session_id):
return await self.get_meta(session_id)
class _Orchestrator:
def __init__(self, running=False, clients=1):
self._running = running
self._clients = clients
def is_running(self, session_id):
return self._running
def client_count(self, session_id):
return self._clients
def _config(primary_window=131072, backup_window=49152):
cloud = Target(name="cloud", host="https://ollama.com", api_key="x")
home = Target(name="home", host="http://192.168.1.129:11434")
return LLMConfig(
targets={"cloud": cloud, "home": home},
routes={
"default": Route(
id="default",
entries=(
RouteEntry(model="glm-5.3-flash:cloud", target=cloud,
context_window=primary_window),
RouteEntry(model="qwen3.6:35b", target=home, context_window=backup_window),
),
)
},
)
def _profile(**kw):
base = dict(
id="server_admin",
name="Server Administrator",
description="",
system_prompt="",
llm_route="default",
model=["glm-5.3-flash:cloud"],
)
base.update(kw)
return AgentProfile(**base)
def _session(profile_id="server_admin"):
return Session(
id="s-1",
profile_id=profile_id,
context_token_count=51013,
db_message_count=3049,
archive_threshold=2050,
created_at=datetime(2026, 10, 28, tzinfo=timezone.utc),
last_active=datetime(2026, 11, 2, tzinfo=timezone.utc),
)
def _registry(profile):
reg = ProfileRegistry()
reg.register(profile)
return reg
async def _call(profile=None, backend="default", session=None, orchestrator=None):
profile = profile if profile is not None else _profile()
return await sessions_mod.get_session_info(
session.id if session is not None else "s-1",
store=_Store(session or _session()),
profiles=_registry(profile),
backends=_Registry(_Backend(_config()) if backend == "default" else backend),
orchestrator=orchestrator or _Orchestrator(),
user=_FAKE_USER,
)
class TestSessionInfo:
async def test_reports_the_active_entry_with_its_host_and_window(self):
info = await _call()
assert info["llm"]["configured"] is True
assert info["llm"]["route"] == "default"
active = info["llm"]["active"]
assert active["model"] == "glm-5.3-flash:cloud"
assert active["target"] == "cloud"
assert active["host"] == "https://ollama.com"
assert active["primary"] is True
assert active["context_window"] == 131072
assert info["max_context_tokens"] == 131072
async def test_the_chain_carries_every_entry_and_its_health(self):
backend = _Backend(
_config(),
health=_Health(targets={
"cloud": {"state": "ok"},
"home": {"state": "degraded", "reason": "connect failed"},
}),
)
info = await _call(backend=backend)
chain = info["llm"]["chain"]
assert [e["model"] for e in chain] == ["glm-5.3-flash:cloud", "qwen3.6:35b"]
assert [e["primary"] for e in chain] == [True, False]
assert chain[1]["state"] == "degraded"
assert chain[1]["reason"] == "connect failed"
async def test_a_missing_model_is_reported_rather_than_a_green_target(self):
backend = _Backend(
_config(),
health=_Health(
missing=[{"target": "home", "model": "qwen3.6:35b", "detail": "404"}]
),
)
info = await _call(backend=backend)
assert info["llm"]["chain"][1]["state"] == "missing"
async def test_session_fields_and_compression_defaults_are_included(self):
info = await _call()
assert info["session_id"] == "s-1"
assert info["profile_name"] == "Server Administrator"
assert info["message_count"] == 3049
assert info["context_token_count"] == 51013
assert info["archive_threshold"] == 2050
assert info["compression"]["enabled"] is True
assert info["compression"]["keep_recent"] == 4
async def test_a_profile_override_replaces_the_compression_default(self):
info = await _call(profile=_profile(compression_keep_recent=3,
compression_max_tokens=1200))
assert info["compression"]["keep_recent"] == 3
assert info["compression"]["summary_max_tokens"] == 1200
async def test_live_state_comes_from_the_orchestrator(self):
info = await _call(orchestrator=_Orchestrator(running=True, clients=2))
assert info["live"] == {"running": True, "clients": 2}
async def test_without_a_route_the_profile_models_are_all_there_is(self):
info = await _call(profile=_profile(llm_route=None), backend=None)
assert info["llm"]["configured"] is False
assert info["llm"]["active"] is None
assert info["llm"]["chain"] == []
assert info["llm"]["models"] == ["glm-5.3-flash:cloud"]
async def test_a_deleted_profile_does_not_break_the_dialog(self):
info = await _call(profile=_profile(llm_route=None), backend=None,
session=_session(profile_id="gone"))
assert info["profile_id"] == "gone"
assert info["profile_name"] is None
assert info["llm"]["configured"] is False
async def test_an_unknown_session_is_a_404(self):
with pytest.raises(HTTPException) as exc:
await sessions_mod.get_session_info(
"nope",
store=_Store(_session()),
profiles=_registry(_profile()),
backends=_Registry(_Backend(_config())),
orchestrator=_Orchestrator(),
user=_FAKE_USER,
)
assert exc.value.status_code == 404
class TestEffectiveWindow:
async def test_meta_reports_the_routes_window_not_the_global_default(self):
meta = await sessions_mod.get_session_meta(
"s-1",
store=_Store(_session()),
profiles=_registry(_profile()),
backends=_Registry(_Backend(_config())),
user=_FAKE_USER,
)
assert meta["max_context_tokens"] == 131072
async def test_the_profile_window_is_used_when_the_backend_has_no_routes(self):
meta = await sessions_mod.get_session_meta(
"s-1",
store=_Store(_session()),
profiles=_registry(_profile(context_window=32000, llm_route=None)),
backends=_Registry(None),
user=_FAKE_USER,
)
assert meta["max_context_tokens"] == 32000
async def test_the_global_default_is_the_last_resort(self):
from navi.config import settings
meta = await sessions_mod.get_session_meta(
"s-1",
store=_Store(_session()),
profiles=_registry(_profile(llm_route=None)),
backends=_Registry(None),
user=_FAKE_USER,
)
assert meta["max_context_tokens"] == settings.ollama_num_ctx
