Afterimage live backend (FastAPI + FastEmbed + embedded Qdrant)

.gitattributes

@@ -33,3 +33,4 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+data/assets/bank/crops/variants/bank_curb_v06.jpg filter=lfs diff=lfs merge=lfs -text

Dockerfile

@@ -0,0 +1,33 @@
+# Afterimage live backend — FastAPI + FastEmbed CLIP + embedded Qdrant.
+# Built for a Hugging Face Docker Space (runs as UID 1000, serves on :7860).
+FROM python:3.11-slim
+
+RUN useradd -m -u 1000 user && mkdir -p /app && chown user:user /app
+USER user
+WORKDIR /app
+
+ENV HOME=/home/user \
+    PATH=/home/user/.local/bin:$PATH \
+    PYTHONUNBUFFERED=1 \
+    AFTERIMAGE_ASSET_ROOT=/app/data/assets \
+    AFTERIMAGE_MANIFEST=/app/data/manifest.json \
+    QDRANT_URL=path:/app/.qdrant-local \
+    AFTERIMAGE_ALLOW_FAKE_EMBEDDINGS=0 \
+    AFTERIMAGE_CORS_ORIGINS=* \
+    HF_HOME=/home/user/.cache/huggingface \
+    FASTEMBED_CACHE_DIR=/home/user/.cache/fastembed
+
+COPY --chown=user backend/requirements.txt ./backend/requirements.txt
+RUN pip install --no-cache-dir --user -r backend/requirements.txt
+
+COPY --chown=user backend/ ./backend/
+COPY --chown=user data/ ./data/
+
+# Bake the model cache + seeded embedded Qdrant into the image so boots are fast.
+# (Downloads CLIP ViT-B/32, embeds the 36 region/baseline/incident points.)
+RUN cd backend && python scripts/seed.py \
+    && python -c "from app.config import get_settings; from app.embedder import ImageEmbedder; ImageEmbedder(get_settings()).embed_text('warm up the clip text tower')" || echo "text-tower preload skipped"
+
+EXPOSE 7860
+WORKDIR /app/backend
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,10 +1,28 @@
 ---
-title: Afterimage
-emoji: 📚
-colorFrom: indigo
-colorTo: yellow
+title: Afterimage API
+emoji: 🛰️
+colorFrom: blue
+colorTo: indigo
 sdk: docker
+app_port: 7860
 pinned: false
+license: mit
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Afterimage — live Qdrant backend
+
+The real FastAPI + FastEmbed CLIP + embedded **Qdrant** backend behind
+[Afterimage](https://afterimage-qdrant.vercel.app) — searchable visual memory for
+physical spaces.
+
+Every request runs live against Qdrant:
+
+- `POST /api/anomaly/scan?scenario=vault` — `query_points` filtered nearest-neighbour;
+  flags the region when its live crop falls below the data-derived floor (`mean − 3σ`).
+- `GET /api/outliers?scenario=vault` — `RecommendQuery(best_score)` outlier vs baselines.
+- `GET /api/matrix?scenario=vault` — `search_matrix_pairs` distance matrix.
+- `GET /api/text_search?q=a+van` — open-vocabulary CLIP text→image search.
+- `GET /api/health` — collection + embedding mode.
+
+512-dim cosine vectors, one `object_memory` collection, no training, no labels.
+Interactive docs at `/docs`.

backend/app/__init__.py

@@ -0,0 +1 @@
+

backend/app/config.py

@@ -0,0 +1,36 @@
+from functools import lru_cache
+from pathlib import Path
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    qdrant_url: str = Field(default="http://localhost:6333", alias="QDRANT_URL")
+    qdrant_api_key: str | None = Field(default=None, alias="QDRANT_API_KEY")
+    asset_root: Path = Field(default=Path("../data/assets"), alias="AFTERIMAGE_ASSET_ROOT")
+    manifest_path: Path = Field(default=Path("../data/manifest.json"), alias="AFTERIMAGE_MANIFEST")
+    allow_fake_embeddings: bool = Field(default=False, alias="AFTERIMAGE_ALLOW_FAKE_EMBEDDINGS")
+    cors_origins: str = Field(default="http://localhost:5173", alias="AFTERIMAGE_CORS_ORIGINS")
+    collection_name: str = "object_memory"
+    vector_size: int = 512
+    model_name: str = "Qdrant/clip-ViT-B-32-vision"
+    text_model_name: str = "Qdrant/clip-ViT-B-32-text"
+
+    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
+
+    @property
+    def cors_origin_list(self) -> list[str]:
+        return [origin.strip() for origin in self.cors_origins.split(",") if origin.strip()]
+
+    @property
+    def resolved_asset_root(self) -> Path:
+        return self.asset_root.expanduser().resolve()
+
+    @property
+    def resolved_manifest_path(self) -> Path:
+        return self.manifest_path.expanduser().resolve()
+
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()

backend/app/embedder.py

@@ -0,0 +1,77 @@
+from pathlib import Path
+import hashlib
+
+import numpy as np
+from PIL import Image
+
+from .config import Settings
+
+
+class ImageEmbedder:
+    def __init__(self, settings: Settings):
+        self.settings = settings
+        self._model = None
+        self._text_model = None
+        self.mode = "fastembed"
+
+    def _load_model(self):
+        if self._model is not None:
+            return self._model
+        try:
+            from fastembed import ImageEmbedding
+
+            self._model = ImageEmbedding(model_name=self.settings.model_name)
+            return self._model
+        except Exception:
+            if not self.settings.allow_fake_embeddings:
+                raise
+            self.mode = "deterministic-fallback"
+            self._model = False
+            return None
+
+    def _load_text_model(self):
+        if self._text_model is not None:
+            return self._text_model
+        from fastembed import TextEmbedding
+
+        self._text_model = TextEmbedding(model_name=self.settings.text_model_name)
+        return self._text_model
+
+    def embed_text(self, query: str) -> list[float]:
+        # CLIP text tower — shares the image vectors' 512-d space, so a text
+        # query can search the visual memory directly.
+        model = self._load_text_model()
+        vector = next(iter(model.embed([query])))
+        return self._normalize(np.asarray(vector, dtype=np.float32))
+
+    def embed_path(self, path: Path) -> list[float]:
+        if not path.exists():
+            raise FileNotFoundError(f"Image not found: {path}")
+        model = self._load_model()
+        if model is None:
+            return self._fallback_embedding(path)
+        image = Image.open(path).convert("RGB")
+        vector = next(iter(model.embed([image])))
+        return self._normalize(np.asarray(vector, dtype=np.float32))
+
+    def _fallback_embedding(self, path: Path) -> list[float]:
+        image = Image.open(path).convert("RGB").resize((64, 64))
+        arr = np.asarray(image, dtype=np.float32) / 255.0
+        means = arr.mean(axis=(0, 1))
+        stds = arr.std(axis=(0, 1))
+        hist = []
+        for channel in range(3):
+            values, _ = np.histogram(arr[:, :, channel], bins=32, range=(0, 1), density=True)
+            hist.extend(values.tolist())
+        digest = hashlib.sha256(path.read_bytes()).digest()
+        noise = np.frombuffer(digest * 16, dtype=np.uint8)[:410].astype(np.float32) / 255.0
+        vector = np.concatenate([means, stds, np.array(hist, dtype=np.float32), noise])
+        return self._normalize(vector[: self.settings.vector_size])
+
+    def _normalize(self, vector: np.ndarray) -> list[float]:
+        if vector.shape[0] != self.settings.vector_size:
+            raise ValueError(f"Expected {self.settings.vector_size}-dim vector, got {vector.shape[0]}")
+        norm = np.linalg.norm(vector)
+        if norm == 0:
+            return vector.tolist()
+        return (vector / norm).astype(float).tolist()

backend/app/main.py

@@ -0,0 +1,124 @@
+from threading import Lock
+
+from fastapi import FastAPI, HTTPException, Query
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.staticfiles import StaticFiles
+
+from .config import get_settings
+from .schemas import ObjectSearchRequest, TrailRequest
+from .services import AfterimageService
+
+
+settings = get_settings()
+_service_lock = Lock()
+_service_instance: AfterimageService | None = None
+app = FastAPI(title="Afterimage API", version="0.1.0")
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=settings.cors_origin_list,
+    allow_credentials=False,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+app.mount("/assets", StaticFiles(directory=settings.resolved_asset_root), name="assets")
+
+
+def service() -> AfterimageService:
+    global _service_instance
+    if _service_instance is None:
+        with _service_lock:
+            if _service_instance is None:
+                _service_instance = AfterimageService(settings)
+    return _service_instance
+
+
+def bad_request(exc: Exception) -> HTTPException:
+    return HTTPException(status_code=400, detail=str(exc))
+
+
+@app.get("/api/health")
+def health():
+    try:
+        return service().health()
+    except Exception as exc:
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+
+
+@app.get("/api/scenarios")
+def scenarios():
+    try:
+        return service().scenarios()
+    except Exception as exc:
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+
+
+@app.get("/api/memories")
+def memories(
+    scenario: str = Query(default="vault", max_length=60),
+    zone: str | None = Query(default=None),
+    region_id: str | None = Query(default=None),
+    is_baseline: bool | None = Query(default=None),
+):
+    try:
+        return service().memories(scenario=scenario, zone=zone, region_id=region_id, is_baseline=is_baseline)
+    except ValueError as exc:
+        raise bad_request(exc) from exc
+
+
+@app.post("/api/anomaly/scan")
+def anomaly_scan(scenario: str = Query(default="vault", max_length=60)):
+    try:
+        return service().scan_anomalies(scenario=scenario)
+    except (FileNotFoundError, ValueError) as exc:
+        raise bad_request(exc) from exc
+
+
+@app.post("/api/object/search")
+def object_search(request: ObjectSearchRequest):
+    try:
+        return service().object_search(
+            scenario=request.scenario,
+            asset_id=request.asset_id or request.crop_id,
+            image_ref=request.image_ref,
+            filters=request.filters,
+            limit=request.limit,
+        )
+    except (FileNotFoundError, ValueError) as exc:
+        raise bad_request(exc) from exc
+
+
+@app.post("/api/trail")
+def trail(request: TrailRequest):
+    try:
+        return service().trail(
+            scenario=request.scenario,
+            asset_id=request.asset_id or request.crop_id,
+            image_ref=request.image_ref,
+            score_cutoff=request.score_cutoff,
+        )
+    except (FileNotFoundError, ValueError) as exc:
+        raise bad_request(exc) from exc
+
+
+@app.get("/api/outliers")
+def outliers(scenario: str = Query(default="vault", max_length=60)):
+    try:
+        return service().outliers(scenario=scenario)
+    except ValueError as exc:
+        raise bad_request(exc) from exc
+
+
+@app.get("/api/matrix")
+def matrix(scenario: str = Query(default="vault", max_length=60), sample: int = Query(default=24, ge=1, le=100)):
+    try:
+        return service().matrix(scenario=scenario, sample=sample)
+    except ValueError as exc:
+        raise bad_request(exc) from exc
+
+
+@app.get("/api/text_search")
+def text_search(q: str = Query(..., max_length=120), scenario: str | None = Query(default=None, max_length=60)):
+    try:
+        return service().text_search(query=q, scenario=scenario)
+    except (FileNotFoundError, ValueError) as exc:
+        raise bad_request(exc) from exc

backend/app/manifest.py

@@ -0,0 +1,92 @@
+from pathlib import Path
+import json
+from typing import Any
+
+from .config import Settings
+
+
+class Manifest:
+    def __init__(self, settings: Settings):
+        self.settings = settings
+        with settings.resolved_manifest_path.open("r", encoding="utf-8") as handle:
+            self.data: dict[str, Any] = json.load(handle)
+
+    @property
+    def scenarios(self) -> list[dict[str, Any]]:
+        if "scenarios" in self.data:
+            return self.data["scenarios"]
+        legacy = {**self.data, "id": "trail", "title": "Trail", "kind": "trail"}
+        return [legacy]
+
+    @property
+    def default_scenario(self) -> str:
+        return self.data.get("default_scenario", self.scenarios[0]["id"])
+
+    def scenario(self, scenario_id: str | None = None) -> dict[str, Any]:
+        selected = scenario_id or self.default_scenario
+        for scenario in self.scenarios:
+            if scenario["id"] == selected:
+                return scenario
+        raise ValueError(f"Unknown scenario: {selected}")
+
+    def scenario_summaries(self) -> list[dict[str, Any]]:
+        return [
+            {
+                "id": item["id"],
+                "title": item.get("title", item["id"]),
+                "setting": item.get("setting", ""),
+                "kind": item.get("kind", "anomaly"),
+                "anomaly_type": item.get("anomaly_type"),
+                "summary": item.get("summary", ""),
+                "clips": item.get("clips"),
+                "hero_frame": item.get("hero_frame") or item.get("cameras", [{}])[0].get("baseline_frame"),
+                "artifact_asset_id": item.get("artifact_asset_id"),
+            }
+            for item in self.scenarios
+        ]
+
+    def cameras(self, scenario_id: str | None = None) -> list[dict[str, Any]]:
+        return self.scenario(scenario_id).get("cameras", [])
+
+    def regions(self, scenario_id: str | None = None) -> list[dict[str, Any]]:
+        return self.scenario(scenario_id).get("regions", [])
+
+    def objects(self, scenario_id: str | None = None) -> list[dict[str, Any]]:
+        return self.scenario(scenario_id).get("objects", [])
+
+    def floorplan(self, scenario_id: str | None = None) -> dict[str, Any]:
+        return self.scenario(scenario_id).get("floorplan", {})
+
+    def artifact_query_crop(self, scenario_id: str | None = None) -> str:
+        return self.scenario(scenario_id)["artifact_query_crop"]
+
+    def asset_path(self, relative: str) -> Path:
+        clean = relative.removeprefix("/assets/").lstrip("/")
+        path = Path(clean)
+        if path.is_absolute() or ".." in path.parts:
+            raise ValueError("Asset path must stay inside data/assets.")
+        if path.suffix.lower() not in {".jpg", ".jpeg", ".png", ".webp"}:
+            raise ValueError("Asset path must reference an image file.")
+        resolved = (self.settings.resolved_asset_root / path).resolve()
+        if not resolved.is_relative_to(self.settings.resolved_asset_root):
+            raise ValueError("Asset path escaped the configured asset root.")
+        return resolved
+
+    def asset_url(self, relative: str) -> str:
+        return f"/assets/{relative}"
+
+    def object_by_asset(self, scenario_id: str, asset_id: str) -> dict[str, Any] | None:
+        for item in self.objects(scenario_id):
+            if item.get("asset_id") == asset_id:
+                return item
+        return None
+
+    def crop_for_ref(self, scenario_id: str, asset_id: str | None, image_ref: str | None) -> Path:
+        if asset_id:
+            item = self.object_by_asset(scenario_id, asset_id)
+            if item is None:
+                raise ValueError(f"Unknown asset_id: {asset_id}")
+            return self.asset_path(item["crop"])
+        if image_ref:
+            return self.asset_path(image_ref.removeprefix("/assets/"))
+        return self.asset_path(self.artifact_query_crop(scenario_id))

backend/app/qdrant_store.py

@@ -0,0 +1,114 @@
+from typing import Any
+from uuid import NAMESPACE_URL, uuid5
+
+from qdrant_client import QdrantClient, models
+
+from .config import Settings
+
+
+def point_id(asset_id: str) -> str:
+    return str(uuid5(NAMESPACE_URL, f"afterimage:{asset_id}"))
+
+
+class QdrantStore:
+    def __init__(self, settings: Settings):
+        self.settings = settings
+        if settings.qdrant_url.startswith("path:"):
+            self.client = QdrantClient(path=settings.qdrant_url.removeprefix("path:"))
+        else:
+            self.client = QdrantClient(url=settings.qdrant_url, api_key=settings.qdrant_api_key or None)
+
+    def ensure_collection(self, recreate: bool = False) -> None:
+        name = self.settings.collection_name
+        exists = self.client.collection_exists(name)
+        if exists and recreate:
+            self.client.delete_collection(name)
+            exists = False
+        if not exists:
+            self.client.create_collection(
+                collection_name=name,
+                vectors_config=models.VectorParams(
+                    size=self.settings.vector_size,
+                    distance=models.Distance.COSINE,
+                ),
+            )
+
+    def upsert(self, points: list[models.PointStruct]) -> None:
+        self.client.upsert(collection_name=self.settings.collection_name, points=points, wait=True)
+
+    def count(self) -> int:
+        return self.client.count(collection_name=self.settings.collection_name, exact=True).count
+
+    def search(self, vector: Any, query_filter: models.Filter | None, limit: int = 8):
+        return self.client.query_points(
+            collection_name=self.settings.collection_name,
+            query=vector,
+            query_filter=query_filter,
+            limit=limit,
+            with_payload=True,
+            with_vectors=False,
+        ).points
+
+    def recommend_best_score(self, negative_ids: list[str], query_filter: models.Filter | None, limit: int = 8):
+        if not negative_ids:
+            raise ValueError("Recommendation query requires at least one negative example.")
+        query = models.RecommendQuery(
+            recommend=models.RecommendInput(
+                negative=negative_ids,
+                strategy=models.RecommendStrategy.BEST_SCORE,
+            )
+        )
+        return self.client.query_points(
+            collection_name=self.settings.collection_name,
+            query=query,
+            query_filter=query_filter,
+            limit=limit,
+            with_payload=True,
+            with_vectors=False,
+        ).points
+
+    def search_matrix_pairs(self, query_filter: models.Filter | None = None, limit: int = 3, sample: int = 24):
+        return self.client.search_matrix_pairs(
+            collection_name=self.settings.collection_name,
+            query_filter=query_filter,
+            limit=limit,
+            sample=sample,
+        )
+
+    def scroll(self, query_filter: models.Filter | None = None, limit: int = 100):
+        points, _ = self.client.scroll(
+            collection_name=self.settings.collection_name,
+            scroll_filter=query_filter,
+            limit=limit,
+            with_payload=True,
+            with_vectors=False,
+        )
+        return points
+
+
+def match_filter(**conditions: Any) -> models.Filter | None:
+    must = []
+    for key, value in conditions.items():
+        if value is None:
+            continue
+        must.append(models.FieldCondition(key=key, match=models.MatchValue(value=value)))
+    return models.Filter(must=must) if must else None
+
+
+def payload_filter(filters: dict[str, Any]) -> models.Filter | None:
+    must = []
+    for key in ("scenario", "camera_id", "zone", "is_baseline", "memory_type"):
+        value = filters.get(key)
+        if value is not None:
+            must.append(models.FieldCondition(key=key, match=models.MatchValue(value=value)))
+    time_range = filters.get("time_range")
+    if time_range:
+        must.append(models.FieldCondition(key="timestamp", range=models.Range(gte=time_range[0], lte=time_range[1])))
+    return models.Filter(must=must) if must else None
+
+
+def inspector_results(points) -> list[dict[str, Any]]:
+    results = []
+    for point in points:
+        results.append({"id": point.id, "score": point.score, "payload": point.payload})
+    return results

backend/app/schemas.py

@@ -0,0 +1,38 @@
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+
+class QueryFilters(BaseModel):
+    scenario: str | None = None
+    camera_id: str | None = None
+    zone: str | None = None
+    is_baseline: bool | None = None
+    memory_type: str | None = None
+    time_range: tuple[int, int] | None = None
+
+
+class ObjectSearchRequest(BaseModel):
+    scenario: str = Field(default="vault", max_length=60)
+    crop_id: str | None = Field(default=None, max_length=140)
+    asset_id: str | None = Field(default=None, max_length=140)
+    image_ref: str | None = Field(default=None, max_length=220)
+    filters: QueryFilters = Field(default_factory=QueryFilters)
+    limit: int = Field(default=8, ge=1, le=50)
+
+
+class TrailRequest(BaseModel):
+    scenario: str = Field(default="vault", max_length=60)
+    crop_id: str | None = Field(default=None, max_length=140)
+    asset_id: str | None = Field(default=None, max_length=140)
+    image_ref: str | None = Field(default=None, max_length=220)
+    score_cutoff: float = Field(default=0.2, ge=0, le=1)
+
+
+class Inspector(BaseModel):
+    api: str
+    collection: str
+    filter: dict[str, Any] | None = None
+    params: dict[str, Any]
+    results: list[dict[str, Any]]
+    took_ms: float | None = None

backend/app/services.py

@@ -0,0 +1,264 @@
+from time import perf_counter
+from typing import Any
+
+from .config import Settings
+from .embedder import ImageEmbedder
+from .manifest import Manifest
+from .qdrant_store import QdrantStore, inspector_results, match_filter, payload_filter
+from .schemas import Inspector, QueryFilters
+
+
+class AfterimageService:
+    def __init__(self, settings: Settings):
+        self.settings = settings
+        self.manifest = Manifest(settings)
+        self.embedder = ImageEmbedder(settings)
+        self.store = QdrantStore(settings)
+
+    def health(self) -> dict[str, Any]:
+        collections = self.store.client.get_collections()
+        return {
+            "ok": True,
+            "collection": self.settings.collection_name,
+            "point_count": self.store.count() if self.store.client.collection_exists(self.settings.collection_name) else 0,
+            "embedding_mode": self.embedder.mode,
+            "collections": [item.name for item in collections.collections],
+            "default_scenario": self.manifest.default_scenario,
+            "scenarios": self.manifest.scenario_summaries(),
+        }
+
+    def scenarios(self) -> dict[str, Any]:
+        return {"default_scenario": self.manifest.default_scenario, "scenarios": self.manifest.scenario_summaries()}
+
+    def memories(self, scenario: str, zone: str | None, region_id: str | None, is_baseline: bool | None) -> dict[str, Any]:
+        self.manifest.scenario(scenario)
+        query_filter = match_filter(scenario=scenario, zone=zone, region_id=region_id, is_baseline=is_baseline)
+        points = self.store.scroll(query_filter=query_filter, limit=200)
+        return {
+            "points": [self._point_view(point) for point in points],
+            "floorplan": self.manifest.floorplan(scenario),
+            "scenario": self.manifest.scenario(scenario),
+        }
+
+    def scan_anomalies(self, scenario: str) -> dict[str, Any]:
+        self.manifest.scenario(scenario)
+        regions = []
+        inspector_blocks = []
+        for region in self.manifest.regions(scenario):
+            crop = self.manifest.asset_path(region["incident_crop"])
+            vector = self.embedder.embed_path(crop)
+            query_filter = match_filter(scenario=scenario, region_id=region["region_id"], is_baseline=True)
+            t0 = perf_counter()
+            points = self.store.search(vector, query_filter=query_filter, limit=8)
+            took_ms = round((perf_counter() - t0) * 1000, 2)
+            top_score = points[0].score if points else 0.0
+            normal_band = self._normal_band(scenario, region["region_id"])
+            threshold = normal_band["floor"] if normal_band["source"] == "baseline_variants" else float(region.get("threshold", 0.78))
+            alarm_enabled = bool(region.get("alarm_enabled", True))
+            status = "anomalous" if alarm_enabled and top_score < threshold else "normal"
+            result = {
+                "region_id": region["region_id"],
+                "region_label": region["region_label"],
+                "camera_id": region["camera_id"],
+                "zone": region["zone"],
+                "score": top_score,
+                "threshold": threshold,
+                "normal_band": normal_band,
+                "alarm_enabled": alarm_enabled,
+                "status": status,
+                "incident_crop_url": self.manifest.asset_url(region["incident_crop"]),
+                "nearest": [self._point_view(point) for point in points],
+            }
+            regions.append(result)
+            inspector_blocks.append(
+                self._inspector(
+                    api="query_points",
+                    params={
+                        "limit": 8,
+                        "query": "incident_region_crop",
+                        "region_id": region["region_id"],
+                        "alarm_enabled": alarm_enabled,
+                        "normal_band": normal_band,
+                    },
+                    query_filter={"scenario": scenario, "region_id": region["region_id"], "is_baseline": True},
+                    points=points,
+                    took_ms=took_ms,
+                )
+            )
+        return {"regions": regions, "inspector": inspector_blocks}
+
+    def object_search(self, scenario: str, asset_id: str | None, image_ref: str | None, filters: QueryFilters, limit: int = 8):
+        self.manifest.scenario(scenario)
+        crop = self.manifest.crop_for_ref(scenario, asset_id, image_ref)
+        vector = self.embedder.embed_path(crop)
+        filter_dict = filters.model_dump()
+        filter_dict["scenario"] = scenario
+        if filter_dict.get("memory_type") is None:
+            filter_dict["memory_type"] = "object_sighting"
+        query_filter = payload_filter(filter_dict)
+        points = self.store.search(vector, query_filter=query_filter, limit=limit)
+        inspector = self._inspector(
+            api="query_points",
+            params={"limit": limit, "query_crop": str(crop.relative_to(self.settings.resolved_asset_root))},
+            query_filter=filter_dict,
+            points=points,
+        )
+        return {"results": [self._point_view(point) for point in points], "inspector": inspector}
+
+    def trail(self, scenario: str, asset_id: str | None, image_ref: str | None, score_cutoff: float):
+        search = self.object_search(
+            scenario=scenario,
+            asset_id=asset_id,
+            image_ref=image_ref,
+            filters=QueryFilters(scenario=scenario, memory_type="object_sighting"),
+            limit=12,
+        )
+        by_zone: dict[str, dict[str, Any]] = {}
+        for result in search["results"]:
+            payload = result["payload"]
+            if result["score"] < score_cutoff:
+                continue
+            zone = payload["zone"]
+            current = by_zone.get(zone)
+            if current is None or result["score"] > current["score"]:
+                by_zone[zone] = result
+        sightings = sorted(by_zone.values(), key=lambda item: item["payload"]["timestamp"])
+        path = []
+        floor_coords = self._floor_coords(scenario)
+        for item in sightings:
+            payload = item["payload"]
+            path.append({**item, "floorplan_xy": floor_coords[payload["zone"]]})
+        return {"trail": path, "ranked_results": search["results"], "inspector": search["inspector"]}
+
+    def outliers(self, scenario: str):
+        self.manifest.scenario(scenario)
+        negative_filter = match_filter(scenario=scenario, memory_type="region_baseline", alarm_enabled=True)
+        negative_points = self.store.scroll(query_filter=negative_filter, limit=50)
+        negative_ids = [str(point.id) for point in negative_points]
+        candidate_filter = match_filter(scenario=scenario, memory_type="region_incident", alarm_enabled=True)
+        t0 = perf_counter()
+        points = self.store.recommend_best_score(
+            negative_ids=negative_ids,
+            query_filter=candidate_filter,
+            limit=8,
+        )
+        took_ms = round((perf_counter() - t0) * 1000, 2)
+        return {
+            "strategy": "RecommendQuery best_score: alarm-zone normal memories as negatives, alarm-zone incidents as candidates",
+            "results": [self._point_view(point) for point in points],
+            "inspector": self._inspector(
+                api="RecommendQuery(best_score)",
+                params={"candidate_scope": "alarm_enabled regions only", "negative_example_count": len(negative_ids), "limit": 8},
+                query_filter={"scenario": scenario, "memory_type": "region_incident", "alarm_enabled": True},
+                points=points,
+                took_ms=took_ms,
+            ),
+        }
+
+    def text_search(self, query: str, scenario: str | None = None, limit: int = 8):
+        query = (query or "").strip()
+        if not query:
+            raise ValueError("Empty query.")
+        vector = self.embedder.embed_text(query)
+        query_filter = match_filter(scenario=scenario) if scenario else None
+        t0 = perf_counter()
+        points = self.store.search(vector, query_filter=query_filter, limit=limit)
+        took_ms = round((perf_counter() - t0) * 1000, 2)
+        return {
+            "query": query,
+            "results": [self._point_view(point) for point in points],
+            "inspector": self._inspector(
+                api="query_points · text",
+                params={"limit": limit, "query": query, "encoder": "clip-ViT-B-32-text"},
+                query_filter={"scenario": scenario} if scenario else None,
+                points=points,
+                took_ms=took_ms,
+            ),
+        }
+
+    def matrix(self, scenario: str, sample: int = 24):
+        self.manifest.scenario(scenario)
+        query_filter = match_filter(scenario=scenario)
+        points = self.store.scroll(query_filter=query_filter, limit=sample)
+        matrix = self.store.search_matrix_pairs(query_filter=query_filter, limit=3, sample=min(sample, len(points) or 1))
+        nodes = []
+        floor_coords = self._floor_coords(scenario)
+        for index, point in enumerate(points):
+            payload = point.payload or {}
+            zone = payload.get("zone", "main_hall")
+            base = floor_coords.get(zone, [120, 120])
+            nodes.append(
+                {
+                    "id": point.id,
+                    "x": base[0] + (index % 4) * 9,
+                    "y": base[1] + (index // 4) * 9,
+                    "payload": payload,
+                }
+            )
+        pairs = [{"a": str(pair.a), "b": str(pair.b), "score": pair.score} for pair in matrix.pairs]
+        return {"nodes": nodes, "pairs": pairs}
+
+    def _inspector(
+        self,
+        api: str,
+        params: dict[str, Any],
+        query_filter: dict[str, Any] | None,
+        points,
+        took_ms: float | None = None,
+    ) -> dict[str, Any]:
+        return Inspector(
+            api=api,
+            collection=self.settings.collection_name,
+            filter=query_filter,
+            params=params,
+            results=inspector_results(points),
+            took_ms=took_ms,
+        ).model_dump()
+
+    def _floor_coords(self, scenario: str) -> dict[str, list[int]]:
+        zones = self.manifest.floorplan(scenario).get("zones", [])
+        return {zone["id"]: zone.get("floorplan_xy", [120, 120]) for zone in zones}
+
+    def _normal_band(self, scenario: str, region_id: str) -> dict[str, Any]:
+        query_filter = match_filter(scenario=scenario, region_id=region_id, is_baseline=True)
+        baseline_points = self.store.scroll(query_filter=query_filter, limit=40)
+        peer_scores = []
+        for point in baseline_points:
+            neighbors = self.store.search(str(point.id), query_filter=query_filter, limit=min(len(baseline_points), 8))
+            for neighbor in neighbors:
+                if str(neighbor.id) != str(point.id) and neighbor.score is not None:
+                    peer_scores.append(float(neighbor.score))
+                    break
+        if len(peer_scores) < 2:
+            return {
+                "source": "manifest_threshold",
+                "count": len(peer_scores),
+                "mean": None,
+                "min": None,
+                "std": None,
+                "floor": None,
+            }
+        mean = sum(peer_scores) / len(peer_scores)
+        variance = sum((score - mean) ** 2 for score in peer_scores) / len(peer_scores)
+        std = variance ** 0.5
+        # Data-derived control limit: three sigma below the mean baseline self-similarity.
+        # No hand-tuned constant — the band widens or tightens with the footage itself.
+        floor = max(0.0, mean - 3.0 * std)
+        return {
+            "source": "baseline_variants",
+            "count": len(peer_scores),
+            "mean": mean,
+            "min": min(peer_scores),
+            "std": std,
+            "floor": floor,
+        }
+
+    def _point_view(self, point) -> dict[str, Any]:
+        payload = dict(point.payload or {})
+        return {
+            "id": point.id,
+            "score": getattr(point, "score", None),
+            "payload": payload,
+            "crop_url": payload.get("crop_url"),
+            "frame_url": payload.get("frame_url"),
+        }

backend/requirements.txt

@@ -0,0 +1,10 @@
+fastapi==0.136.3
+starlette==1.0.1
+uvicorn[standard]==0.38.0
+qdrant-client[fastembed]==1.18.0
+fastembed==0.8.0
+pillow==12.2.0
+numpy==2.3.5
+pydantic==2.12.5
+pydantic-settings==2.12.0
+python-dotenv==1.2.2

backend/scripts/seed.py

@@ -0,0 +1,113 @@
+from pathlib import Path
+import sys
+
+from qdrant_client import models
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from app.config import get_settings  # noqa: E402
+from app.embedder import ImageEmbedder  # noqa: E402
+from app.manifest import Manifest  # noqa: E402
+from app.qdrant_store import QdrantStore, point_id  # noqa: E402
+
+
+def payload_base(manifest: Manifest, scenario: dict, item: dict, crop: str, memory_type: str) -> dict:
+    camera = next(cam for cam in manifest.cameras(scenario["id"]) if cam["id"] == item["camera_id"])
+    return {
+        "scenario": scenario["id"],
+        "camera_id": item["camera_id"],
+        "zone": item.get("zone", camera["zone"]),
+        "timestamp": int(item.get("timestamp", 1716900000)),
+        "frame_url": manifest.asset_url(item.get("frame", camera.get("incident_frame", camera["baseline_frame"]))),
+        "crop_url": manifest.asset_url(crop),
+        "bbox": item.get("bbox", [0, 0, 0, 0]),
+        "asset_id": item["asset_id"],
+        "memory_type": memory_type,
+        "is_baseline": bool(item.get("is_baseline", False)),
+        "is_incident": bool(item.get("is_incident", False)),
+        "object_label": item.get("object_label", "region"),
+        "track_id": item.get("track_id"),
+        "region_id": item.get("region_id"),
+        "region_label": item.get("region_label"),
+        "alarm_enabled": item.get("alarm_enabled"),
+    }
+
+
+def region_points(manifest: Manifest, embedder: ImageEmbedder, scenario: dict):
+    points = []
+    for region in manifest.regions(scenario["id"]):
+        camera = next(cam for cam in manifest.cameras(scenario["id"]) if cam["id"] == region["camera_id"])
+        for state, crop, baseline in [
+            ("baseline", region["baseline_crop"], True),
+            ("incident", region["incident_crop"], False),
+        ]:
+            asset_id = f"{region['region_id']}_{state}"
+            payload = payload_base(
+                manifest,
+                scenario,
+                {
+                    **region,
+                    "asset_id": asset_id,
+                    "frame": camera[f"{state}_frame"],
+                    "is_baseline": baseline,
+                    "is_incident": not baseline,
+                    "object_label": region.get(f"{state}_label", region["region_id"]),
+                },
+                crop,
+                f"region_{state}",
+            )
+            vector = embedder.embed_path(manifest.asset_path(crop))
+            points.append(models.PointStruct(id=point_id(f"{scenario['id']}:{asset_id}"), vector=vector, payload=payload))
+            if baseline:
+                for index, variant_crop in enumerate(baseline_variants(manifest, crop, region.get("variant_prefix")), start=1):
+                    variant_id = f"{asset_id}_v{index}"
+                    variant_payload = {
+                        **payload,
+                        "asset_id": variant_id,
+                        "crop_url": manifest.asset_url(variant_crop),
+                        "memory_variant": index,
+                    }
+                    vector = embedder.embed_path(manifest.asset_path(variant_crop))
+                    points.append(models.PointStruct(id=point_id(f"{scenario['id']}:{variant_id}"), vector=vector, payload=variant_payload))
+    return points
+
+
+def baseline_variants(manifest: Manifest, crop: str, variant_prefix: str | None = None) -> list[str]:
+    path = Path(crop)
+    stem = variant_prefix or path.stem
+    variant_dir = manifest.settings.resolved_asset_root / path.parent / "variants"
+    if not variant_dir.exists():
+        return []
+    return [
+        f"{path.parent.as_posix()}/variants/{variant.name}"
+        for variant in sorted(variant_dir.glob(f"{stem}_v*.jpg"))
+    ]
+
+
+def object_points(manifest: Manifest, embedder: ImageEmbedder, scenario: dict):
+    points = []
+    for item in manifest.objects(scenario["id"]):
+        payload = payload_base(manifest, scenario, item, item["crop"], item["memory_type"])
+        vector = embedder.embed_path(manifest.asset_path(item["crop"]))
+        points.append(models.PointStruct(id=point_id(f"{scenario['id']}:{item['asset_id']}"), vector=vector, payload=payload))
+    return points
+
+
+def main() -> int:
+    settings = get_settings()
+    manifest = Manifest(settings)
+    embedder = ImageEmbedder(settings)
+    store = QdrantStore(settings)
+    store.ensure_collection(recreate=True)
+    points = []
+    for scenario in manifest.scenarios:
+        points.extend(region_points(manifest, embedder, scenario))
+        points.extend(object_points(manifest, embedder, scenario))
+    store.upsert(points)
+    print(f"Seeded {len(points)} points into {settings.collection_name} using {embedder.mode}.")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

backend/scripts/verify.py

@@ -0,0 +1,110 @@
+from pathlib import Path
+import sys
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from app.config import get_settings  # noqa: E402
+from app.qdrant_store import match_filter  # noqa: E402
+from app.services import AfterimageService  # noqa: E402
+
+
+# scenario -> (region_id, expected sole outlier asset_id)
+SCENARIOS = {
+    "vault": ("vault_pedestal", "vault_pedestal_incident"),
+    "museum": ("museum_pedestal", "museum_pedestal_incident"),
+    "bank": ("bank_curb", "bank_curb_incident"),
+}
+
+
+def fail(message: str) -> int:
+    print(f"FAIL: {message}")
+    return 1
+
+
+def held_out_normal_score(service: AfterimageService, scenario: str, region_id: str) -> float | None:
+    """Embed the canonical baseline crop — a held-out NORMAL frame, distinct from
+    the seeded baseline variants — and score it against the region's learned band.
+    This is the false-positive control: a normal frame must stay above the floor."""
+    region = next(
+        (r for r in service.manifest.regions(scenario) if r["region_id"] == region_id),
+        None,
+    )
+    if not region or not region.get("baseline_crop"):
+        return None
+    vector = service.embedder.embed_path(service.manifest.asset_path(region["baseline_crop"]))
+    query_filter = match_filter(scenario=scenario, region_id=region_id, is_baseline=True)
+    points = service.store.search(vector, query_filter=query_filter, limit=8)
+    # Exclude the exact self-match (~1.0) so this is a true held-out peer
+    # similarity — a normal frame compared against the *other* baselines.
+    peer = next((p for p in points if p.score is not None and p.score < 0.9999), None)
+    return float(peer.score) if peer else None
+
+
+def verify_scenario(service: AfterimageService, scenario: str, region_id: str, incident_id: str) -> int:
+    scan = service.scan_anomalies(scenario)
+    regions = {item["region_id"]: item for item in scan["regions"]}
+    region = regions.get(region_id)
+    if region is None:
+        return fail(f"{scenario}: missing region {region_id}")
+
+    band = region["normal_band"]
+    print(
+        f"{scenario}/{region_id}: score={region['score']:.4f} "
+        f"floor={region['threshold']:.4f} margin={region['threshold'] - region['score']:+.4f} "
+        f"status={region['status']}"
+    )
+    if band["source"] != "baseline_variants":
+        return fail(f"{scenario}: normal band is not data-derived (source={band['source']})")
+    if region["status"] != "anomalous" or region["score"] >= region["threshold"]:
+        return fail(f"{scenario}: incident did not cross the learned normal band")
+
+    # False-positive control: a held-out NORMAL frame must stay inside the band.
+    # Without this, the detector could be one that simply always alarms.
+    normal_top = held_out_normal_score(service, scenario, region_id)
+    if normal_top is not None:
+        if normal_top < region["threshold"]:
+            return fail(
+                f"{scenario}: held-out NORMAL frame scored {normal_top:.4f} < floor "
+                f"{region['threshold']:.4f} — false positive"
+            )
+        print(
+            f"  control: normal frame {normal_top:.4f} >= floor {region['threshold']:.4f} "
+            f"(stays in band) | incident {region['score']:.4f} breaches"
+        )
+
+    outliers = service.outliers(scenario)
+    if outliers["inspector"]["api"] != "RecommendQuery(best_score)":
+        return fail(f"{scenario}: outliers not backed by RecommendQuery(best_score)")
+    outlier_ids = [item["payload"]["asset_id"] for item in outliers["results"]]
+    if outlier_ids != [incident_id]:
+        return fail(f"{scenario}: expected sole outlier {incident_id!r}, got {outlier_ids}")
+    return 0
+
+
+def main() -> int:
+    service = AfterimageService(get_settings())
+    health = service.health()
+
+    if health["embedding_mode"] != "fastembed":
+        return fail(f"embeddings are not real CLIP (mode={health['embedding_mode']})")
+    scenario_ids = {item["id"] for item in health["scenarios"]}
+    if set(SCENARIOS).issubset(scenario_ids) is False:
+        return fail(f"missing required scenarios, got {sorted(scenario_ids)}")
+    if health["point_count"] < 30:
+        return fail(f"expected at least 30 points, got {health['point_count']}")
+
+    for scenario, (region_id, incident_id) in SCENARIOS.items():
+        result = verify_scenario(service, scenario, region_id, incident_id)
+        if result:
+            return result
+
+    print(
+        "Verification passed: all three incidents flagged, all three held-out "
+        "normals stayed in band, all three outliers correct, real CLIP."
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

data/assets/.gitkeep

@@ -0,0 +1 @@
+