app/adapters/base.py

"""What every adapter agrees to, and the two things none of them may do.

The zero-training directive (§3, §4, §40.1) names a stack of pretrained models —
SAM, DINOv3, MegaDescriptor, Grounding DINO, CountGD, a hosted reasoner — plus a
deterministic OpenCV/NumPy path that §4 says to prefer whenever it is better.
Those have almost nothing in common at the point of use: one returns masks, one
returns a 384-dimensional vector, one returns a breath rate. So this file does
**not** try to give them a single `run`.

What they do have in common is governance, and that is what is unified here:

**An adapter says what it costs, and `None` means nobody measured it.**
`MeasuredCost` has no defaults and no published-figure fallback. A latency copied
from a paper is a claim about somebody else's GPU, and the farms this serves run
a 2 vCPU / 4 GiB container.

**An adapter cannot produce a result it has no model for.** There is deliberately
no `run` on this class — the same reason `providers.InferenceProvider` has none.
A base implementation would be a way to return something plausible with nothing
behind it, and that is the single failure this service exists to prevent. Work
happens on the object `load()` returns, and `load()` raises when the artefact is
absent.

The task protocols below are the narrow interfaces callers actually use. They are
kept as small as `detectors.Detector` is, for the same reason: everything added
here is something the replacement adapter has to reimplement on the day a licence
forces a swap, and ADR 0017 is the record of that day arriving.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum
from typing import Protocol, runtime_checkable

import numpy as np
from PIL import Image


class Task(str, Enum):
    """What an adapter produces. A model may do several."""

    DETECT = "detect"
    SEGMENT = "segment"
    EMBED = "embed"
    COUNT = "count"
    TRACK = "track"
    POSE = "pose"
    #: Structured reasoning from a hosted multimodal model. Named apart from the
    #: rest because §4 is explicit that it is "an experimental visual reasoner,
    #: not an authority", and a caller should have to type the difference.
    REASON = "reason"
    #: Deterministic signal processing — optical flow, FFT, contour geometry.
    #: No weights, no licence question, and §4 says to prefer it where it wins.
    MEASURE = "measure"


class Modality(str, Enum):
    IMAGE = "image"
    VIDEO = "video"
    AUDIO = "audio"


@dataclass(frozen=True)
class MeasuredCost:
    """Latency and memory from a run that actually happened.

    Every field is required. There is no `estimated` variant and no default,
    because the only thing worse than not knowing what an adapter costs on a
    2 vCPU box is believing a number nobody produced there.

    `hardware` is free text on purpose: it has to be able to say "MacBook, 8
    performance cores, not the target" as easily as it says the container SKU,
    and a reader needs to see which one they are looking at.
    """

    hardware: str
    threads: int
    #: What it ran on, specifically enough to re-run. A slug from
    #: `evaluation/dataset.json`, or a count of frames from a named set.
    sample: str
    runs: int
    median_seconds: float
    peak_rss_mb: float
    measured_on: str

    @property
    def fits_cpu_service(self) -> bool:
        """Whether this would survive the 2 vCPU / 4 GiB CPU worker.

        **This is a placement hint, not a verdict on the model.** A model that
        returns False here belongs on a GPU host, and that is a deployment
        decision rather than a reason to drop a capability. The distinction is
        recorded because the earlier version of this file got it wrong and would
        have excluded most of directive §3 on the strength of a container size.
        """
        return (
            self.peak_rss_mb <= CPU_SERVICE_MEMORY_CEILING_MB
            and self.median_seconds <= INLINE_LATENCY_CEILING_SECONDS
        )


#: Peak RSS above which an adapter will not sit comfortably beside the API on the
#: existing CPU worker. ADR 0018 measured YOLOX-m at 591 MB and YOLOX-x at
#: 1,003 MB. 2,000 MB leaves the Python process, onnxruntime's arenas and
#: Pillow's decode buffers room inside 4 GiB.
CPU_SERVICE_MEMORY_CEILING_MB = 2000.0

#: Wall-clock above which a capability cannot run inline on a request, wherever
#: it is hosted. ADR 0018's phrasing: "A 16-second inline request is not a
#: request; it is a timeout with a result attached." Past this a capability needs
#: a queue, not a bigger box.
INLINE_LATENCY_CEILING_SECONDS = 8.0


class Placement(str, Enum):
    """Where a leg of the stack should run.

    Three tiers, and the choice between them is a product decision as much as an
    engineering one. Animap is offline-first (ADR 0002): a capability that needs
    a round trip is one a farm cannot use in a shed with no signal, so pushing
    work off the phone has a cost that a latency table does not show.
    """

    #: On the phone. The only tier that works with no signal at all.
    ON_DEVICE = "on_device"
    #: The existing 2 vCPU / 4 GiB CPU container, beside the API.
    CPU_SERVICE = "cpu_service"
    #: A GPU host. Available, and the right answer for most of directive §3 —
    #: the models it names are GPU-class work and it was written knowing that.
    GPU_SERVICE = "gpu_service"


@dataclass(frozen=True)
class AdapterSpec:
    """An adapter's identity, licence position and measured cost.

    This is committed code rather than a JSON card, and that is the point.
    `providers.ModelArtefact` reads a card, and a card is written by whoever
    writes the card — ADR 0017 records a watchdog defeating the licence gate by
    declaring `Apache-2.0` over a path to AGPL weights. The `runtime` here names
    which loader runs, which is a fact about the code and not a claim about
    terms, and `adapters.licences` holds what that runtime's weights are really
    licensed under.
    """

    adapter_id: str
    #: Which loader runs. The key into `licences.RUNTIME_LICENCES`, and the only
    #: field the licence gate trusts.
    runtime: str
    tasks: tuple[Task, ...]
    modalities: tuple[Modality, ...]
    #: What the zero-training directive asks this model for, quoted closely
    #: enough that a reader can find the section.
    directive_role: str
    #: False for the deterministic methods — optical flow, FFT, contour
    #: geometry. They need no weights, so they have no artefact to be absent and
    #: no licence to refuse, which is most of why §4 prefers them.
    requires_artefact: bool = True
    #: `None` until somebody runs it and writes the number down. Reported as
    #: "not measured", never filled in from a paper.
    measured: MeasuredCost | None = None
    #: Where this leg should run. A recommendation with a reason, not a
    #: constraint — see `Placement`.
    placement: Placement = Placement.CPU_SERVICE
    #: Whether a GPU is needed for this to be usable at all, as opposed to
    #: merely faster. Recorded separately from `placement` because "runs on CPU
    #: but slowly" and "does not run on CPU" are different facts and only the
    #: second one closes a door.
    requires_gpu: bool = False
    placement_reason: str = ""
    notes: str = ""


@dataclass(frozen=True)
class Availability:
    """Whether an adapter can run, and if not, what would change that.

    `remedy` exists because "unavailable" without it is the answer that gets
    read as "broken". The service already distinguishes *"no validated model
    exists"* from *"this is not planned"* in `main._unavailable_reason`, and an
    adapter that cannot say which of those it is has lost the distinction.
    """

    ready: bool
    #: Empty when ready. Otherwise says what is missing, not what went wrong.
    reason: str = ""
    remedy: str = ""

    def __post_init__(self) -> None:
        if not self.ready and not self.reason:
            raise ValueError(
                "An unavailable adapter must say why. A bare False is what a "
                "caller renders as a silent failure."
            )


class AdapterError(RuntimeError):
    """The adapter is present but could not do the work."""


class AdapterUnavailable(AdapterError):
    """No model behind this adapter, so there is nothing to run.

    Raised by `load()`, never returned as a result. A caller that catches this
    reports `unavailable` — the state `JobState.UNAVAILABLE` already exists for,
    and which is the honest answer for most of the stack today.
    """

    def __init__(self, availability: Availability) -> None:
        self.availability = availability
        message = availability.reason
        if availability.remedy:
            message = f"{message} {availability.remedy}"
        super().__init__(message)


class Adapter:
    """A pretrained model, or a deterministic method, behind one interface.

    **There is no `run` here, and adding one would be the bug.** Subclasses
    expose whichever task protocol they satisfy — `Embedder`, `Segmenter`,
    `Reasoner` — and only after `load()` has succeeded against a real artefact.
    A default implementation on this class would be a way to answer a farmer
    with no model in the loop.
    """

    spec: AdapterSpec

    def availability(self) -> Availability:
        """Whether this adapter could run right now.

        Must not load anything. Called on `/health` and `/capabilities`, which
        a platform probe hits often enough that reading a hundred megabytes of
        weights to answer it would be its own outage.
        """
        raise NotImplementedError

    def load(self) -> "Adapter":
        """Prepare the runtime, or raise `AdapterUnavailable`.

        Returns self so a caller can write `adapter.load().embed(image)` and
        have no path to `embed` that skipped the check.
        """
        raise NotImplementedError

    def describe(self) -> dict[str, object]:
        """Everything a governance reader needs, including what is unmeasured."""
        from app.adapters import licences

        availability = self.availability()
        licence = licences.RUNTIME_LICENCES.get(self.spec.runtime)
        cost = self.spec.measured
        return {
            "adapter_id": self.spec.adapter_id,
            "runtime": self.spec.runtime,
            "tasks": [t.value for t in self.spec.tasks],
            "modalities": [m.value for m in self.spec.modalities],
            "directive_role": self.spec.directive_role,
            "ready": availability.ready,
            "reason": availability.reason,
            "remedy": availability.remedy,
            "licence": licence.licence if licence else "unknown runtime",
            "licence_source": licence.source_url if licence else "",
            "servable": bool(licence and licence.servable),
            "placement": self.spec.placement.value,
            "requires_gpu": self.spec.requires_gpu,
            "placement_reason": self.spec.placement_reason,
            # The absence is the finding, so it is spelled rather than nulled.
            "measured": (
                {
                    "hardware": cost.hardware,
                    "threads": cost.threads,
                    "sample": cost.sample,
                    "runs": cost.runs,
                    "median_seconds": cost.median_seconds,
                    "peak_rss_mb": cost.peak_rss_mb,
                    "measured_on": cost.measured_on,
                    "fits_cpu_service": cost.fits_cpu_service,
                }
                if cost is not None
                else "not measured"
            ),
            "notes": self.spec.notes,
        }


# --- Task protocols. Narrow on purpose. --------------------------------------


@dataclass(frozen=True)
class Region:
    """A box, a mask, or both. The common currency of detection and segmentation.

    `box` is in source-image pixels, matching `detectors.Detection`, so a caller
    that already knows how to read a YOLOX box does not learn a second convention.
    `mask` is a boolean array at source-image resolution, or `None` when the
    adapter only localises.
    """

    label: str
    score: float
    box: tuple[float, float, float, float]
    mask: np.ndarray | None = None
    area_fraction: float = 0.0

    @property
    def has_mask(self) -> bool:
        return self.mask is not None


@runtime_checkable
class Embedder(Protocol):
    """Frozen features. §3's instruction for DINOv3 is explicit that this comes
    before any fine-tuning: "Start with frozen embeddings + nearest-neighbor
    retrieval."
    """

    #: Length of the vector `embed` returns. Recorded because a retrieval index
    #: built at one dimension and queried at another fails silently.
    dimensions: int

    def embed(self, image: Image.Image) -> np.ndarray:
        """One L2-normalised float32 vector.

        Normalised by the adapter rather than the caller, so cosine similarity
        is a dot product everywhere and no index has to remember which
        convention it was built under.
        """
        ...


@runtime_checkable
class Segmenter(Protocol):
    def segment(
        self, image: Image.Image, *, concepts: tuple[str, ...] = ()
    ) -> list[Region]:
        ...


@runtime_checkable
class OpenVocabularyDetector(Protocol):
    """Text-prompted detection. §4 names Grounding DINO as the fallback for when
    SAM's concept prompting is weak."""

    def detect_text(
        self, image: Image.Image, prompts: tuple[str, ...]
    ) -> list[Region]:
        ...


@runtime_checkable
class ExemplarCounter(Protocol):
    """Zero-shot counting, optionally guided by example boxes (§4, CountGD)."""

    def count(
        self,
        image: Image.Image,
        *,
        text: str = "",
        exemplars: tuple[tuple[float, float, float, float], ...] = (),
    ) -> "CountEstimate":
        ...


@dataclass(frozen=True)
class CountEstimate:
    """A count, or an honest refusal to publish one.

    `value` is `None` when the method ran and the result should not be shown —
    the same shape `app/counting.py` already uses, where withholding is a first
    class outcome rather than an exception. §6.3 requires the distinction
    between *visible count*, *unique birds observed* and *reconciled population*
    to survive to the UI, so `kind` carries it.
    """

    value: float | None
    kind: str
    withheld_reason: str = ""
    confidence: float | None = None


@runtime_checkable
class Reasoner(Protocol):
    """A hosted multimodal model.

    §4: "All calls must return structured JSON" and "The multimodal model is an
    **experimental visual reasoner**, not an authority." Both are enforced in
    the adapter rather than left to the prompt — see `adapters/multimodal.py`.
    """

    def reason(
        self,
        images: list[Image.Image],
        *,
        schema: dict,
        rubric: str,
    ) -> dict:
        ...


@dataclass(frozen=True)
class Measurement:
    """A number a deterministic method produced, with its own quality verdict.

    Signal processing fails differently from a model: it does not become
    uncertain, it becomes wrong in a way that still returns a float. So a
    measurement carries the evidence that the signal was there at all —
    `support` is whatever the method uses to know it measured something rather
    than measuring noise, and `usable` is its own judgement about that.
    """

    kind: str
    value: float | None
    unit: str
    usable: bool
    support: dict[str, float] = field(default_factory=dict)
    detail: str = ""