Upload folder using huggingface_hub

backend/config.py

@@ -0,0 +1,65 @@
+"""Runtime configuration from environment. Phase 5.
+
+All secrets come from env / .env (never hardcoded — see CLAUDE.md). Everything degrades
+gracefully when a secret is absent so the app runs locally without Supabase / R2 / weights:
+  - no Supabase creds  -> logging is a no-op
+  - no R2 creds        -> uploads return a local placeholder url
+  - weights not on Hub -> inference runs in STUB mode (contract-valid synthetic output)
+"""
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+
+
+def _flag(name: str) -> bool:
+    return os.getenv(name, "").lower() in ("1", "true", "yes")
+
+
+@dataclass(frozen=True)
+class Settings:
+    # --- model weights (HF Hub) ---
+    hf_repo: str = os.getenv("HF_MODEL_REPO", "HrishiKabra/reefscan-dinov2-coral")
+    hf_stage: str = os.getenv("HF_MODEL_STAGE", "linear_probe")
+    hf_token: str | None = os.getenv("HF_TOKEN") or None
+
+    # --- Supabase (logging) ---
+    # Use the SERVICE_ROLE key here (server-side only) — tables have RLS enabled, so the
+    # service_role key (which bypasses RLS) is required to read/write. SUPABASE_KEY is the
+    # canonical name; SUPABASE_ANON_KEY is accepted for backward-compat.
+    supabase_url: str | None = os.getenv("SUPABASE_URL") or None
+    supabase_key: str | None = os.getenv("SUPABASE_KEY") or os.getenv("SUPABASE_ANON_KEY") or None
+
+    # --- object storage ---
+    # Default = Supabase Storage (reuses the Supabase service_role client; no extra account).
+    # Optionally override with Cloudflare R2 by setting all four R2_* vars.
+    storage_bucket: str = os.getenv("STORAGE_BUCKET", "reefscan-uploads")
+    r2_endpoint: str | None = os.getenv("R2_ENDPOINT") or None
+    r2_key_id: str | None = os.getenv("R2_ACCESS_KEY_ID") or None
+    r2_secret: str | None = os.getenv("R2_SECRET_ACCESS_KEY") or None
+    r2_bucket: str | None = os.getenv("R2_BUCKET") or None
+
+    # --- locked AMG config (Phase 1.5; do NOT inline elsewhere) ---
+    amg_points_per_side: int = 16
+    amg_longest_edge: int = 512
+
+    # --- conformal ---
+    coverage_alpha: float = 0.10  # 90% target; qhat loaded from HF conformal.json
+
+    # --- modeling ---
+    classes: tuple[str, ...] = ("healthy", "bleached")
+    input_size: int = 224
+
+    # force synthetic inference (also auto-enabled if weights fail to load)
+    stub_mode: bool = _flag("REEFSCAN_STUB")
+
+    @property
+    def supabase_enabled(self) -> bool:
+        return bool(self.supabase_url and self.supabase_key)
+
+    @property
+    def r2_enabled(self) -> bool:
+        return bool(self.r2_endpoint and self.r2_key_id and self.r2_secret and self.r2_bucket)
+
+
+settings = Settings()

backend/data/__init__.py

@@ -0,0 +1,17 @@
+"""ReefScan data pipeline (Phase 2): imagefolder patch Dataset, transforms, EDA.
+
+CoralPatchDataset (ImageFolder) is the loader for the current 2-class NOAA dataset.
+backend.data.split is retained for FUTURE point-annotated sources (not used here).
+"""
+from .label_mapping import CLASSES, CLASS_TO_IDX, IDX_TO_CLASS, LABEL_MAPPING, map_label
+from .patch_dataset import CoralPatchDataset, scan_imagefolder
+
+__all__ = [
+    "CoralPatchDataset",
+    "scan_imagefolder",
+    "CLASSES",
+    "CLASS_TO_IDX",
+    "IDX_TO_CLASS",
+    "LABEL_MAPPING",
+    "map_label",
+]

backend/data/eda.py

@@ -0,0 +1,121 @@
+"""EDA over the CoralNet annotations. Phase 2.
+
+Its main job is to produce exactly what's needed to author backend/data/label_mapping.py
+and choose a loss strategy. When run against real data it prints:
+
+  1. Raw label frequency table  — every CoralNet label found, sorted by count, printed
+     VERBATIM (no normalization/stripping) so the strings can be pasted into LABEL_MAPPING.
+  2. Per-image label distribution — points/image and distinct-labels/image stats, plus a
+     full image x label count matrix written to data/eda_per_image.csv.
+  3. Class imbalance ratio — on the 4-class mapped distribution if LABEL_MAPPING is filled,
+     otherwise on raw labels as a proxy.
+  4. A recommendation — standard CE vs. class-weighting/oversampling vs. focal loss.
+
+Run:  python -m backend.data.eda [--csv data/annotations.csv] [--per-image-out data/eda_per_image.csv]
+"""
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+import pandas as pd
+
+from .label_mapping import CLASSES, LABEL_MAPPING, map_label
+
+
+def _imbalance_report(counts: dict[str, int], basis: str) -> str:
+    vals = [c for c in counts.values() if c > 0]
+    if not vals:
+        return f"imbalance ({basis}): no labeled samples."
+    hi, lo = max(vals), min(vals)
+    ratio = hi / lo
+    lines = [f"imbalance ratio ({basis}) = {ratio:.1f}:1   (largest {hi} / smallest {lo})"]
+    if ratio < 3:
+        lines.append("  -> roughly balanced. Standard cross-entropy; no resampling needed.")
+    elif ratio < 10:
+        lines.append("  -> MODERATE imbalance. Use class-weighted cross-entropy "
+                     "(weight = 1/freq) OR oversample minority classes.")
+    else:
+        lines.append("  -> SEVERE imbalance. Recommend focal loss (gamma~2) and/or "
+                     "oversampling minority classes; consider undersampling the dominant class.")
+    if lo < 50:
+        lines.append(f"  !! smallest class has only {lo} samples — likely too few for a "
+                     "reliable linear probe. Consider merging classes or collecting more.")
+    return "\n".join(lines)
+
+
+def run_eda(csv_path: str | Path, per_image_out: str | Path = "data/eda_per_image.csv") -> None:
+    # dtype=str on label keeps the raw CoralNet strings EXACTLY as written (no coercion).
+    df = pd.read_csv(csv_path, dtype={"label": str})
+    n = len(df)
+    n_images = df["image_name"].nunique()
+
+    print(f"\n=== ReefScan EDA :: {csv_path} ===")
+    print(f"annotations: {n}   images: {n_images}   "
+          f"avg points/image: {n / max(n_images, 1):.1f}")
+
+    # --- 1. Raw label frequency table (verbatim strings, sorted by count) -------------
+    print("\n--- 1. RAW CoralNet label frequency (verbatim, sorted) ---")
+    raw_counts = df["label"].value_counts()  # already sorted desc
+    width = max((len(str(l)) for l in raw_counts.index), default=8)
+    print(f"  {'label'.ljust(width)}  {'count':>7}  {'pct':>6}  -> mapped")
+    for label, c in raw_counts.items():
+        mapped = map_label(label)
+        if mapped is not None:
+            tag = mapped
+        elif label in LABEL_MAPPING:
+            tag = "DROP"
+        else:
+            tag = "UNMAPPED"
+        print(f"  {str(label).ljust(width)}  {c:>7}  {100 * c / n:>5.1f}%  -> {tag}")
+
+    unmapped = sorted(l for l in raw_counts.index if l not in LABEL_MAPPING)
+    if unmapped:
+        print(f"\n  !! {len(unmapped)} raw label(s) UNMAPPED — add each to LABEL_MAPPING:")
+        print(f"     {unmapped}")
+
+    # --- 2. Per-image label distribution ----------------------------------------------
+    print("\n--- 2. Per-image label distribution ---")
+    points_per_image = df.groupby("image_name").size()
+    labels_per_image = df.groupby("image_name")["label"].nunique()
+    print(f"  points/image   : min {points_per_image.min()}  median "
+          f"{points_per_image.median():.0f}  mean {points_per_image.mean():.1f}  "
+          f"max {points_per_image.max()}")
+    print(f"  distinct labels/image: min {labels_per_image.min()}  median "
+          f"{labels_per_image.median():.0f}  max {labels_per_image.max()}")
+    matrix = pd.crosstab(df["image_name"], df["label"])  # image x label counts
+    Path(per_image_out).parent.mkdir(parents=True, exist_ok=True)
+    matrix.to_csv(per_image_out)
+    print(f"  full image x label count matrix ({matrix.shape[0]} images x "
+          f"{matrix.shape[1]} labels) written to: {per_image_out}")
+    print("  head:")
+    print(matrix.head(10).to_string().replace("\n", "\n    "))
+
+    # --- 3 & 4. Imbalance ratio + recommendation --------------------------------------
+    print("\n--- 3/4. Class imbalance + loss recommendation ---")
+    if LABEL_MAPPING:
+        mapped_counts = {cls: 0 for cls in CLASSES}
+        for label, c in raw_counts.items():
+            m = map_label(label)
+            if m is not None:
+                mapped_counts[m] += int(c)
+        total_mapped = sum(mapped_counts.values()) or 1
+        nclass = len(CLASSES)
+        print(f"  {nclass}-class distribution:")
+        for cls in CLASSES:
+            c = mapped_counts[cls]
+            print(f"    {cls:<16} {c:>7}  ({100 * c / total_mapped:.1f}%)")
+        print(_imbalance_report(mapped_counts, basis=f"{nclass}-class"))
+    else:
+        print("  LABEL_MAPPING is EMPTY — reporting imbalance on RAW labels as a proxy.")
+        print("  (Recompute after filling the mapping; the 4-class ratio is what matters.)")
+        print(_imbalance_report(dict(raw_counts), basis="raw labels, PROXY"))
+    print()
+
+
+if __name__ == "__main__":
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--csv", default="data/annotations.csv")
+    ap.add_argument("--per-image-out", default="data/eda_per_image.csv")
+    args = ap.parse_args()
+    run_eda(args.csv, args.per_image_out)

backend/data/label_mapping.py

@@ -0,0 +1,29 @@
+"""Raw dataset label -> ReefScan class mapping.
+
+INITIAL MODEL IS 2-CLASS: healthy | bleached
+(Locked from EDA of NMFS-OSI/NOAA-PIFSC-ESD-CORAL-Bleaching-Dataset, which contains only
+two health states: CORAL = healthy, CORAL_BL = bleached. No taxonomy, no dead/algae.)
+
+The DB `coral_label` enum still reserves `dead` and `algae_covered` for a future
+extension (e.g. ReefNet supplementation) — those are NOT modeled yet, but keeping the
+enum values means no DB migration is needed when they arrive. The MODEL head, UI, and
+conformal sets are all 2-class for now.
+"""
+from __future__ import annotations
+
+# The valid target classes for the CURRENT model. 2-class initial model.
+# (The DB enum reserves "dead" and "algae_covered" for future extension — see schema.sql.)
+CLASSES: tuple[str, ...] = ("healthy", "bleached")
+CLASS_TO_IDX: dict[str, int] = {c: i for i, c in enumerate(CLASSES)}
+IDX_TO_CLASS: dict[int, str] = {i: c for i, c in enumerate(CLASSES)}
+
+# raw dataset label -> one of CLASSES, or None to explicitly drop.
+LABEL_MAPPING: dict[str, str | None] = {
+    "CORAL": "healthy",
+    "CORAL_BL": "bleached",
+}
+
+
+def map_label(raw_label: str) -> str | None:
+    """Return the ReefScan class for a raw dataset label, or None if unmapped/dropped."""
+    return LABEL_MAPPING.get(raw_label)

backend/data/patch_dataset.py

@@ -0,0 +1,118 @@
+"""Coral colony-patch classification dataset (ImageFolder format). Phase 2.
+
+Built for NMFS-OSI/NOAA-PIFSC-ESD-CORAL-Bleaching-Dataset, which ships as an imagefolder:
+
+    <root>/<split>/<RAW_LABEL>/<image>.PNG     e.g. train/CORAL/FFS-B013_2019_15_1024.PNG
+
+These are pre-cropped, image-level colony patches (one health label per image) — NOT point
+annotations. So training uses the WHOLE patch (resize -> 224, ImageNet-normalize); there is
+no point/centroid crop here. Raw folder labels (CORAL, CORAL_BL) are collapsed to the
+ReefScan classes via label_mapping.LABEL_MAPPING.
+
+IMPORTANT: use the dataset's NATIVE train/val/test splits (the directory names). Do NOT
+re-split with backend.data.split — the NOAA splits are site/year-controlled to prevent
+leakage. backend.data.split is retained only for FUTURE point-annotated sources.
+
+The inference-time bridge (Phase 5): SAM2 mask -> bbox crop -> resize 224 -> classify. The
+training distribution (colony-level crops) closely matches those mask-bbox crops, so this
+imagefolder design is a cleaner train/inference match than the original centroid-patch plan.
+"""
+from __future__ import annotations
+
+import logging
+from collections import Counter
+from pathlib import Path
+
+from PIL import Image
+from torch.utils.data import Dataset
+
+from .label_mapping import CLASS_TO_IDX, LABEL_MAPPING, map_label
+from .transforms import build_transform
+
+logger = logging.getLogger(__name__)
+
+IMG_EXTENSIONS = (".png", ".jpg", ".jpeg", ".tif", ".tiff")
+
+
+def scan_imagefolder(root: str | Path, split: str) -> list[tuple[Path, str]]:
+    """Return [(image_path, raw_label), ...] for <root>/<split>/<RAW_LABEL>/*.
+
+    raw_label is the immediate parent directory name (e.g. "CORAL", "CORAL_BL").
+    """
+    split_dir = Path(root) / split
+    if not split_dir.is_dir():
+        raise FileNotFoundError(f"split dir not found: {split_dir}")
+    samples: list[tuple[Path, str]] = []
+    for label_dir in sorted(p for p in split_dir.iterdir() if p.is_dir()):
+        raw_label = label_dir.name
+        for img in label_dir.rglob("*"):
+            if img.suffix.lower() in IMG_EXTENSIONS:
+                samples.append((img, raw_label))
+    return samples
+
+
+class CoralPatchDataset(Dataset):
+    """ImageFolder-style health-state classification dataset."""
+
+    def __init__(
+        self,
+        samples: list[tuple[Path, str]],
+        train: bool = False,
+        allow_empty_mapping: bool = False,
+    ) -> None:
+        """
+        Args:
+            samples: list of (image_path, raw_label) — build with `from_imagefolder`.
+            train: apply train-time augmentations if True.
+            allow_empty_mapping: TEST-MODE ESCAPE HATCH. Must be True to construct while
+                LABEL_MAPPING is empty. Real training/inference leaves this False so an
+                unfilled mapping fails loudly instead of passing raw label strings as
+                targets into Phase 3.
+        """
+        # Guard: never let an unfilled label_mapping.py reach training as a no-op.
+        if not LABEL_MAPPING and not allow_empty_mapping:
+            raise RuntimeError(
+                "LABEL_MAPPING is empty — fill backend/data/label_mapping.py from the EDA "
+                "raw-label table before building a real CoralPatchDataset. "
+                "Pass allow_empty_mapping=True ONLY in tests/smoke checks."
+            )
+
+        self.transform = build_transform(train=train)
+
+        kept: list[tuple[Path, int]] = []
+        dropped: Counter = Counter()
+        for path, raw in samples:
+            mapped = map_label(raw)
+            if mapped is None:
+                dropped[raw] += 1
+                continue
+            kept.append((path, CLASS_TO_IDX[mapped]))
+        if dropped:
+            logger.warning("Dropping %d annotations with unmapped/None labels: %s",
+                           sum(dropped.values()), dict(dropped))
+        self.samples = kept
+        if not kept:
+            logger.warning(
+                "CoralPatchDataset is EMPTY — LABEL_MAPPING covers none of the raw labels."
+            )
+
+    @classmethod
+    def from_imagefolder(
+        cls, root: str | Path, split: str, train: bool | None = None,
+        allow_empty_mapping: bool = False,
+    ) -> "CoralPatchDataset":
+        """Build from <root>/<split>/<RAW_LABEL>/*. Augments iff split == 'train' (override
+        with `train`)."""
+        samples = scan_imagefolder(root, split)
+        do_train = (split == "train") if train is None else train
+        return cls(samples, train=do_train, allow_empty_mapping=allow_empty_mapping)
+
+    def __len__(self) -> int:
+        return len(self.samples)
+
+    def __getitem__(self, idx: int):
+        path, label_idx = self.samples[idx]
+        with Image.open(path) as im:
+            im = im.convert("RGB")
+            tensor = self.transform(im)
+        return tensor, label_idx

backend/data/split.py

@@ -0,0 +1,92 @@
+"""Train/val/test split, stratified BY IMAGE on the image's majority class. Phase 2.
+
+RESERVED FOR FUTURE POINT-ANNOTATED SOURCES — NOT used for the current NOAA dataset.
+The NOAA-PIFSC bleaching dataset ships with native, site/year-controlled train/val/test
+splits, so we use those directly (see patch_dataset.CoralPatchDataset.from_imagefolder)
+and must NOT re-split it here (that would risk the very site leakage NOAA already
+controlled for). This module is kept for a future CoralNet-style point-annotated source.
+
+Splitting by image (not by point) prevents leakage — many points share one image, so a
+point-level split would put correlated patches from the same image in both train and val.
+Each image is assigned its majority mapped label and stratified on that.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+import pandas as pd
+from sklearn.model_selection import train_test_split
+
+from .label_mapping import map_label
+
+DEFAULT_RATIOS = (0.70, 0.15, 0.15)  # train, val, test
+
+
+def _split_once(frame: pd.DataFrame, test_size: float, random_state: int,
+                stratify_col: str) -> tuple[pd.DataFrame, pd.DataFrame]:
+    """train_test_split, stratified when possible, falling back to random.
+
+    Stratification needs >=2 samples per class AND each resulting partition to hold at
+    least one sample per class; on tiny/dev data that can't hold, so we fall back rather
+    than crash. On real data (thousands of images) the stratified path is taken.
+    """
+    strat = frame[stratify_col]
+    if strat.value_counts().min() >= 2:
+        try:
+            return train_test_split(frame, test_size=test_size,
+                                    random_state=random_state, stratify=strat)
+        except ValueError:
+            pass  # partition too small to carry every class -> fall back
+    return train_test_split(frame, test_size=test_size, random_state=random_state)
+
+
+def _image_majority_labels(df: pd.DataFrame) -> pd.DataFrame:
+    """One row per image with its majority mapped label (unmapped points ignored)."""
+    d = df.copy()
+    d["mapped"] = d["label"].map(map_label)
+    d = d[d["mapped"].notna()]
+    if len(d) == 0:
+        return pd.DataFrame(columns=["image_name", "majority"])
+    majority = (
+        d.groupby("image_name")["mapped"]
+        .agg(lambda s: s.value_counts().idxmax())
+        .reset_index()
+        .rename(columns={"mapped": "majority"})
+    )
+    return majority
+
+
+def stratified_split_by_image(
+    annotations_csv: str | Path,
+    ratios: tuple[float, float, float] = DEFAULT_RATIOS,
+    random_state: int = 42,
+) -> dict[str, pd.DataFrame]:
+    """Return {'train','val','test'} -> annotation DataFrames (point rows).
+
+    Stratification falls back to a non-stratified split for any class too small to
+    appear in every split (common with tiny/dev data).
+    """
+    assert abs(sum(ratios) - 1.0) < 1e-6, "ratios must sum to 1"
+    df = pd.read_csv(annotations_csv)
+    img_labels = _image_majority_labels(df)
+    if len(img_labels) < 3:
+        raise ValueError(
+            f"Need >=3 labeled images to split; got {len(img_labels)}. "
+            "This is expected with the synthetic stub data — provide real data."
+        )
+
+    train_r, val_r, test_r = ratios
+    train_imgs, hold_imgs = _split_once(
+        img_labels, test_size=(val_r + test_r),
+        random_state=random_state, stratify_col="majority",
+    )
+    rel_test = test_r / (val_r + test_r)
+    val_imgs, test_imgs = _split_once(
+        hold_imgs, test_size=rel_test,
+        random_state=random_state, stratify_col="majority",
+    )
+
+    def rows_for(imgs: pd.DataFrame) -> pd.DataFrame:
+        return df[df["image_name"].isin(imgs["image_name"])].reset_index(drop=True)
+
+    return {"train": rows_for(train_imgs), "val": rows_for(val_imgs), "test": rows_for(test_imgs)}

backend/data/transforms.py

@@ -0,0 +1,58 @@
+"""Preprocessing transforms for DINOv2-B. Phase 2.
+
+Patch geometry (locked in CLAUDE.md, must be IDENTICAL in train and inference):
+  - Training (current NOAA 2-class data): the dataset images ARE colony patches, so we use
+    the WHOLE image — resize to 224x224, ImageNet-normalize. No point/centroid crop.
+  - Inference: crop the SAM2 mask's bounding box, then resize 224 + normalize (same tail).
+DINOv2 expects inputs sized to multiples of 14; 224 = 16 * 14.
+
+`crop_patch_around_point` below is retained for FUTURE point-annotated sources (paired with
+backend.data.split); it is NOT used by the current imagefolder pipeline.
+"""
+from __future__ import annotations
+
+from PIL import Image
+from torchvision import transforms
+
+# DINOv2 uses ImageNet normalization.
+IMAGENET_MEAN = (0.485, 0.456, 0.406)
+IMAGENET_STD = (0.229, 0.224, 0.225)
+INPUT_SIZE = 224
+
+# Default square window (in source pixels) cropped around a point before resizing.
+# CoralNet-style patch classification; tune in Phase 3 if needed.
+DEFAULT_CROP_SIZE = 224
+
+
+def build_transform(train: bool) -> transforms.Compose:
+    """Return the tensor transform applied to an already-cropped PIL patch."""
+    if train:
+        return transforms.Compose([
+            transforms.Resize((INPUT_SIZE, INPUT_SIZE)),
+            transforms.RandomHorizontalFlip(),
+            transforms.ColorJitter(brightness=0.2, contrast=0.2),
+            transforms.ToTensor(),
+            transforms.Normalize(IMAGENET_MEAN, IMAGENET_STD),
+        ])
+    return transforms.Compose([
+        transforms.Resize((INPUT_SIZE, INPUT_SIZE)),
+        transforms.ToTensor(),
+        transforms.Normalize(IMAGENET_MEAN, IMAGENET_STD),
+    ])
+
+
+def crop_patch_around_point(img: Image.Image, row: int, col: int,
+                            crop_size: int = DEFAULT_CROP_SIZE) -> Image.Image:
+    """Crop a square `crop_size` window centered on (row, col), clamped to image bounds.
+
+    (row, col) follow CoralNet convention: row = y (vertical), col = x (horizontal).
+    """
+    half = crop_size // 2
+    width, height = img.size
+    left = max(0, min(col - half, width - crop_size))
+    top = max(0, min(row - half, height - crop_size))
+    left = max(0, left)
+    top = max(0, top)
+    right = min(width, left + crop_size)
+    bottom = min(height, top + crop_size)
+    return img.crop((left, top, right, bottom))

backend/db/schema.sql

@@ -0,0 +1,173 @@
+-- ReefScan — Supabase (Postgres) schema
+-- Phase 1 deliverable.
+--
+-- Tables:
+--   jobs             — async inference jobs (POST /infer enqueues, GET /infer/{job_id} polls)
+--   reef_locations   — named reef sites; uploads attach to one
+--   inference_logs   — one row per classified SEGMENT (grouped by request_id);
+--                      powers observability + drift views
+--   review_queue     — uncertain predictions (conformal set size > 1) awaiting human label
+--   human_labels     — confirmed labels from /admin/review; feeds manual retraining
+--   health_snapshots — per-upload aggregate per reef; powers the temporal tracker
+--                      (stores BOTH per-class counts and per-class pixel areas)
+--
+-- Conventions: uuid PKs, timestamptz, jsonb for variable-shape fields.
+-- Apply via Supabase SQL editor or `supabase db push`.
+
+-- ---------------------------------------------------------------------------
+-- Enums
+-- ---------------------------------------------------------------------------
+-- coral_label keeps all 4 values, but the INITIAL model is 2-class (healthy/bleached);
+-- 'dead' and 'algae_covered' are RESERVED for a future extension (ReefNet supplementation)
+-- so no enum migration is needed when they arrive. See CLAUDE.md / label_mapping.py.
+do $$ begin
+  create type coral_label as enum ('healthy', 'bleached', 'dead', 'algae_covered');
+exception when duplicate_object then null; end $$;
+
+do $$ begin
+  create type review_status as enum ('pending', 'confirmed', 'rejected');
+exception when duplicate_object then null; end $$;
+
+do $$ begin
+  create type job_status as enum ('queued', 'processing', 'complete', 'failed');
+exception when duplicate_object then null; end $$;
+
+-- ---------------------------------------------------------------------------
+-- reef_locations  (defined first — referenced by jobs/inference_logs/etc.)
+-- ---------------------------------------------------------------------------
+create table if not exists reef_locations (
+  id           uuid primary key default gen_random_uuid(),
+  name         text not null,
+  latitude     double precision,
+  longitude    double precision,
+  description  text,
+  created_at   timestamptz not null default now()
+);
+
+-- ---------------------------------------------------------------------------
+-- jobs  (async inference; POST /infer enqueues, GET /infer/{job_id} polls)
+-- Same pipeline for image and video (a video is just N frames).
+-- ---------------------------------------------------------------------------
+create table if not exists jobs (
+  job_id            uuid primary key default gen_random_uuid(),
+  status            job_status not null default 'queued',
+  reef_location_id  uuid references reef_locations(id) on delete set null,
+  source_kind       text,                              -- 'image' | 'video'
+  source_url        text,                              -- R2 url of the uploaded file
+  created_at        timestamptz not null default now(),
+  completed_at      timestamptz,
+  result_json       jsonb,                             -- structured pipeline output when complete
+  error_message     text                               -- populated when status = 'failed'
+);
+
+create index if not exists idx_jobs_status  on jobs (status);
+create index if not exists idx_jobs_created on jobs (created_at);
+
+-- ---------------------------------------------------------------------------
+-- inference_logs  (one row per classified segment)
+-- ---------------------------------------------------------------------------
+create table if not exists inference_logs (
+  id                  uuid primary key default gen_random_uuid(),
+  request_id          uuid not null,                 -- groups all segments of one /infer call
+  image_id            text not null,                 -- R2 object key / source image id
+  segment_id          int  not null,                 -- index of the SAM2 mask within the image
+  reef_location_id    uuid references reef_locations(id) on delete set null,
+
+  ts                  timestamptz not null default now(),
+  latency_ms          integer not null,              -- full-image pipeline latency (same per request)
+
+  -- raw softmax per class (kept as explicit columns for easy SQL aggregation)
+  conf_healthy        real not null,
+  conf_bleached       real not null,
+  conf_dead           real not null,
+  conf_algae_covered  real not null,
+
+  -- conformal output
+  prediction_set      jsonb not null,                -- e.g. ["healthy","algae_covered"]
+  prediction_set_size int  not null,                 -- 1 = confident, >1 = uncertain
+  predicted_label     coral_label,                   -- argmax / point label for convenience
+
+  model_version       text not null,
+
+  created_at          timestamptz not null default now()
+);
+
+create index if not exists idx_inflogs_ts        on inference_logs (ts);
+create index if not exists idx_inflogs_request   on inference_logs (request_id);
+create index if not exists idx_inflogs_location  on inference_logs (reef_location_id);
+create index if not exists idx_inflogs_setsize   on inference_logs (prediction_set_size);
+create index if not exists idx_inflogs_version   on inference_logs (model_version);
+
+-- ---------------------------------------------------------------------------
+-- review_queue  (uncertain predictions -> human review)
+-- ---------------------------------------------------------------------------
+create table if not exists review_queue (
+  id                  uuid primary key default gen_random_uuid(),
+  request_id          uuid not null,
+  image_id            text not null,
+  segment_id          int  not null,
+  reef_location_id    uuid references reef_locations(id) on delete set null,
+
+  patch_url           text,                          -- R2 url of the centroid patch crop
+  image_url           text,                          -- R2 url of the full source image
+
+  prediction_set      jsonb not null,                -- candidate labels shown to the human
+  conf_healthy        real,
+  conf_bleached       real,
+  conf_dead           real,
+  conf_algae_covered  real,
+
+  model_version       text not null,
+  status              review_status not null default 'pending',
+
+  created_at          timestamptz not null default now()
+);
+
+create index if not exists idx_review_status   on review_queue (status);
+create index if not exists idx_review_created  on review_queue (created_at);
+
+-- ---------------------------------------------------------------------------
+-- human_labels  (confirmed labels; retrain trigger at 100 new rows)
+-- ---------------------------------------------------------------------------
+create table if not exists human_labels (
+  id               uuid primary key default gen_random_uuid(),
+  review_queue_id  uuid references review_queue(id) on delete set null,
+  image_id         text not null,
+  segment_id       int  not null,
+  confirmed_label  coral_label not null,
+  labeled_by       text,                             -- reviewer id/email
+  used_in_training boolean not null default false,   -- flips true once consumed by a retrain
+  created_at       timestamptz not null default now()
+);
+
+create index if not exists idx_humanlabels_unused  on human_labels (used_in_training);
+create index if not exists idx_humanlabels_created on human_labels (created_at);
+
+-- ---------------------------------------------------------------------------
+-- health_snapshots  (per-upload aggregate; temporal tracker source)
+-- Stores BOTH counts and pixel areas per class so the UI can show count-% or area-%.
+-- ---------------------------------------------------------------------------
+create table if not exists health_snapshots (
+  id                  uuid primary key default gen_random_uuid(),
+  reef_location_id    uuid references reef_locations(id) on delete cascade,
+  request_id          uuid not null,
+  source_image_id     text not null,
+  snapshot_time       timestamptz not null default now(),
+
+  total_segments      int  not null,
+  healthy_count       int  not null default 0,
+  bleached_count      int  not null default 0,
+  dead_count          int  not null default 0,
+  algae_covered_count int  not null default 0,
+
+  total_area_px            bigint not null default 0,
+  healthy_area_px          bigint not null default 0,
+  bleached_area_px         bigint not null default 0,
+  dead_area_px             bigint not null default 0,
+  algae_covered_area_px    bigint not null default 0,
+
+  created_at          timestamptz not null default now()
+);
+
+create index if not exists idx_snapshots_location_time
+  on health_snapshots (reef_location_id, snapshot_time);

backend/db/seed_demo.sql

@@ -0,0 +1,85 @@
+-- ReefScan demo seed — populates a fresh Supabase so the tracker + dashboard render with
+-- data before any real uploads. Run AFTER schema.sql, in the Supabase SQL editor.
+-- Safe to re-run: it clears the demo rows first (by the fixed reef UUIDs / 'demo%' ids).
+
+delete from inference_logs where image_id like 'demo%';
+delete from review_queue where image_id like 'demo%';
+delete from health_snapshots where source_image_id like 'demo%';
+delete from reef_locations where id in (
+  '11111111-1111-1111-1111-111111111111',
+  '22222222-2222-2222-2222-222222222222',
+  '33333333-3333-3333-3333-333333333333');
+
+-- ---- reef_locations ----
+insert into reef_locations (id, name, latitude, longitude, description) values
+  ('11111111-1111-1111-1111-111111111111', 'Kāneʻohe Bay — Patch Reef 12', 21.45, -157.79, 'demo'),
+  ('22222222-2222-2222-2222-222222222222', 'Molokini Crater', 20.63, -156.49, 'demo'),
+  ('33333333-3333-3333-3333-333333333333', 'Hanauma Bay', 21.27, -157.69, 'demo');
+
+-- ---- health_snapshots: 9 monthly surveys per reef, healthy % declining toward now ----
+insert into health_snapshots (reef_location_id, request_id, source_image_id, snapshot_time,
+  total_segments, healthy_count, bleached_count, dead_count, algae_covered_count,
+  total_area_px, healthy_area_px, bleached_area_px, dead_area_px, algae_covered_area_px)
+select r.id, gen_random_uuid(), 'demo-survey',
+  date_trunc('month', now()) - (m || ' months')::interval,
+  20, g.gh, 20 - g.gh, 0, 0,
+  200000, (200000 * g.gh / 20.0)::bigint, (200000 * (20 - g.gh) / 20.0)::bigint, 0, 0
+from (values
+  ('11111111-1111-1111-1111-111111111111'::uuid, 17),
+  ('22222222-2222-2222-2222-222222222222'::uuid, 18),
+  ('33333333-3333-3333-3333-333333333333'::uuid, 15)
+) as r(id, base)
+cross join generate_series(0, 8) as m
+cross join lateral (
+  select greatest(6, least(20, r.base - (8 - m) + (random() * 2 - 1)::int)) as gh
+) as g;
+
+-- ---- inference_logs: 14 days x 3 reefs x 6 segments ----
+-- Health declines + uncertainty (set size) rises toward now (drift story). Confidences are
+-- bimodal (mostly near 0/1) like a real model, so ~10-20% land in the conformal band
+-- [1-qhat, qhat] = [0.372, 0.628] and become uncertain. NB the per-row random()s live in a
+-- MATERIALIZED CTE — a plain lateral subquery gets hoisted and evaluated ONCE for the whole
+-- insert (every row identical). Don't "simplify" it back to a lateral.
+insert into inference_logs (request_id, image_id, segment_id, reef_location_id, ts, latency_ms,
+  conf_healthy, conf_bleached, conf_dead, conf_algae_covered, prediction_set, prediction_set_size,
+  predicted_label, model_version)
+with base as materialized (
+  select r.id as reef, d, s, random() as u, random() as ra,
+    (0.62 - (13 - d) * 0.012) as p_h,   -- healthy prob declines toward now
+    (0.08 + (13 - d) * 0.009) as p_u    -- uncertain prob rises toward now (drift)
+  from (values
+    ('11111111-1111-1111-1111-111111111111'::uuid),
+    ('22222222-2222-2222-2222-222222222222'::uuid),
+    ('33333333-3333-3333-3333-333333333333'::uuid)
+  ) r(id)
+  cross join generate_series(0, 13) d
+  cross join generate_series(1, 6) s
+),
+draw as (
+  select *,
+    case
+      when u < p_u       then 0.45 + ra * 0.10   -- uncertain band
+      when u < p_u + p_h then 0.70 + ra * 0.28   -- confident healthy
+      else                    0.02 + ra * 0.28   -- confident bleached
+    end as rh
+  from base
+)
+select gen_random_uuid(), 'demo-' || d || '-' || s, s, reef,
+  now() - (d || ' days')::interval - (s || ' minutes')::interval,
+  16000 + (13 - d) * 250 + (random() * 4000)::int,
+  round(rh::numeric, 3), round((1 - rh)::numeric, 3), 0, 0,
+  case when rh between 0.372 and 0.628 then '["healthy","bleached"]'::jsonb
+       when rh >= 0.5 then '["healthy"]'::jsonb else '["bleached"]'::jsonb end,
+  case when rh between 0.372 and 0.628 then 2 else 1 end,
+  (case when rh >= 0.5 then 'healthy' else 'bleached' end)::coral_label,
+  'reefscan-dinov2-coral-v1-linearprobe'
+from draw;
+
+-- ---- review_queue: a few pending uncertain segments ----
+insert into review_queue (request_id, image_id, segment_id, reef_location_id, image_url,
+  prediction_set, conf_healthy, conf_bleached, model_version, status)
+select gen_random_uuid(), 'demo-review-' || g, g, '11111111-1111-1111-1111-111111111111'::uuid,
+  'local://demo.jpg', '["healthy","bleached"]'::jsonb,
+  round((0.45 + random() * 0.1)::numeric, 3), round((0.45 + random() * 0.1)::numeric, 3),
+  'reefscan-dinov2-coral-v1-linearprobe', 'pending'
+from generate_series(1, 5) as g;

backend/inference/__init__.py

@@ -0,0 +1 @@
+"""ReefScan inference pipeline (Phase 5): enhance -> frames -> segment -> classify -> log."""

backend/inference/classifier.py

@@ -0,0 +1,126 @@
+"""DINOv2-B classifier + MAPIE-style split conformal. Phase 4/5.
+
+Loads the trained head + backbone and the conformal calibration (qhat, LAC) from the HF
+weights repo. Each SAM2 mask is classified on its bbox crop (resize 224 + ImageNet norm —
+identical to training). Conformal LAC builds the prediction SET: include class k iff
+1 - p_k <= qhat; never emit an empty set. Set size > 1 => uncertain => review_queue.
+
+When weights are unavailable (stub mode), deterministic softmax is produced so the pipeline
+and the uncertainty/review path are fully exercisable.
+"""
+from __future__ import annotations
+
+import logging
+
+from PIL import Image
+
+from ..config import settings
+
+logger = logging.getLogger(__name__)
+
+CLASSES = list(settings.classes)  # ["healthy", "bleached"]
+_STUB_VERSION = "reefscan-stub-v0"
+_STUB_QHAT = 0.6  # threshold 0.4 -> segments within ~40-60 become uncertain (demo-visible)
+
+_model = None
+_tf = None
+_qhat: float = _STUB_QHAT
+_model_version: str = _STUB_VERSION
+
+
+# ---------------------------------------------------------------------------
+# Model (mirrors notebooks/01_train_dinov2 DINOv2Classifier)
+# ---------------------------------------------------------------------------
+def _build_model(num_classes: int):
+    import torch.nn as nn
+    from transformers import AutoModel
+
+    class DINOv2Classifier(nn.Module):
+        def __init__(self) -> None:
+            super().__init__()
+            self.backbone = AutoModel.from_pretrained("facebook/dinov2-base")
+            self.head = nn.Linear(self.backbone.config.hidden_size, num_classes)
+
+        def forward(self, x):
+            o = self.backbone(pixel_values=x)
+            cls = getattr(o, "pooler_output", None)
+            if cls is None:
+                cls = o.last_hidden_state[:, 0]
+            return self.head(cls)
+
+    return DINOv2Classifier()
+
+
+def load() -> None:
+    """Load DINOv2 head/backbone + conformal.json from the HF weights repo. Raises if absent."""
+    global _model, _tf, _qhat, _model_version
+    import json
+
+    import torch
+    from huggingface_hub import hf_hub_download
+    from safetensors.torch import load_file
+    from torchvision import transforms
+
+    stage = settings.hf_stage
+    w = hf_hub_download(settings.hf_repo, f"{stage}/model.safetensors", token=settings.hf_token)
+    model = _build_model(len(CLASSES))
+    model.load_state_dict(load_file(w))
+    model.eval()
+    _model = model
+
+    _tf = transforms.Compose([
+        transforms.Resize((settings.input_size, settings.input_size)),
+        transforms.ToTensor(),
+        transforms.Normalize((0.485, 0.456, 0.406), (0.229, 0.224, 0.225)),
+    ])
+
+    cj = hf_hub_download(settings.hf_repo, f"{stage}/conformal.json", token=settings.hf_token)
+    meta = json.load(open(cj))
+    _qhat = float(meta["qhat"])
+    _model_version = meta.get("model_version", f"reefscan-dinov2-coral-{stage}")
+    logger.info("DINOv2 classifier loaded (version=%s, qhat=%.4f)", _model_version, _qhat)
+
+
+def is_loaded() -> bool:
+    return _model is not None
+
+
+def qhat() -> float:
+    return _qhat
+
+
+def model_version() -> str:
+    return _model_version
+
+
+def classify(crop: Image.Image) -> dict[str, float]:
+    """Return softmax {healthy, bleached} for a bbox crop."""
+    if _model is None:
+        return _stub_probs(crop)
+    import torch
+
+    x = _tf(crop.convert("RGB")).unsqueeze(0)
+    with torch.inference_mode():
+        p = torch.softmax(_model(x), dim=1)[0].tolist()
+    return {CLASSES[i]: float(p[i]) for i in range(len(CLASSES))}
+
+
+def conformal_set(probs: dict[str, float], q: float) -> tuple[list[str], int]:
+    """LAC split-conformal set: include class k iff (1 - p_k) <= qhat. Never empty."""
+    keep = [c for c in CLASSES if (1.0 - probs[c]) <= q]
+    if not keep:
+        keep = [max(probs, key=probs.get)]
+    # order by descending confidence for stable display
+    keep.sort(key=lambda c: probs[c], reverse=True)
+    return keep, len(keep)
+
+
+def _stub_probs(crop: Image.Image) -> dict[str, float]:
+    """Deterministic softmax from the crop's mean color — yields a mix incl. near-even
+    (uncertain) segments so the review path is exercised."""
+    import numpy as np
+
+    arr = np.asarray(crop.convert("RGB").resize((16, 16)), dtype="float32") / 255.0
+    # 'bluer / darker' -> leans bleached-ish; just a deterministic spread, not real signal
+    h = float(np.clip(0.5 + (arr[..., 1].mean() - arr[..., 2].mean()) * 1.4, 0.05, 0.95))
+    return {"healthy": h, "bleached": 1.0 - h}

backend/inference/enhancer.py

@@ -0,0 +1,14 @@
+"""WaterNet underwater enhancement. Phase 5.
+
+Currently an identity passthrough (real WaterNet weights are vendored later — the
+WaterNet-over-CLAHE rationale stays a README talking point). Kept as a seam so the real
+model drops in without touching the pipeline.
+"""
+from __future__ import annotations
+
+from PIL import Image
+
+
+def enhance(img: Image.Image) -> Image.Image:
+    """Return an enhanced copy of the image. Identity for now."""
+    return img

backend/inference/frame_extractor.py

@@ -0,0 +1,57 @@
+"""Frame extraction. Phase 5.
+
+Image input passes through as a single frame. Video input is split on scene changes via
+PySceneDetect (NOT fixed-fps). Both feed the same downstream pipeline — the image/video
+distinction collapses (CLAUDE.md). Video decoding degrades gracefully if scenedetect/opencv
+are unavailable.
+"""
+from __future__ import annotations
+
+import io
+import logging
+import tempfile
+from pathlib import Path
+
+from PIL import Image
+
+logger = logging.getLogger(__name__)
+
+MAX_VIDEO_FRAMES = 12  # cap representative frames per clip (free-CPU budget)
+
+
+def extract(data: bytes, kind: str) -> list[Image.Image]:
+    """Return a list of frames (PIL RGB) for an image or video upload."""
+    if kind == "video":
+        return _video_frames(data)
+    return [Image.open(io.BytesIO(data)).convert("RGB")]
+
+
+def _video_frames(data: bytes) -> list[Image.Image]:
+    try:
+        import cv2  # type: ignore
+        from scenedetect import ContentDetector, SceneManager, open_video  # type: ignore
+    except Exception as e:  # noqa: BLE001
+        raise RuntimeError(f"video support needs scenedetect+opencv: {e}")
+
+    with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+        f.write(data)
+        path = f.name
+    try:
+        video = open_video(path)
+        sm = SceneManager()
+        sm.add_detector(ContentDetector())
+        sm.detect_scenes(video)
+        scenes = sm.get_scene_list()
+        cap = cv2.VideoCapture(path)
+        frames: list[Image.Image] = []
+        # one representative frame per scene (scene start), capped
+        targets = [s[0].get_frames() for s in scenes] or [0]
+        for fno in targets[:MAX_VIDEO_FRAMES]:
+            cap.set(cv2.CAP_PROP_POS_FRAMES, fno)
+            ok, frame = cap.read()
+            if ok:
+                frames.append(Image.fromarray(frame[:, :, ::-1]))  # BGR -> RGB
+        cap.release()
+        return frames or [Image.new("RGB", (512, 512))]
+    finally:
+        Path(path).unlink(missing_ok=True)

backend/inference/pipeline.py

@@ -0,0 +1,146 @@
+"""Full inference pipeline orchestration. Phase 5.
+
+enhance -> frames (video) / passthrough (image) -> SAM2 AMG segment -> per-segment bbox
+crop -> DINOv2 + conformal classify -> assemble the frozen contract -> log to Supabase +
+store source to R2 -> return result dict.
+
+Returns the InferenceResponse fields (minus job_id/status, which the JobStore adds).
+Runs synchronously in a threadpool (called via run_in_executor) — CPU-bound.
+"""
+from __future__ import annotations
+
+import logging
+import time
+
+from ..config import settings
+from ..persistence import new_request_id, r2, supabase
+from . import classifier, enhancer, frame_extractor, segmenter
+
+logger = logging.getLogger(__name__)
+
+_last_latency_ms: int | None = None
+
+
+def load_models() -> None:
+    """Load SAM2 + DINOv2 once at startup (resident, no lazy-load — Phase 1.5). On any
+    failure (e.g. weights not yet on the Hub) we log and fall back to STUB mode so the
+    service still answers with contract-valid output."""
+    if settings.stub_mode:
+        logger.warning("REEFSCAN_STUB=1 -> running inference in STUB mode")
+        return
+    try:
+        segmenter.load()
+        classifier.load()
+    except Exception as e:  # noqa: BLE001
+        logger.warning("model load failed (%s) -> STUB mode (contract-valid synthetic output)", e)
+
+
+def models_loaded() -> bool:
+    return segmenter.is_loaded() and classifier.is_loaded()
+
+
+def model_version() -> str:
+    return classifier.model_version()
+
+
+def last_latency_ms() -> int | None:
+    return _last_latency_ms
+
+
+def run(job_id: str, data: bytes, kind: str, reef_location_id: str | None) -> dict:
+    global _last_latency_ms
+    t0 = time.perf_counter()
+    request_id = new_request_id()
+
+    content_type = "video/mp4" if kind == "video" else "image/jpeg"
+    image_url = r2.upload(f"{request_id}/source", data, content_type)
+
+    frames = frame_extractor.extract(data, kind)
+    img = enhancer.enhance(frames[0])  # scaffold: classify on first frame (N-frame agg = future)
+    W, H = img.size
+
+    raw = segmenter.segment(img)
+    q = classifier.qhat()
+    total_area = sum(s["mask_area_px"] for s in raw) or 1
+
+    segments: list[dict] = []
+    for i, s in enumerate(raw, start=1):
+        crop = img.crop(tuple(s["bbox"]))
+        probs = classifier.classify(crop)
+        pset, size = classifier.conformal_set(probs, q)
+        predicted = max(probs, key=probs.get)
+        segments.append({
+            "segment_id": i,
+            "mask_area_px": s["mask_area_px"],
+            "bbox": s["bbox"],
+            "predicted_class": predicted,
+            "prediction_set": pset,
+            "prediction_set_size": size,
+            "confidence_scores": {k: round(v, 4) for k, v in probs.items()},
+            "coverage_pct": round(s["mask_area_px"] / total_area * 100, 1),
+        })
+
+    summary = _summarize(segments, total_area)
+    latency = int((time.perf_counter() - t0) * 1000)
+    _last_latency_ms = latency
+    version = classifier.model_version()
+
+    _log_all(request_id, image_url, reef_location_id, segments, summary,
+             total_area, latency, version)
+
+    return {
+        "processing_time_ms": latency,
+        "image_url": image_url,
+        "model_version": version,
+        "image_width": W,
+        "image_height": H,
+        "segments": segments,
+        "summary": summary,
+    }
+
+
+def _summarize(segments: list[dict], total_area: int) -> dict:
+    healthy_area = sum(s["mask_area_px"] for s in segments if s["predicted_class"] == "healthy")
+    healthy_pct = round(healthy_area / total_area * 100, 1) if total_area else 0.0
+    return {
+        "total_segments": len(segments),
+        "area_weighted": {"healthy_pct": healthy_pct, "bleached_pct": round(100 - healthy_pct, 1)},
+        "uncertain_segments": sum(1 for s in segments if s["prediction_set_size"] > 1),
+        "dominant_status": "healthy" if healthy_pct >= 50 else "bleached",
+    }
+
+
+def _log_all(request_id, image_url, reef_id, segments, summary, total_area, latency, version):
+    """Best-effort Supabase logging (no-op without creds). Mirrors backend/db/schema.sql."""
+    inf_rows, review_rows = [], []
+    counts = {"healthy": 0, "bleached": 0}
+    areas = {"healthy": 0, "bleached": 0}
+    for s in segments:
+        cs = s["confidence_scores"]
+        inf_rows.append({
+            "request_id": request_id, "image_id": image_url, "segment_id": s["segment_id"],
+            "reef_location_id": reef_id, "latency_ms": latency,
+            "conf_healthy": cs.get("healthy", 0.0), "conf_bleached": cs.get("bleached", 0.0),
+            "conf_dead": 0.0, "conf_algae_covered": 0.0,
+            "prediction_set": s["prediction_set"], "prediction_set_size": s["prediction_set_size"],
+            "predicted_label": s["predicted_class"], "model_version": version,
+        })
+        counts[s["predicted_class"]] += 1
+        areas[s["predicted_class"]] += s["mask_area_px"]
+        if s["prediction_set_size"] > 1:
+            review_rows.append({
+                "request_id": request_id, "image_id": image_url, "segment_id": s["segment_id"],
+                "reef_location_id": reef_id, "image_url": image_url,
+                "prediction_set": s["prediction_set"],
+                "conf_healthy": cs.get("healthy", 0.0), "conf_bleached": cs.get("bleached", 0.0),
+                "model_version": version, "status": "pending",
+            })
+    supabase.log_segments(inf_rows)
+    supabase.log_review(review_rows)
+    supabase.log_snapshot({
+        "reef_location_id": reef_id, "request_id": request_id, "source_image_id": image_url,
+        "total_segments": summary["total_segments"],
+        "healthy_count": counts["healthy"], "bleached_count": counts["bleached"],
+        "total_area_px": total_area,
+        "healthy_area_px": areas["healthy"], "bleached_area_px": areas["bleached"],
+    })

backend/inference/segmenter.py

@@ -0,0 +1,91 @@
+"""SAM2 segmentation. Phase 5.
+
+SAM2-Hiera-Small via the Automatic Mask Generator (grid point prompts) — NOT manual
+clicks, NOT a trained detection head (that is documented future work). Returns masks +
+centroids; centroids feed classifier.py's patch crop.
+
+AMG config is LOCKED by the Phase 1.5 sweep (read from config.settings, never inlined):
+    points_per_side = 16 ; AMG input downscaled so longest edge = 512 px.
+
+`segment()` returns a list of {bbox:[x0,y0,x1,y1], mask_area_px:int} in ORIGINAL image px.
+When SAM2 isn't loaded (stub mode / weights absent), a deterministic grid of boxes is
+returned so the whole pipeline + contract are exercisable without the heavy model.
+"""
+from __future__ import annotations
+
+import logging
+
+from PIL import Image
+
+from ..config import settings
+
+logger = logging.getLogger(__name__)
+
+# Locked AMG config (Phase 1.5). Keep in sync with CLAUDE.md.
+AMG_POINTS_PER_SIDE = settings.amg_points_per_side  # 16
+AMG_INPUT_LONGEST_EDGE = settings.amg_longest_edge  # 512
+
+_amg = None  # loaded SAM2AutomaticMaskGenerator
+
+
+def load() -> None:
+    """Load SAM2-Hiera-Small + AMG once at startup. Raises if deps/weights unavailable."""
+    global _amg
+    from sam2.automatic_mask_generator import SAM2AutomaticMaskGenerator  # type: ignore
+    from sam2.build_sam import build_sam2_hf  # type: ignore
+
+    model = build_sam2_hf("facebook/sam2-hiera-small", device="cpu")
+    _amg = SAM2AutomaticMaskGenerator(model, points_per_side=AMG_POINTS_PER_SIDE)
+    logger.info("SAM2 AMG loaded (pps=%d)", AMG_POINTS_PER_SIDE)
+
+
+def is_loaded() -> bool:
+    return _amg is not None
+
+
+def _scale_to_longest(img: Image.Image, longest: int) -> tuple[Image.Image, float]:
+    w, h = img.size
+    scale = longest / max(w, h)
+    if scale >= 1.0:
+        return img, 1.0
+    return img.resize((max(1, round(w * scale)), max(1, round(h * scale)))), scale
+
+
+def segment(img: Image.Image) -> list[dict]:
+    if _amg is None:
+        return _stub_segments(img)
+
+    import numpy as np
+
+    small, scale = _scale_to_longest(img, AMG_INPUT_LONGEST_EDGE)
+    masks = _amg.generate(np.array(small))
+    out: list[dict] = []
+    for m in masks:
+        x, y, w, h = m["bbox"]  # XYWH in downscaled coords
+        out.append({
+            "bbox": [round(x / scale), round(y / scale),
+                     round((x + w) / scale), round((y + h) / scale)],
+            "mask_area_px": int(m["area"] / (scale * scale)),
+        })
+    return out
+
+
+def _stub_segments(img: Image.Image) -> list[dict]:
+    """Deterministic 3x2 grid of boxes — contract-valid output without SAM2."""
+    W, H = img.size
+    cols, rows = 3, 2
+    pad_x, pad_y = W * 0.04, H * 0.05
+    cw, ch = (W - pad_x * (cols + 1)) / cols, (H - pad_y * (rows + 1)) / rows
+    out: list[dict] = []
+    sid = 0
+    for r in range(rows):
+        for c in range(cols):
+            sid += 1
+            x0 = pad_x * (c + 1) + cw * c
+            y0 = pad_y * (r + 1) + ch * r
+            # vary box size a little so coverage % differs per segment
+            shrink = 0.82 + 0.12 * ((sid * 7) % 3) / 2
+            bw, bh = cw * shrink, ch * shrink
+            bbox = [round(x0), round(y0), round(x0 + bw), round(y0 + bh)]
+            out.append({"bbox": bbox, "mask_area_px": round(bw * bh * 0.7)})
+    return out

backend/jobs.py

@@ -0,0 +1,59 @@
+"""Async job store + background worker. Phase 5.
+
+POST /infer creates a job (status 'queued') and schedules run_job; GET /infer/{job_id}
+reads back status + results. In-memory store is the source of truth for polling; the
+Supabase `jobs` table is mirrored best-effort (single-process worker on HF Spaces).
+The CPU-bound pipeline runs in a threadpool so the event loop stays responsive.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any, Optional
+
+from .inference import pipeline
+from .persistence import new_request_id, supabase
+from .schemas import InferenceResponse
+
+logger = logging.getLogger(__name__)
+
+
+class JobStore:
+    def __init__(self) -> None:
+        self._jobs: dict[str, dict[str, Any]] = {}
+
+    def create(self, kind: str, reef_location_id: Optional[str], source_url: str = "") -> str:
+        job_id = new_request_id()
+        self._jobs[job_id] = {"status": "queued", "result": None, "error": None,
+                              "kind": kind, "reef": reef_location_id}
+        supabase.upsert_job({"job_id": job_id, "status": "queued", "source_kind": kind,
+                             "source_url": source_url, "reef_location_id": reef_location_id})
+        return job_id
+
+    def update(self, job_id: str, **kw: Any) -> None:
+        if job_id in self._jobs:
+            self._jobs[job_id].update(kw)
+            row = {"job_id": job_id, "status": self._jobs[job_id]["status"]}
+            if kw.get("error"):
+                row["error_message"] = kw["error"]
+            supabase.upsert_job(row)
+
+    def response(self, job_id: str) -> Optional[InferenceResponse]:
+        j = self._jobs.get(job_id)
+        if j is None:
+            return None
+        result = j.get("result") or {}
+        return InferenceResponse(job_id=job_id, status=j["status"],
+                                 error_message=j.get("error"), **result)
+
+
+async def run_job(store: JobStore, job_id: str, data: bytes, kind: str,
+                  reef_location_id: Optional[str]) -> None:
+    store.update(job_id, status="processing")
+    loop = asyncio.get_event_loop()
+    try:
+        result = await loop.run_in_executor(None, pipeline.run, job_id, data, kind, reef_location_id)
+        store.update(job_id, status="complete", result=result)
+    except Exception as e:  # noqa: BLE001
+        logger.exception("job %s failed", job_id)
+        store.update(job_id, status="failed", error=str(e))

backend/main.py

@@ -0,0 +1,148 @@
+"""ReefScan FastAPI app — async-job inference. Phase 5.
+
+  POST /infer            multipart file (image|video) or form `url` -> {job_id}  (enqueue)
+  GET  /infer/{job_id}   -> InferenceResponse (status + results when complete)  (poll)
+  GET  /health           -> model load status + last inference latency
+
+Images and video share the SAME async path (CLAUDE.md). Models are loaded once at startup
+and kept resident (Phase 1.5: RAM cleared, no lazy-load). The CPU-bound pipeline runs in a
+threadpool so polling stays responsive. Phase 6 frontend swaps lib/api.ts onto these routes.
+"""
+from __future__ import annotations
+
+import asyncio
+import io
+import logging
+from contextlib import asynccontextmanager
+from typing import Optional
+
+from fastapi import Body, FastAPI, File, Form, HTTPException, UploadFile
+from fastapi.middleware.cors import CORSMiddleware
+
+from . import observability
+from .config import settings
+from .inference import pipeline
+from .jobs import JobStore, run_job
+from .persistence import supabase
+from .schemas import HealthResponse, InferenceResponse, SubmitResponse
+
+logging.basicConfig(level=logging.INFO)
+store = JobStore()
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    pipeline.load_models()  # resident SAM2 + DINOv2 (or stub fallback)
+    yield
+
+
+app = FastAPI(title="ReefScan", version="0.1.0", lifespan=lifespan)
+app.add_middleware(
+    CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"],
+)
+
+
+@app.get("/health", response_model=HealthResponse)
+async def health() -> HealthResponse:
+    loaded = pipeline.models_loaded()
+    return HealthResponse(
+        status="ok",
+        models_loaded=loaded,
+        stub_mode=not loaded,
+        model_version=pipeline.model_version(),
+        last_inference_latency_ms=pipeline.last_latency_ms(),
+    )
+
+
+async def _read_input(file: Optional[UploadFile], url: Optional[str]) -> tuple[bytes, str]:
+    """Return (bytes, kind) for an uploaded file or a url. kind in {'image','video'}."""
+    if file is not None:
+        data = await file.read()
+        kind = "video" if (file.content_type or "").startswith("video") else "image"
+        return data, kind
+    if url:
+        try:
+            import httpx
+            async with httpx.AsyncClient(timeout=30) as client:
+                r = await client.get(url)
+                r.raise_for_status()
+            ct = r.headers.get("content-type", "")
+            return r.content, ("video" if ct.startswith("video") else "image")
+        except Exception as e:  # noqa: BLE001
+            raise HTTPException(status_code=400, detail=f"could not fetch url: {e}")
+    raise HTTPException(status_code=400, detail="provide a file upload or a `url`")
+
+
+@app.post("/infer", response_model=SubmitResponse, status_code=202)
+async def infer(
+    file: Optional[UploadFile] = File(default=None),
+    url: Optional[str] = Form(default=None),
+    reef_location_id: Optional[str] = Form(default=None),
+) -> SubmitResponse:
+    data, kind = await _read_input(file, url)
+    if not data:
+        raise HTTPException(status_code=400, detail="empty upload")
+    job_id = store.create(kind, reef_location_id, source_url=url or "")
+    asyncio.create_task(run_job(store, job_id, data, kind, reef_location_id))
+    return SubmitResponse(job_id=job_id)
+
+
+@app.get("/infer/{job_id}", response_model=InferenceResponse)
+async def get_job(job_id: str) -> InferenceResponse:
+    resp = store.response(job_id)
+    if resp is None:
+        raise HTTPException(status_code=404, detail="unknown job_id")
+    return resp
+
+
+# --- read endpoints (Phase 6 review/tracker + Phase 7 dashboard) ---
+# All read from Supabase; return [] when Supabase creds are absent (the frontend keeps
+# its own mock for that case, selected by NEXT_PUBLIC_REEFSCAN_API being unset).
+
+@app.get("/review-queue")
+async def review_queue(limit: int = 50) -> list[dict]:
+    return [{
+        "id": r.get("id"), "image_id": r.get("image_id"), "segment_id": r.get("segment_id"),
+        "patch_url": r.get("patch_url") or "", "image_url": r.get("image_url") or "",
+        "prediction_set": r.get("prediction_set") or [],
+        "confidence_scores": {"healthy": r.get("conf_healthy") or 0.0,
+                              "bleached": r.get("conf_bleached") or 0.0},
+        "model_version": r.get("model_version") or "",
+        "reef_location": r.get("reef_location_id") or "",
+        "created_at": r.get("created_at") or "", "status": r.get("status") or "pending",
+    } for r in supabase.review_queue(limit)]
+
+
+@app.post("/review-queue/{review_id}/confirm")
+async def confirm(review_id: str, label: str = Body(..., embed=True)) -> dict:
+    if label not in settings.classes:
+        raise HTTPException(status_code=400, detail=f"label must be one of {settings.classes}")
+    ok = supabase.confirm_label(review_id, label)
+    return {"ok": ok, "review_id": review_id, "label": label}
+
+
+@app.get("/reef-locations")
+async def reef_locations() -> list[dict]:
+    return [{"id": r.get("id"), "name": r.get("name"),
+             "lat": r.get("latitude"), "lng": r.get("longitude")}
+            for r in supabase.reef_locations()]
+
+
+@app.get("/reef-locations/{reef_id}/snapshots")
+async def snapshots(reef_id: str) -> list[dict]:
+    out = []
+    for r in supabase.snapshots(reef_id):
+        total_area = r.get("total_area_px") or 0
+        healthy_pct = round((r.get("healthy_area_px") or 0) / total_area * 100, 1) if total_area else 0.0
+        out.append({
+            "date": str(r.get("snapshot_time", ""))[:10],
+            "healthy_pct": healthy_pct, "bleached_pct": round(100 - healthy_pct, 1),
+            "total_segments": r.get("total_segments") or 0,
+            "avg_set_size": None,  # drift lives in /observability, not health_snapshots
+        })
+    return out
+
+
+@app.get("/observability")
+async def get_observability() -> dict:
+    return observability.build(supabase.recent_logs(), settings.classes)

backend/models/train_dinov2.py

@@ -0,0 +1,317 @@
+"""DINOv2-B health classifier — canonical training library. Phase 3.
+
+This is the maintained, importable implementation. The Kaggle notebook
+(notebooks/01_train_dinov2_kaggle.ipynb) mirrors this logic inline so it can run
+self-contained on Kaggle; keep the two in sync.
+
+Two stages (CLAUDE.md):
+  - "linear_probe": freeze the DINOv2 backbone, train only the linear head (~30 min).
+  - "finetune":     unfreeze the last 2 transformer blocks + final norm + head (~4 hr).
+
+Hard Kaggle constraints honored:
+  - checkpoint EVERY epoch to <work>/checkpoints/ (persisted as Kaggle output)
+  - resume from the latest checkpoint if one exists (input dataset or working dir)
+  - W&B logging throughout
+  - 2-class head (healthy/bleached) driven by backend.data.CLASSES
+
+Data: ImageFolder via backend.data.CoralPatchDataset using the NOAA NATIVE train/val/test
+splits (do NOT re-split — see CLAUDE.md).
+"""
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import asdict, dataclass, field
+from pathlib import Path
+
+import numpy as np
+import torch
+import torch.nn as nn
+from sklearn.metrics import classification_report, f1_score
+from torch.utils.data import DataLoader
+from transformers import AutoModel
+
+from backend.data import CLASSES, CoralPatchDataset
+
+BACKBONE_NAME = "facebook/dinov2-base"
+
+
+# ---------------------------------------------------------------------------
+# Config
+# ---------------------------------------------------------------------------
+@dataclass
+class TrainConfig:
+    stage: str = "linear_probe"            # "linear_probe" | "finetune"
+    data_root: str = "/kaggle/input/noaa-coral-bleaching"  # imagefolder root (train/val/test)
+    work_dir: str = "/kaggle/working"      # checkpoints written under <work_dir>/checkpoints
+    resume_dir: str | None = None          # extra dir (e.g. prior run's output) to resume from
+    backbone_name: str = BACKBONE_NAME
+    epochs: int = 10                       # linear_probe ~10; finetune ~8 (fits 9h session)
+    batch_size: int = 64
+    num_workers: int = 4
+    lr_head: float = 1e-3                  # linear_probe head lr; finetune head lr below
+    lr_head_finetune: float = 1e-4
+    lr_backbone: float = 1e-5              # finetune: last-2-block lr
+    weight_decay: float = 0.05
+    unfreeze_last_n: int = 2               # finetune: number of trailing transformer blocks
+    label_smoothing: float = 0.0
+    seed: int = 42
+    # Hugging Face Hub
+    hf_repo: str | None = None             # e.g. "user/reefscan-dinov2b" (provided before run)
+    push_to_hub: bool = False
+    # W&B
+    wandb_project: str = "reefscan"
+    wandb_run_name: str | None = None
+    classes: list[str] = field(default_factory=lambda: list(CLASSES))
+
+    @property
+    def ckpt_dir(self) -> Path:
+        return Path(self.work_dir) / "checkpoints" / self.stage
+
+
+# ---------------------------------------------------------------------------
+# Model
+# ---------------------------------------------------------------------------
+class DINOv2Classifier(nn.Module):
+    """DINOv2 backbone -> CLS pooled embedding -> linear head."""
+
+    def __init__(self, num_classes: int, backbone_name: str = BACKBONE_NAME):
+        super().__init__()
+        self.backbone = AutoModel.from_pretrained(backbone_name)
+        hidden = self.backbone.config.hidden_size
+        self.head = nn.Linear(hidden, num_classes)
+
+    def forward(self, pixel_values: torch.Tensor) -> torch.Tensor:
+        out = self.backbone(pixel_values=pixel_values)
+        # DINOv2 returns pooler_output (CLS token); fall back to CLS of last_hidden_state.
+        cls = getattr(out, "pooler_output", None)
+        if cls is None:
+            cls = out.last_hidden_state[:, 0]
+        return self.head(cls)
+
+
+def set_trainable(model: DINOv2Classifier, cfg: TrainConfig) -> list[dict]:
+    """Freeze/unfreeze per stage; return optimizer param groups (with per-group lr)."""
+    # Head is always trainable.
+    for p in model.backbone.parameters():
+        p.requires_grad = False
+    for p in model.head.parameters():
+        p.requires_grad = True
+
+    if cfg.stage == "linear_probe":
+        return [{"params": model.head.parameters(), "lr": cfg.lr_head}]
+
+    if cfg.stage == "finetune":
+        # Unfreeze the last N transformer blocks + the final layernorm.
+        blocks = model.backbone.encoder.layer
+        for blk in blocks[-cfg.unfreeze_last_n:]:
+            for p in blk.parameters():
+                p.requires_grad = True
+        if hasattr(model.backbone, "layernorm"):
+            for p in model.backbone.layernorm.parameters():
+                p.requires_grad = True
+        backbone_trainable = [p for p in model.backbone.parameters() if p.requires_grad]
+        return [
+            {"params": model.head.parameters(), "lr": cfg.lr_head_finetune},
+            {"params": backbone_trainable, "lr": cfg.lr_backbone},
+        ]
+
+    raise ValueError(f"unknown stage: {cfg.stage!r}")
+
+
+# ---------------------------------------------------------------------------
+# Data
+# ---------------------------------------------------------------------------
+def get_dataloaders(cfg: TrainConfig) -> dict[str, DataLoader]:
+    loaders: dict[str, DataLoader] = {}
+    for split in ("train", "val", "test"):
+        try:
+            ds = CoralPatchDataset.from_imagefolder(cfg.data_root, split)
+        except FileNotFoundError:
+            continue  # test split optional
+        loaders[split] = DataLoader(
+            ds, batch_size=cfg.batch_size, shuffle=(split == "train"),
+            num_workers=cfg.num_workers, pin_memory=True, drop_last=(split == "train"),
+        )
+    if "train" not in loaders or "val" not in loaders:
+        raise RuntimeError(f"need train+val imagefolders under {cfg.data_root}")
+    return loaders
+
+
+# ---------------------------------------------------------------------------
+# Checkpointing
+# ---------------------------------------------------------------------------
+def save_checkpoint(cfg: TrainConfig, model, optimizer, scheduler, scaler,
+                    epoch: int, best_f1: float) -> Path:
+    cfg.ckpt_dir.mkdir(parents=True, exist_ok=True)
+    path = cfg.ckpt_dir / f"epoch_{epoch:03d}.pt"
+    torch.save({
+        "epoch": epoch,
+        "best_f1": best_f1,
+        "model": model.state_dict(),
+        "optimizer": optimizer.state_dict(),
+        "scheduler": scheduler.state_dict() if scheduler else None,
+        "scaler": scaler.state_dict() if scaler else None,
+        "config": asdict(cfg),
+        "classes": cfg.classes,
+    }, path)
+    # Also write/refresh a 'last.pt' pointer for easy resume.
+    torch.save(torch.load(path, map_location="cpu"), cfg.ckpt_dir / "last.pt")
+    return path
+
+
+def _find_latest_checkpoint(cfg: TrainConfig) -> Path | None:
+    candidates: list[Path] = []
+    for base in [cfg.ckpt_dir, *( [Path(cfg.resume_dir)] if cfg.resume_dir else [] )]:
+        if base and base.exists():
+            last = base / "last.pt"
+            if last.exists():
+                candidates.append(last)
+            candidates += sorted(base.glob("epoch_*.pt"))
+    return candidates[-1] if candidates else None
+
+
+def maybe_resume(cfg: TrainConfig, model, optimizer, scheduler, scaler) -> tuple[int, float]:
+    ckpt = _find_latest_checkpoint(cfg)
+    if ckpt is None:
+        return 0, -1.0
+    state = torch.load(ckpt, map_location="cpu")
+    model.load_state_dict(state["model"])
+    optimizer.load_state_dict(state["optimizer"])
+    if scheduler and state.get("scheduler"):
+        scheduler.load_state_dict(state["scheduler"])
+    if scaler and state.get("scaler"):
+        scaler.load_state_dict(state["scaler"])
+    start_epoch = int(state["epoch"]) + 1
+    best_f1 = float(state.get("best_f1", -1.0))
+    print(f"[resume] from {ckpt} -> start_epoch={start_epoch}, best_f1={best_f1:.4f}")
+    return start_epoch, best_f1
+
+
+# ---------------------------------------------------------------------------
+# Train / eval
+# ---------------------------------------------------------------------------
+@torch.inference_mode()
+def evaluate(model, loader, device, criterion) -> dict:
+    model.eval()
+    losses, ys, ps = [], [], []
+    for x, y in loader:
+        x, y = x.to(device, non_blocking=True), y.to(device, non_blocking=True)
+        logits = model(x)
+        losses.append(criterion(logits, y).item())
+        ys.append(y.cpu().numpy())
+        ps.append(logits.argmax(1).cpu().numpy())
+    y_true, y_pred = np.concatenate(ys), np.concatenate(ps)
+    acc = float((y_true == y_pred).mean())
+    macro_f1 = float(f1_score(y_true, y_pred, average="macro"))
+    report = classification_report(y_true, y_pred, target_names=CLASSES,
+                                   output_dict=True, zero_division=0)
+    return {"loss": float(np.mean(losses)), "acc": acc, "macro_f1": macro_f1, "report": report}
+
+
+def run_training(cfg: TrainConfig, wandb=None) -> dict:
+    torch.manual_seed(cfg.seed)
+    np.random.seed(cfg.seed)
+    device = "cuda" if torch.cuda.is_available() else "cpu"
+
+    loaders = get_dataloaders(cfg)
+    model = DINOv2Classifier(len(cfg.classes), cfg.backbone_name).to(device)
+    param_groups = set_trainable(model, cfg)
+    optimizer = torch.optim.AdamW(param_groups, weight_decay=cfg.weight_decay)
+    scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=cfg.epochs)
+    scaler = torch.cuda.amp.GradScaler(enabled=(device == "cuda"))
+    criterion = nn.CrossEntropyLoss(label_smoothing=cfg.label_smoothing)
+
+    start_epoch, best_f1 = maybe_resume(cfg, model, optimizer, scheduler, scaler)
+
+    n_trainable = sum(p.numel() for p in model.parameters() if p.requires_grad)
+    print(f"[train] stage={cfg.stage} device={device} trainable_params={n_trainable:,} "
+          f"epochs {start_epoch}->{cfg.epochs}")
+
+    for epoch in range(start_epoch, cfg.epochs):
+        model.train()
+        running = 0.0
+        for x, y in loaders["train"]:
+            x, y = x.to(device, non_blocking=True), y.to(device, non_blocking=True)
+            optimizer.zero_grad(set_to_none=True)
+            with torch.cuda.amp.autocast(enabled=(device == "cuda")):
+                loss = criterion(model(x), y)
+            scaler.scale(loss).backward()
+            scaler.step(optimizer)
+            scaler.update()
+            running += loss.item()
+        scheduler.step()
+        train_loss = running / max(len(loaders["train"]), 1)
+        val = evaluate(model, loaders["val"], device, criterion)
+        best_f1 = max(best_f1, val["macro_f1"])
+        print(f"[epoch {epoch:03d}] train_loss={train_loss:.4f} "
+              f"val_loss={val['loss']:.4f} val_acc={val['acc']:.4f} "
+              f"val_macroF1={val['macro_f1']:.4f} (best {best_f1:.4f})")
+
+        if wandb is not None:
+            wandb.log({"epoch": epoch, "train_loss": train_loss, "val_loss": val["loss"],
+                       "val_acc": val["acc"], "val_macro_f1": val["macro_f1"],
+                       "lr": optimizer.param_groups[0]["lr"]})
+
+        # CHECKPOINT EVERY EPOCH (Kaggle constraint).
+        save_checkpoint(cfg, model, optimizer, scheduler, scaler, epoch, best_f1)
+
+    # Final eval on test if present.
+    results = {"best_val_macro_f1": best_f1}
+    if "test" in loaders:
+        results["test"] = evaluate(model, loaders["test"], device, criterion)
+        print(f"[test] acc={results['test']['acc']:.4f} "
+              f"macroF1={results['test']['macro_f1']:.4f}")
+
+    _write_config(cfg, results)
+    if cfg.push_to_hub and cfg.hf_repo:
+        push_to_hub(cfg, model, results)
+    return results
+
+
+def _write_config(cfg: TrainConfig, results: dict) -> None:
+    cfg.ckpt_dir.mkdir(parents=True, exist_ok=True)
+    with open(cfg.ckpt_dir / "config.json", "w") as f:
+        json.dump({"config": asdict(cfg), "classes": cfg.classes, "results": _slim(results)},
+                  f, indent=2)
+
+
+def _slim(results: dict) -> dict:
+    out = {k: v for k, v in results.items() if k != "test"}
+    if "test" in results:
+        out["test"] = {k: results["test"][k] for k in ("loss", "acc", "macro_f1")}
+    return out
+
+
+def push_to_hub(cfg: TrainConfig, model, results: dict) -> None:
+    from huggingface_hub import HfApi
+    api = HfApi(token=os.environ.get("HF_TOKEN"))
+    api.create_repo(cfg.hf_repo, exist_ok=True)
+    weights = cfg.ckpt_dir / "model.safetensors"
+    try:
+        from safetensors.torch import save_file
+        save_file(model.state_dict(), str(weights))
+    except Exception:
+        weights = cfg.ckpt_dir / "model.pt"
+        torch.save(model.state_dict(), weights)
+    api.upload_file(path_or_fileobj=str(weights), path_in_repo=weights.name, repo_id=cfg.hf_repo)
+    api.upload_file(path_or_fileobj=str(cfg.ckpt_dir / "config.json"),
+                    path_in_repo=f"{cfg.stage}_config.json", repo_id=cfg.hf_repo)
+    print(f"[hub] pushed weights + config to {cfg.hf_repo}")
+
+
+if __name__ == "__main__":
+    import argparse
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--stage", default="linear_probe", choices=["linear_probe", "finetune"])
+    ap.add_argument("--data-root", default="/kaggle/input/noaa-coral-bleaching")
+    ap.add_argument("--work-dir", default="/kaggle/working")
+    ap.add_argument("--epochs", type=int, default=10)
+    ap.add_argument("--batch-size", type=int, default=64)
+    ap.add_argument("--hf-repo", default=None)
+    ap.add_argument("--push-to-hub", action="store_true")
+    a = ap.parse_args()
+    run_training(TrainConfig(
+        stage=a.stage, data_root=a.data_root, work_dir=a.work_dir, epochs=a.epochs,
+        batch_size=a.batch_size, hf_repo=a.hf_repo, push_to_hub=a.push_to_hub,
+    ))

backend/observability.py

@@ -0,0 +1,80 @@
+"""Observability aggregations over inference_logs. Phase 7.
+
+Pure functions (testable) that turn raw inference_logs rows into the three dashboard views
+computed entirely from Supabase data — no external observability tool:
+  - rolling mean prediction_set_size per day  -> drift proxy (rising = distribution shift)
+  - latency p50 / p95 per day
+  - class distribution: current 7-day window vs the prior 7-day baseline
+"""
+from __future__ import annotations
+
+from collections import defaultdict
+from datetime import date, timedelta
+
+
+def _day(ts) -> str:
+    return str(ts)[:10]
+
+
+def _percentile(values: list[float], p: float) -> float:
+    if not values:
+        return 0.0
+    v = sorted(values)
+    k = (len(v) - 1) * p / 100.0
+    f = int(k)
+    if f + 1 < len(v):
+        return round(v[f] + (v[f + 1] - v[f]) * (k - f), 1)
+    return round(v[f], 1)
+
+
+def rolling_set_size(logs: list[dict]) -> list[dict]:
+    by: dict[str, list[float]] = defaultdict(list)
+    for r in logs:
+        by[_day(r["ts"])].append(float(r.get("prediction_set_size", 1)))
+    return [{"date": d, "avg_set_size": round(sum(v) / len(v), 3), "n": len(v)}
+            for d, v in sorted(by.items())]
+
+
+def latency_percentiles(logs: list[dict]) -> list[dict]:
+    by: dict[str, list[float]] = defaultdict(list)
+    for r in logs:
+        by[_day(r["ts"])].append(float(r.get("latency_ms", 0)))
+    return [{"date": d, "p50": _percentile(v, 50), "p95": _percentile(v, 95), "n": len(v)}
+            for d, v in sorted(by.items())]
+
+
+def class_distribution(logs: list[dict], classes: tuple[str, ...]) -> dict:
+    days = sorted({_day(r["ts"]) for r in logs})
+    if not days:
+        return {"current": {}, "baseline": {}, "current_window": None, "baseline_window": None}
+    anchor = date.fromisoformat(days[-1])
+    cur_lo = anchor - timedelta(days=6)
+    base_hi = cur_lo - timedelta(days=1)
+    base_lo = base_hi - timedelta(days=6)
+
+    def frac(lo: date, hi: date) -> dict:
+        counts = {c: 0 for c in classes}
+        for r in logs:
+            d = date.fromisoformat(_day(r["ts"]))
+            if lo <= d <= hi:
+                lab = r.get("predicted_label")
+                if lab in counts:
+                    counts[lab] += 1
+        total = sum(counts.values()) or 1
+        return {c: round(counts[c] / total * 100, 1) for c in classes}
+
+    return {
+        "current": frac(cur_lo, anchor),
+        "baseline": frac(base_lo, base_hi),
+        "current_window": [cur_lo.isoformat(), anchor.isoformat()],
+        "baseline_window": [base_lo.isoformat(), base_hi.isoformat()],
+    }
+
+
+def build(logs: list[dict], classes: tuple[str, ...]) -> dict:
+    return {
+        "drift": rolling_set_size(logs),
+        "latency": latency_percentiles(logs),
+        "class_distribution": class_distribution(logs, classes),
+        "total_logs": len(logs),
+    }

backend/persistence.py

@@ -0,0 +1,157 @@
+"""Persistence wrappers: Cloudflare R2 (objects) + Supabase (logging). Phase 5.
+
+Both degrade to safe no-ops when their credentials are absent, so the pipeline runs
+end-to-end locally. Tables match backend/db/schema.sql.
+"""
+from __future__ import annotations
+
+import logging
+import uuid
+from typing import Any, Optional
+
+from .config import settings
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Object storage — Supabase Storage by default (no extra account), R2 if configured.
+# ---------------------------------------------------------------------------
+class ObjectStore:
+    def __init__(self, supabase_client=None) -> None:
+        self._r2 = None
+        self._sb = supabase_client          # reuse the Supabase service_role client
+        self._bucket = settings.storage_bucket
+        if settings.r2_enabled:
+            try:
+                import boto3  # type: ignore
+                self._r2 = boto3.client(
+                    "s3", endpoint_url=settings.r2_endpoint,
+                    aws_access_key_id=settings.r2_key_id,
+                    aws_secret_access_key=settings.r2_secret,
+                )
+                logger.info("object store: Cloudflare R2 (bucket=%s)", settings.r2_bucket)
+            except Exception as e:  # noqa: BLE001
+                logger.warning("R2 init failed: %s", e)
+        elif self._sb is not None:
+            logger.info("object store: Supabase Storage (bucket=%s)", self._bucket)
+
+    def upload(self, key: str, data: bytes, content_type: str) -> str:
+        """Store bytes, return a public url. Falls back to a local placeholder if neither
+        backend is configured (the frontend then shows its gradient placeholder)."""
+        # 1) Cloudflare R2
+        if self._r2 is not None:
+            self._r2.put_object(Bucket=settings.r2_bucket, Key=key, Body=data,
+                                ContentType=content_type)
+            ep = (settings.r2_endpoint or "").rstrip("/")
+            return f"{ep}/{settings.r2_bucket}/{key}"
+        # 2) Supabase Storage (via the service_role client; bucket must be public)
+        if self._sb is not None:
+            try:
+                self._sb.storage.from_(self._bucket).upload(
+                    key, data, {"content-type": content_type, "upsert": "true"})
+                url = self._sb.storage.from_(self._bucket).get_public_url(key)
+                return url if isinstance(url, str) else key
+            except Exception as e:  # noqa: BLE001
+                logger.warning("supabase storage upload failed: %s", e)
+        # 3) local placeholder
+        return f"local://uploads/{key}"
+
+
+# ---------------------------------------------------------------------------
+# Supabase logging
+# ---------------------------------------------------------------------------
+class SupabaseLogger:
+    def __init__(self) -> None:
+        self._client = None
+        if settings.supabase_enabled:
+            try:
+                from supabase import create_client  # type: ignore
+                self._client = create_client(settings.supabase_url, settings.supabase_key)
+                logger.info("Supabase logging enabled")
+            except Exception as e:  # noqa: BLE001
+                logger.warning("Supabase init failed, logging disabled: %s", e)
+
+    @property
+    def enabled(self) -> bool:
+        return self._client is not None
+
+    def _insert(self, table: str, rows: list[dict[str, Any]]) -> None:
+        if self._client is None or not rows:
+            return
+        try:
+            self._client.table(table).insert(rows).execute()
+        except Exception as e:  # noqa: BLE001
+            logger.warning("supabase insert into %s failed: %s", table, e)
+
+    # ---- jobs ----
+    def upsert_job(self, row: dict[str, Any]) -> None:
+        if self._client is None:
+            return
+        try:
+            self._client.table("jobs").upsert(row).execute()
+        except Exception as e:  # noqa: BLE001
+            logger.warning("supabase upsert job failed: %s", e)
+
+    # ---- inference_logs (one row per classified segment) ----
+    def log_segments(self, rows: list[dict[str, Any]]) -> None:
+        self._insert("inference_logs", rows)
+
+    # ---- review_queue (uncertain predictions) ----
+    def log_review(self, rows: list[dict[str, Any]]) -> None:
+        self._insert("review_queue", rows)
+
+    # ---- health_snapshots (per-upload aggregate) ----
+    def log_snapshot(self, row: dict[str, Any]) -> None:
+        self._insert("health_snapshots", [row])
+
+    # ---- reads (Phase 6 review/tracker + Phase 7 dashboard) ----
+    def _select(self, table: str, build) -> list[dict[str, Any]]:
+        if self._client is None:
+            return []
+        try:
+            return build(self._client.table(table).select("*")).execute().data or []
+        except Exception as e:  # noqa: BLE001
+            logger.warning("supabase select from %s failed: %s", table, e)
+            return []
+
+    def review_queue(self, limit: int = 50) -> list[dict[str, Any]]:
+        return self._select("review_queue", lambda q: q.eq("status", "pending")
+                            .order("created_at", desc=True).limit(limit))
+
+    def reef_locations(self) -> list[dict[str, Any]]:
+        return self._select("reef_locations", lambda q: q.order("name"))
+
+    def snapshots(self, reef_id: str) -> list[dict[str, Any]]:
+        return self._select("health_snapshots", lambda q: q.eq("reef_location_id", reef_id)
+                            .order("snapshot_time"))
+
+    def recent_logs(self, limit: int = 5000) -> list[dict[str, Any]]:
+        return self._select("inference_logs", lambda q: q.order("ts", desc=True).limit(limit))
+
+    def confirm_label(self, review_id: str, label: str, labeled_by: str = "admin") -> bool:
+        if self._client is None:
+            return False
+        try:
+            row = self._client.table("review_queue").select("*").eq("id", review_id).execute().data
+            if not row:
+                return False
+            r = row[0]
+            self._client.table("human_labels").insert({
+                "review_queue_id": review_id, "image_id": r.get("image_id"),
+                "segment_id": r.get("segment_id"), "confirmed_label": label,
+                "labeled_by": labeled_by,
+            }).execute()
+            self._client.table("review_queue").update({"status": "confirmed"}).eq("id", review_id).execute()
+            return True
+        except Exception as e:  # noqa: BLE001
+            logger.warning("confirm_label failed: %s", e)
+            return False
+
+
+def new_request_id() -> str:
+    return str(uuid.uuid4())
+
+
+supabase = SupabaseLogger()
+r2 = ObjectStore(supabase_client=supabase._client)  # exported as `r2` for pipeline import

backend/requirements.txt

@@ -0,0 +1,48 @@
+# ReefScan backend — Python 3.11+
+# Install (CPU inference target, e.g. HF Spaces free CPU):
+#   pip install --extra-index-url https://download.pytorch.org/whl/cpu -r requirements.txt
+#
+# NOTE: torch/sam2/WaterNet notes below. Versions are pinned with lower bounds and will
+# be hard-pinned once the Phase 1.5 feasibility spike confirms what actually loads/runs
+# inside the HF Spaces free CPU memory budget.
+
+# --- Web / API ---
+fastapi>=0.111
+uvicorn[standard]>=0.30
+python-multipart>=0.0.9        # file uploads
+httpx>=0.27                    # POST /infer url-fetch path + TestClient
+pydantic>=2.7
+python-dotenv>=1.0
+
+# --- ML core (CPU wheels via --extra-index-url above) ---
+torch>=2.4
+torchvision>=0.19
+numpy>=1.26
+pillow>=10.3
+opencv-python-headless>=4.10   # headless: no GUI libs on Spaces
+
+# --- Models ---
+transformers>=4.44             # DINOv2 (facebook/dinov2-base)
+# SAM2 (Hiera-Small) — installed from source, no PyPI release:
+#   pip install "git+https://github.com/facebookresearch/segment-anything-2.git"
+# WaterNet — pretrained inference only; no PyPI package. Vendor weights + minimal
+#   forward pass under backend/inference/ (source documented in enhancer.py, Phase 5).
+
+# --- Uncertainty / conformal ---
+mapie>=1.0                     # SplitConformalClassifier, LAC/APS scoring
+scikit-learn>=1.5
+
+# --- Video frame extraction ---
+scenedetect[opencv]>=0.6.4     # PySceneDetect, scene-change based
+
+# --- Data / training ---
+pandas>=2.2
+wandb>=0.17                    # experiment tracking (free academic tier)
+huggingface_hub>=0.24         # pull/push model weights
+
+# --- Storage / DB ---
+supabase>=2.6                 # Supabase Postgres client
+boto3>=1.34                   # Cloudflare R2 (S3-compatible)
+
+# --- Tests ---
+pytest>=8.2

backend/schemas.py

@@ -0,0 +1,65 @@
+"""Pydantic models for the FROZEN inference response contract. Phase 5.
+
+These mirror frontend/lib/types.ts exactly — the field names/shape are the integration
+seam (Phase 6 mocks it, Phase 5 fills it). Do not rename fields without changing both.
+"""
+from __future__ import annotations
+
+from typing import Literal, Optional
+
+from pydantic import BaseModel, Field
+
+CoralClass = Literal["healthy", "bleached"]
+JobStatus = Literal["queued", "processing", "complete", "failed"]
+
+
+class Segment(BaseModel):
+    segment_id: int
+    mask_area_px: int
+    bbox: list[int]  # [x0, y0, x1, y1] in source-image px
+    predicted_class: CoralClass
+    prediction_set: list[CoralClass]
+    prediction_set_size: int
+    confidence_scores: dict[str, float]  # {"healthy": .., "bleached": ..}
+    coverage_pct: float
+
+
+class AreaWeighted(BaseModel):
+    healthy_pct: float
+    bleached_pct: float
+
+
+class Summary(BaseModel):
+    total_segments: int
+    area_weighted: AreaWeighted
+    uncertain_segments: int
+    dominant_status: CoralClass
+
+
+class InferenceResponse(BaseModel):
+    """Returned by GET /infer/{job_id}. When status != 'complete', segments/summary are
+    empty/None and the client keeps polling."""
+    job_id: str
+    status: JobStatus
+    processing_time_ms: int = 0
+    image_url: str = ""
+    model_version: str = ""
+    image_width: Optional[int] = None
+    image_height: Optional[int] = None
+    segments: list[Segment] = Field(default_factory=list)
+    summary: Optional[Summary] = None
+    error_message: Optional[str] = None
+
+
+class SubmitResponse(BaseModel):
+    """Returned by POST /infer — the async enqueue ack."""
+    job_id: str
+    status: JobStatus = "queued"
+
+
+class HealthResponse(BaseModel):
+    status: str
+    models_loaded: bool
+    stub_mode: bool
+    model_version: str
+    last_inference_latency_ms: Optional[int] = None

backend/tests/test_api.py

@@ -0,0 +1,82 @@
+"""Phase 5 async-pipeline contract tests (stub mode — no weights/Supabase/R2 needed).
+
+Drives the real async path: POST /infer enqueues, GET /infer/{job_id} polls to completion,
+and the response is validated against the frozen contract (frontend/lib/types.ts).
+"""
+import io
+import os
+import time
+
+os.environ["REEFSCAN_STUB"] = "1"  # force synthetic inference before app import
+
+from PIL import Image  # noqa: E402
+from fastapi.testclient import TestClient  # noqa: E402
+
+from backend.main import app  # noqa: E402
+
+SEGMENT_KEYS = {
+    "segment_id", "mask_area_px", "bbox", "predicted_class", "prediction_set",
+    "prediction_set_size", "confidence_scores", "coverage_pct",
+}
+
+
+def _png(w=600, h=380):
+    b = io.BytesIO()
+    Image.new("RGB", (w, h), (20, 80, 90)).save(b, format="PNG")
+    return b.getvalue()
+
+
+def _run_to_completion(client, job_id, tries=60):
+    for _ in range(tries):
+        resp = client.get(f"/infer/{job_id}").json()
+        if resp["status"] == "complete":
+            return resp
+        if resp["status"] == "failed":
+            raise AssertionError(f"job failed: {resp.get('error_message')}")
+        time.sleep(0.1)
+    raise AssertionError("job did not complete in time")
+
+
+def test_health_stub_mode():
+    with TestClient(app) as client:
+        h = client.get("/health").json()
+        assert h["status"] == "ok"
+        assert h["stub_mode"] is True  # no weights -> stub
+
+
+def test_async_infer_contract():
+    with TestClient(app) as client:
+        r = client.post("/infer", files={"file": ("reef.png", _png(), "image/png")})
+        assert r.status_code == 202
+        job_id = r.json()["job_id"]
+        assert client.get(f"/infer/{job_id}").json()["status"] in ("queued", "processing", "complete")
+
+        data = _run_to_completion(client, job_id)
+
+        assert data["image_width"] == 600 and data["image_height"] == 380
+        assert data["model_version"]
+        assert len(data["segments"]) >= 1
+        for seg in data["segments"]:
+            assert set(seg) == SEGMENT_KEYS
+            assert set(seg["confidence_scores"]) == {"healthy", "bleached"}
+            assert seg["predicted_class"] in ("healthy", "bleached")
+            assert seg["prediction_set_size"] == len(seg["prediction_set"]) >= 1
+            assert abs(sum(seg["confidence_scores"].values()) - 1.0) < 1e-3
+
+        s = data["summary"]
+        assert s["total_segments"] == len(data["segments"])
+        assert abs(s["area_weighted"]["healthy_pct"] + s["area_weighted"]["bleached_pct"] - 100) < 0.2
+        assert s["dominant_status"] in ("healthy", "bleached")
+        # uncertain accounting must match the per-segment sets
+        unc = sum(1 for x in data["segments"] if x["prediction_set_size"] > 1)
+        assert s["uncertain_segments"] == unc
+
+
+def test_unknown_job_404():
+    with TestClient(app) as client:
+        assert client.get("/infer/does-not-exist").status_code == 404
+
+
+def test_infer_requires_input():
+    with TestClient(app) as client:
+        assert client.post("/infer").status_code == 400

backend/tests/test_observability.py

@@ -0,0 +1,65 @@
+"""Phase 7 observability aggregation tests + read-endpoint fallbacks (no Supabase)."""
+import os
+
+os.environ["REEFSCAN_STUB"] = "1"
+
+from fastapi.testclient import TestClient  # noqa: E402
+
+from backend import observability  # noqa: E402
+from backend.main import app  # noqa: E402
+
+CLASSES = ("healthy", "bleached")
+
+
+def _logs():
+    # two days, varied set sizes / latencies / labels
+    rows = []
+    for i in range(10):
+        rows.append({"ts": "2026-06-20T10:00:00Z", "prediction_set_size": 1 if i % 2 else 2,
+                     "latency_ms": 100 + i * 10, "predicted_label": "healthy" if i < 7 else "bleached"})
+    for i in range(10):
+        rows.append({"ts": "2026-06-27T10:00:00Z", "prediction_set_size": 2 if i % 2 else 1,
+                     "latency_ms": 200 + i * 10, "predicted_label": "bleached" if i < 6 else "healthy"})
+    return rows
+
+
+def test_rolling_set_size():
+    out = observability.rolling_set_size(_logs())
+    assert [r["date"] for r in out] == ["2026-06-20", "2026-06-27"]
+    assert all(1.0 <= r["avg_set_size"] <= 2.0 for r in out)
+    assert out[0]["n"] == 10
+
+
+def test_latency_percentiles():
+    out = observability.latency_percentiles(_logs())
+    assert out[0]["p50"] <= out[0]["p95"]
+    assert out[1]["p95"] >= out[1]["p50"] >= 200
+
+
+def test_class_distribution_windows():
+    cd = observability.class_distribution(_logs(), CLASSES)
+    assert abs(cd["current"]["healthy"] + cd["current"]["bleached"] - 100) < 0.2
+    assert cd["current_window"] and cd["baseline_window"]
+
+
+def test_build_shape():
+    o = observability.build(_logs(), CLASSES)
+    assert set(o) == {"drift", "latency", "class_distribution", "total_logs"}
+    assert o["total_logs"] == 20
+
+
+def test_read_endpoints_empty_without_supabase():
+    with TestClient(app) as client:
+        assert client.get("/review-queue").json() == []
+        assert client.get("/reef-locations").json() == []
+        assert client.get("/reef-locations/x/snapshots").json() == []
+        obs = client.get("/observability").json()
+        assert obs["total_logs"] == 0 and obs["drift"] == []
+
+
+def test_confirm_validates_label():
+    with TestClient(app) as client:
+        assert client.post("/review-queue/abc/confirm", json={"label": "nope"}).status_code == 400
+        # valid label but no supabase -> ok False, still 200
+        r = client.post("/review-queue/abc/confirm", json={"label": "healthy"})
+        assert r.status_code == 200 and r.json()["ok"] is False