app/identification.py

"""Matching a capture against a farm's enrolled animals, and refusing to name one.

Directive §6.4, and the counterpart to `app/counting.py`: that module turns boxes
into a claim, this one turns a nearest neighbour into a candidate. Both exist
because the hard part is not the model — it is being honest about the cases the
model cannot carry.

**The measurement this runner ships against is the open-set one, not the
closed-set one.** Run `8db9e0bd1b30`
(`experiments/cattle_identity/metrics.json`) enrolled 169 animals from the
muzzle268 database and measured, on 789 probes of animals that *were* enrolled:
top-1 0.9772, top-3 0.9937, MRR 0.9853, against a chance rate of 0.005917. That
is the number worth quoting and it is not the number that decides the product.

On 416 probes of animals nobody enrolled, **the unthresholded false-accept rate
is 1.000**. Every unenrolled animal comes back as somebody, because every query
has a nearest neighbour and nothing about a nearest neighbour knows the right
answer was absent. The two similarity distributions overlap badly — enrolled
probes median 0.9707, unenrolled median 0.9042 with an unenrolled *maximum* of
0.9776 — so the cutoff that admits no impostor at all sits at 0.98 and accepts
**24.08%** of the correct matches.

`MEASURED_POLICY` is that operating point, and choosing it is the whole design:

- Three quarters of the time an enrolled animal is photographed, this runner
  claims `no_confident_match` **and still returns the ranked candidates**, so a
  person sees the names and picks one. That is a worse headline and the same
  information.
- The alternative — no threshold, always name the top candidate — is the one
  that reads better and puts a neighbour's cow in a farm's records 100% of the
  time an unenrolled animal is photographed.

`§6.4`'s confirm step is therefore load-bearing rather than decorative.

**What carries that requirement to a device, precisely.** This paragraph used to
say `IdentityResult.to_json` states `requires_confirmation: True` on every path.
It does — and this runner never calls it. The served result is
`schemas.InferenceResult`, which has no such field, so **the flag reaches
nobody** and citing it here described a guarantee that was not on the wire.

What actually crosses is the capability row. `Requirement.HUMAN_CONFIRMATION` is
on the registry entry, travels verbatim to the API, and is published as
`requirements: ["guided_capture", "human_confirmation"]` by `GET /capabilities`,
where a database constraint forces §6.4's four buttons to travel beside it. That
is a real mechanism and it is a **different endpoint from the result**: a client
that renders a result without having read `/capabilities` has nothing in the
payload telling it to ask. Closing that gap means a field on `InferenceResult`,
which is a contract change for every capability and is not made here.

What this runner does guarantee on every path is narrower and worth stating
exactly: no result carries an interpretation, an `observation_confidence`, or a
non-`None` confidence on any observation; `identity_candidate` is emitted only
when the measured policy accepted; and `_result` raises on
`identity_without_confirmation`. A result from here cannot *look* settled. It
relies on the client having read the capability to know it must ask.

**What is not measured here.** The database is US beef breeds — Angus, Angus x
Hereford, Continental x British crosses. No Nigerian and no zebu animal has been
through this, and a White Fulani is white all over, so the coat-pattern signal
the published re-identification literature leans on is absent for the herds this
product is for. The registry carries `UNVALIDATED_GEOGRAPHY`, which is what stops
`may_be_promoted_to_production` mechanically.

## The cold-gallery cost, which is a real limit and not a solved problem

Measured on 2026-08-23 over this service's own HTTP path by
`scripts/verify_identity_wiring.py`, on an Apple M-series laptop, single process,
enrolling five muzzle photographs per animal:

| Register | Enrolment photographs | Cold request | Per photograph | Warm median |
|---|---|---|---|---|
| 20 animals | 100 | **11.13 s** | 0.11 s | **0.059 s** |
| 60 animals | 300 | **60.35 s** | 0.20 s | **0.07 s** |

`_VECTOR_CACHE` is what separates the cold column from the warm one, and it only
helps the warm one. **A farm's first identification pays for embedding its whole
register.**

**Both per-photograph rates are recorded because they disagree**, and the honest
reading is that this is a loaded shared laptop rather than a controlled
measurement — the 0.20 s run had a test suite beside it. Extrapolate on the
slower one; a latency figure from a shared machine is not a latency figure, which
is a rule this repository already writes down in
`experiments/cattle_respiratory/README.md`. At 0.20 s a 169-animal register is
about **169 seconds** cold.

Neither rate is the 59 ms in `backbones.py`. That is the embed call alone on a
frame somebody already decoded; this path also reads the blob and decodes a JPEG.
An earlier version of this paragraph extrapolated on 59 ms and so put a
169-animal farm at "about a minute" — roughly three times better than measured,
in a paragraph whose entire purpose is to be honest about a limit.

**Even the small register does not fit the inline job model.** `app/main.py` runs
jobs synchronously, and this service declined to wire an eighteen-second
capability inline at all; eleven seconds for twenty animals is the same order and
a hundred and sixty-nine seconds is not close. So the honest statement is not
"wired for the farm sizes Animap has" — it is that **every** cold request is slow
and only the warm path is fast, so a deployment has to keep the cache warm rather
than treat the cold case as an edge.

The fix is not a bigger cache. It is for the API to store each enrolment vector
beside its `MediaAsset` and send vectors instead of media ids, which moves the
cost to enrolment time where a person is already waiting. That is a change to the
API's schema and is not made here.

**Where these numbers come from, since they cite no run record.** They are a
session measurement, not a benchmark, and the distinction matters in a repository
whose whole discipline is that a figure names a run id. The script is
`scripts/verify_identity_wiring.py`; it is reproducible and it has not been
through the experiment harness, so nothing here may be quoted as a measured
accuracy. What it does establish is that the wiring reproduces the benchmark
rather than merely citing it: over 122 held-out probes against a 60-animal
gallery, top-1 similarity ran min 0.8336, median 0.9764, max 0.9930 against the
run record's min 0.8336, median 0.9707, max 0.9930, and rank-1 was right 121
times out of 122. The accept rate was 41% against the benchmark's 24.08%, which
is the direction a 60-animal gallery moves it against 169.
"""

from __future__ import annotations

import threading
from collections import OrderedDict
from uuid import UUID, uuid4

import numpy as np
from PIL import Image

from app.adapters.embedding import DINOV3_SPEC, OnnxEmbeddingAdapter
from app.adapters.embedding.identity import (
    ENROLMENT_VIEWS,
    PRIMARY_VIEW,
    Embedding,
    IdentityIndex,
    IndexMismatch,
    OpenSetPolicy,
)
from app.capabilities import FORBIDDEN_CLAIMS, Capability
from app.media import MediaRef, MediaStore
from app.providers import ModelArtefact
from app.quality import assess
from app.schemas import (
    EnrolledAnimal,
    InferenceLocation,
    InferenceRequest,
    InferenceResult,
    Observation,
    QualityCheck,
)

#: The operating point measured in run `8db9e0bd1b30`, and the only policy this
#: runner ships.
#:
#: `accept_similarity` 0.98 is the lowest cutoff in the published sweep whose
#: false-accept rate is 0.0. `accept_margin` 0.0 is the measured companion and it
#: never rejects anything — a margin can only be non-negative — which is
#: deliberate rather than an oversight: `open_set_margin_sweep` in the same run
#: shows the margin rule buys nothing here. At margin 0.07 it holds the
#: false-accept rate to 0.0072 while accepting 23.45% of correct matches, which
#: is *worse on both axes* than the similarity rule alone. The rule stays wired
#: because a future backbone may separate the two distributions differently, and
#: it is set where the evidence puts it.
#:
#: **Both numbers cite the run that produced them and `OpenSetPolicy` will not
#: accept them otherwise.** Its `__post_init__` raises `UnmeasuredThreshold` on a
#: threshold with no run id, which is what stops a cutoff being tuned by whoever
#: is looking at a demo that morning.
MEASURED_POLICY = OpenSetPolicy.measured(
    accept_similarity=0.98,
    accept_margin=0.0,
    run_id="8db9e0bd1b30",
    measured_on="muzzle268-169enrolled",
)

#: Said on every run, whatever the result, in the shape `CountingProfile`
#: established. §6.4's product statement, cut to what a person reads.
STANDING_WARNING = (
    "A suggestion, not a record. Animap proposes an animal and you confirm it — "
    "an unconfirmed match is never written to this animal's history."
)

#: Said whenever candidates are returned at all, because the open-set result is
#: not something a reader can infer from a ranked list.
UNENROLLED_WARNING = (
    "An animal that has never been enrolled will still produce a nearest match. "
    "Measured on 416 photographs of unenrolled cattle, every one of them came "
    "back as somebody. If this animal is new, register it rather than picking "
    "the closest name."
)

#: How many candidates a result carries. §6.4 shows a ranked list rather than one
#: answer, and `evidence_correction.selected_interpretation` is the reason: a
#: farmer picking the second name records *"rank 2 was right"*, which its own
#: docstring calls the highest-value training signal in the table. Three, because
#: `closed_set_top3_accuracy` is 0.9937 and a fourth carries no measurement.
TOP_K = 3

#: Enrolment vectors held between requests, keyed by media id and artefact.
#:
#: **Without this the design does not fit the inline job model.** The gallery
#: travels on the request so the service holds no farm state
#: (`schemas.EnrolledAnimal` says why), and the cost of that is re-embedding
#: every enrolled photograph on every identification. At the **measured 0.20 s**
#: per enrolment photograph — end to end, including decode, not the 59 ms embed
#: call `backbones.py` reports on a frame somebody already decoded — a farm with
#: a hundred animals and the five-shot enrolment the accuracy was measured on is
#: 500 photographs, or **about a hundred seconds a request**. That is far past
#: the inline ceiling `app/main.py` describes and far worse than the
#: eighteen-second capability this service declined to wire inline.
#:
#: **A media id is immutable, which is what makes caching it correct rather than
#: merely fast.** `MediaAsset` rows are write-once and the blob behind one is
#: never rewritten, so the same id is the same bytes forever and its vector
#: cannot go stale. The artefact digest is in the key because a re-export with
#: different pooling produces a vector of the same width and a different meaning
#: — the case `IdentityIndex` pins its backbone id to catch — so vectors from two
#: artefacts must never collide here.
#:
#: **The farm id is in the key, and it was not for one commit.** The argument for
#: leaving it out was that nothing can be read from the cache without already
#: holding the media id. A watchdog showed that is not quite the property that
#: matters: `AzureBlobMediaStore` scopes an unpathed lookup to `farm/{farm_id}/`,
#: so *without* a cache a request naming another farm's media id fails to resolve
#: — and *with* one it would hit and score. What leaked was a similarity against
#: a foreign photograph rather than the photograph or the name, and it needed the
#: service token plus a known foreign UUID, so it was narrow. It was also free to
#: close, and a cache must not be the reason a farm boundary that the media store
#: enforces stops being enforced.
_VECTOR_CACHE: OrderedDict[tuple[str, str, str], np.ndarray] = OrderedDict()

#: Entries kept. 768 float32 values is 3 KB, so this is about 12 MB — small
#: against the container's 4 GiB, and enough for a few hundred animals at five
#: shots each. Least-recently-used is evicted, so the farms being worked today
#: stay warm.
_CACHE_LIMIT = 4096

#: `submit()` runs in FastAPI's threadpool, so two captures can be scored at
#: once and an `OrderedDict` is not safe under that on its own.
_CACHE_LOCK = threading.Lock()


def cache_clear() -> None:
    """Empty the vector cache. For tests, and for a deployment that swaps an
    artefact without a restart."""
    with _CACHE_LOCK:
        _VECTOR_CACHE.clear()


def _cached_vector(key: tuple[str, str, str]) -> np.ndarray | None:
    with _CACHE_LOCK:
        vector = _VECTOR_CACHE.get(key)
        if vector is not None:
            _VECTOR_CACHE.move_to_end(key)
        return vector


def _remember(key: tuple[str, str, str], vector: np.ndarray) -> None:
    with _CACHE_LOCK:
        _VECTOR_CACHE[key] = vector
        _VECTOR_CACHE.move_to_end(key)
        while len(_VECTOR_CACHE) > _CACHE_LIMIT:
            _VECTOR_CACHE.popitem(last=False)


#: Muzzle photographs per animal in the protocol the accuracy was measured under.
#:
#: Run `8db9e0bd1b30` enrolled each animal from five, all under the `muzzle` view
#: name, and `IdentityIndex.candidates` scores an animal by its **best** view —
#: so the five are five chances to match rather than five samples averaged
#: together. An animal enrolled from one photograph is being asked a harder
#: question than the benchmark asked, and measured that way on twelve animals,
#: **0 of 23 held-out queries cleared the 0.98 cutoff**.
#:
#: Used only to report a thin enrolment, never to refuse one.
MEASURED_ENROLMENT_SHOTS = 5


class IdentityRunner:
    """Runs `cattle_identity` against one capture and one farm's register."""

    #: The view a single capture is compared against. §6.4 leads with the muzzle
    #: and the re-identification measurement is entirely muzzle-to-muzzle, so
    #: comparing a muzzle print against a side-body vector would contribute
    #: nothing but a chance of a spurious high score. Restricting the pool is
    #: what keeps the runtime behaviour inside the measured one.
    #:
    #: **Nothing checks that the capture actually is a muzzle, and a watchdog was
    #: right to call that a gap.** The registry declares `muzzle_not_visible` in
    #: `reject_if`, but `app/quality.py` implements resolution, illumination and
    #: motion blur and nothing that looks for an anatomical part — so the
    #: condition is declared and never evaluated. A caller that sends the five
    #: enrolment views in the order the registry lists them sends `front_face`
    #: first, and it would be scored against a gallery of muzzles.
    #:
    #: What that costs is bounded in the safe direction: a front-face photograph
    #: is *less* similar to every enrolled muzzle, so it drives the top score down
    #: and towards `no_confident_match`. It degrades to a refusal rather than to a
    #: wrong name. But it is a refusal for a reason nobody is told, and the fix is
    #: a real one — either the capture flow states which view it took, or
    #: something evaluates `muzzle_not_visible`. Neither exists today.
    query_view = PRIMARY_VIEW

    def run(
        self,
        *,
        request: InferenceRequest,
        capability: Capability,
        artefact: ModelArtefact,
        store: MediaStore,
        request_id: UUID | None = None,
    ) -> InferenceResult:
        request_id = request_id or uuid4()
        warnings: list[str] = [STANDING_WARNING]

        if len(request.media_ids) > capability.frames_required:
            warnings.append(
                f"{len(request.media_ids)} frames were supplied; this capability "
                f"reads {capability.frames_required}."
            )

        image = store.open_image(self._ref(request, request.media_ids[0]))
        verdict = assess(image)
        checks = list(verdict.checks)

        if verdict.blocked:
            failure = verdict.first_failure
            warnings.append(
                failure.detail if failure and failure.detail
                else "The capture was not usable."
            )
            return self._result(
                request=request, capability=capability, artefact=artefact,
                request_id=request_id, observations=[], checks=checks,
                warnings=warnings, recapture=True,
            )

        adapter = OnnxEmbeddingAdapter(
            artefact, DINOV3_SPEC,
            input_size=artefact.input_size or 224,
            mean=artefact.image_mean or (0.485, 0.456, 0.406),
            std=artefact.image_std or (0.229, 0.224, 0.225),
            dimensions=artefact.embedding_dimensions or 768,
        ).load()

        index = IdentityIndex(
            farm_id=str(request.farm_id),
            backbone_id=DINOV3_SPEC.adapter_id,
            dimensions=adapter.dimensions,
            artefact_sha256=artefact.sha256,
        )
        names: dict[str, str] = {}
        enrolled_views, skipped = self._enrol(
            index, adapter, artefact, request, store, names,
        )
        warnings.extend(skipped)

        query = self._embed(adapter, artefact, image)
        result = index.match(
            query,
            policy=MEASURED_POLICY,
            top_k=TOP_K,
            # Compared against muzzles only. See `query_view`.
            restrict_to_views=(self.query_view,),
            names=names,
        )

        checks.append(self._enrolment_check(index, enrolled_views))

        observations: list[Observation] = [
            # Emitted on every path, refusals included, for the reason
            # `counting.py` emits its grid: a threshold that has to be re-derived
            # later is re-derived from stored results or not at all.
            Observation(
                type="enrolled_animals",
                value=float(result.enrolled_animals),
                unit="animals",
                confidence=None,
            ),
            Observation(
                type="open_set_verified",
                value=1.0 if result.open_set_verified else 0.0,
                confidence=None,
            ),
        ]

        top = result.top
        if top is not None:
            observations.append(Observation(
                # **Uncalibrated, and named so.** Nothing mapped cosine
                # similarity onto a probability that the animal is Kofi, so this
                # is a diagnostic and the release declares no sentence for it —
                # exactly as `counting_grid` carries none. §37's whole complaint
                # is an uncalibrated model score reaching a person as a promise.
                type="top_candidate_similarity",
                value=round(top.similarity, 4),
                confidence=None,
            ))

        accepted = result.claim == "identity_candidate"
        if not accepted:
            observations.append(Observation(
                type="no_confident_match", value=None, confidence=None,
            ))

        # **The ranked list is emitted on both paths, under different names.**
        # `IdentityResult` returns candidates whether or not the policy accepted
        # the top one, and dropping them on the refusal path would throw away the
        # thing §6.4 is shaped around: `evidence_correction.selected_
        # interpretation` records *"rank 2 was right"*, and its own docstring
        # calls that the highest-value training signal in the table. At the
        # measured operating point three quarters of correct matches are refused,
        # so the refusal path is where most of that signal lives — emitting
        # nothing there would collect it almost nowhere.
        #
        # The type differs because the claim differs. `identity_candidate` is
        # §6.4's *"This looks like Kofi"*; `closest_candidate` is *"not a
        # confident match, and these are the nearest on your register"*. One
        # observation type carrying both would leave the device deciding which
        # sentence to show from a field that does not say, and the release
        # declares a separate sentence for each.
        #
        # ## Exactly one settled name, ever
        #
        # An accepted run used to publish `identity_candidate` for **all three**
        # ranked candidates, so `deploy/semantics` rendered *"This looks like
        # Cow 0100"*, *"This looks like Cow 0200"* and *"This looks like Cow
        # 0300"* — three settled names for one animal, on one `captured_at`,
        # which no ordering on the device can pick between. The policy accepted
        # **the top candidate**; it said nothing whatever about ranks 2 and 3.
        #
        # So rank 1 is the proposal and the rest are what §6.4's screen calls
        # *"Also considered"* — the same list, under the type that does not
        # assert. Picking one of them is `choose_another_animal`, which is a
        # correction and the signal the ledger is built to keep.
        for position, candidate in enumerate(result.candidates):
            settled = accepted and position == 0
            observations.append(Observation(
                type="identity_candidate" if settled else "closest_candidate",
                # The animal's own name. `identity_candidate` is one of three
                # claims in the whole registry that carries free text, and
                # `app/capabilities.py` says why: a registry cannot enumerate a
                # farm's animals. The value comes from the farm's own register
                # via the request, never from this service.
                value=candidate.display_name,
                # **`confidence` stays None on both paths.**
                # `Candidate.confidence` is documented as never set by the
                # matcher, and promoting a similarity into it here would defeat
                # that in one line.
                confidence=None,
            ))

        if result.candidates:
            warnings.append(UNENROLLED_WARNING)

        warnings.extend(result.warnings)

        if index.unverified_queries:
            # Should be zero on this path: every vector is built as an
            # `Embedding` below. Reported rather than asserted, because the count
            # existing at all is the gap between what the index checks and what
            # it would like to.
            warnings.append(
                f"{index.unverified_queries} vectors were compared without "
                f"provenance. Every vector this service builds carries it, so "
                f"this is a defect rather than a capture problem."
            )

        return self._result(
            request=request, capability=capability, artefact=artefact,
            request_id=request_id, observations=observations, checks=checks,
            warnings=warnings,
            # A refusal here is not a bad photograph. The animal may simply not
            # be enrolled, and asking for a recapture would send a worker back to
            # a pen to re-photograph a cow the system has never seen.
            recapture=verdict.degraded,
        )

    def _ref(self, request: InferenceRequest, media_id: UUID,
             object_path: str | None = None) -> MediaRef:
        return MediaRef(
            media_id=media_id,
            farm_id=request.farm_id,
            captured_at=request.captured_at,
            object_path=object_path or request.path_for(media_id),
        )

    def _embed(
        self, adapter: OnnxEmbeddingAdapter, artefact: ModelArtefact,
        image: Image.Image,
    ) -> Embedding:
        """A vector that can prove where it came from.

        Always an `Embedding`, never a bare array. `IdentityIndex._accept` can
        only check the *width* of an array, and a watchdog scored a foreign
        vector against a `dinov3-vits16` index at similarity 1.0 through exactly
        that gap.
        """
        return Embedding(
            vector=adapter.embed(image),
            backbone_id=DINOV3_SPEC.adapter_id,
            artefact_sha256=artefact.sha256,
        )

    def _enrol(
        self,
        index: IdentityIndex,
        adapter: OnnxEmbeddingAdapter,
        artefact: ModelArtefact,
        request: InferenceRequest,
        store: MediaStore,
        names: dict[str, str],
    ) -> tuple[int, list[str]]:
        """Build this farm's gallery from the request. Returns views enrolled.

        A gallery entry that cannot be read is **skipped with a warning rather
        than failing the job**: one unreadable enrolment photograph out of two
        hundred should not stop a farmer identifying an animal, and the animals
        that did load are still a gallery. The count that reaches the result is
        the count that actually enrolled, so a farm cannot be told it was
        compared against more animals than it was.
        """
        enrolled_views = 0
        skipped: list[str] = []

        for entry in request.enrolled:
            usable = self._views_for(entry, skipped)
            if not usable:
                continue
            who = entry.display_name or entry.animal_id
            stored = 0

            for view, media_ids in usable.items():
                for position, media_id in enumerate(media_ids):
                    key = (str(request.farm_id), str(media_id), artefact.sha256)
                    cached = _cached_vector(key)
                    if cached is not None:
                        vector = Embedding(
                            vector=cached,
                            backbone_id=DINOV3_SPEC.adapter_id,
                            artefact_sha256=artefact.sha256,
                        )
                    else:
                        try:
                            image = store.open_image(self._ref(
                                request, media_id,
                                entry.path_for_view(view, position),
                            ))
                        except Exception as exc:  # MediaError, or any store error
                            skipped.append(
                                f"An enrolled {view} photograph of {who} could "
                                f"not be read ({exc}), so it was left out of "
                                f"the comparison."
                            )
                            continue
                        vector = self._embed(adapter, artefact, image)
                        _remember(key, vector.vector)
                    try:
                        # **One call per photograph, under the one view name.**
                        # `enrol` takes a mapping of view to vector, so a single
                        # call can only hold one shot per view — and the measured
                        # protocol is five muzzles per animal, scored by maximum.
                        # Calling it per photograph is how those five reach the
                        # index, and it is what run 8db9e0bd1b30 did.
                        index.enrol(
                            entry.animal_id, {view: vector},
                            media_ids={view: str(media_id)},
                        )
                    except (IndexMismatch, ValueError) as exc:
                        skipped.append(
                            f"A {view} photograph of {who} was left out of the "
                            f"comparison: {exc}"
                        )
                        continue
                    stored += 1

            if not stored:
                continue
            names[entry.animal_id] = who
            enrolled_views += stored

        return enrolled_views, skipped

    def _views_for(
        self, entry: EnrolledAnimal, skipped: list[str]
    ) -> dict[str, list[UUID]]:
        """The views of one animal this runner will compare against.

        Unknown view names are dropped here with a warning rather than left for
        `enrol` to raise on, because one typo in one animal's record should cost
        that animal's view and not the whole request.
        """
        usable: dict[str, list[UUID]] = {}
        for view, media_ids in entry.views.items():
            if view not in ENROLMENT_VIEWS:
                skipped.append(
                    f"{entry.display_name or entry.animal_id} has a view called "
                    f"{view!r}, which is not one of the five §6.4 enrols "
                    f"({', '.join(ENROLMENT_VIEWS)}). It was ignored."
                )
                continue
            if media_ids:
                usable[view] = list(media_ids)
        return usable

    def _enrolment_check(
        self, index: IdentityIndex, enrolled_views: int
    ) -> QualityCheck:
        """Whether this farm had anything to compare the capture against.

        A quality check rather than a warning, because the answer is about the
        *farm's register* rather than about the photograph, and a capture flow
        that can distinguish the two can say *"register this animal"* instead of
        *"take it again"*.
        """
        muzzles = sum(1 for v in index.views if v.view == self.query_view)
        animals = len(index.animal_ids)
        if muzzles:
            detail = (
                f"{animals} animals enrolled, {enrolled_views} views, "
                f"{muzzles} of them muzzles."
            )
            thin = [
                a for a in index.animal_ids
                if sum(1 for v in index.views
                       if v.animal_id == a and v.view == self.query_view)
                < MEASURED_ENROLMENT_SHOTS
            ]
            if thin:
                # **Reported, never a refusal.** A thin enrolment still matches,
                # and one good muzzle photograph is worth more than a refused
                # enrolment — `IdentityIndex.enrol` takes that position and this
                # agrees with it. What it is not is the protocol the accuracy was
                # measured under, and a farm whose animals never quite match
                # deserves to know the reason is its register rather than its
                # camera.
                detail += (
                    f" {len(thin)} of them carry fewer than "
                    f"{MEASURED_ENROLMENT_SHOTS} muzzle photographs, which is "
                    f"the enrolment the accuracy was measured on."
                )
            return QualityCheck(check="gallery", passed=True, detail=detail)
        return QualityCheck(
            check="gallery",
            passed=False,
            detail=(
                f"No enrolled {self.query_view} to compare against. "
                f"{len(index.animal_ids)} animals were sent and none carries "
                f"the view this capability matches on."
            ),
        )

    def _result(
        self, *, request, capability, artefact, request_id, observations,
        checks, warnings, recapture,
    ) -> InferenceResult:
        forbidden = [o.type for o in observations if o.type in FORBIDDEN_CLAIMS]
        if forbidden:
            raise ValueError(
                f"{capability.key} tried to emit a forbidden claim: {forbidden}"
            )
        banned = set(capability.acquisition.forbidden_claims)
        offending = [o.type for o in observations if o.type in banned]
        if offending:
            # `identity_without_confirmation` is the one thing §6.4 forbids, and
            # it is checked here as well as in the registry because this runner
            # is the only thing that could emit it.
            raise ValueError(
                f"{capability.key} tried to emit {offending}, which its own "
                f"registry entry forbids by name."
            )

        return InferenceResult(
            request_id=request_id,
            capability_key=capability.key,
            model_id=artefact.model_id,
            model_version=artefact.version,
            inference_location=InferenceLocation.REMOTE,
            subject_type=request.subject_type,
            subject_id=request.subject_id,
            observations=observations,
            # **No interpretation, and none is possible.** A name is a fact about
            # the register, not a judgement about the animal, and there is
            # nothing for a vet to review. `observation_confidence` is likewise
            # None: the three-level label would be read as a confidence in the
            # name, and nothing calibrated one.
            interpretations=[],
            observation_confidence=None,
            interpretation_confidence=None,
            quality_checks=checks,
            warnings=warnings,
            recommended_recapture=recapture,
        )


#: Capabilities with an implemented adapter, in the shape `app/counting.py`
#: publishes. `app/main.py` merges the two: a capability with a validated
#: artefact and no entry in either returns 501 rather than a plausible result.
RUNNERS: dict[str, IdentityRunner] = {
    "cattle_identity": IdentityRunner(),
}