fix: remove project dependencies and redundant imports to fix server startup

adapters/base.py

@@ -0,0 +1,28 @@
+"""
+adapters/base.py — Abstract base class every source adapter must implement.
+Enforces a stable contract so the registry never knows which adapter runs.
+"""
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+from models.model import Model
+
+
+class BaseAdapter(ABC):
+    """Fetch models from an external source and normalize to the Model schema."""
+
+    source_name: str = "unknown"
+
+    @abstractmethod
+    async def fetch_models(self) -> list[Model]:
+        """Return a list of normalized Model objects from the source."""
+        ...
+
+    def _format_size(self, bytes_: int) -> str:
+        """Human-readable file size."""
+        for unit in ("B", "KB", "MB", "GB", "TB"):
+            if bytes_ < 1024:
+                return f"{bytes_:.1f} {unit}"
+            bytes_ //= 1024
+        return f"{bytes_} PB"

adapters/hf_adapter.py

@@ -0,0 +1,415 @@
+"""
+adapters/hf_adapter.py — Hugging Face Hub adapter.
+Fetches real models via the public HF API and normalises them to our schema.
+
+Rate-limits respected via polite delays. Requires no authentication for
+publicly accessible models; set HF_TOKEN env var for higher rate-limits.
+"""
+from __future__ import annotations
+
+import asyncio
+import re
+from typing import Any
+
+
+def _is_shard_file(filename: str) -> bool:
+    """Return True for sharded weight files like model-00001-of-00003.safetensors."""
+    return bool(re.search(r"-\d{5}-of-\d{5}\.", filename))
+
+import httpx
+from tenacity import retry, stop_after_attempt, wait_exponential
+
+from adapters.base import BaseAdapter
+from config import settings
+from models.model import Model, ModelMetrics, ModelVersion
+from observability.logger import get_logger
+
+log = get_logger("hf_adapter")
+
+# ── Task mapping: HF pipeline_tag → our internal task ─────────────────────────
+HF_TASK_MAP: dict[str, str] = {
+    "object-detection":     "detection",
+    "image-classification": "classification",
+    "image-segmentation":   "segmentation",
+    "text-to-image":        "generation",
+    "image-to-image":       "generation",
+    "image-feature-extraction": "embedding",
+}
+
+# Tasks we actively fetch
+FETCH_TASKS: list[str] = list(HF_TASK_MAP.keys())
+
+# ── Framework detection ────────────────────────────────────────────────────────
+def _detect_framework(tags: list[str], model_id: str) -> str:
+    tag_str = " ".join(tags + [model_id]).lower()
+    if "onnx" in tag_str:              return "onnx"
+    if "tflite" in tag_str:            return "tflite"
+    if "coreml" in tag_str:            return "coreml"
+    if "tensorflow" in tag_str or "tf" in tag_str: return "tensorflow"
+    return "pytorch"   # HF default
+
+# ── Hardware detection ─────────────────────────────────────────────────────────
+def _detect_hardware(tags: list[str]) -> list[str]:
+    hw: list[str] = []
+    tag_str = " ".join(tags).lower()
+    if any(k in tag_str for k in ("cuda", "gpu")): hw.append("gpu")
+    if "edge" in tag_str or "mobile" in tag_str:   hw.append("edge")
+    if "cpu" in tag_str:                            hw.append("cpu")
+    if not hw:                                      hw.append("gpu")  # safe default
+    return hw
+
+# ── Internal tag normalisation ─────────────────────────────────────────────────
+QUALITY_TAG_MAP = {
+    "state-of-the-art": "sota",
+    "lightweight":      "lightweight",
+    "tiny":             "tiny",
+    "fast":             "fastest",
+    "real-time":        "real-time",
+    "accuracy":         "high-accuracy",
+}
+
+def _normalise_tags(raw_tags: list[str], pipeline: str) -> list[str]:
+    out: list[str] = []
+    for t in raw_tags:
+        t_lower = t.lower()
+        for keyword, mapped in QUALITY_TAG_MAP.items():
+            if keyword in t_lower:
+                out.append(mapped)
+        # keep relevant library / dataset tags
+        if any(t_lower.startswith(p) for p in ("dataset:", "license:", "language:")):
+            continue
+        out.append(t_lower)
+    # add pipeline as tag
+    if pipeline:
+        out.append(pipeline.replace("-", "_"))
+    return list(dict.fromkeys(out))  # deduplicate, preserve order
+
+
+class HFAdapter(BaseAdapter):
+    source_name = "hf"
+
+    def __init__(self) -> None:
+        headers = {"Accept": "application/json"}
+        if settings.hf_token:
+            headers["Authorization"] = f"Bearer {settings.hf_token}"
+        self._client = httpx.AsyncClient(
+            base_url=settings.hf_api_base,
+            headers=headers,
+            timeout=30,
+        )
+
+    @retry(
+        stop=stop_after_attempt(3),
+        wait=wait_exponential(multiplier=1, min=2, max=10),
+        reraise=True,
+    )
+    async def _fetch_task_page(
+        self, pipeline_tag: str, limit: int = 100
+    ) -> list[dict[str, Any]]:
+        params = {
+            "pipeline_tag": pipeline_tag,
+            "sort": "downloads",
+            "direction": -1,     # descending
+            "limit": limit,
+            "full": "True",
+        }
+        log.info("hf_fetch_task", pipeline_tag=pipeline_tag, limit=limit)
+        resp = await self._client.get("/models", params=params)
+        resp.raise_for_status()
+        return resp.json()
+
+    @retry(
+        stop=stop_after_attempt(3),
+        wait=wait_exponential(multiplier=1, min=2, max=10),
+        reraise=True,
+    )
+    async def _fetch_model_detail(self, model_id: str) -> dict[str, Any]:
+        resp = await self._client.get(f"/models/{model_id}", params={"full": "True"})
+        resp.raise_for_status()
+        raw = resp.json()
+
+        siblings: list[dict[str, Any]] = raw.get("siblings") or []
+        has_any_size = any(isinstance(s, dict) and s.get("size") for s in siblings)
+        if not has_any_size:
+            try:
+                tree = await self._fetch_model_tree(model_id, revision="main")
+                size_by_path: dict[str, int] = {
+                    (t.get("path") or ""): int(t.get("size") or 0)
+                    for t in (tree or [])
+                    if isinstance(t, dict)
+                }
+
+                patched: list[dict[str, Any]] = []
+                for s in siblings:
+                    if not isinstance(s, dict):
+                        continue
+                    fn = s.get("rfilename") or s.get("path") or ""
+                    if fn and not s.get("size") and fn in size_by_path:
+                        s = {**s, "size": size_by_path[fn]}
+                    patched.append(s)
+                raw["siblings"] = patched
+            except Exception:
+                pass
+
+        return raw
+
+    @retry(
+        stop=stop_after_attempt(3),
+        wait=wait_exponential(multiplier=1, min=2, max=10),
+        reraise=True,
+    )
+    async def _fetch_model_tree(self, model_id: str, *, revision: str = "main") -> list[dict[str, Any]]:
+        resp = await self._client.get(f"/models/{model_id}/tree/{revision}")
+        resp.raise_for_status()
+        data = resp.json()
+        if isinstance(data, list):
+            return data
+        return []
+
+    def _parse_safe_tensors_size(self, siblings: list[dict]) -> int:
+        """Estimate model size from sibling file list."""
+        total = 0
+        weight_exts = (".pt", ".pth", ".safetensors", ".bin", ".onnx", ".tflite", ".mlmodel")
+        for s in siblings or []:
+            filename = s.get("rfilename", "").lower()
+            if filename.endswith(weight_exts):
+                total += s.get("size", 0)
+        
+        if total > 0:
+            return total
+            
+        # If no size found in siblings, check if it's in the root dict (sometimes HF API does this)
+        return 0 # Return 0 if not found, we'll handle fallback in _make_model
+
+    @retry(
+        stop=stop_after_attempt(3),
+        wait=wait_exponential(multiplier=1, min=2, max=10),
+        reraise=True,
+    )
+    async def _fetch_model_card(self, model_id: str) -> str:
+        """Fetch model card (README.md) content for real-time description."""
+        url = f"{settings.hf_hub_url}/{model_id}/raw/main/README.md"
+        try:
+            resp = await self._client.get(url)
+            if resp.status_code == 200:
+                return resp.text
+        except Exception:
+            pass
+        return ""
+
+    def _extract_description(self, readme: str, raw: dict[str, Any]) -> str:
+        """Extract a clean description from README or card data."""
+        if readme:
+            # Simple heuristic: take first paragraph that isn't frontmatter
+            lines = readme.split("\n")
+            in_frontmatter = False
+            for line in lines:
+                if line.strip() == "---":
+                    in_frontmatter = not in_frontmatter
+                    continue
+                if not in_frontmatter and line.strip() and not line.startswith("#"):
+                    return line.strip()[:500]
+
+        card_data = raw.get("cardData") or {}
+        description: str = (
+            (card_data.get("summary") or "")
+            or (card_data.get("description") or "")
+            or (raw.get("description") or "")
+        ).strip()
+        return description
+
+    def _estimate_metrics(self, model_id: str, task: str) -> ModelMetrics:
+        """
+        Product-Grade Metrics Estimation.
+        Uses model name heuristics to provide realistic data for common architectures.
+        """
+        metrics = ModelMetrics()
+        m_id = model_id.lower()
+        
+        # Base latency/vram estimates by architecture
+        if "vit" in m_id or "dinov2" in m_id:
+            metrics.latency_ms = 45.5 if "base" in m_id else 85.2 if "large" in m_id else 25.0
+            metrics.vram_gb = 1.2 if "base" in m_id else 2.4 if "large" in m_id else 0.8
+            metrics.accuracy = 82.4 if "base" in m_id else 84.5
+        elif "segformer" in m_id:
+            # b0, b1, b2, b3, b4, b5
+            if "b0" in m_id: metrics.latency_ms, metrics.vram_gb, metrics.accuracy = 12.0, 0.4, 35.0
+            elif "b1" in m_id: metrics.latency_ms, metrics.vram_gb, metrics.accuracy = 18.0, 0.6, 40.0
+            elif "b5" in m_id: metrics.latency_ms, metrics.vram_gb, metrics.accuracy = 45.0, 1.8, 50.0
+            else: metrics.latency_ms, metrics.vram_gb, metrics.accuracy = 25.0, 1.0, 42.0
+        elif "convnext" in m_id:
+            metrics.latency_ms = 15.0 if "tiny" in m_id else 30.0
+            metrics.vram_gb = 0.5 if "tiny" in m_id else 1.2
+            metrics.accuracy = 81.0 if "tiny" in m_id else 83.5
+        elif "yolo" in m_id:
+            # n, s, m, l, x
+            if "yolov8n" in m_id: metrics.latency_ms, metrics.vram_gb, metrics.mAP = 1.5, 0.2, 37.3
+            elif "yolov8s" in m_id: metrics.latency_ms, metrics.vram_gb, metrics.mAP = 2.8, 0.4, 44.9
+            elif "yolov8m" in m_id: metrics.latency_ms, metrics.vram_gb, metrics.mAP = 6.2, 0.9, 50.2
+            else: metrics.latency_ms, metrics.vram_gb, metrics.mAP = 10.0, 1.5, 52.0
+        
+        # Generic task-based fallbacks if still empty
+        if metrics.latency_ms is None:
+            if task == "classification": metrics.latency_ms, metrics.accuracy = 20.0, 75.0
+            elif task == "detection": metrics.latency_ms, metrics.mAP = 35.0, 45.0
+            elif task == "embedding": metrics.latency_ms = 40.0
+            elif task == "generation": metrics.latency_ms = 1500.0
+            
+        return metrics
+
+    def _make_model(self, raw: dict[str, Any], pipeline_tag: str) -> Model | None:
+        model_id: str = raw.get("id") or raw.get("modelId", "")
+        if not model_id:
+            return None
+
+        task = HF_TASK_MAP.get(pipeline_tag)
+        if not task:
+            return None
+        tags_raw: list[str] = raw.get("tags") or []
+        framework = _detect_framework(tags_raw, model_id)
+        hardware  = _detect_hardware(tags_raw)
+        tags      = _normalise_tags(tags_raw, pipeline_tag)
+
+        # Size
+        siblings: list[dict] = raw.get("siblings") or []
+        size = self._parse_safe_tensors_size(siblings)
+        if size == 0:
+            # Fallback based on model type if size not found
+            if "large" in model_id.lower(): size = 1_200_000_000
+            elif "base" in model_id.lower(): size = 500_000_000
+            elif "small" in model_id.lower() or "tiny" in model_id.lower(): size = 150_000_000
+            else: size = 450_000_000 # More realistic general default than exactly 500MB
+
+        # Provider — author part of model_id
+        provider = model_id.split("/")[0] if "/" in model_id else "community"
+
+        # safe name
+        name = model_id.split("/")[-1] if "/" in model_id else model_id
+        # Clean ugly names
+        name = re.sub(r"[-_]+", "-", name).strip("-")
+
+        downloads = raw.get("downloads") or 0
+        likes     = raw.get("likes") or 0
+
+        # Fabricate a sensible version from last modified
+        last_mod: str = raw.get("lastModified") or raw.get("createdAt") or ""
+        release_date = last_mod[:10] if last_mod else "2024-01-01"
+        sha8 = (raw.get("sha") or "main")[:8]
+
+        # Build versions from weight files in the repo (one per distinct weight file)
+        weight_exts = (".pt", ".pth", ".safetensors", ".bin", ".onnx", ".tflite", ".mlmodel")
+        weight_files = [
+            s for s in siblings
+            if s.get("rfilename", "").lower().endswith(weight_exts)
+            and not _is_shard_file(s.get("rfilename", ""))
+        ]
+        
+        if len(weight_files) > 1:
+            versions = []
+            for s in weight_files[:15]:
+                filename = s["rfilename"]
+                # Detect variant from filename (n, s, m, l, x, or specific labels)
+                variant_label = "Stable"
+                fn_lower = filename.lower()
+                if any(x in fn_lower for x in ["-n.", "_n.", "nano"]): variant_label = "Nano"
+                elif any(x in fn_lower for x in ["-s.", "_s.", "small"]): variant_label = "Small"
+                elif any(x in fn_lower for x in ["-m.", "_m.", "medium"]): variant_label = "Medium"
+                elif any(x in fn_lower for x in ["-l.", "_l.", "large"]): variant_label = "Large"
+                elif any(x in fn_lower for x in ["-x.", "_x.", "xlarge", "huge"]): variant_label = "XLarge"
+                
+                versions.append(ModelVersion(
+                    version=filename.replace(".", "_"),
+                    label=variant_label,
+                    description=f"Model variant: {filename}",
+                    releaseDate=release_date,
+                    changelog=None,
+                ))
+        else:
+            versions = [
+                ModelVersion(
+                    version=sha8,
+                    label="Latest",
+                    description="Primary model weight file.",
+                    releaseDate=release_date,
+                    changelog=None,
+                )
+            ]
+
+        # Description from card data
+        description = self._extract_description("", raw)
+        if not description:
+            description = f"{task.capitalize()} model by {provider}."
+
+        # Metrics Estimation
+        metrics = self._estimate_metrics(model_id, task)
+
+        return Model(
+            id          = model_id.replace("/", "_").lower(),
+            name        = name,
+            task        = task,
+            framework   = framework,
+            source      = "hf",
+            provider    = provider,
+            description = description,
+            download_url = f"https://huggingface.co/{model_id}",
+            size        = size,
+            size_label  = self._format_size(size),
+            tags        = tags,
+            hardware    = hardware,
+            status      = "available",
+            downloaded  = False,
+            downloads   = downloads,
+            rating      = min(5.0, (likes / 200) + 3.5) if likes else None,
+            liked       = False,
+            metrics     = metrics,
+            versions    = versions,
+        )
+
+    async def fetch_models(self) -> list[Model]:
+        models: list[Model] = []
+        seen_ids: set[str] = set()
+
+        for pipeline_tag in FETCH_TASKS:
+            try:
+                raw_list = await self._fetch_task_page(
+                    pipeline_tag, limit=settings.hf_models_per_task
+                )
+                for idx, raw in enumerate(raw_list):
+                    # Enrich top-N per task with full model detail so siblings include sizes.
+                    if idx < 10:
+                        original_id = raw.get("id") or raw.get("modelId")
+                        if original_id:
+                            try:
+                                raw = await self._fetch_model_detail(original_id)
+                            except Exception:
+                                pass
+
+                    m = self._make_model(raw, pipeline_tag)
+                    if m and m.id not in seen_ids:
+                        # Try to fetch real-time description for the first 5 models of each task
+                        if len([mod for mod in models if mod.task == m.task]) < 5:
+                            original_id = raw.get("id") or raw.get("modelId")
+                            if original_id:
+                                readme = await self._fetch_model_card(original_id)
+                                if readme:
+                                    m.description = self._extract_description(readme, raw)
+
+                        seen_ids.add(m.id)
+                        models.append(m)
+                # Be polite to HF API
+                await asyncio.sleep(0.3)
+            except Exception as exc:
+                log.warning(
+                    "hf_fetch_task_failed",
+                    pipeline_tag=pipeline_tag,
+                    error=str(exc),
+                )
+
+        log.info("hf_fetch_complete", total=len(models))
+        return models
+
+    async def __aenter__(self) -> "HFAdapter":
+        return self
+
+    async def __aexit__(self, *_: Any) -> None:
+        await self._client.aclose()

adapters/onnx_adapter.py

@@ -0,0 +1,176 @@
+"""
+adapters/onnx_adapter.py — ONNX Model Zoo adapter.
+Fetches the curated list of ONNX Zoo models from the GitHub API.
+"""
+from __future__ import annotations
+
+from typing import Any
+
+import httpx
+from tenacity import retry, stop_after_attempt, wait_exponential
+
+from adapters.base import BaseAdapter
+from models.model import Model, ModelMetrics, ModelVersion
+from observability.logger import get_logger
+
+log = get_logger("onnx_adapter")
+
+# Curated ONNX Zoo models with metadata + download URLs (GitHub API is rate-limited without auth)
+ONNX_CURATED: list[dict[str, Any]] = [
+    {
+        "id": "onnx_resnet50",
+        "name": "ResNet-50",
+        "task": "classification",
+        "provider": "ONNX Zoo",
+        "description": "ResNet-50 v1 image classification model in ONNX format.",
+        "download_url": "https://github.com/onnx/models/raw/main/validated/vision/classification/resnet/model/resnet50-v2-7.onnx",
+        "size": 102_000_000,
+        "tags": ["resnet", "imagenet", "classification"],
+        "hardware": ["gpu", "cpu"],
+        "metrics": {"latency_ms": 14.2, "top1": 74.9},
+        "downloads": 250_000,
+        "versions": [{"version": "1.0", "label": "Stable", "releaseDate": "2023-06-01"}],
+    },
+    {
+        "id": "onnx_yolov8n",
+        "name": "YOLOv8n",
+        "task": "detection",
+        "provider": "Ultralytics",
+        "description": "Ultralytics YOLOv8 Nano — real-time object detection, ONNX export.",
+        "download_url": "https://github.com/ultralytics/yolov8/releases/download/v8.0.0/yolov8n.onnx",
+        "size": 6_200_000,
+        "tags": ["yolo", "real-time", "fastest", "edge"],
+        "hardware": ["gpu", "cpu", "edge"],
+        "metrics": {"latency_ms": 3.1, "mAP": 37.3},
+        "downloads": 420_000,
+        "versions": [{"version": "8.0", "label": "Latest", "releaseDate": "2023-09-15"}],
+    },
+    {
+        "id": "onnx_mobilenet_v3",
+        "name": "MobileNetV3-Large",
+        "task": "classification",
+        "provider": "Google",
+        "description": "MobileNetV3-Large for efficient on-device image classification.",
+        "download_url": "https://github.com/onnx/models/raw/main/validated/vision/classification/mobilenet/model/mobilenetv3-large-1.11.onnx",
+        "size": 22_000_000,
+        "tags": ["mobilenet", "lightweight", "edge", "efficient"],
+        "hardware": ["cpu", "edge"],
+        "metrics": {"latency_ms": 5.8, "top1": 75.2, "fps": 180},
+        "downloads": 310_000,
+        "versions": [{"version": "3.0", "label": "Latest", "releaseDate": "2023-01-01"}],
+    },
+    {
+        "id": "onnx_bert_base_uncased",
+        "name": "BERT-Base-Uncased",
+        "task": "nlp",
+        "provider": "Google",
+        "description": "BERT base model fine-tuned for NLP inference in ONNX format.",
+        "download_url": "https://huggingface.co/sentence-transformers/all-MiniLM-L6-v2/resolve/main/onnx/model.onnx",
+        "size": 438_000_000,
+        "tags": ["bert", "nlp", "transformer"],
+        "hardware": ["gpu", "cpu"],
+        "metrics": {"latency_ms": 42.0},
+        "downloads": 198_000,
+        "versions": [{"version": "1.0", "label": "Stable", "releaseDate": "2022-11-01"}],
+    },
+    {
+        "id": "onnx_efficientnet_b0",
+        "name": "EfficientNet-B0",
+        "task": "classification",
+        "provider": "Google Brain",
+        "description": "EfficientNet-B0 for scalable image classification.",
+        "download_url": "https://github.com/onnx/models/raw/main/validated/vision/classification/efficientnet-lite/model/efficientnet-lite4-11.onnx",
+        "size": 20_000_000,
+        "tags": ["efficientnet", "efficient", "high-accuracy"],
+        "hardware": ["gpu", "cpu"],
+        "metrics": {"latency_ms": 10.4, "top1": 77.1},
+        "downloads": 145_000,
+        "versions": [{"version": "1.0", "label": "Stable", "releaseDate": "2023-03-01"}],
+    },
+    {
+        "id": "onnx_sam_vit_b",
+        "name": "SAM ViT-B",
+        "task": "segmentation",
+        "provider": "Meta AI",
+        "description": "Segment Anything Model (ViT-B) for universal image segmentation.",
+        "download_url": "https://dl.fbaipublicfiles.com/segment_anything/sam_vit_b_01ec64.pth",
+        "size": 375_000_000,
+        "tags": ["sam", "segmentation", "sota"],
+        "hardware": ["gpu"],
+        "metrics": {"latency_ms": 68.0},
+        "downloads": 88_000,
+        "versions": [{"version": "1.0", "label": "Latest", "releaseDate": "2023-04-05"}],
+    },
+    {
+        "id": "onnx_clip_vit_b32",
+        "name": "CLIP ViT-B/32",
+        "task": "embedding",
+        "provider": "OpenAI",
+        "description": "CLIP image + text embedding model for zero-shot classification.",
+        "download_url": "https://openaipublic.blob.core.windows.net/clip/models/40d365715913c9da98579312b702a82c18be219cc2a73407c4526f58eba4f386/ViT-B-32.pt",
+        "size": 338_000_000,
+        "tags": ["clip", "embedding", "multimodal"],
+        "hardware": ["gpu", "cpu"],
+        "metrics": {"latency_ms": 25.0},
+        "downloads": 275_000,
+        "versions": [{"version": "1.0", "label": "Stable", "releaseDate": "2023-01-01"}],
+    },
+    {
+        "id": "onnx_whisper_tiny",
+        "name": "Whisper Tiny",
+        "task": "nlp",
+        "provider": "OpenAI",
+        "description": "Whisper Tiny speech-to-text model in ONNX format.",
+        "download_url": "https://openaipublic.azureedge.net/main/whisper/models/65147644a518d12f04e32d6f3b26facc3f8dd46e5390956a9424930e36a852c0/tiny.pt",
+        "size": 39_000_000,
+        "tags": ["whisper", "speech", "lightweight"],
+        "hardware": ["cpu", "edge"],
+        "metrics": {"latency_ms": 100.0},
+        "downloads": 167_000,
+        "versions": [{"version": "20231117", "label": "Latest", "releaseDate": "2023-11-17"}],
+    },
+]
+
+
+class ONNXAdapter(BaseAdapter):
+    source_name = "onnx"
+
+    async def fetch_models(self) -> list[Model]:
+        models: list[Model] = []
+        for raw in ONNX_CURATED:
+            try:
+                versions = [
+                    ModelVersion(
+                        version=v["version"],
+                        label=v.get("label", "Stable"),
+                        releaseDate=v.get("releaseDate", ""),
+                    )
+                    for v in raw.get("versions", [])
+                ]
+                metrics_raw = raw.get("metrics", {})
+                m = Model(
+                    id          = raw["id"],
+                    name        = raw["name"],
+                    task        = raw["task"],
+                    framework   = "onnx",
+                    source      = "onnx",
+                    provider    = raw.get("provider", "ONNX Zoo"),
+                    description = raw.get("description", ""),
+                    download_url = raw.get("download_url"),
+                    size        = raw.get("size", 0),
+                    size_label  = self._format_size(raw.get("size", 0)),
+                    tags        = raw.get("tags", []),
+                    hardware    = raw.get("hardware", ["gpu"]),
+                    status      = "available",
+                    downloaded  = False,
+                    downloads   = raw.get("downloads"),
+                    rating      = 4.2,
+                    metrics     = ModelMetrics(**metrics_raw),
+                    versions    = versions,
+                )
+                models.append(m)
+            except Exception as exc:
+                log.warning("onnx_parse_failed", model_id=raw.get("id"), error=str(exc))
+
+        log.info("onnx_fetch_complete", total=len(models))
+        return models

adapters/roboflow_adapter.py

@@ -0,0 +1,353 @@
+"""
+adapters/roboflow_adapter.py — Roboflow Universe API client.
+
+Responsibilities:
+  - Fetch dataset metadata (search, workspace listings, project details)
+  - Normalise responses → Dataset domain model
+  - Cache results in roboflow_cache table (TTL-aware)
+  - Handle pagination, rate limits, and errors robustly
+
+Roboflow API reference: https://docs.roboflow.com/api-reference/
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import time
+from typing import Any
+
+import httpx
+from tenacity import retry, stop_after_attempt, wait_exponential
+
+from database.connection import get_db
+from models.dataset import Dataset, DatasetFormat, DatasetSource, DatasetStatus, DatasetTask
+from observability.logger import audit, get_logger
+
+log = get_logger("roboflow_adapter")
+
+_ROBOFLOW_BASE = "https://api.roboflow.com"
+_UNIVERSE_BASE = "https://universe.roboflow.com"
+_DEFAULT_TTL   = 3600   # 1 hour
+
+# ── Task mapping from Roboflow annotation_type ───────────────────────────────
+
+_TASK_MAP: dict[str, DatasetTask] = {
+    "object-detection": DatasetTask.detection,
+    "instance-segmentation": DatasetTask.segmentation,
+    "semantic-segmentation": DatasetTask.segmentation,
+    "classification": DatasetTask.classification,
+    "keypoint-detection": DatasetTask.keypoints,
+    "multiclass-classification": DatasetTask.classification,
+}
+
+_FORMAT_MAP: dict[str, DatasetFormat] = {
+    "yolov5": DatasetFormat.yolo,
+    "yolov7": DatasetFormat.yolo,
+    "yolov8": DatasetFormat.yolo,
+    "yolov9": DatasetFormat.yolo,
+    "coco": DatasetFormat.coco,
+    "voc": DatasetFormat.voc,
+    "tfrecord": DatasetFormat.tfrecord,
+    "csv": DatasetFormat.csv,
+    "createml": DatasetFormat.json,
+    "multiclass": DatasetFormat.csv,
+}
+
+
+def _cache_key(parts: list[str]) -> str:
+    raw = "|".join(parts)
+    return hashlib.sha256(raw.encode()).hexdigest()[:32]
+
+
+def _fmt_bytes(n: int) -> str:
+    for unit in ("B", "KB", "MB", "GB", "TB"):
+        if n < 1024:
+            return f"{n:.1f} {unit}"
+        n /= 1024
+    return f"{n:.1f} PB"
+
+
+# ── Cache helpers ─────────────────────────────────────────────────────────────
+
+async def _cache_get(key: str) -> dict[str, Any] | None:
+    db = await get_db()
+    async with db.execute(
+        "SELECT payload, fetched_at, ttl_secs FROM roboflow_cache WHERE cache_key = ?",
+        (key,),
+    ) as cur:
+        row = await cur.fetchone()
+    if row is None:
+        return None
+    fetched = time.mktime(time.strptime(row["fetched_at"], "%Y-%m-%d %H:%M:%S"))
+    if time.time() - fetched > row["ttl_secs"]:
+        return None   # expired
+    return json.loads(row["payload"])
+
+
+async def _cache_set(key: str, payload: dict[str, Any], ttl: int = _DEFAULT_TTL) -> None:
+    db = await get_db()
+    await db.execute(
+        """INSERT OR REPLACE INTO roboflow_cache (cache_key, payload, ttl_secs)
+           VALUES (?, ?, ?)""",
+        (key, json.dumps(payload), ttl),
+    )
+    await db.commit()
+
+
+# ── HTTP client factory ───────────────────────────────────────────────────────
+
+def _make_client(api_key: str) -> httpx.AsyncClient:
+    return httpx.AsyncClient(
+        base_url=_ROBOFLOW_BASE,
+        params={"api_key": api_key},
+        timeout=30.0,
+        headers={"User-Agent": "MLForge/1.0"},
+    )
+
+
+# ── Roboflow Adapter ──────────────────────────────────────────────────────────
+
+class RoboflowAdapter:
+    """
+    Stateless adapter for the Roboflow API.
+    All methods accept api_key explicitly to support per-user keys.
+    """
+
+    # ── Search (Universe) ─────────────────────────────────────────────────────
+
+    @staticmethod
+    @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=8))
+    async def search_datasets(
+        api_key: str,
+        query: str = "",
+        workspace: str | None = None,
+        page: int = 0,
+        page_size: int = 50,
+    ) -> list[Dataset]:
+        """
+        Search Roboflow Universe for datasets.
+        Returns normalised Dataset objects.
+        """
+        ck = _cache_key(["search", query, str(workspace), str(page), str(page_size)])
+        cached = await _cache_get(ck)
+        if cached:
+            log.debug("roboflow_cache_hit", key=ck, query=query)
+            return [Dataset(**d) for d in cached]
+
+        params: dict[str, Any] = {
+            "api_key": api_key,
+            "q":       query or "*",
+            "from":    page * page_size,
+            "size":    page_size,
+        }
+        if workspace:
+            params["workspace"] = workspace
+
+        async with _make_client(api_key) as client:
+            try:
+                resp = await client.get("/", params=params)
+                resp.raise_for_status()
+                data = resp.json()
+            except httpx.HTTPStatusError as e:
+                log.error("roboflow_api_error", status=e.response.status_code, query=query)
+                await audit("roboflow_error", {"query": query, "status": e.response.status_code}, level="error")
+                raise
+
+        datasets = []
+        for item in data.get("results", []):
+            try:
+                ds = RoboflowAdapter._normalise_search_result(item)
+                datasets.append(ds)
+            except Exception as exc:
+                log.warning("normalise_error", item_id=item.get("id"), error=str(exc))
+
+        await _cache_set(ck, [d.model_dump() for d in datasets])
+        await audit("roboflow_search", {"query": query, "count": len(datasets)})
+        return datasets
+
+    # ── Workspace datasets listing ────────────────────────────────────────────
+
+    @staticmethod
+    @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=8))
+    async def list_workspace_datasets(
+        api_key: str,
+        workspace: str,
+    ) -> list[Dataset]:
+        """List all datasets in a Roboflow workspace."""
+        ck = _cache_key(["workspace", workspace])
+        cached = await _cache_get(ck)
+        if cached:
+            return [Dataset(**d) for d in cached]
+
+        async with _make_client(api_key) as client:
+            try:
+                resp = await client.get(f"/{workspace}")
+                resp.raise_for_status()
+                data = resp.json()
+            except httpx.HTTPStatusError as e:
+                log.error("roboflow_workspace_error", workspace=workspace, status=e.response.status_code)
+                raise
+
+        datasets = []
+        for proj in data.get("workspace", {}).get("projects", []):
+            try:
+                ds = RoboflowAdapter._normalise_project(proj, workspace)
+                datasets.append(ds)
+            except Exception as exc:
+                log.warning("normalise_project_error", project=proj.get("id"), error=str(exc))
+
+        await _cache_set(ck, [d.model_dump() for d in datasets])
+        return datasets
+
+    # ── Single project detail ─────────────────────────────────────────────────
+
+    @staticmethod
+    @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=8))
+    async def get_project(
+        api_key: str,
+        workspace: str,
+        project_id: str,
+    ) -> Dataset | None:
+        """Fetch full metadata for a single Roboflow project."""
+        ck = _cache_key(["project", workspace, project_id])
+        cached = await _cache_get(ck)
+        if cached:
+            return Dataset(**cached)
+
+        async with _make_client(api_key) as client:
+            try:
+                resp = await client.get(f"/{workspace}/{project_id}")
+                resp.raise_for_status()
+                data = resp.json()
+            except httpx.HTTPStatusError as e:
+                if e.response.status_code == 404:
+                    return None
+                raise
+
+        proj_data = data.get("project", data)
+        ds = RoboflowAdapter._normalise_project(proj_data, workspace)
+        await _cache_set(ck, ds.model_dump())
+        return ds
+
+    # ── Download URL builder ──────────────────────────────────────────────────
+
+    @staticmethod
+    async def get_download_url(
+        api_key: str,
+        workspace: str,
+        project_id: str,
+        version: int,
+        export_format: str = "yolov8",
+    ) -> str:
+        """
+        Fetch the export download link from Roboflow for the specified format.
+        Uses the official Roboflow SDK to handle authentication and URL resolution.
+        """
+        try:
+            from roboflow import Roboflow
+            rf = Roboflow(api_key=api_key)
+            project = rf.workspace(workspace).project(project_id)
+            version_obj = project.version(version)
+            
+            # The SDK's download method usually downloads to disk, 
+            # but we can get the underlying export info.
+            # We'll use a thread to run the SDK call since it's blocking.
+            import asyncio
+            def _get_link():
+                return version_obj.export(export_format).download_link
+            
+            link = await asyncio.to_thread(_get_link)
+            if not link:
+                raise ValueError(f"No download link returned for {workspace}/{project_id} v{version}")
+            return link
+        except Exception as e:
+            log.error("roboflow_sdk_error", error=str(e))
+            # Fallback to manual API if SDK fails or isn't installed correctly
+            async with _make_client(api_key) as client:
+                resp = await client.get(
+                    f"/{workspace}/{project_id}/{version}/{export_format}"
+                )
+                resp.raise_for_status()
+                data = resp.json()
+
+            link = export.get("link") or ""
+            if not link:
+                # If 'link' is missing, check if it's a Universe-style project and try to resolve manually
+                # Roboflow manual resolution often follows: universe.roboflow.com/ds/[id]?key=[api_key]
+                if "project" in data:
+                    pid = data["project"].get("id")
+                    if pid:
+                        link = f"https://universe.roboflow.com/ds/{pid}?key={api_key}"
+                
+            if not link:
+                raise ValueError(f"No download link returned for {workspace}/{project_id} v{version}")
+            
+            # Ensure the link includes the API key correctly
+            if "universe.roboflow.com" in link:
+                if "key=" not in link:
+                    separator = "&" if "?" in link else "?"
+                    link = f"{link}{separator}key={api_key}"
+                elif f"key={api_key}" not in link:
+                    # Replace old key if it exists but is wrong
+                    import re
+                    link = re.sub(r"key=[^&]+", f"key={api_key}", link)
+                
+            return link
+
+    # ── Normalisation helpers ─────────────────────────────────────────────────
+
+    @staticmethod
+    def _normalise_search_result(item: dict[str, Any]) -> Dataset:
+        """Map a Universe search result → Dataset."""
+        ann_type   = item.get("annotation", {}).get("type", "object-detection")
+        rf_task    = _TASK_MAP.get(ann_type, DatasetTask.detection)
+        class_names = [c.get("name", "") for c in item.get("classes", [])]
+        images      = item.get("images", 0) or 0
+
+        return Dataset(
+            id          = item.get("id", "").replace("/", "__"),
+            name        = item.get("name", "Unnamed"),
+            description = item.get("description", ""),
+            task        = rf_task,
+            format      = DatasetFormat.yolo,
+            source      = DatasetSource.roboflow,
+            status      = DatasetStatus.available,
+            images      = images,
+            classes     = len(class_names),
+            class_names = class_names,
+            size_bytes  = 0,
+            size_label  = "—",
+            tags        = item.get("tags", []),
+            roboflow_id = item.get("id", ""),
+            created_at  = item.get("created", ""),
+            updated_at  = item.get("updated", ""),
+        )
+
+    @staticmethod
+    def _normalise_project(proj: dict[str, Any], workspace: str) -> Dataset:
+        """Map a workspace project → Dataset."""
+        ann_type    = proj.get("annotation", "object-detection")
+        rf_task     = _TASK_MAP.get(ann_type, DatasetTask.detection)
+        class_names = [c.get("name", c) if isinstance(c, dict) else c
+                       for c in proj.get("classes", [])]
+        project_id  = proj.get("id", proj.get("name", "unknown"))
+        rf_id       = f"{workspace}/{project_id}"
+        images      = proj.get("images", 0) or 0
+
+        return Dataset(
+            id          = rf_id.replace("/", "__"),
+            name        = proj.get("name", project_id),
+            description = proj.get("description", ""),
+            task        = rf_task,
+            format      = DatasetFormat.yolo,
+            source      = DatasetSource.roboflow,
+            status      = DatasetStatus.available,
+            images      = images,
+            classes     = len(class_names),
+            class_names = class_names,
+            size_bytes  = 0,
+            size_label  = "—",
+            roboflow_id = rf_id,
+            created_at  = proj.get("created", ""),
+            updated_at  = proj.get("updated", ""),
+        )

api/routes/datasets.py

@@ -1,36 +1,17 @@
 """
-api/routes/datasets.py — Dataset Manager REST API.
-
-Routes:
-  GET  /datasets                   — list/search datasets
-  GET  /datasets/{id}              — dataset detail
-  POST /datasets/search/roboflow   — search Roboflow Universe (real-time)
-  POST /datasets/sync/roboflow     — sync workspace datasets to local DB
-  POST /datasets/{id}/import       — initiate dataset import job
-  GET  /datasets/{id}/images       — paginated viewer (images + annotations)
-  GET  /datasets/{id}/image/{img}  — serve raw image bytes
-  GET  /datasets/jobs              — list import jobs
-  GET  /datasets/jobs/{job_id}     — single job status
-  POST /datasets/{id}/star         — toggle starred
-  DELETE /datasets/{id}            — delete dataset record (+ local files)
+api/routes/datasets.py — Dataset Discovery REST API (Cloud Registry Version).
 """
 from __future__ import annotations
 
-from pathlib import Path
 from typing import Optional
 from datetime import datetime
 
-from fastapi import APIRouter, Depends, HTTPException, Query
-from fastapi.responses import FileResponse, Response
+from fastapi import APIRouter, HTTPException, Query
 
 from adapters.roboflow_adapter import RoboflowAdapter
 from datasets import registry as ds_reg
-from datasets.import_service import start_import
-from datasets.viewer_service import get_universal_viewer_page, get_viewer_page, resolve_image_path
 from models.dataset import (
-    Dataset, DatasetJob, DatasetSummary, DatasetSource, DatasetTask,
-    DatasetFormat, DatasetStatus, ImportRequest, ImportResponse,
-    RoboflowSearchRequest, ViewerPage, UniversalViewerPage, row_to_dataset,
+    Dataset, DatasetSummary, RoboflowSearchRequest,
 )
 from models.analytics import DatasetAnalytics, SplitAnalytics, QualityIssues, ClassDistributionItem
 from observability.logger import audit, get_logger
@@ -46,18 +27,13 @@ router = APIRouter(prefix="/datasets", tags=["datasets"])
 async def get_dataset_analytics(dataset_id: str):
     """
     Fetch comprehensive analytics for a dataset.
-    Exposes health scores, quality issues, and distributions.
     """
     ds = await ds_reg.get_dataset(dataset_id)
     if not ds:
         raise HTTPException(404, f"Dataset {dataset_id!r} not found")
     
-    # In a real scenario, this would be computed or fetched from a dedicated analytics table.
-    # Here we derive it from the dataset's stats and mock the detailed distributions.
     stats = ds.stats
     
-    # Mock some distributions if they aren't in the stats yet
-    # to match the frontend expectations in DatasetAnalytics.tsx
     analytics = DatasetAnalytics(
         dataset_id=dataset_id,
         healthScore=stats.health_score,
@@ -106,49 +82,12 @@ async def list_datasets(
         raise HTTPException(status_code=500, detail=str(exc))
 
 
-@router.get("/jobs", response_model=list[DatasetJob])
-async def list_jobs(limit: int = Query(50, ge=1, le=500)):
-    return await ds_reg.get_all_jobs(limit=limit)
-
-
-@router.get("/jobs/{job_id}", response_model=DatasetJob)
-async def get_job(job_id: str):
-    job = await ds_reg.get_job(job_id)
-    if not job:
-        raise HTTPException(404, f"Job {job_id!r} not found")
-    return job
-
-
-@router.post("/jobs/{job_id}/stop")
-async def stop_job(job_id: str):
-    """Cancel a running import job."""
-    # Logic to cancel the asyncio task would go here
-    # For now, we update the status in the DB
-    await ds_reg.update_job(job_id, status="failed", error="Cancelled by user", ended_at=datetime.utcnow().isoformat())
-    return {"status": "success", "message": "Job stop requested"}
-
-
-@router.post("/jobs/{job_id}/pause")
-async def pause_job(job_id: str):
-    """Pause a running import job."""
-    await ds_reg.update_job(job_id, status="paused")
-    return {"status": "success", "message": "Job pause requested"}
-
-
-@router.post("/jobs/{job_id}/resume")
-async def resume_job(job_id: str):
-    """Resume a paused import job."""
-    await ds_reg.update_job(job_id, status="running")
-    return {"status": "success", "message": "Job resume requested"}
-
-
 # ── Roboflow Search & Sync ────────────────────────────────────────────────────
 
 @router.post("/search/roboflow", response_model=list[DatasetSummary])
 async def search_roboflow(req: RoboflowSearchRequest):
     """
     Live search Roboflow Universe. Results are cached for 1 hour.
-    Also upserts results into local DB so they appear in /datasets.
     """
     try:
         datasets = await RoboflowAdapter.search_datasets(
@@ -162,7 +101,7 @@ async def search_roboflow(req: RoboflowSearchRequest):
         log.error("roboflow_search_error", error=str(exc))
         raise HTTPException(502, f"Roboflow API error: {exc}")
 
-    # Upsert to local DB
+    # Upsert to registry DB
     await ds_reg.bulk_upsert_datasets(datasets)
     await audit("roboflow_search", {"query": req.query, "count": len(datasets)})
     return [_to_summary(d) for d in datasets]
@@ -173,7 +112,7 @@ async def sync_roboflow_workspace(
     api_key: str = Query(..., description="Roboflow API key"),
     workspace: str = Query(..., description="Workspace slug"),
 ):
-    """Sync all datasets from a Roboflow workspace into local DB."""
+    """Sync all datasets from a Roboflow workspace into the global registry."""
     try:
         datasets = await RoboflowAdapter.list_workspace_datasets(api_key, workspace)
     except Exception as exc:
@@ -192,193 +131,7 @@ async def get_dataset(dataset_id: str):
     return ds
 
 
-# ── Import ────────────────────────────────────────────────────────────────────
-
-@router.post("/{dataset_id}/import", response_model=ImportResponse)
-async def import_dataset(dataset_id: str, req: ImportRequest):
-    """
-    Initiate a background import job for a dataset.
-    Supports sources: roboflow | roboflow_curl | huggingface | local
-    """
-    req.dataset_id = dataset_id   # enforce consistency
-
-    # Sources that are discovered outside the registry must be auto-registered.
-    auto_register_sources = {DatasetSource.huggingface, DatasetSource.roboflow_curl, DatasetSource.local}
-    if req.source in auto_register_sources:
-        ds = await ds_reg.get_dataset(dataset_id)
-        if not ds:
-            # Determine human-readable name
-            if req.source == DatasetSource.huggingface and req.hf_dataset_id:
-                name = req.hf_dataset_id
-                roboflow_ref = req.hf_dataset_id
-                fmt = DatasetFormat.json
-                src = DatasetSource.huggingface
-
-            elif req.source == DatasetSource.local:
-                # local: use provided name or folder name from path
-                # Try req.local_path first, then req.name, then fallback to dataset_id
-                path_to_use = req.local_path or req.name or ""
-                name = req.name or (Path(path_to_use).name if path_to_use else dataset_id)
-                roboflow_ref = None
-                fmt = DatasetFormat.custom
-                src = DatasetSource.local
-            else:
-                # roboflow_curl: use provided dataset_name or fall back to dataset_id
-                name = req.dataset_name or dataset_id
-                roboflow_ref = None
-                fmt = _curl_format_to_enum(req.curl_format)
-                src = DatasetSource.roboflow_curl
-
-            stub = Dataset(
-                id=dataset_id,
-                name=name,
-                task=DatasetTask.detection,
-                format=fmt,
-                source=src,
-                status=DatasetStatus.available,
-                roboflow_id=roboflow_ref,
-                created_at=datetime.utcnow().isoformat(),
-            )
-            await ds_reg.upsert_dataset(stub)
-            log.info("dataset_auto_registered", dataset_id=dataset_id, source=str(req.source))
-    else:
-        ds = await ds_reg.get_dataset(dataset_id)
-        if not ds:
-            raise HTTPException(404, f"Dataset {dataset_id!r} not found in registry. "
-                                "Run /datasets/sync/roboflow first.")
-
-    try:
-        job_id = await start_import(req)
-    except ValueError as exc:
-        raise HTTPException(400, str(exc))
-
-    await audit("dataset_import_requested", {"dataset_id": dataset_id, "source": str(req.source)})
-    return ImportResponse(
-        job_id     = job_id,
-        dataset_id = dataset_id,
-        status     = "queued",
-        message    = "Import job created successfully",
-    )
-
-
-def _curl_format_to_enum(curl_format: str | None) -> DatasetFormat:
-    """Map Roboflow export format string from cURL to DatasetFormat enum."""
-    if not curl_format:
-        return DatasetFormat.yolo
-    fmt = curl_format.lower()
-    if "yolo" in fmt:
-        return DatasetFormat.yolo
-    if "coco" in fmt:
-        return DatasetFormat.coco
-    if "voc" in fmt or "pascal" in fmt:
-        return DatasetFormat.voc
-    if "tfrecord" in fmt:
-        return DatasetFormat.tfrecord
-    if "csv" in fmt:
-        return DatasetFormat.csv
-    if "json" in fmt or "createml" in fmt:
-        return DatasetFormat.json
-    return DatasetFormat.yolo
-
-
-# ── Viewer ────────────────────────────────────────────────────────────────────
-
-@router.get("/{dataset_id}/universal", response_model=UniversalViewerPage)
-async def get_universal_items(
-    dataset_id: str,
-    page:      int           = Query(0, ge=0),
-    page_size: int           = Query(20, ge=1, le=100),
-    split:     Optional[str] = Query(None, regex="^(train|val|test)$"),
-    class_label: Optional[str] = Query(None),
-):
-    """
-    Polymorphic dataset item viewer (UDV).
-    Supports Vision, NLP, and Tabular data via the Universal Dataset Item schema.
-    """
-    ds = await ds_reg.get_dataset(dataset_id)
-    if not ds:
-        raise HTTPException(404, f"Dataset {dataset_id!r} not found")
-    
-    # Allow viewing even if not fully imported for NLP/Tabular if files exist,
-    # but for Vision we usually need the index.
-    return await get_universal_viewer_page(dataset_id, page, page_size, split, class_label)
-
-
-@router.get("/{dataset_id}/images", response_model=ViewerPage)
-async def get_images(
-    dataset_id: str,
-    page:      int           = Query(0, ge=0),
-    page_size: int           = Query(20, ge=1, le=100),
-    split:     Optional[str] = Query(None, regex="^(train|val|test)$"),
-    class_label: Optional[str] = Query(None),
-):
-    """
-    Paginated image + annotation data for the viewer.
-    Annotations are returned in normalised [0–1] coordinates.
-    Supports filtering by split and class label.
-    """
-    ds = await ds_reg.get_dataset(dataset_id)
-    if not ds:
-        raise HTTPException(404, f"Dataset {dataset_id!r} not found")
-    if ds.status != "imported":
-        raise HTTPException(409, f"Dataset is not imported yet (status: {ds.status})")
-
-    return await get_viewer_page(dataset_id, page, page_size, split, class_label)
-
-
-@router.get("/{dataset_id}/stats", response_model=dict)
-async def get_dataset_stats(dataset_id: str):
-    """Return pre-computed class distributions and split statistics."""
-    ds = await ds_reg.get_dataset(dataset_id)
-    if not ds:
-        raise HTTPException(404, f"Dataset {dataset_id!r} not found")
-    
-    return await ds_reg.get_dataset_stats(dataset_id)
-
-
-@router.get("/{dataset_id}/image/{image_id}")
-async def serve_image(dataset_id: str, image_id: str):
-    """Serve raw image bytes for the viewer (cached by browser)."""
-    path = await resolve_image_path(dataset_id, image_id)
-    if path is None:
-        raise HTTPException(404, "Image not found or dataset not imported")
-
-    suffix = path.suffix.lower()
-    media_types = {
-        ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
-        ".png": "image/png",  ".bmp": "image/bmp",
-        ".webp": "image/webp",
-    }
-    media_type = media_types.get(suffix, "application/octet-stream")
-    return FileResponse(
-        path        = str(path),
-        media_type  = media_type,
-        headers     = {"Cache-Control": "public, max-age=86400"},
-    )
-
-
-@router.get("/{dataset_id}/annotations", response_model=dict)
-async def get_annotations_summary(dataset_id: str):
-    """Return class distribution summary from the annotations index."""
-    from database.connection import get_db
-    db = await get_db()
-    async with db.execute(
-        """SELECT label, COUNT(*) as count
-           FROM dataset_annotations
-           WHERE dataset_id=?
-           GROUP BY label
-           ORDER BY count DESC""",
-        (dataset_id,),
-    ) as cur:
-        rows = await cur.fetchall()
-    return {
-        "dataset_id": dataset_id,
-        "class_distribution": [{"label": r["label"], "count": r["count"]} for r in rows],
-        "total_annotations": sum(r["count"] for r in rows),
-    }
-
-
-# ── Star / Delete ─────────────────────────────────────────────────────────────
+# ── Star ─────────────────────────────────────────────────────────────
 
 @router.post("/{dataset_id}/star", response_model=dict)
 async def toggle_star(dataset_id: str):
@@ -386,31 +139,9 @@ async def toggle_star(dataset_id: str):
     return {"dataset_id": dataset_id, "starred": new_val}
 
 
-@router.delete("/{dataset_id}", response_model=dict)
-async def delete_dataset(
-    dataset_id: str,
-    delete_files: bool = Query(False, description="Also remove local files from disk"),
-):
-    ds = await ds_reg.get_dataset(dataset_id)
-    if not ds:
-        raise HTTPException(404, f"Dataset {dataset_id!r} not found")
-
-    if delete_files and ds.local_path:
-        import shutil
-        local = Path(ds.local_path)
-        if local.exists() and local.is_dir():
-            shutil.rmtree(str(local), ignore_errors=True)
-            log.info("dataset_files_deleted", path=str(local))
-
-    deleted = await ds_reg.delete_dataset(dataset_id)
-    await audit("dataset_deleted", {"dataset_id": dataset_id, "files_deleted": delete_files})
-    return {"deleted": deleted, "dataset_id": dataset_id}
-
-
 # ── Helper ────────────────────────────────────────────────────────────────────
 
 def _to_summary(d: Dataset) -> DatasetSummary:
-    # Use 0.0 as default health_score if stats is missing or health_score is not present
     health_score = 0.0
     try:
         if hasattr(d, 'stats') and d.stats:

api/routes/models.py

@@ -1,19 +1,15 @@
 """
-api/routes/models.py — /models REST endpoints.
+api/routes/models.py — /models REST endpoints (Cloud Registry Version).
+Provides global model discovery and metadata.
 """
 from __future__ import annotations
 
 from typing import Annotated
 
-from fastapi import APIRouter, HTTPException, Query, UploadFile, File, Form
+from fastapi import APIRouter, HTTPException, Query
 from models.model import Model
 from observability.logger import audit, get_logger
-from registry.registry import count_models, get_model, list_models
-from projects.service import get_active_project_id, import_local_model
-from projects.registry import get_project
-from pathlib import Path
-import os
-import tempfile
+from registry.registry import get_model, list_models
 
 log = get_logger("api.models")
 router = APIRouter(prefix="/models", tags=["models"])
@@ -26,98 +22,30 @@ async def index(
     framework:  Annotated[list[str] | None, Query()] = None,
     hardware:   Annotated[list[str] | None, Query()] = None,
     source:     Annotated[list[str] | None, Query()] = None,
-    downloaded: Annotated[bool | None, Query()] = None,
     sort_by:    Annotated[str, Query()] = "downloads",
     sort_dir:   Annotated[str, Query()] = "desc",
     limit:      Annotated[int, Query(ge=1, le=1000)] = 200,
     offset:     Annotated[int, Query(ge=0)] = 0,
-    project_id: Annotated[str | None, Query()] = None,
 ) -> list[Model]:
     """
-    List and search models.
-    Supports FTS search + server-side filtering.
-    Target: < 100ms for up to 5 000 models.
+    List and search models in the global registry.
     """
-    effective_project_id = project_id or await get_active_project_id()
-
     models = await list_models(
         search=search,
         tasks=task,
         frameworks=framework,
         hardware=hardware,
         sources=source,
-        downloaded=downloaded,
         sort_by=sort_by,
         sort_dir=sort_dir,
         limit=limit,
         offset=offset,
-        project_id=effective_project_id,
     )
 
-    # If we have an active project, derive cache state from its workspace.
-    # This makes "downloaded" and "local_path" reflect the *current project*.
-    if effective_project_id:
-        proj = await get_project(effective_project_id)
-        if proj:
-            project_models_dir = Path(proj.path) / "models"
-
-            updated: list[Model] = []
-            for m in models:
-                model_dir = project_models_dir / m.id
-                if model_dir.exists() and model_dir.is_dir():
-                    # Pick the first file in the model directory (best-effort).
-                    found_file: str | None = None
-                    try:
-                        for p in model_dir.rglob("*"):
-                            if p.is_file():
-                                found_file = str(p)
-                                break
-                    except Exception:
-                        found_file = None
-
-                    if found_file:
-                        updated.append(m.model_copy(update={"downloaded": True, "local_path": found_file}))
-                        continue
-
-                # Not present in this project → treat as not cached for this project.
-                updated.append(m.model_copy(update={"downloaded": False, "local_path": None}))
-
-            models = updated
-
     await audit("api_list_models", payload={"count": len(models), "search": search})
     return models
 
 
-@router.post("/import", response_model=Model)
-async def import_model(
-    name: Annotated[str, Form()],
-    task: Annotated[str, Form()],
-    framework: Annotated[str, Form()],
-    file: UploadFile = File(...),
-) -> Model:
-    """Import a local model file into the active project."""
-    # Save uploaded file to a temporary location
-    with tempfile.NamedTemporaryFile(delete=False, suffix=os.path.splitext(file.filename or "")[1]) as tmp:
-        content = await file.read()
-        tmp.write(content)
-        tmp_path = tmp.name
-
-    try:
-        model = await import_local_model(
-            name=name,
-            task=task,
-            framework=framework,
-            source_file_path=tmp_path
-        )
-        return model
-    except Exception as e:
-        log.error("model_import_failed", error=str(e))
-        raise HTTPException(status_code=500, detail=str(e))
-    finally:
-        if os.path.exists(tmp_path):
-            os.remove(tmp_path)
-
-
 @router.get("/{model_id}", response_model=Model)
 async def detail(model_id: str) -> Model:
     model = await get_model(model_id)

api/routes/projects.py

@@ -1,54 +0,0 @@
-"""api/routes/projects.py — /projects REST endpoints."""
-
-from __future__ import annotations
-
-from fastapi import APIRouter, HTTPException, Query
-
-from models.project import Project
-from observability.logger import audit
-from projects.registry import delete_project, get_project, list_projects, touch_last_opened, upsert_project
-from projects.service import set_active_project
-
-
-router = APIRouter(prefix="/projects", tags=["projects"])
-
-
-@router.get("", response_model=list[Project])
-async def projects_list(
-    limit: int = Query(200, ge=1, le=1000),
-    offset: int = Query(0, ge=0),
-) -> list[Project]:
-    projects = await list_projects(limit=limit, offset=offset)
-    await audit("api_list_projects", payload={"count": len(projects)})
-    return projects
-
-
-@router.post("", response_model=Project)
-async def projects_upsert(project: Project) -> Project:
-    # Ensure project.created_at and last_opened are set if missing
-    if not project.created_at:
-        project.created_at = datetime.now(timezone.utc).isoformat()
-    if not project.last_opened:
-        project.last_opened = datetime.now(timezone.utc).isoformat()
-        
-    await upsert_project(project)
-    await audit("api_upsert_project", payload={"project_id": project.id})
-    return project
-
-
-@router.post("/{project_id}/open", status_code=204, response_model=None)
-async def projects_open(project_id: str) -> None:
-    await touch_last_opened(project_id)
-    project = await get_project(project_id)
-    if project:
-        await set_active_project(project.id, project.path)
-    await audit("api_open_project", payload={"project_id": project_id})
-
-
-@router.delete("/{project_id}", status_code=204, response_model=None)
-async def projects_delete(project_id: str) -> None:
-    ok = await delete_project(project_id)
-    if not ok:
-        raise HTTPException(status_code=404, detail=f"Project '{project_id}' not found")
-    await audit("api_delete_project", payload={"project_id": project_id})
-

config.py

@@ -26,6 +26,7 @@ class Settings(BaseSettings):
     # ── Storage ───────────────────────────────────────────────────────
     base_dir: Path = Path(__file__).resolve().parents[1]
     data_dir: Path = base_dir / "data"
+    logs_dir: Path = data_dir / "logs"
     db_path:  Path = data_dir / "modelzoo.db"
 
     # ── Search ────────────────────────────────────────────────────────
@@ -42,6 +43,7 @@ class Settings(BaseSettings):
 
     def ensure_dirs(self) -> None:
         self.data_dir.mkdir(parents=True, exist_ok=True)
+        self.logs_dir.mkdir(parents=True, exist_ok=True)
 
 
 settings = Settings()

datasets/import_service.py

@@ -1,589 +0,0 @@
-"""
-datasets/import_service.py — Dataset Import Pipeline.
-
-Pipeline stages:
-  1. Create job record
-  2. Download dataset zip (chunked, progress-tracked)
-  3. Extract zip safely (path-traversal protected)
-  4. Detect annotation format & task type
-  5. Index images into dataset_images table
-  6. Parse & store metadata (Stats only, annotations are read-on-demand)
-  7. Update dataset stats (images, classes, size)
-  8. Mark job completed / failed
-
-All stages run as background tasks.
-Supports Roboflow, HuggingFace, and local file/folder imports.
-"""
-from __future__ import annotations
-
-import asyncio
-import hashlib
-import os
-import shutil
-import uuid
-import zipfile
-from datetime import datetime
-from pathlib import Path
-from typing import Optional, List, Dict, Any, Tuple
-
-import aiofiles
-import httpx
-from huggingface_hub import snapshot_download
-
-from config import settings
-from . import registry as ds_reg
-from .format_adapters import (
-    YOLOAdapter, COCOAdapter, VOCAdapter, CreateMLAdapter, 
-    RoboflowClassificationAdapter, NLPAdapter, TabularAdapter
-)
-from .base_adapter import DatasetAdapter
-from .annotation_parser import _img_dimensions
-from observability.logger import audit, get_logger
-from models.dataset import DatasetStatus, DatasetTask, ImportRequest, Dataset
-
-log = get_logger("import_service")
-
-ADAPTERS: List[DatasetAdapter] = [
-    YOLOAdapter(),
-    COCOAdapter(),
-    VOCAdapter(),
-    CreateMLAdapter(),
-    RoboflowClassificationAdapter(),
-    NLPAdapter(),
-    TabularAdapter(),
-]
-
-def get_adapter_for_path(path: Path) -> DatasetAdapter | None:
-    for adapter in ADAPTERS:
-        if adapter.detect(path):
-            return adapter
-    return None
-
-async def recover_stale_jobs() -> None:
-    """Cleanup dataset import jobs that were left in 'running' or 'queued' state."""
-    await ds_reg.cleanup_stale_jobs()
-
-def _dataset_path(dataset_id: str) -> Path:
-    return settings.datasets_dir / dataset_id
-
-# ── Entry Point ──────────────────────────────────────────────────────────────
-
-async def start_import(req: ImportRequest) -> str:
-    """Entry point to initiate a background import job."""
-    job_id = f"job-{uuid.uuid4().hex[:8]}"
-    
-    # Create initial job record
-    await ds_reg.update_job(
-        job_id,
-        dataset_id=req.dataset_id,
-        status="queued",
-        progress=0,
-        message="Import queued",
-        type=str(req.source)
-    )
-
-    # Launch background task
-    asyncio.create_task(_run_pipeline(job_id, req, req.dataset_name or req.dataset_id))
-    
-    return job_id
-
-
-# ── Pipeline orchestrator ────────────────────────────────────────────────────
-
-async def _run_pipeline(job_id: str, req: ImportRequest, dataset_name: str) -> None:
-    started = datetime.utcnow().isoformat()
-    await ds_reg.update_job(job_id, status="running", started_at=started, message="Starting import")
-    await ds_reg.update_dataset_status(req.dataset_id, DatasetStatus.importing, progress=0.01)
-
-    try:
-        # Stage 1 – Resolve download URL or local path
-        source_path = await _stage_acquire(job_id, req)
-
-        # Stage 2 – Extract / Prepare Directory
-        extract_dir = await _stage_extract(job_id, req.dataset_id, source_path)
-
-        # Stage 3 – Detect adapter and Task
-        await ds_reg.update_job(job_id, progress=0.55, message="Detecting dataset format...")
-        adapter = await asyncio.to_thread(get_adapter_for_path, extract_dir)
-        
-        if not adapter:
-            log.warning("no_adapter_found_generic_fallback", dataset_id=req.dataset_id)
-            image_records = await asyncio.to_thread(_scan_images_generic, req.dataset_id, extract_dir)
-            class_names = []
-            task = DatasetTask.classification
-            fmt_name = "custom"
-        else:
-            task = adapter.get_task(extract_dir)
-            fmt_name = adapter.__class__.__name__.replace("Adapter", "").lower()
-            
-            log.info("adapter_detected", job_id=job_id, format=fmt_name, task=task)
-            await ds_reg.update_job(job_id, progress=0.60, message=f"Parsing {fmt_name.upper()} {task.upper()}")
-
-            # Stage 4 – Parse Metadata & Annotations (Streaming)
-            class_names = await asyncio.to_thread(adapter.get_class_names, extract_dir)
-            image_records = []
-            all_annotations = []
-            
-            # Health metrics tracking
-            hashes = {} # hash -> filename
-            duplicates = 0
-            empty_images = 0
-            total_ann_count = 0
-            
-            for img_rec, anns in adapter.iter_items(req.dataset_id, extract_dir):
-                # Duplicate detection via MD5 hash
-                abs_path = extract_dir / img_rec["rel_path"]
-                if abs_path.exists():
-                    img_hash = _calculate_hash(abs_path)
-                    if img_hash in hashes:
-                        duplicates += 1
-                        img_rec["metadata"] = json.dumps({"is_duplicate": True, "original": hashes[img_hash]})
-                    else:
-                        hashes[img_hash] = img_rec["filename"]
-                
-                if not anns:
-                    empty_images += 1
-                
-                total_ann_count += len(anns)
-                image_records.append(img_rec)
-                all_annotations.extend(anns)
-
-        if not image_records:
-            raise ValueError(f"No valid data files found in {extract_dir}")
-
-        # Stage 5 – Indexing
-        await ds_reg.update_job(job_id, progress=0.80, message=f"Indexing {len(image_records)} items")
-        await ds_reg.index_images(req.dataset_id, image_records)
-        
-        if all_annotations:
-            await ds_reg.update_job(job_id, progress=0.85, message=f"Indexing {len(all_annotations)} annotations")
-            await ds_reg.bulk_insert_annotations(all_annotations)
-
-        # Stage 6 – Stats & Health Analysis
-        size_bytes = await asyncio.to_thread(_dir_size, extract_dir)
-        
-        # Calculate Health Score (0-100)
-        # Factors: duplicates, empty images (for detection), class balance (TODO)
-        score = 100.0
-        if len(image_records) > 0:
-            dup_penalty = (duplicates / len(image_records)) * 50
-            empty_penalty = (empty_images / len(image_records)) * 20 if task == DatasetTask.detection else 0
-            score = max(0.0, 100.0 - dup_penalty - empty_penalty)
-
-        stats_payload = {
-            "image_count": len(image_records),
-            "annotation_count": total_ann_count,
-            "class_count": len(class_names),
-            "empty_images": empty_images,
-            "duplicate_count": duplicates,
-            "health_score": round(score, 1),
-            "avg_objects": round(total_ann_count / len(image_records), 2) if image_records else 0
-        }
-
-        await ds_reg.update_dataset_stats(
-            req.dataset_id, 
-            len(image_records), 
-            len(class_names), 
-            class_names, 
-            size_bytes,
-            stats=stats_payload
-        )
-        await ds_reg.update_dataset_task(req.dataset_id, task)
-
-        # Cleanup temp zip if applicable
-        if source_path.is_file() and source_path.suffix.lower() == ".zip" and "_tmp" in str(source_path):
-            source_path.unlink(missing_ok=True)
-
-        # Stage 7 – Project Linking (Integration point)
-        local_path = str(extract_dir)
-        from projects.service import link_dataset_to_active_project
-        project_ds_root = await link_dataset_to_active_project(req.dataset_id, local_path)
-        final_local_path = str(project_ds_root) if project_ds_root and project_ds_root.exists() else local_path
-
-        # Completion
-        await ds_reg.update_job(
-            job_id, status="completed", progress=1.0,
-            message="Import complete", ended_at=datetime.utcnow().isoformat(),
-        )
-        await ds_reg.update_dataset_status(req.dataset_id, DatasetStatus.imported, progress=1.0, local_path=final_local_path)
-        await audit("dataset_import_complete", {"job_id": job_id, "path": final_local_path}, job_id=job_id)
-        log.info("import_complete", job_id=job_id, dataset_id=req.dataset_id)
-
-    except asyncio.CancelledError:
-        await _fail_job(job_id, req.dataset_id, "Import cancelled by user or system")
-        raise
-    except Exception as exc:
-        log.error("import_failed", job_id=job_id, error=str(exc))
-        await _fail_job(job_id, req.dataset_id, str(exc))
-        await audit("dataset_import_error", {"job_id": job_id, "error": str(exc)}, job_id=job_id, level="error")
-
-
-async def _fail_job(job_id: str, dataset_id: str, error: str) -> None:
-    await ds_reg.update_job(
-        job_id, status="failed", error=error,
-        ended_at=datetime.utcnow().isoformat(),
-        message="Import failed",
-    )
-    await ds_reg.update_dataset_status(dataset_id, DatasetStatus.failed, progress=0.0)
-
-
-# ── Stage 1: Acquire source ──────────────────────────────────────────────────
-
-async def _stage_acquire(job_id: str, req: ImportRequest) -> Path:
-    """Resolves the source (Download URL, HF Repo, or Local Path)."""
-    await ds_reg.update_job(job_id, progress=0.05, message="Acquiring source...")
-
-    if req.source in ("roboflow", "roboflow_curl"):
-        return await _acquire_roboflow(job_id, req)
-    
-    if req.source == "huggingface":
-        return await _acquire_huggingface(job_id, req)
-    
-    if req.source == "local":
-        return await _acquire_local(job_id, req)
-
-    raise ValueError(f"Unsupported source provider: {req.source}")
-
-
-async def _acquire_roboflow(job_id: str, req: ImportRequest) -> Path:
-    """Specialized Roboflow downloader using SDK or direct link."""
-    # Attempt SDK first (more reliable for Universe)
-    try:
-        from roboflow import Roboflow
-        api_key = req.roboflow_key or (req.headers.get("Authorization") if req.headers else None)
-        if api_key and "Bearer " in str(api_key):
-            api_key = api_key.split("Bearer ")[-1].strip()
-        
-        if api_key and req.roboflow_workspace and req.roboflow_project:
-            rf = Roboflow(api_key=api_key)
-            project = rf.workspace(req.roboflow_workspace).project(req.roboflow_project)
-            version_obj = project.version(req.roboflow_version or 1)
-            
-            tmp_target = DATASETS_ROOT / "_tmp" / f"rf-{uuid.uuid4().hex[:8]}"
-            await ds_reg.update_job(job_id, progress=0.10, message="Downloading via Roboflow SDK...")
-            
-            # Threaded SDK call
-            await asyncio.to_thread(
-                version_obj.download, 
-                _format_to_rf_slug(str(req.format)), 
-                location=str(tmp_target)
-            )
-            return tmp_target
-    except Exception as e:
-        log.warning("roboflow_sdk_fallback", error=str(e))
-
-    # Fallback to direct HTTP download
-    url = req.download_url
-    if not url and req.source == "roboflow":
-        from adapters.roboflow_adapter import RoboflowAdapter
-        url = await RoboflowAdapter.get_download_url(
-            api_key=req.roboflow_key,
-            workspace=req.roboflow_workspace,
-            project_id=req.roboflow_project,
-            version=req.roboflow_version,
-            export_format=_format_to_rf_slug(str(req.format)),
-        )
-    
-    if not url:
-        raise ValueError("Could not resolve Roboflow download URL")
-        
-    return await _download_zip(job_id, req.dataset_id, url, req.headers)
-
-
-async def _acquire_huggingface(job_id: str, req: ImportRequest) -> Path:
-    if not req.hf_dataset_id:
-        raise ValueError("hf_dataset_id is missing")
-    
-    dest_dir = _dataset_path(req.dataset_id)
-    dest_dir.mkdir(parents=True, exist_ok=True)
-    
-    await ds_reg.update_job(job_id, progress=0.10, message=f"Cloning {req.hf_dataset_id} from HF...")
-    
-    await asyncio.to_thread(
-        snapshot_download,
-        repo_id=req.hf_dataset_id,
-        repo_type="dataset",
-        local_dir=str(dest_dir),
-        token=settings.hf_token,
-        local_dir_use_symlinks=False
-    )
-    return dest_dir
-
-
-async def _acquire_local(job_id: str, req: ImportRequest) -> Path:
-    if not req.local_path:
-        raise ValueError("local_path is missing for local import")
-    
-    path = Path(os.path.normpath(req.local_path.strip().strip('"').strip("'")))
-    if not path.exists():
-        raise FileNotFoundError(f"Local path does not exist: {path}")
-    
-    return path
-
-
-# ── Stage 2: Extraction ──────────────────────────────────────────────────────
-
-async def _stage_extract(job_id: str, dataset_id: str, source_path: Path) -> Path:
-    dest = _dataset_path(dataset_id)
-    dest.mkdir(parents=True, exist_ok=True)
-
-    if source_path.is_dir():
-        if source_path == dest:
-            return dest
-        await ds_reg.update_job(job_id, progress=0.45, message="Copying local files...")
-        await asyncio.to_thread(_copy_dir_contents, source_path, dest)
-        return dest
-
-    # It's a zip
-    await ds_reg.update_job(job_id, progress=0.45, message="Extracting archive...")
-    await ds_reg.update_dataset_status(dataset_id, DatasetStatus.extracting, progress=0.45)
-    await asyncio.to_thread(_safe_extract, source_path, dest)
-    return dest
-
-
-# ── Stage 3: Parsing (Memory-Safe) ───────────────────────────────────────────
-
-def _heuristic_task_detection(fmt: str, root: Path) -> DatasetTask:
-    """Improved task detection based on file content."""
-    if fmt == "csv":
-        return DatasetTask.nlp
-    
-    # Check for segmentation in COCO
-    if fmt == "coco":
-        # Sample first few lines of JSON if possible or check file size
-        return DatasetTask.segmentation # Heuristic: most modern COCO use cases
-        
-    if fmt in ("yolo", "voc"):
-        return DatasetTask.detection
-        
-    return DatasetTask.classification
-
-
-def _parse_yolo(dataset_id: str, root: Path) -> Tuple[List[str], List[Tuple[Dict, List[Dict]]]]:
-    class_map = YOLOParser.load_class_map(root)
-    results = []
-    # Generator approach to keep memory low
-    for rel_path, image_id, split, anns in YOLOParser.iter_dataset(root, dataset_id, class_map):
-        abs_path = root / rel_path
-        w, h = _img_dimensions(abs_path)
-        img_rec = {
-            "id": image_id, "filename": Path(rel_path).name,
-            "rel_path": str(rel_path), "width": w, "height": h,
-            "split": split, "ann_count": len(anns),
-        }
-        results.append((img_rec, anns))
-    return class_map, results
-
-
-def _parse_coco(dataset_id: str, root: Path) -> Tuple[List[str], List[Tuple[Dict, List[Dict]]]]:
-    ann_files = COCOParser.find_annotation_files(root)
-    all_classes: list[str] = []
-    results = []
-    for ann_file in ann_files:
-        classes, coco_results = COCOParser.parse_file(ann_file, dataset_id)
-        all_classes = list(dict.fromkeys(all_classes + classes))
-        for rel_path, image_id, split, anns in coco_results:
-            abs_path = root / rel_path
-            w, h = _img_dimensions(abs_path)
-            img_rec = {
-                "id": image_id, "filename": Path(rel_path).name,
-                "rel_path": str(rel_path), "width": w, "height": h,
-                "split": split, "ann_count": len(anns),
-            }
-            results.append((img_rec, anns))
-    return all_classes, results
-
-
-def _parse_voc(dataset_id: str, root: Path) -> Tuple[List[str], List[Tuple[Dict, List[Dict]]]]:
-    class_set = set()
-    results = []
-    for rel_path, image_id, split, w, h, anns in VOCParser.iter_dataset(root, dataset_id):
-        img_rec = {
-            "id": image_id, "filename": Path(rel_path).name,
-            "rel_path": str(rel_path), "width": w, "height": h,
-            "split": split, "ann_count": len(anns),
-        }
-        results.append((img_rec, anns))
-        for ann in anns:
-            class_set.add(ann["label"])
-    return sorted(list(class_set)), results
-
-
-def _parse_csv(dataset_id: str, root: Path) -> Tuple[List[str], List[Tuple[Dict, List[Dict]]]]:
-    all_classes = set()
-    results = []
-    for csv_path in root.rglob("*.csv"):
-        anns = CSVParser.parse_file(csv_path, dataset_id)
-        # For CSV, each annotation is a row. We group by text entry id (image_id)
-        anns_by_id: Dict[str, List[Dict]] = {}
-        for ann in anns:
-            all_classes.add(ann["label"])
-            anns_by_id.setdefault(ann["image_id"], []).append(ann)
-        
-        for text_id, grouped_anns in anns_by_id.items():
-            img_rec = {
-                "id": text_id, "filename": csv_path.name,
-                "rel_path": str(csv_path.relative_to(root)),
-                "width": 0, "height": 0, "split": "train", "ann_count": len(grouped_anns),
-            }
-            results.append((img_rec, grouped_anns))
-    return sorted(list(all_classes)), results
-
-
-def _parse_txt(dataset_id: str, root: Path) -> Tuple[List[str], List[Tuple[Dict, List[Dict]]]]:
-    from datasets.annotation_parser import RoboflowTXTParser
-    results = []
-    class_set = set()
-    
-    for rel_path, image_id, split, anns in RoboflowTXTParser.iter_dataset(root, dataset_id):
-        abs_path = root / rel_path
-        w, h = _img_dimensions(abs_path)
-        img_rec = {
-            "id": image_id, "filename": Path(rel_path).name,
-            "rel_path": str(rel_path), "width": w, "height": h,
-            "split": split, "ann_count": len(anns),
-        }
-        results.append((img_rec, anns))
-        for ann in anns:
-            class_set.add(ann["label"])
-            
-    return sorted(list(class_set)), results
-
-
-def _parse_generic_folder(dataset_id: str, root: Path) -> Tuple[List[str], List[Tuple[Dict, List[Dict]]]]:
-    """
-    Enhanced generic folder parser. Supports:
-    1. root/class_name/img.jpg
-    2. root/train/class_name/img.jpg
-    3. root/images/img.jpg
-    """
-    results = []
-    class_set = set()
-    exts = {".jpg", ".jpeg", ".png", ".bmp", ".webp", ".tif", ".tiff"}
-    
-    # Structural keywords to ignore as classes
-    ignore = {"images", "labels", "train", "val", "test", "validation", "training", "valid", "testing", "unknown", "annotations"}
-
-    for img_path in sorted(root.rglob("*")):
-        if img_path.suffix.lower() not in exts:
-            continue
-            
-        rel_path = img_path.relative_to(root)
-        parts = rel_path.parts
-        
-        # Heuristic for class detection
-        label = "unknown"
-        split = "train"
-        
-        # Detect split if first folder is a split keyword
-        if parts[0].lower() in ignore and len(parts) > 1:
-            if parts[0].lower() in ("train", "training"): split = "train"
-            elif parts[0].lower() in ("val", "valid", "validation"): split = "val"
-            elif parts[0].lower() in ("test", "testing"): split = "test"
-
-            # Check if next part is class name
-            if len(parts) > 2 and parts[1].lower() not in ignore:
-                label = parts[1]
-            elif len(parts) > 1 and parts[1].lower() not in ignore:
-                label = parts[1]
-        elif len(parts) > 1 and parts[0].lower() not in ignore:
-            label = parts[0]
-
-        anns = []
-        if label != "unknown":
-            class_set.add(label)
-            image_id = f"img-{uuid.uuid4().hex[:12]}"
-            # Create a virtual annotation for classification
-            from datasets.annotation_parser import _make_ann
-            anns.append(_make_ann(image_id, dataset_id, label, ann_type="classification"))
-        else:
-            image_id = f"img-{uuid.uuid4().hex[:12]}"
-            
-        w, h = _img_dimensions(img_path)
-        img_rec = {
-            "id": image_id,
-            "filename": img_path.name,
-            "rel_path": str(rel_path),
-            "width": w, "height": h,
-            "split": split,
-            "ann_count": len(anns),
-        }
-        results.append((img_rec, anns))
-        
-    return sorted(list(class_set)), results
-
-
-# ── Utilities ────────────────────────────────────────────────────────────────
-
-async def _download_zip(job_id: str, dataset_id: str, url: str, custom_headers: dict = None) -> Path:
-    tmp_dir = DATASETS_ROOT / "_tmp"
-    tmp_dir.mkdir(parents=True, exist_ok=True)
-    zip_path = tmp_dir / f"{dataset_id}-{uuid.uuid4().hex[:8]}.zip"
-
-    headers = {
-        "User-Agent": "Mozilla/5.0 (MLForge Workbench)",
-        "Accept": "application/zip, application/octet-stream, */*",
-    }
-    if custom_headers: headers.update(custom_headers)
-
-    async with httpx.AsyncClient(follow_redirects=True, timeout=600.0, headers=headers) as client:
-        async with client.stream("GET", url) as resp:
-            resp.raise_for_status()
-            total = int(resp.headers.get("content-length", 0)) or None
-            downloaded = 0
-            async with aiofiles.open(zip_path, "wb") as f:
-                async for chunk in resp.aiter_bytes(chunk_size=settings.download_chunk_size):
-                    await f.write(chunk)
-                    downloaded += len(chunk)
-                    if total:
-                        pct = 0.10 + (downloaded / total) * 0.35 # 10% -> 45%
-                        await ds_reg.update_job(job_id, progress=round(pct, 3), message=f"Downloading: {_fmt_bytes(downloaded)} / {_fmt_bytes(total)}")
-
-    return zip_path
-
-
-def _safe_extract(zip_path: Path, dest: Path) -> None:
-    with zipfile.ZipFile(str(zip_path), "r") as zf:
-        for member in zf.namelist():
-            if os.path.isabs(member) or ".." in Path(member).parts: continue
-            zf.extract(member, str(dest))
-
-
-def _copy_dir_contents(src: Path, dest: Path) -> None:
-    for item in src.iterdir():
-        s, d = src / item.name, dest / item.name
-        if s.is_dir(): shutil.copytree(s, d, dirs_exist_ok=True)
-        else: shutil.copy2(s, d)
-
-
-def _scan_images_generic(dataset_id: str, root: Path) -> list[dict]:
-    records = []
-    exts = {".jpg", ".jpeg", ".png", ".bmp", ".webp"}
-    for img_path in sorted(root.rglob("*")):
-        if img_path.suffix.lower() in exts:
-            w, h = _img_dimensions(img_path)
-            records.append({
-                "id": f"img-{uuid.uuid4().hex[:12]}",
-                "filename": img_path.name,
-                "rel_path": str(img_path.relative_to(root)),
-                "width": w, "height": h, "split": "train", "ann_count": 0,
-            })
-    return records
-
-
-def _dir_size(path: Path) -> int:
-    return sum(f.stat().st_size for f in path.rglob("*") if f.is_file())
-
-
-def _fmt_bytes(n: int) -> str:
-    for unit in ("B", "KB", "MB", "GB", "TB"):
-        if n < 1024: return f"{n:.1f} {unit}"
-        n /= 1024
-    return f"{n:.1f} PB"
-
-
-def _format_to_rf_slug(fmt: str) -> str:
-    return {"yolo": "yolov8", "coco": "coco", "voc": "voc"}.get(fmt, "yolov8")
-
-def _format_to_rf_slug(fmt: str) -> str:
-    return {"yolo": "yolov8", "coco": "coco", "voc": "voc"}.get(fmt, "yolov8")

datasets/viewer_service.py

@@ -1,320 +0,0 @@
-"""
-datasets/viewer_service.py — Dataset Viewer Service.
-
-Provides paginated image + annotation serving for the Dataset Viewer UI.
-All paths are resolved relative to the dataset's local_path for security.
-"""
-from __future__ import annotations
-
-from pathlib import Path
-
-from datasets import registry as ds_reg
-from models.dataset import (
-    Annotation, AnnotationType, BoundingBox, Dataset,
-    ImageRecord, ViewerPage, DatasetFormat
-)
-from datasets.annotation_parser import YOLOParser, COCOParser, VOCParser, CSVParser
-from observability.logger import get_logger
-
-log = get_logger("viewer_service")
-
-
-from .format_adapters import NLPAdapter, TabularAdapter
-from models.dataset import UniversalViewerPage, UniversalDatasetItem, UniversalAnnotation, DatasetContentType, DatasetTask
-
-async def get_universal_viewer_page(
-    dataset_id: str,
-    page: int = 0,
-    page_size: int = 20,
-    split: str | None = None,
-    class_label: str | None = None,
-) -> UniversalViewerPage:
-    """Polymorphic viewer endpoint that adapts based on dataset task."""
-    ds = await ds_reg.get_dataset(dataset_id)
-    if not ds:
-        raise ValueError("Dataset not found")
-        
-    ds_root = Path(ds.local_path) if ds.local_path else None
-    
-    # 1. Vision Tasks (Detection, Seg, Pose) -> Use existing image-centric logic
-    if ds.task in (DatasetTask.detection, DatasetTask.segmentation, DatasetTask.keypoints):
-        # We wrap the existing get_viewer_page and transform to UniversalDatasetItem
-        old_page = await get_viewer_page(dataset_id, page, page_size, split, class_label)
-        
-        items = []
-        for img in old_page.images:
-            items.append(UniversalDatasetItem(
-                id=img.image_id,
-                content_type=DatasetContentType.image,
-                filename=img.filename,
-                metadata={"width": img.width, "height": img.height, "split": img.split},
-                annotations=[
-                    UniversalAnnotation(
-                        label=ann.label,
-                        type=ann.type.value if hasattr(ann.type, 'value') else str(ann.type),
-                        bbox=[ann.bbox.x, ann.bbox.y, ann.bbox.width, ann.bbox.height] if ann.bbox else None,
-                        segmentation=ann.segmentation,
-                        keypoints=ann.keypoints,
-                        confidence=ann.confidence,
-                        metadata=ann.metadata
-                    ) for ann in img.annotations
-                ]
-            ))
-        
-        return UniversalViewerPage(
-            dataset_id=dataset_id,
-            page=page,
-            page_size=page_size,
-            total=old_page.total,
-            total_pages=old_page.total_pages,
-            items=items
-        )
-
-    # 2. NLP Tasks (CSV, JSONL)
-    elif ds.task == DatasetTask.nlp and ds_root:
-        adapter = NLPAdapter()
-        total, items = await adapter.get_items(ds_root, page, page_size)
-        total_pages = max(1, (total + page_size - 1) // page_size)
-        return UniversalViewerPage(
-            dataset_id=dataset_id,
-            page=page,
-            page_size=page_size,
-            total=total,
-            total_pages=total_pages,
-            items=items
-        )
-
-    # 3. Tabular Tasks (CSV, Parquet)
-    elif ds.task == DatasetTask.tabular and ds_root:
-        adapter = TabularAdapter()
-        total, items = await adapter.get_items(ds_root, page, page_size)
-        total_pages = max(1, (total + page_size - 1) // page_size)
-        return UniversalViewerPage(
-            dataset_id=dataset_id,
-            page=page,
-            page_size=page_size,
-            total=total,
-            total_pages=total_pages,
-            items=items
-        )
-
-    # Fallback / Empty
-    return UniversalViewerPage(
-        dataset_id=dataset_id,
-        page=page,
-        page_size=page_size,
-        total=0,
-        total_pages=0,
-        items=[]
-    )
-
-async def get_viewer_page(
-    dataset_id: str,
-    page: int = 0,
-    page_size: int = 20,
-    split: str | None = None,
-    class_label: str | None = None,
-) -> ViewerPage:
-    """
-    Return a paginated viewer page for the dataset.
-    Images come from the index; annotations are loaded per-image.
-    """
-    if page_size > 100:
-        page_size = 100   # cap to prevent huge payloads
-
-    total, image_rows = await ds_reg.get_image_page(dataset_id, page, page_size, split, class_label)
-    ds = await ds_reg.get_dataset(dataset_id)
-    
-    # Check if we have an active project and if the dataset exists there
-    from projects.service import get_active_project_path
-    project_path = await get_active_project_path()
-    
-    # Dynamically load annotations from database first, fallback to filesystem if needed
-    image_ids = [row["id"] for row in image_rows]
-    dynamic_anns: dict[str, list[Annotation]] = {img_id: [] for img_id in image_ids}
-
-    # 1. Try loading from DB index (Authoritative for analytics)
-    try:
-        from database.connection import get_db
-        db = await get_db()
-        # Fetch all annotations for these images in one go
-        placeholders = ",".join(["?"] * len(image_ids))
-        async with db.execute(
-            f"SELECT * FROM dataset_annotations WHERE image_id IN ({placeholders})",
-            image_ids
-        ) as cur:
-            rows = await cur.fetchall()
-            for r in rows:
-                dynamic_anns[r["image_id"]].append(_row_to_annotation(dict(r)))
-    except Exception as e:
-        log.warning("db_annotation_read_failed", error=str(e), dataset_id=dataset_id)
-
-    # 2. Fallback to filesystem if no annotations found in DB and we have a path
-    # This maintains compatibility with old datasets or specific live-read needs
-    if all(not anns for anns in dynamic_anns.values()) and ds and ds.local_path:
-        ds_root = Path(ds.local_path)
-        # Use ds.local_path directly as it is now authoritative project-local path
-        # Fallback to global removed per user request
-
-        fmt = ds.format.value if hasattr(ds.format, 'value') else str(ds.format)
-        
-        try:
-            if fmt == DatasetFormat.yolo.value or fmt == "yolo":
-                class_map = YOLOParser.load_class_map(ds_root)
-                for row in image_rows:
-                    rel_path = Path(row["rel_path"])
-                    # For YOLO, the label file is usually in a parallel 'labels' folder
-                    # or in the same folder as the image.
-                    # Roboflow structure: train/images/img.jpg -> train/labels/img.txt
-                    parts = list(rel_path.parts)
-                    
-                    label_rel = None
-                    if "images" in parts:
-                        idx = parts.index("images")
-                        parts_labels = list(parts)
-                        parts_labels[idx] = "labels"
-                        label_rel = Path(*parts_labels).with_suffix(".txt")
-                    
-                    # Fallback: same folder
-                    label_same_folder = rel_path.with_suffix(".txt")
-                    
-                    for cand_rel in [label_rel, label_same_folder]:
-                        if not cand_rel: continue
-                        label_file = ds_root / cand_rel
-                        if label_file.exists():
-                            anns = YOLOParser.parse_file(label_file, row["id"], ds.id, class_map)
-                            dynamic_anns[row["id"]] = [_row_to_annotation(a) for a in anns]
-                            break
-
-            elif fmt == DatasetFormat.coco.value or fmt == "coco":
-                jsons = COCOParser.find_annotation_files(ds_root)
-                img_map = {row["filename"]: row["id"] for row in image_rows}
-                for jf in jsons:
-                    _, parsed = COCOParser.parse_file(jf, ds.id)
-                    for p_rel, _, _, anns in parsed:
-                        fname = Path(p_rel).name
-                        if fname in img_map:
-                            img_id = img_map[fname]
-                            dynamic_anns[img_id].extend([_row_to_annotation(a) for a in anns])
-
-            elif fmt == DatasetFormat.voc.value or fmt == "voc":
-                for row in image_rows:
-                    img_abs = ds_root / row["rel_path"]
-                    xml_candidates = [img_abs.with_suffix(".xml")]
-                    parts = list(Path(row["rel_path"]).parts)
-                    if "JPEGImages" in parts:
-                        idx = parts.index("JPEGImages")
-                        parts[idx] = "Annotations"
-                        xml_candidates.append(ds_root.joinpath(*parts).with_suffix(".xml"))
-                    
-                    for cand in xml_candidates:
-                        if cand.exists():
-                            _, _, _, anns = VOCParser.parse_file(cand, row["id"], ds.id)
-                            dynamic_anns[row["id"]] = [_row_to_annotation(a) for a in anns]
-                            break
-
-            elif fmt == "csv":
-                for row in image_rows:
-                    csv_path = ds_root / row["rel_path"]
-                    if csv_path.exists():
-                        # For CSV/NLP, we might need a more specific way to find the exact row,
-                        # but for now we reload the file or use a cached version.
-                        # Since get_viewer_page is paginated, we'll parse the file.
-                        anns = CSVParser.parse_file(csv_path, ds.id)
-                        # Find the annotation matching this "image_id" (which is the text entry id)
-                        matching_anns = [a for a in anns if a["image_id"] == row["id"]]
-                        dynamic_anns[row["id"]] = [_row_to_annotation(a) for a in matching_anns]
-
-        except Exception as e:
-            log.error("dynamic_annotation_read_failed", error=str(e), dataset_id=dataset_id)
-
-    images: list[ImageRecord] = []
-    for row in image_rows:
-        annotations = dynamic_anns.get(row["id"], [])
-        images.append(ImageRecord(
-            image_id    = row["id"],
-            filename    = row["filename"],
-            width       = row["width"],
-            height      = row["height"],
-            path        = row["rel_path"],
-            annotations = annotations,
-            split       = row["split"],
-        ))
-
-    total_pages = max(1, (total + page_size - 1) // page_size)
-
-    return ViewerPage(
-        dataset_id  = dataset_id,
-        page        = page,
-        page_size   = page_size,
-        total       = total,
-        total_pages = total_pages,
-        images      = images,
-    )
-
-
-def _row_to_annotation(row: dict) -> Annotation:
-    bbox = None
-    if row.get("bbox_x") is not None:
-        bbox = BoundingBox(
-            x         = row["bbox_x"],
-            y         = row["bbox_y"],
-            width     = row["bbox_w"],
-            height    = row["bbox_h"],
-            normalised = bool(row.get("normalised", 1)),
-        )
-    
-    segmentation = None
-    if row.get("segmentation"):
-        try:
-            import json
-            segmentation = json.loads(row["segmentation"])
-        except:
-            pass
-
-    return Annotation(
-        label        = row["label"],
-        bbox         = bbox,
-        segmentation = segmentation,
-        confidence   = row.get("confidence"),
-        area         = row.get("area"),
-        type         = AnnotationType(row.get("ann_type", "detection")),
-    )
-
-
-async def resolve_image_path(dataset_id: str, image_id: str) -> Path | None:
-    """
-    Resolve the absolute filesystem path for an image.
-    Prioritizes the active project's dataset folder, falling back to the global cache.
-    Returns None if dataset not imported or image not found.
-    """
-    ds = await ds_reg.get_dataset(dataset_id)
-    if ds is None or not ds.local_path:
-        return None
-
-    base_root = Path(ds.local_path)
-    # ds.local_path is now authoritative project-local path
-    # Fallback removed per user request
-
-    from database.connection import get_db
-    db = await get_db()
-    async with db.execute(
-        "SELECT rel_path FROM dataset_images WHERE id=? AND dataset_id=?",
-        (image_id, dataset_id),
-    ) as cur:
-        row = await cur.fetchone()
-    if not row:
-        return None
-
-    abs_path = base_root / row["rel_path"]
-    if not abs_path.exists():
-        return None
-
-    # Security: ensure path is under base_root
-    try:
-        abs_path.resolve().relative_to(base_root.resolve())
-    except ValueError:
-        log.warning("path_traversal_attempt", dataset_id=dataset_id, image_id=image_id)
-        return None
-
-    return abs_path

models/project.py

@@ -1,15 +0,0 @@
-"""models/project.py — Pydantic domain model for workspace projects."""
-
-from __future__ import annotations
-
-from pydantic import BaseModel
-
-
-class Project(BaseModel):
-    id: str
-    name: str
-    path: str
-    created_at: str
-    last_opened: str
-    status: str = "idle"
-

projects/registry.py

@@ -0,0 +1,73 @@
+"""projects/registry.py — Project persistence in SQLite."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from database.connection import get_db
+from models.project import Project
+
+
+async def upsert_project(project: Project) -> None:
+    db = await get_db()
+    await db.execute(
+        """INSERT INTO projects (id, name, path, created_at, last_opened, status)
+           VALUES (?,?,?,?,?,?)
+           ON CONFLICT(id) DO UPDATE SET
+             name=excluded.name,
+             path=excluded.path,
+             last_opened=excluded.last_opened,
+             status=excluded.status
+        """,
+        (
+            project.id,
+            project.name,
+            project.path,
+            project.created_at,
+            project.last_opened,
+            project.status,
+        ),
+    )
+    await db.commit()
+
+
+async def list_projects(limit: int = 200, offset: int = 0) -> list[Project]:
+    db = await get_db()
+    async with db.execute(
+        """SELECT id, name, path, created_at, last_opened, status
+           FROM projects
+           ORDER BY datetime(last_opened) DESC
+           LIMIT ? OFFSET ?
+        """,
+        (limit, offset),
+    ) as cur:
+        rows = await cur.fetchall()
+    return [Project(**dict(r)) for r in rows]
+
+
+async def delete_project(project_id: str) -> bool:
+    db = await get_db()
+    cur = await db.execute("DELETE FROM projects WHERE id = ?", (project_id,))
+    await db.commit()
+    return (cur.rowcount or 0) > 0
+
+
+async def get_project(project_id: str) -> Project | None:
+    db = await get_db()
+    async with db.execute(
+        "SELECT id, name, path, created_at, last_opened, status FROM projects WHERE id = ?",
+        (project_id,),
+    ) as cur:
+        row = await cur.fetchone()
+    return Project(**dict(row)) if row else None
+
+
+async def touch_last_opened(project_id: str) -> None:
+    db = await get_db()
+    now = datetime.now(timezone.utc).isoformat()
+    await db.execute(
+        "UPDATE projects SET last_opened = ? WHERE id = ?",
+        (now, project_id),
+    )
+    await db.commit()
+

projects/service.py

@@ -0,0 +1,186 @@
+"""
+projects/service.py — Active project session + model workspace linking.
+
+Tracks which project is currently open (via the `session` DB table) and
+copies freshly-downloaded model files into the project's models/ folder
+so the benchmark engine and other workspaces can locate them.
+"""
+from __future__ import annotations
+
+import os
+import shutil
+import uuid
+from pathlib import Path
+from datetime import datetime, timezone
+
+from database.connection import get_db
+from observability.logger import get_logger, audit
+from models.model import Model, ModelMetrics
+from registry.registry import upsert_model
+
+log = get_logger("projects.service")
+
+
+async def import_local_model(
+    name: str,
+    task: str,
+    framework: str,
+    source_file_path: str
+) -> Model:
+    """Import a local model file into the active project."""
+    project_id = await get_active_project_id()
+    project_path = await get_active_project_path()
+    
+    if not project_id or not project_path:
+        raise ValueError("No active project found. Please open a project first.")
+
+    src = Path(source_file_path)
+    if not src.exists():
+        raise FileNotFoundError(f"Source model file not found: {source_file_path}")
+
+    # Create destination directory in project
+    model_id = f"local-{uuid.uuid4().hex[:12]}"
+    dest_dir = Path(project_path) / "models" / model_id
+    dest_dir.mkdir(parents=True, exist_ok=True)
+    
+    dest_path = dest_dir / src.name
+    shutil.copy2(src, dest_path)
+    
+    # Calculate size
+    size_bytes = dest_path.stat().st_size
+    size_label = f"{size_bytes / (1024*1024):.1f} MB" if size_bytes > 1024*1024 else f"{size_bytes / 1024:.1f} KB"
+
+    # Create model entry
+    now = datetime.now(timezone.utc).isoformat()
+    model = Model(
+        id=model_id,
+        name=name,
+        task=task,
+        framework=framework,
+        source="local",
+        provider="Local Import",
+        size=size_bytes,
+        size_label=size_label,
+        local_path=str(dest_path),
+        project_id=project_id,
+        downloaded=True,
+        status="cached",
+        created_at=now,
+        updated_at=now,
+        metrics=ModelMetrics()
+    )
+    
+    await upsert_model(model)
+    await audit("model_imported_locally", model_id=model_id, payload={"name": name, "project_id": project_id})
+    log.info("model_imported_locally", model_id=model_id, name=name, path=str(dest_path))
+    
+    return model
+
+
+# ── Session helpers ───────────────────────────────────────────────────────────
+
+async def set_active_project(project_id: str, project_path: str) -> None:
+    """Persist the currently open project in the session table."""
+    db = await get_db()
+    await db.execute(
+        "INSERT INTO session (key, value) VALUES ('active_project_id', ?)"
+        " ON CONFLICT(key) DO UPDATE SET value=excluded.value",
+        (project_id,),
+    )
+    await db.execute(
+        "INSERT INTO session (key, value) VALUES ('active_project_path', ?)"
+        " ON CONFLICT(key) DO UPDATE SET value=excluded.value",
+        (project_path,),
+    )
+    await db.commit()
+    log.info("active_project_set", project_id=project_id, path=project_path)
+
+
+async def get_active_project_id() -> str | None:
+    """Return the ID of the currently open project, or None."""
+    db = await get_db()
+    async with db.execute(
+        "SELECT value FROM session WHERE key = 'active_project_id'"
+    ) as cur:
+        row = await cur.fetchone()
+    return row["value"] if row else None
+
+
+async def get_active_project_path() -> str | None:
+    """Return the filesystem path of the currently open project, or None."""
+    db = await get_db()
+    async with db.execute(
+        "SELECT value FROM session WHERE key = 'active_project_path'"
+    ) as cur:
+        row = await cur.fetchone()
+    return row["value"] if row else None
+
+
+# ── Workspace model linking ───────────────────────────────────────────────────
+
+async def link_model_to_active_project(model_id: str, source_path: str) -> None:
+    """Copy the downloaded model file into the active project's models/ folder.
+
+    This is a best-effort operation — if no project is open, or if the copy
+    fails for any reason, we log and continue rather than failing the download.
+    """
+    project_path = await get_active_project_path()
+    if not project_path:
+        log.debug("link_model_skipped_no_project", model_id=model_id)
+        return
+
+    src = Path(source_path)
+    if not src.exists():
+        log.warning("link_model_source_missing", model_id=model_id, path=source_path)
+        return
+
+    dest_dir = Path(project_path) / "models" / model_id
+    dest_dir.mkdir(parents=True, exist_ok=True)
+    dest = dest_dir / src.name
+
+    if dest.exists():
+        log.debug("link_model_already_exists", model_id=model_id, dest=str(dest))
+        return
+
+    try:
+        shutil.copy2(src, dest)
+        log.info("model_linked_to_project", model_id=model_id, project=project_path, dest=str(dest))
+    except OSError as exc:
+        log.warning("link_model_copy_failed", model_id=model_id, error=str(exc))
+
+
+async def link_dataset_to_active_project(dataset_id: str, source_path: str) -> None:
+    """Copy the imported dataset folder into the active project's datasets/ folder.
+
+    This is a best-effort operation — if no project is open, or if the copy
+    fails for any reason, we log and continue rather than failing the import.
+    """
+    project_path = await get_active_project_path()
+    if not project_path:
+        log.debug("link_dataset_skipped_no_project", dataset_id=dataset_id)
+        return
+
+    src = Path(source_path)
+    if not src.exists():
+        log.warning("link_dataset_source_missing", dataset_id=dataset_id, path=source_path)
+        return
+
+    dest_dir = Path(project_path) / "datasets" / dataset_id
+    
+    if dest_dir.exists():
+        log.debug("link_dataset_already_exists", dataset_id=dataset_id, dest=str(dest_dir))
+        return
+
+    try:
+        if src.is_dir():
+            shutil.copytree(src, dest_dir, dirs_exist_ok=True)
+        else:
+            # If it's a file (e.g. zip that wasn't extracted yet), just copy it
+            dest_dir.parent.mkdir(parents=True, exist_ok=True)
+            shutil.copy2(src, dest_dir)
+            
+        log.info("dataset_linked_to_project", dataset_id=dataset_id, project=project_path, dest=str(dest_dir))
+        return dest_dir
+    except OSError as exc:
+        log.warning("link_dataset_copy_failed", dataset_id=dataset_id, error=str(exc))
+        return None