feat: AI server status in the sidebar

backend/app/api/ai_status.py

@@ -0,0 +1,123 @@
+import logging
+import time
+from typing import Annotated, Literal
+
+import httpx
+from fastapi import APIRouter, Depends
+from pydantic import BaseModel
+
+from app.config import get_settings
+from app.database import DbSession
+from app.models.user import User
+from app.services.user_ai import ai_service_for
+from app.utils.auth import get_current_user
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/ai", tags=["AI"])
+
+# The sidebar asks every minute from every open tab; one probe per window is plenty. Keyed by
+# user, because each user can point the app at their own models.
+CACHE_SECONDS = 20
+_cache: dict[str, tuple[float, "AIStatusResponse"]] = {}
+
+
+class AIStatusResponse(BaseModel):
+    status: Literal["online", "model_missing", "offline", "disabled"]
+    vision_model: str | None = None
+    text_model: str | None = None
+    missing_models: list[str] = []
+    # Ollama only: models in memory right now (GET /api/ps). A configured model that is not
+    # loaded is fine: Ollama loads it on the first request, which is just slower.
+    url: str | None = None
+    loaded_models: list[str] = []
+    model_loaded: bool | None = None
+
+
+def _matches(model: str, available: list[str]) -> bool:
+    """A configured model setting is there when any of its models is: the setting can be a
+    comma-separated rotation, and Ollama lists "name:tag" (no tag means ":latest")."""
+    for choice in (part.strip() for part in model.split(",")):
+        if not choice:
+            continue
+        candidates = {choice, choice if ":" in choice else f"{choice}:latest"}
+        if any(name in candidates for name in available):
+            return True
+    return False
+
+
+def _wanted(vision_model: str | None, text_model: str | None) -> list[str]:
+    """The models in use: a capability that is switched off cannot be missing."""
+    settings = get_settings()
+    wanted = []
+    if vision_model and settings.effective_ai_vision_enabled:
+        wanted.append(vision_model)
+    if text_model and settings.effective_ai_text_enabled:
+        wanted.append(text_model)
+    return wanted
+
+
+def summarize(raw: dict) -> AIStatusResponse:
+    endpoints = raw.get("endpoints", [])
+    healthy = [ep for ep in endpoints if ep.get("status") == "healthy"]
+    if not healthy:
+        first = endpoints[0] if endpoints else {}
+        return AIStatusResponse(
+            status="offline",
+            vision_model=first.get("vision_model"),
+            text_model=first.get("text_model"),
+        )
+    endpoint = healthy[0]
+    available = endpoint.get("available_models") or []
+    wanted = _wanted(endpoint.get("vision_model"), endpoint.get("text_model"))
+    missing = sorted({m for m in wanted if not _matches(m, available)})
+    return AIStatusResponse(
+        status="model_missing" if missing else "online",
+        vision_model=endpoint.get("vision_model"),
+        text_model=endpoint.get("text_model"),
+        missing_models=missing,
+        url=endpoint.get("url"),
+    )
+
+
+async def loaded_models(base_url: str) -> list[str] | None:
+    """Models an Ollama server holds in memory, or None when the server is not Ollama (or
+    does not answer /api/ps). Read only: asking never loads anything."""
+    root = base_url.rstrip("/").removesuffix("/v1")
+    try:
+        async with httpx.AsyncClient(timeout=3) as client:
+            response = await client.get(f"{root}/api/ps")
+        if response.status_code != 200:
+            return None
+        return [m.get("name") or m.get("model", "") for m in response.json().get("models", [])]
+    except (httpx.HTTPError, ValueError) as e:
+        logger.debug(f"Could not read loaded models from {root}: {e}")
+        return None
+
+
+async def with_loaded_models(result: AIStatusResponse) -> AIStatusResponse:
+    if result.status != "online" or not result.url:
+        return result
+    loaded = await loaded_models(result.url)
+    if loaded is None:
+        return result
+    wanted = _wanted(result.vision_model, result.text_model)
+    result.loaded_models = loaded
+    result.model_loaded = bool(wanted) and all(_matches(m, loaded) for m in wanted)
+    return result
+
+
+@router.get("/status", response_model=AIStatusResponse)
+async def ai_status(
+    db: DbSession, current_user: Annotated[User, Depends(get_current_user)]
+) -> AIStatusResponse:
+    if not get_settings().ai_enabled:
+        return AIStatusResponse(status="disabled")
+    key = str(current_user.id)
+    cached = _cache.get(key)
+    if cached and time.monotonic() - cached[0] < CACHE_SECONDS:
+        return cached[1]
+    ai = await ai_service_for(db, current_user.id)
+    result = await with_loaded_models(summarize(await ai.check_health()))
+    _cache[key] = (time.monotonic(), result)
+    return result

backend/app/api/router.py

@@ -2,6 +2,7 @@
 
 from fastapi import APIRouter
 
+from app.api.ai_status import router as ai_status_router
 from app.api.analytics import router as analytics_router
 from app.api.auth import router as auth_router
 from app.api.families import router as families_router
@@ -32,3 +33,4 @@
 api_router.include_router(notifications_router, prefix="/notifications", tags=["notifications"])
 api_router.include_router(analytics_router)
 api_router.include_router(learning_router)
+api_router.include_router(ai_status_router)

backend/app/services/user_ai.py

@@ -0,0 +1,33 @@
+"""The AI service as the user configured it.
+
+Settings > AI lets each user pick their own endpoints and models (stored in
+UserPreference.ai_endpoints). Tagging, outfit suggestions and pairings already honour them;
+every feature that talks to the model should, so it goes through here. The server's
+AI_BASE_URL / AI_*_MODEL stay as the fallback after the user's endpoints, as in AIService.
+"""
+
+from uuid import UUID
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.preference import UserPreference
+from app.services.ai_service import AIService, get_ai_service
+
+
+async def user_ai_endpoints(db: AsyncSession, user_id: UUID) -> list[dict]:
+    endpoints = await db.scalar(
+        select(UserPreference.ai_endpoints).where(UserPreference.user_id == user_id)
+    )
+    return [ep for ep in endpoints or [] if ep.get("enabled", True)]
+
+
+async def ai_service_for(db: AsyncSession, user_id: UUID) -> AIService:
+    """AIService using the user's own endpoints first; the shared default when they set none.
+
+    Raises AIDisabledError when internal AI is off, like get_ai_service.
+    """
+    endpoints = await user_ai_endpoints(db, user_id)
+    if not endpoints:
+        return get_ai_service()
+    return AIService(endpoints=endpoints)

backend/tests/test_ai_status.py

@@ -0,0 +1,128 @@
+from io import BytesIO
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from httpx import AsyncClient
+from PIL import Image
+
+from app.api import ai_status
+from app.config import get_settings
+
+
+def _jpeg() -> bytes:
+    buffer = BytesIO()
+    Image.new("RGB", (400, 600), (90, 90, 90)).save(buffer, format="JPEG")
+    return buffer.getvalue()
+
+
+class TestAIStatus:
+    def test_summarize_reports_a_missing_model(self):
+        raw = {
+            "endpoints": [
+                {
+                    "status": "healthy",
+                    "vision_model": "Ornith-1.5-9B",
+                    "text_model": "hf.co/ornith-ai/Ornith-1.5-9B-GGUF:Q5_K_M",
+                    "available_models": ["hf.co/ornith-ai/Ornith-1.5-9B-GGUF:Q5_K_M"],
+                }
+            ]
+        }
+        result = ai_status.summarize(raw)
+        assert result.status == "model_missing"
+        assert result.missing_models == ["Ornith-1.5-9B"]
+
+    def test_summarize_online_and_offline(self):
+        online = {
+            "endpoints": [
+                {
+                    "status": "healthy",
+                    "vision_model": "gemma3",
+                    "text_model": "gemma3",
+                    "available_models": ["gemma3:latest"],
+                }
+            ]
+        }
+        assert ai_status.summarize(online).status == "online"
+        offline = {"endpoints": [{"status": "unhealthy", "vision_model": "gemma3"}]}
+        assert ai_status.summarize(offline).status == "offline"
+
+    @pytest.mark.asyncio
+    async def test_endpoint_needs_sign_in_and_answers(self, client: AsyncClient, auth_headers):
+        assert (await client.get("/api/v1/ai/status")).status_code == 401
+        ai_status._cache.clear()
+        health = {
+            "status": "unhealthy",
+            "endpoints": [{"status": "unhealthy", "vision_model": "m", "text_model": "m"}],
+        }
+        check = AsyncMock(return_value=health)
+        with (
+            patch.object(type(get_settings()), "ai_enabled", property(lambda self: True)),
+            patch("app.services.ai_service.AIService.check_health", check),
+        ):
+            response = await client.get("/api/v1/ai/status", headers=auth_headers)
+        ai_status._cache.clear()
+        assert response.status_code == 200
+        assert response.json()["status"] == "offline"
+        check.assert_awaited_once()
+
+    def test_a_rotation_is_there_when_any_of_its_models_is(self):
+        health = {
+            "endpoints": [
+                {
+                    "status": "healthy",
+                    "vision_model": "llava:7b, gemma3",
+                    "text_model": "gemma3",
+                    "available_models": ["gemma3:latest"],
+                }
+            ]
+        }
+        assert ai_status.summarize(health).status == "online"
+
+    def test_a_switched_off_capability_cannot_be_missing(self):
+        health = {
+            "endpoints": [
+                {
+                    "status": "healthy",
+                    "vision_model": "not-installed",
+                    "text_model": "gemma3",
+                    "available_models": ["gemma3:latest"],
+                }
+            ]
+        }
+        settings = type(get_settings())
+        with (
+            patch.object(settings, "effective_ai_vision_enabled", property(lambda self: False)),
+            patch.object(settings, "effective_ai_text_enabled", property(lambda self: True)),
+        ):
+            result = ai_status.summarize(health)
+        assert result.status == "online"
+        assert result.missing_models == []
+
+
+class TestLoadedModels:
+    def _online(self):
+        return ai_status.AIStatusResponse(
+            status="online",
+            vision_model="qwen2.5vl:7b",
+            text_model="qwen2.5vl:7b",
+            url="http://ollama:11434/v1",
+        )
+
+    @pytest.mark.asyncio
+    async def test_reports_whether_the_model_is_in_memory(self):
+        with patch("app.api.ai_status.loaded_models", AsyncMock(return_value=["qwen2.5vl:7b"])):
+            assert (await ai_status.with_loaded_models(self._online())).model_loaded is True
+        with patch("app.api.ai_status.loaded_models", AsyncMock(return_value=[])):
+            result = await ai_status.with_loaded_models(self._online())
+        # Reachable with the model installed but not loaded: still online, loads on first use.
+        assert result.status == "online"
+        assert result.model_loaded is False
+
+    @pytest.mark.asyncio
+    async def test_a_server_without_api_ps_leaves_it_unknown(self):
+        with patch("app.api.ai_status.loaded_models", AsyncMock(return_value=None)):
+            assert (await ai_status.with_loaded_models(self._online())).model_loaded is None
+
+    @pytest.mark.asyncio
+    async def test_an_unreachable_server_is_unknown_not_an_error(self):
+        assert await ai_status.loaded_models("http://127.0.0.1:9/v1") is None

backend/tests/test_user_ai.py

@@ -0,0 +1,41 @@
+"""Features that call the model use the endpoints and models the user picked in Settings."""
+
+import pytest
+
+from app.models.preference import UserPreference
+from app.services.ai_service import get_ai_service
+from app.services.user_ai import ai_service_for
+
+
+class TestUserModels:
+    @pytest.mark.asyncio
+    async def test_the_users_own_endpoint_comes_first(self, db_session, test_user):
+        db_session.add(
+            UserPreference(
+                user_id=test_user.id,
+                ai_endpoints=[
+                    {
+                        "name": "mine",
+                        "url": "http://ollama.local:11434/v1",
+                        "vision_model": "qwen2.5vl:7b",
+                        "text_model": "qwen3:8b",
+                        "enabled": True,
+                    },
+                    {"name": "off", "url": "http://off/v1", "enabled": False},
+                ],
+            )
+        )
+        await db_session.commit()
+
+        ai = await ai_service_for(db_session, test_user.id)
+        first = ai._endpoints[0]
+        assert (first.url, first.vision_model, first.text_model) == (
+            "http://ollama.local:11434/v1",
+            "qwen2.5vl:7b",
+            "qwen3:8b",
+        )
+        assert all(ep.name != "off" for ep in ai._endpoints)
+
+    @pytest.mark.asyncio
+    async def test_without_own_endpoints_the_server_default_is_used(self, db_session, test_user):
+        assert await ai_service_for(db_session, test_user.id) is get_ai_service()

frontend/components/ai-status-dot.tsx

@@ -0,0 +1,39 @@
+'use client';
+
+import { useTranslations } from 'next-intl';
+import { useAIStatus } from '@/lib/hooks/use-ai-status';
+import { cn } from '@/lib/utils';
+
+const DOT: Record<string, string> = {
+  online: 'bg-green-500',
+  model_missing: 'bg-amber-500',
+  offline: 'bg-red-500',
+  disabled: 'bg-muted-foreground',
+  unknown: 'bg-muted-foreground',
+};
+
+/** Green when the AI server answers with the configured models, amber when a model is missing,
+ * red when the server does not answer. */
+export function AIStatusDot({ className }: { className?: string }) {
+  const t = useTranslations('nav.aiStatus');
+  const { data, isError } = useAIStatus();
+  const state = isError ? 'unknown' : (data?.status ?? 'unknown');
+  const detail =
+    state === 'model_missing' && data?.missing_models.length
+      ? t('missingDetail', { models: data.missing_models.join(', ') })
+      : data?.vision_model
+        ? t('modelDetail', { model: data.vision_model })
+        : t('unknownDetail');
+
+  return (
+    <div className={cn('flex items-center gap-2 text-xs text-muted-foreground', className)} title={detail}>
+      <span className="relative flex h-2.5 w-2.5">
+        {state === 'online' && (
+          <span className="absolute inline-flex h-full w-full animate-ping rounded-full bg-green-500 opacity-60" />
+        )}
+        <span className={cn('relative inline-flex h-2.5 w-2.5 rounded-full', DOT[state])} />
+      </span>
+      <span>{t(state)}</span>
+    </div>
+  );
+}

frontend/components/mobile-sidebar.tsx

@@ -6,6 +6,7 @@ import { usePathname } from 'next/navigation';
 import { X, Home, Shirt, Sparkles, Layers, LayoutGrid, History, BarChart3, Brain, Settings, Users, Bell, HeartHandshake } from 'lucide-react';
 import { cn } from '@/lib/utils';
 import { useTranslations } from 'next-intl';
+import { AIStatusDot } from '@/components/ai-status-dot';
 
 interface MobileSidebarProps {
   open: boolean;
@@ -145,6 +146,7 @@ export function MobileSidebar({ open, onClose }: MobileSidebarProps) {
               </li>
             </ul>
           </nav>
+          <AIStatusDot className="pt-4" />
         </div>
       </div>
     </div>