feat(studio): add dashboard studio routes and catalog

aisbf/routes/dashboard/providers.py

@@ -7,6 +7,7 @@ from datetime import datetime, timedelta
 from aisbf.database import DatabaseRegistry
 from aisbf.database import _hash_password as _db_hash_password
 from aisbf import __version__
+from aisbf.studio import build_studio_catalog
 from aisbf.app.templates import url_for, get_base_url
 from aisbf.app.startup import _reload_global_config, _apply_condense_defaults_provider, _apply_condense_defaults_rotation, _providers_json_path, _rotations_json_path, _autoselect_json_path, _claude_cli_mode
 from aisbf.app.middleware import _is_local_client
@@ -255,6 +256,26 @@ async def dashboard_studio(request: Request):
     )
 
 
+@router.get("/dashboard/studio/catalog")
+async def dashboard_studio_catalog(request: Request):
+    """Return Studio catalog for the current dashboard principal."""
+    auth_check = require_dashboard_auth(request)
+    if auth_check:
+        return JSONResponse({"entries": [], "error": "unauthorized"}, status_code=401)
+
+    current_user_id = request.session.get("user_id")
+    scope = "admin" if current_user_id is None else "user"
+    db = None if scope == "admin" else DatabaseRegistry.get_config_database()
+
+    catalog = build_studio_catalog(
+        scope=scope,
+        owner_id=current_user_id,
+        config=_config,
+        db=db,
+    )
+    return JSONResponse(catalog)
+
+
 async def _auto_detect_provider_models(provider_key: str, provider: dict) -> list:
     """
     Auto-detect models from a provider's API endpoint.

aisbf/studio.py

 from __future__ import annotations
 
 from dataclasses import dataclass
+from pathlib import Path
+import json
 from typing import Any, Dict, Iterable, List, Optional
 
 
@@ -211,3 +213,184 @@ def build_catalog_entry(
         "availability_reason": availability_reason,
         "metadata": metadata or {},
     }
+
+
+def _coerce_model_dict(model: Any) -> Dict[str, Any]:
+    if isinstance(model, dict):
+        return model
+
+    data: Dict[str, Any] = {}
+    for key in (
+        "name",
+        "id",
+        "description",
+        "capabilities",
+        "context_length",
+        "architecture",
+        "pricing",
+        "supported_parameters",
+        "default_parameters",
+    ):
+        if hasattr(model, key):
+            data[key] = getattr(model, key)
+    return data
+
+
+def _provider_models_from_config(provider_config: Any) -> List[Dict[str, Any]]:
+    models = getattr(provider_config, "models", None)
+    if models is None and isinstance(provider_config, dict):
+        models = provider_config.get("models")
+    return [_coerce_model_dict(model) for model in (models or [])]
+
+
+def _load_global_providers_from_disk() -> Dict[str, Dict[str, Any]]:
+    config_path = Path.home() / ".aisbf" / "providers.json"
+    if not config_path.exists():
+        return {}
+
+    with open(config_path) as handle:
+        payload = json.load(handle)
+
+    providers = payload.get("providers") if isinstance(payload, dict) else None
+    if isinstance(providers, dict):
+        return providers
+    if isinstance(payload, dict):
+        return {key: value for key, value in payload.items() if key != "condensation"}
+    return {}
+
+
+def _build_provider_entries(scope: str, owner_id: Optional[int], providers: Dict[str, Any]) -> List[Dict[str, Any]]:
+    entries: List[Dict[str, Any]] = []
+    for provider_id, provider_config in providers.items():
+        provider_type = getattr(provider_config, "type", None)
+        if provider_type is None and isinstance(provider_config, dict):
+            provider_type = provider_config.get("type", "openai")
+        provider_type = provider_type or "openai"
+
+        for model in _provider_models_from_config(provider_config):
+            target_id = model.get("name") or model.get("id")
+            if not target_id:
+                continue
+
+            capability_result = infer_model_capabilities(
+                model_name=target_id,
+                provider_type=provider_type,
+                explicit_capabilities=model.get("capabilities"),
+                architecture=model.get("architecture"),
+                provider_metadata=model,
+            )
+            metadata = {
+                "provider_type": provider_type,
+            }
+            if model.get("context_length") is not None:
+                metadata["context_length"] = model.get("context_length")
+            if model.get("architecture") is not None:
+                metadata["architecture"] = model.get("architecture")
+            if capability_result.source:
+                metadata["capability_source"] = capability_result.source
+            if capability_result.notes:
+                metadata["capability_notes"] = capability_result.notes
+
+            entries.append(
+                build_catalog_entry(
+                    scope=scope,
+                    owner_id=owner_id,
+                    kind="provider_model",
+                    source_id=provider_id,
+                    target_id=target_id,
+                    label=model.get("name") or target_id,
+                    description=model.get("description"),
+                    capabilities=capability_result.capabilities,
+                    availability_state="ready",
+                    availability_reason=None,
+                    metadata=metadata,
+                )
+            )
+    return entries
+
+
+def _build_rotation_entries(scope: str, owner_id: Optional[int], rotations: Dict[str, Any]) -> List[Dict[str, Any]]:
+    entries: List[Dict[str, Any]] = []
+    for rotation_id, rotation_config in rotations.items():
+        config_data = rotation_config if isinstance(rotation_config, dict) else rotation_config.model_dump()
+        entries.append(
+            {
+                "id": f"rotation/{rotation_id}",
+                "kind": "rotation",
+                "owner_scope": scope,
+                "owner_id": owner_id,
+                "source_id": rotation_id,
+                "target_id": rotation_id,
+                "label": config_data.get("model_name") or rotation_id,
+                "description": config_data.get("description"),
+                "capabilities": normalize_capabilities(config_data.get("capabilities")),
+                "availability_state": "ready",
+                "availability_reason": None,
+                "metadata": {
+                    "provider_count": len(config_data.get("providers") or []),
+                    "context_length": config_data.get("context_length"),
+                },
+            }
+        )
+    return entries
+
+
+def _build_autoselect_entries(scope: str, owner_id: Optional[int], autoselects: Dict[str, Any]) -> List[Dict[str, Any]]:
+    entries: List[Dict[str, Any]] = []
+    for autoselect_id, autoselect_config in autoselects.items():
+        config_data = autoselect_config if isinstance(autoselect_config, dict) else autoselect_config.model_dump()
+        available_models = config_data.get("available_models") or []
+        entries.append(
+            {
+                "id": f"autoselect/{autoselect_id}",
+                "kind": "autoselect",
+                "owner_scope": scope,
+                "owner_id": owner_id,
+                "source_id": autoselect_id,
+                "target_id": autoselect_id,
+                "label": config_data.get("model_name") or autoselect_id,
+                "description": config_data.get("description"),
+                "capabilities": normalize_capabilities(config_data.get("capabilities")),
+                "availability_state": "ready",
+                "availability_reason": None,
+                "metadata": {
+                    "available_model_count": len(available_models),
+                    "fallback": config_data.get("fallback"),
+                    "selection_model": config_data.get("selection_model"),
+                },
+            }
+        )
+    return entries
+
+
+def build_studio_catalog(
+    *,
+    scope: str,
+    owner_id: Optional[int],
+    config: Any = None,
+    db: Any = None,
+) -> Dict[str, Any]:
+    if scope == "user":
+        provider_rows = db.get_user_providers(owner_id) if db and owner_id is not None else []
+        rotation_rows = db.get_user_rotations(owner_id) if db and owner_id is not None else []
+        autoselect_rows = db.get_user_autoselects(owner_id) if db and owner_id is not None else []
+        providers = {row["provider_id"]: row.get("config", {}) for row in provider_rows}
+        rotations = {row["rotation_id"]: row.get("config", {}) for row in rotation_rows}
+        autoselects = {row["autoselect_id"]: row.get("config", {}) for row in autoselect_rows}
+    else:
+        providers = getattr(config, "providers", None) or _load_global_providers_from_disk()
+        rotations = getattr(config, "rotations", None) or {}
+        autoselects = getattr(config, "autoselect", None) or {}
+
+    entries = [
+        *_build_provider_entries(scope, owner_id, providers),
+        *_build_rotation_entries(scope, owner_id, rotations),
+        *_build_autoselect_entries(scope, owner_id, autoselects),
+    ]
+
+    entries.sort(key=lambda entry: (entry["kind"], entry["label"], entry["id"]))
+    return {
+        "scope": scope,
+        "owner_id": owner_id,
+        "entries": entries,
+    }

tests/routes/test_dashboard_studio.py

@@ -2,6 +2,7 @@ import json
 from pathlib import Path
 import sys
 from base64 import b64encode
+from uuid import uuid4
 
 import pytest
 from fastapi.testclient import TestClient
@@ -10,6 +11,8 @@ from itsdangerous import TimestampSigner
 sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
 
 from aisbf.routes.dashboard import providers as dashboard_providers
+from aisbf.database import DatabaseRegistry
+from aisbf.studio import build_studio_catalog
 from main import app
 from main import templates
 
@@ -80,6 +83,159 @@ def test_dashboard_studio_renders_empty_diagnostics_contract_for_shell_boot():
     assert '<script id="studio-bootstrap" type="application/json">{}</script>' in response.text
 
 
+def test_dashboard_studio_catalog_returns_global_resources_for_admin(monkeypatch):
+    client = TestClient(app)
+    _login_as_admin(client)
+
+    monkeypatch.setattr(
+        dashboard_providers,
+        "build_studio_catalog",
+        lambda **kwargs: {
+            "scope": kwargs["scope"],
+            "owner_id": kwargs["owner_id"],
+            "entries": [{"id": "provider/openai/gpt-4o", "owner_scope": "admin"}],
+        },
+    )
+
+    response = client.get("/dashboard/studio/catalog")
+
+    assert response.status_code == 200
+    assert response.json() == {
+        "scope": "admin",
+        "owner_id": None,
+        "entries": [{"id": "provider/openai/gpt-4o", "owner_scope": "admin"}],
+    }
+
+
+def test_dashboard_studio_catalog_returns_user_resources_for_user(monkeypatch):
+    client = TestClient(app)
+    db = DatabaseRegistry.get_config_database()
+    user_id = db.create_user(f"studio-demo-{uuid4().hex}", "not-used", role="user")
+    _set_session_cookie(
+        client,
+        {
+            "logged_in": True,
+            "username": "demo",
+            "role": "user",
+            "user_id": user_id,
+            "expires_at": 4102444800,
+        },
+    )
+
+    monkeypatch.setattr(
+        dashboard_providers,
+        "build_studio_catalog",
+        lambda **kwargs: {
+            "scope": kwargs["scope"],
+            "owner_id": kwargs["owner_id"],
+            "entries": [{"id": "provider/demo/gpt-4o-mini", "owner_scope": "user"}],
+        },
+    )
+
+    response = client.get("/dashboard/studio/catalog")
+
+    assert response.status_code == 200
+    assert response.json() == {
+        "scope": "user",
+        "owner_id": user_id,
+        "entries": [{"id": "provider/demo/gpt-4o-mini", "owner_scope": "user"}],
+    }
+
+
+def test_build_studio_catalog_uses_global_config_for_admin_scope():
+    class ModelStub:
+        def __init__(self, name, description=None, capabilities=None, context_length=None, architecture=None):
+            self.name = name
+            self.description = description
+            self.capabilities = capabilities
+            self.context_length = context_length
+            self.architecture = architecture
+
+    class ProviderStub:
+        def __init__(self, provider_type, models):
+            self.type = provider_type
+            self.models = models
+
+    class ConfigStub:
+        providers = {
+            "openai": ProviderStub(
+                "openai",
+                [ModelStub("gpt-4o", description="Flagship", capabilities=["chat", "vision"], context_length=128000)],
+            )
+        }
+        rotations = {
+            "team-default": {
+                "model_name": "Team default",
+                "providers": [{"provider": "openai", "model": "gpt-4o"}],
+                "capabilities": ["chat"],
+            }
+        }
+        autoselect = {
+            "writer": {
+                "model_name": "Writer",
+                "description": "General writing",
+                "fallback": "openai/gpt-4o",
+                "selection_model": "internal",
+                "available_models": [{"model_id": "openai/gpt-4o", "description": "Primary"}],
+                "capabilities": ["chat"],
+            }
+        }
+
+    catalog = build_studio_catalog(scope="admin", owner_id=None, config=ConfigStub())
+
+    assert catalog["scope"] == "admin"
+    assert catalog["owner_id"] is None
+    assert {entry["kind"] for entry in catalog["entries"]} == {"provider_model", "rotation", "autoselect"}
+    provider_entry = next(entry for entry in catalog["entries"] if entry["kind"] == "provider_model")
+    assert provider_entry["id"] == "provider/openai/gpt-4o"
+    assert provider_entry["owner_scope"] == "admin"
+    assert provider_entry["metadata"]["context_length"] == 128000
+
+
+def test_build_studio_catalog_uses_user_owned_resources_for_user_scope():
+    class DbStub:
+        def get_user_providers(self, user_id):
+            assert user_id == 17
+            return [{
+                "provider_id": "local-openai",
+                "config": {
+                    "type": "openai",
+                    "models": [{"name": "gpt-4o-mini", "description": "Mini", "capabilities": ["chat"]}],
+                },
+            }]
+
+        def get_user_rotations(self, user_id):
+            assert user_id == 17
+            return [{
+                "rotation_id": "my-rotation",
+                "config": {"model_name": "My rotation", "providers": [{"provider": "local-openai", "model": "gpt-4o-mini"}]},
+            }]
+
+        def get_user_autoselects(self, user_id):
+            assert user_id == 17
+            return [{
+                "autoselect_id": "my-autoselect",
+                "config": {
+                    "model_name": "My autoselect",
+                    "description": "Pick best model",
+                    "fallback": "local-openai/gpt-4o-mini",
+                    "selection_model": "internal",
+                    "available_models": [{"model_id": "local-openai/gpt-4o-mini", "description": "Mini"}],
+                },
+            }]
+
+    catalog = build_studio_catalog(scope="user", owner_id=17, db=DbStub())
+
+    assert catalog["scope"] == "user"
+    assert catalog["owner_id"] == 17
+    assert all(entry["owner_scope"] == "user" for entry in catalog["entries"])
+    assert {entry["id"] for entry in catalog["entries"]} == {
+        "provider/local-openai/gpt-4o-mini",
+        "rotation/my-rotation",
+        "autoselect/my-autoselect",
+    }
+
+
 def _find_session_secret() -> str:
     for middleware in app.user_middleware:
         kwargs = getattr(middleware, "kwargs", {})