python/elidedb/identity.py

"""OBJECT IDENTITY on the write path, text-free, riding on tracks.

Identity is a property of a physical object, not of a name and not of a
category. The same spoon across every demo gets one id. A spatula gets
a different id. A red pepper and a green pepper get different ids, even
though every text-aligned encoder calls them both "pepper".

Nothing here turns pixels into language. Regions come from a
class-agnostic segmenter, identity from a ReID encoder; both vision
only, both real time. This is separate from objects.py, which proposes
regions for SEMANTIC retrieval - that channel answers "does this look
like a cloth", this one answers "is this the same object again".

CONTINUITY IS THE ARCHITECTURE
------------------------------
The earlier version sampled 3 frames per episode and asked the gallery
about every crop independently. That is the wrong question asked ~500
times: it made identity a threshold problem, and the threshold had no
good setting - 0.55 built attractor ids that swallowed a green box, a
broccoli, a toy mouse and a carrot; 0.85 fragmented to 1.15 sightings
per object. Neither end works because a single crop of an object from
one angle simply is not enough evidence.

A tracker answers it for free instead. While an object stays visible,
IoU + Kalman association says "same object" from GEOMETRY, at 0.25
ms/frame, with no appearance model and no threshold. The object store
is consulted only when continuity BREAKS - a new track appears - and
then it is asked once, about a descriptor pooled over the whole track,
not once per frame about one view.

Measured over 24 episodes / 883 frames of Bridge:

    median track length      24-48 frames (episodes are ~37)
    singleton tracks         5%
    object-store uploads     19, against 415 tracked detections
                             = 95% fewer questions asked

That 95% is the design. Fewer questions, each with far more evidence
behind it.

COST, and where it goes
-----------------------
Detection is STATELESS, so it batches across episode boundaries;
association is STATEFUL, so it runs per episode, sequentially. Splitting
them is what makes every-frame tracking affordable - ultralytics' own
track() ties them together and pays batch-size-1 prices for both.

    per-frame track() calls          15.8 ms/frame
    batched track() (list per ep)     9.3
    detect batched + associate        4.8   <- this module
      of which detect  (fp16, 448)    4.16
      of which associate              0.13
      of which reid                   0.50  (was 4.1 per FRAME)

At 18,000 frames per hour of video that is 1.43 min/hour marginal,
against a 1 min/hour write budget of which 0.23 is already spent by the
existing pass. STATED STRAIGHT: identity does not fit the budget. What
riding on tracks bought is the ReID half, which used to dominate and is
now 0.1 min/hour; the detector is the entire remaining cost, so closing
the gap is a detector question, not a design one.

fp16 is worth 14% of detect (4.94 -> 4.16 ms/frame) with the fitted cut
and track count unchanged. That took interleaved runs to establish - a
first attempt compared an fp16 run against fp32 runs taken earlier in a
long session and read the thermal drift as fp16 being SLOWER.

DEVICE. Defaults to the GPU, and this is now measured rather than
assumed. The old 3-frames-per-episode path put CPU and MPS within 5%
(19.3 vs 20.3 ms/frame) because per-call overhead dominated a batch of
three. On real batches the gap is 3.7x - mps 6.7 ms/frame vs cpu 24.8
at imgsz 640 - and the CPU gets WORSE with batching (93.9 ms/frame at
batch 16) as threads contend. Frames are already resident on the GPU
for the FDNN-V pass, and on CUDA a host round trip per frame costs more
than the inference. CPU is the fallback, never the default.
"""
from __future__ import annotations

import os
from types import SimpleNamespace

import numpy as np

_M: dict = {}

# WEIGHTS LIVE IN models/weights, not the repo root. ultralytics
# downloads into the CWD by default, which is how 86 MB of checkpoints
# ended up beside the README with 62 MB of it committed to git. _weights
# resolves a bare name against models/weights and falls back to the bare
# name so a missing file still auto-downloads - into models/weights,
# because that is where we then look.
def _weights(name):
    from pathlib import Path as _P
    d = _P(__file__).resolve().parents[2] / "models" / "weights"
    d.mkdir(parents=True, exist_ok=True)
    p = d / name
    return str(p) if p.exists() else str(d / name)


SEG_MODEL = os.environ.get("ELIDEDB_SEG") or _weights("yolo11n-seg.pt")
REID_MODEL = os.environ.get("ELIDEDB_REID") or _weights("yolo26n-reid.onnx")
# imgsz 448 rather than 384, deliberately paying 0.5 ms/frame for it.
# 384 is cheaper (3.76 vs 4.23 ms/frame with fp16) and finds MORE boxes
# (2.63 vs 2.44 per frame), but the extra boxes are marginal ones: its
# proven-different pairs reach 0.757 against 0.555 at 448, i.e. noisier
# negatives, and it recovered 72 multi-episode objects against 81. A
# cheaper size that degrades the identity is not cheaper.
DET_SZ = int(os.environ.get("ELIDEDB_DET_SZ", "448"))
# free: same 2.44 boxes/frame, 4.74 -> 4.23 ms/frame
DET_FP16 = os.environ.get("ELIDEDB_DET_FP16", "1") not in ("0", "")
# detection is stateless, so the batch may cross episode boundaries;
# 3.87 ms/frame at ~37 (one episode) against 3.32 at 64
DET_BATCH = int(os.environ.get("ELIDEDB_DET_BATCH", "64"))
# Global motion compensation estimates camera movement before
# association. Default OFF from MEASUREMENT, not from an assumption
# about this corpus: with sparseOptFlow, median track length was 47 at
# imgsz 448 and 38 at 640; with none, 48 and 38 - identical - while the
# optical flow cost 1.8 ms/frame, 7x the entire association step. A
# corpus with a moving camera must re-measure and will likely want it
# back; that is what the knob is for.
GMC = os.environ.get("ELIDEDB_GMC", "none")
# how many views of a track to embed. The pooled descriptor is what the
# object store sees, so this trades evidence against ReID calls.
EXEMPLARS = int(os.environ.get("ELIDEDB_OBJ_VIEWS", "4"))

# "same physical object" cut. Not a constant to be tuned by hand - see
# calibrate(), which fits it from the corpus being ingested. This value
# is only the fallback for a corpus too small to fit on.
MATCH = float(os.environ.get("ELIDEDB_OBJ_MATCH", "0.85"))
# quantile of the free-negative distribution used as the cut
CAL_Q = float(os.environ.get("ELIDEDB_OBJ_CAL_Q", "99.5"))
# Adding an exemplar WIDENS the id's acceptance region, because matching
# takes a max over the gallery. Unchecked that is a runaway: absorb ->
# widen -> absorb more. A new view is admitted only if it is close to
# EVERY exemplar already held, which keeps an id tight instead of
# letting it sprawl across a chain of intermediate appearances.
MAX_EXEMPLARS = 5
ADMIT = float(os.environ.get("ELIDEDB_OBJ_ADMIT", "0.70"))
MIN_SIDE = 16
MAX_AREA = 0.5


def device():
    d = os.environ.get("ELIDEDB_DEVICE")
    if d:
        return d
    try:
        import torch
        if torch.cuda.is_available():
            return "cuda"
        if torch.backends.mps.is_available():
            return "mps"
    except Exception:
        pass
    return "cpu"


def _load():
    # check for THIS function's own keys, not for _M being non-empty:
    # propose() also caches into _M, so `if _M` short-circuited here and
    # left reid unloaded the moment a proposer had run first.
    if "reid" in _M:
        return _M
    from ultralytics import YOLO
    from ultralytics.trackers.utils.reid import ReID
    dev = device()
    _M["seg"] = YOLO(SEG_MODEL)
    # the ReID encoder runs under onnxruntime and picks its own
    # execution provider; it does not share the torch device
    _M["reid"] = ReID(REID_MODEL, device="cpu" if dev == "mps" else dev)
    _M["dev"] = dev
    return _M


PROPOSER = os.environ.get("ELIDEDB_PROPOSER") or _weights("FastSAM-s.pt")


def propose(frames, imgsz=DET_SZ, batch=DET_BATCH, conf=0.25):
    """CLASS-AGNOSTIC region proposal. Not YOLO-with-single_cls.

    MEASURED, same frames, same model: single_cls=True and
    single_cls=False give byte-identical output - 3.10 det/frame, all of
    it {oven, sink, bowl, person, spoon, wine glass}. single_cls is a
    TRAINING flag; at inference the detector still fires only on
    COCO-shaped things. Everything built on it inherited an 80-class
    prior that was never intended, which is why looking for a container
    found 0.7 regions per episode and they were ovens.

    FastSAM is trained on SA-1B with no class list at all, and on this
    corpus returns 45.3 regions/frame against COCO YOLO's 2.6, with 3.5
    large structures per frame against 1.1.

    It costs 24.93 ms/frame batched fp16 = 7.48 min per hour of video,
    which is 7x the write budget - so this is a TEACHER. It is paid once
    to make training pairs and then distilled, exactly like every other
    teacher here. Never put it on the write path directly.
    """
    from ultralytics import FastSAM
    if "prop" not in _M:
        _M["prop"] = FastSAM(PROPOSER)
    out = []
    kw = {"quantize": "fp16"} if DET_FP16 else {}
    for i in range(0, len(frames), batch):
        chunk = frames[i:i + batch]
        res = _M["prop"].predict(chunk, device=device(), verbose=False,
                                 imgsz=imgsz, conf=conf, **kw)
        for im, r in zip(chunk, res):
            H, W = im.shape[:2]
            if r.boxes is None or not len(r.boxes):
                out.append(np.zeros((0, 4), np.int32))
                continue
            b = r.boxes.xyxy.cpu().numpy()
            b = np.stack([b[:, 0].clip(0, W), b[:, 1].clip(0, H),
                          b[:, 2].clip(0, W), b[:, 3].clip(0, H)],
                         1).astype(np.int32)
            wh = (b[:, 2] - b[:, 0]), (b[:, 3] - b[:, 1])
            out.append(b[(wh[0] >= MIN_SIDE) & (wh[1] >= MIN_SIDE)])
    return out


def detect(frames, imgsz=DET_SZ, batch=DET_BATCH, conf=0.25):
    """Class-agnostic regions per frame: [(boxes Nx4, conf N, area N)].

    Stateless and therefore batched - the caller may hand in frames from
    several episodes at once. single_cls collapses the 80 COCO classes
    into one anonymous class, so the model reports "a thing is here" and
    never a label.
    """
    m = _load()
    out = []
    kw = {"quantize": "fp16"} if DET_FP16 else {}
    for i in range(0, len(frames), batch):
        chunk = frames[i:i + batch]
        res = m["seg"].predict(chunk, device=m["dev"], single_cls=True,
                               verbose=False, imgsz=imgsz, conf=conf, **kw)
        for im, r in zip(chunk, res):
            H, W = im.shape[:2]
            if r.boxes is None or not len(r.boxes):
                out.append((np.zeros((0, 4), np.int32), np.zeros(0, np.float32),
                            np.zeros(0, np.float32)))
                continue
            xy = r.boxes.xyxy.cpu().numpy()
            cf = r.boxes.conf.cpu().numpy()
            ar = (r.masks.data.sum((1, 2)).cpu().numpy()
                  if r.masks is not None else np.zeros(len(xy)))
            b = np.stack([xy[:, 0].clip(0, W), xy[:, 1].clip(0, H),
                          xy[:, 2].clip(0, W), xy[:, 3].clip(0, H)],
                         1).astype(np.int32)
            wh = (b[:, 2] - b[:, 0]), (b[:, 3] - b[:, 1])
            ok = ((wh[0] >= MIN_SIDE) & (wh[1] >= MIN_SIDE)
                  & (wh[0] * wh[1] <= MAX_AREA * W * H))
            out.append((b[ok], cf[ok].astype(np.float32),
                        ar[ok].astype(np.float32)))
    return out


class _Dets:
    """The minimal shape ultralytics' trackers consume: xywh/conf/cls,
    a length, and boolean-mask indexing (they split detections into
    high- and low-confidence subsets). Handing them this instead of a
    Results object is what lets detection run batched somewhere else."""

    __slots__ = ("xywh", "conf", "cls")

    def __init__(self, xywh, conf, cls):
        self.xywh, self.conf, self.cls = xywh, conf, cls

    def __len__(self):
        return len(self.conf)

    def __getitem__(self, m):
        return _Dets(self.xywh[m], self.conf[m], self.cls[m])

    @property
    def xyxy(self):
        """Corner boxes, which global motion compensation asks for.

        byte_tracker.py calls `self.gmc.apply(img, results_high.xyxy)` to
        mask moving objects out before estimating camera motion, inside a
        try/except that WARNS and falls back to an identity warp. Without
        this property every frame took that fallback: 1,434 warnings in
        the first two minutes of a write, and camera motion silently not
        compensated.

        It was invisible on this corpus because Bridge's camera is fixed,
        where identity is the correct warp anyway - so the bug cost
        nothing here and everything on a corpus that moves. A robot that
        drives or a vehicle camera would have had its tracks fragmented
        by exactly the motion GMC exists to remove.

        xywh is centre-based (built that way in Stream.push, and what
        ultralytics' xywh2ltwh assumes), so the corners are centre +/-
        half-extent.
        """
        cx, cy, w, h = (self.xywh[:, 0], self.xywh[:, 1],
                        self.xywh[:, 2], self.xywh[:, 3])
        return np.stack([cx - w / 2, cy - h / 2,
                         cx + w / 2, cy + h / 2], 1)


def _tracker(gmc=None):
    from ultralytics.trackers.bot_sort import BOTSORT
    from ultralytics.utils import YAML
    import ultralytics
    from pathlib import Path
    cfg = YAML.load(Path(ultralytics.__file__).parent
                    / "cfg/trackers/botsort.yaml")
    cfg["gmc_method"] = gmc or GMC
    # BoT-SORT's own ReID would embed every detection on every frame,
    # which is the cost this design exists to avoid. Association here is
    # geometry only; appearance is consulted once per track, later.
    cfg["with_reid"] = False
    return BOTSORT(args=SimpleNamespace(**cfg))


class Stream:
    """CONTINUOUS tracking over a stream. No episode boundaries.

    link() below takes one episode's frames and builds a fresh tracker
    for each - correct when a corpus ships discrete demos, and an
    assumption the engine has no right to make. Raw capture is a
    continuous recording; episodes are something the engine must
    PRODUCE. A driving log, a surveillance feed and a surgical recording
    have no cuts to reset on.

    So the tracker runs for the life of the stream and a track ends when
    the OBJECT does - it leaves frame, is occluded past the buffer, or
    is carried away. That track's span is exactly a PRESENCE INTERVAL:
    "this object was here, from t0 to t1", one row however long it
    lasted. An object sitting still for four hours is one row, not
    72,000, and it is finally recorded at all - the motion-triggered
    element path never produced a row for anything that did not move.

    Online by construction: detections go in frame by frame, closed
    tracks come out as they close, and only the open tracks are held.
    A stream that does not fit in memory is the normal case, not an
    edge case.

    Exemplar crops are retained per OPEN track and embedded once, at
    close - the same "ask the object store one question per track, with
    a whole sighting behind it" that made identity work, now without
    needing the episode to know when to ask.
    """

    def __init__(self, gmc=None, views=EXEMPLARS, buffer_s=6.0):
        self.tr = _tracker(gmc)
        self.views = views
        # how long a track survives with no detection before it is
        # declared ended, in SECONDS not frames: a frame count means
        # different things at 5 fps and 30 fps, and the corpus chooses
        # the frame rate.
        self.buffer_s = buffer_s
        self.open: dict[int, dict] = {}
        self.last_ts = None

    def update(self, ts, det, frame=None):
        """One frame in; the tracks that CLOSED at this frame out.

        `det` is (boxes Nx4, conf N, area N) from detect()/propose().
        """
        b, c, a = det
        self.last_ts = int(ts)
        seen = set()
        if len(b):
            xywh = np.stack([(b[:, 0] + b[:, 2]) / 2,
                             (b[:, 1] + b[:, 3]) / 2,
                             b[:, 2] - b[:, 0], b[:, 3] - b[:, 1]], 1)
            rows = self.tr.update(
                _Dets(xywh.astype(np.float32), c,
                      np.zeros(len(b), np.float32)), frame)
            for row in rows:
                tid, di = int(row[-4]), int(row[-1])
                if di >= len(b):
                    continue
                seen.add(tid)
                t = self.open.setdefault(tid, {
                    "ts": int(ts), "t1": int(ts), "n": 0, "t": [],
                    "box": [], "conf": [], "area": [], "crops": []})
                t["t1"] = int(ts)
                t["n"] += 1
                # the box list without its timestamps is a SHAPE, not a
                # trajectory; every consumer that wanted motion had to
                # guess the time axis back from (ts, t1, n)
                t["t"].append(int(ts))
                t["box"].append(b[di])
                t["conf"].append(float(c[di]))
                t["area"].append(float(a[di]))
                # retain a bounded, spread set of views for the one
                # ReID call this track will ever cost
                if frame is not None and len(t["crops"]) < self.views:
                    x0, y0, x1, y1 = (int(v) for v in b[di])
                    if x1 > x0 and y1 > y0:
                        t["crops"].append((b[di], frame[y0:y1, x0:x1]))
        return self._reap(seen)

    def _reap(self, seen):
        gap = self.buffer_s * 1e9
        done = []
        for tid in list(self.open):
            if tid in seen:
                continue
            if self.last_ts - self.open[tid]["t1"] > gap:
                done.append((tid, self.open.pop(tid)))
        return done

    def flush(self):
        """End of stream: everything still open is still real."""
        out = list(self.open.items())
        self.open = {}
        return out


def link(dets, frames=None, gmc=None):
    """Associate one episode's consecutive detections into tracks.

    Returns {track_id: {"f": [frame idx], "box": [...], "conf": [...],
    "area": [...]}}. A fresh tracker per call, which is right only when
    the caller really does have a scene cut. Prefer Stream for
    continuous capture; this remains for corpora that ship discrete
    clips, where resetting at a genuine cut avoids carrying Kalman state
    across it.
    """
    tr = _tracker(gmc)
    out: dict[int, dict] = {}
    for j, (b, c, a) in enumerate(dets):
        if not len(b):
            continue
        xywh = np.stack([(b[:, 0] + b[:, 2]) / 2, (b[:, 1] + b[:, 3]) / 2,
                         b[:, 2] - b[:, 0], b[:, 3] - b[:, 1]], 1)
        det = _Dets(xywh.astype(np.float32), c, np.zeros(len(b), np.float32))
        rows = tr.update(det, None if frames is None else frames[j])
        for row in rows:
            # ultralytics packs [*xyxy, track_id, conf, cls, det_idx]
            tid, di = int(row[-4]), int(row[-1])
            if di >= len(b):
                continue
            t = out.setdefault(tid, {"f": [], "box": [], "conf": [],
                                     "area": []})
            t["f"].append(j)
            t["box"].append(b[di])
            t["conf"].append(float(c[di]))
            t["area"].append(float(a[di]))
    return out


def _views(track, k=EXEMPLARS):
    """Pick up to k frames of a track: the best-scoring view in each of
    k equal slices of the track's life. Spread matters more than score -
    k views of the same instant carry one view's worth of evidence."""
    n = len(track["f"])
    if n <= k:
        return list(range(n))
    edges = np.linspace(0, n, k + 1).round().astype(int)
    conf = np.asarray(track["conf"])
    return [int(a + conf[a:b].argmax())
            for a, b in zip(edges[:-1], edges[1:]) if b > a]


def features(frame, boxes):
    """Appearance embedding per region. Vision only, no text."""
    if not len(boxes):
        return np.zeros((0, 512), np.float32)
    m = _load()
    b = np.asarray(boxes, np.float32)
    xywh = np.stack([(b[:, 0] + b[:, 2]) / 2, (b[:, 1] + b[:, 3]) / 2,
                     b[:, 2] - b[:, 0], b[:, 3] - b[:, 1]], 1)
    f = np.asarray(m["reid"](frame, xywh))
    f = f.reshape(len(boxes), -1).astype(np.float32)
    return f / (np.linalg.norm(f, axis=1, keepdims=True) + 1e-8)


def descriptors(frames, tracks, k=EXEMPLARS):
    """One pooled appearance vector per track - the unit of upload.

    This is the "break in continuity" moment: while a track holds, the
    object store hears nothing about it. When the track is done, it is
    described ONCE, by a mean over k views spread across its life, and
    that single descriptor is what the store is asked about.
    """
    ids, out = [], []
    for tid, t in tracks.items():
        V = []
        for i in _views(t, k):
            V.append(features(frames[t["f"][i]], [t["box"][i]])[0])
        v = np.mean(V, 0)
        ids.append(tid)
        out.append(v / (np.linalg.norm(v) + 1e-8))
    return ids, (np.stack(out) if out else np.zeros((0, 512), np.float32))


def _iou(a, b):
    x0, y0 = max(a[0], b[0]), max(a[1], b[1])
    x1, y1 = min(a[2], b[2]), min(a[3], b[3])
    inter = max(x1 - x0, 0) * max(y1 - y0, 0)
    ua = ((a[2] - a[0]) * (a[3] - a[1])
          + (b[2] - b[0]) * (b[3] - b[1]) - inter)
    return inter / ua if ua > 0 else 0.0


def free_negatives(tracks, iou_max=0.1):
    """Track-id pairs that are CERTAINLY different physical objects.

    Two tracks occupying DISJOINT REGIONS OF THE SAME FRAME are two
    different objects: one object cannot be in two places at one
    instant, and the tracker has already linked each object to itself,
    so a second track is a second thing. No annotation, no dataset
    knowledge, no prior - geometry alone.

    The disjointness test is not decoration. Co-existence ALONE is not
    enough, because a detector will sometimes put two boxes on one
    object; those two tracks co-exist and look identical, and they land
    exactly in the high tail that a calibration quantile reads. Fitting
    on co-existence alone drove the cut to its 0.95 ceiling on a 300-
    episode run and fragmented the store to 1.21 sightings per object.
    Requiring the boxes to be apart removes the double detections, which
    are the only way two tracks of the same object can co-exist.
    """
    out, ids = [], list(tracks)
    for x in range(len(ids)):
        a = tracks[ids[x]]
        fa = {f: i for i, f in enumerate(a["f"])}
        for y in range(x + 1, len(ids)):
            b = tracks[ids[y]]
            shared = [f for f in b["f"] if f in fa]
            if not shared:
                continue
            bf = {f: i for i, f in enumerate(b["f"])}
            if all(_iou(a["box"][fa[f]], b["box"][bf[f]]) <= iou_max
                   for f in shared):
                out.append((ids[x], ids[y]))
    return out


def free_positives(tracks, iou_min=0.8):
    """Track-id pairs that are CERTAINLY the SAME physical object.

    The mirror of free_negatives, from the mirror of its argument. That
    one says two tracks in DISJOINT regions of one frame are two objects,
    because one object cannot be in two places at once. This one says two
    tracks in the SAME region of one frame are ONE object, because two
    objects cannot occupy one place at once. Both are geometry; neither
    needs an annotation, a label or a dataset prior.

    These pairs are DOUBLE DETECTIONS - the detector put two boxes on one
    thing - which is why free_negatives works to exclude them. They were
    treated as waste. They are not waste: they are the only proven-same
    evidence this corpus can produce for free, and without them the cut
    was fitted from one side of a two-sided decision.

    iou_min is high on purpose. Overlap alone does not prove identity:
    nested and contacting things - a lid on a jar, a hand on a pot -
    overlap heavily and are two objects. Measured on fresh_bench, the
    positive sample gets monotonically cleaner as the bar rises, and the
    AUC it implies rises with it (0.860 at IoU>0.5 to 0.902 at IoU>0.95),
    i.e. the loose bands are contaminated by exactly those nested pairs.
    0.8 keeps thousands of pairs while paying most of that gap.

    BIAS, stated, because it bounds what this can conclude: a same-frame
    positive is the EASIEST positive there is - one instant, one
    viewpoint, one exposure - while a same-frame negative is the HARDEST
    negative. So the fitted cut is bracketed by two optimistic samples
    pulling in opposite directions, and a genuine cross-episode
    re-identification is harder than anything measured here.
    """
    out, ids = [], list(tracks)
    for x in range(len(ids)):
        a = tracks[ids[x]]
        fa = {f: i for i, f in enumerate(a["f"])}
        for y in range(x + 1, len(ids)):
            b = tracks[ids[y]]
            shared = [f for f in b["f"] if f in fa]
            if not shared:
                continue
            bf = {f: i for i, f in enumerate(b["f"])}
            if all(_iou(a["box"][fa[f]], b["box"][bf[f]]) >= iou_min
                   for f in shared):
                out.append((ids[x], ids[y]))
    return out


def interval_pairs(rows, iou_diff=0.1, iou_same=0.8, start_frac=0.25):
    """free_negatives and free_positives over CLOSED intervals.

    The write path does not hold tracks - it holds closed presence
    intervals, each reduced to one box. Same two geometric arguments,
    applied to what the writer actually has.

    APPROXIMATION, stated: the retained box is the track's FIRST box, so
    two boxes are only comparable when the tracks START together. That is
    always true of the double detections the positive test is looking
    for, and start_frac enforces it rather than assuming it. The negative
    test does not need the guard - a pair wrongly called disjoint is a
    pair dropped, not a pair mislabelled.

    Args:
        rows: (stream, meta) or (stream, meta, vec) as close_track emits;
              meta carries ts, t1 and box.
    Returns:
        (neg, pos) index-pair lists into `rows`.
    """
    n = len(rows)
    key = sorted(range(n), key=lambda i: (rows[i][0], rows[i][1]["ts"]))
    neg, pos = [], []
    for a in range(len(key)):
        i = key[a]
        si, mi = rows[i][0], rows[i][1]
        for b in range(a + 1, len(key)):
            j = key[b]
            sj, mj = rows[j][0], rows[j][1]
            if sj != si or mj["ts"] > mi["t1"]:
                break                       # sorted: no later one overlaps
            ov = _iou(mi["box"], mj["box"])
            # the sweep runs in time order, so emit (min, max) rather
            # than (earlier, later) - a pair is unordered and callers
            # should not have to know which way round it came out.
            e = (i, j) if i < j else (j, i)
            if ov <= iou_diff:
                neg.append(e)
            elif ov >= iou_same:
                span = min(mi["t1"] - mi["ts"], mj["t1"] - mj["ts"])
                if abs(mj["ts"] - mi["ts"]) <= start_frac * max(span, 1):
                    pos.append(e)
    return neg, pos


def handoff_pairs(rows, gap_s=1.0, iou_min=0.5):
    """Proven-same evidence every corpus produces in BULK: track handoffs.

    An interval that ends and another that begins at the same place
    within `gap_s` is one object twice - things do not teleport, and
    nothing else can arrive at that exact spot that fast. This is the
    within-group continuity that double-detection positives cannot
    supply (a clean proposer produces almost none: 77 pairs on the sim
    corpus vs 927k negatives), and it is exactly the decision event
    binding depends on: "is the block after the occlusion the block
    from before it".

    The retained box is the track's FIRST box, so a pair is emitted only
    when the dying track's origin still overlaps the resuming track's
    origin - true for the static objects that dominate any corpus, a
    miss (never a mislabel) for objects whose track broke mid-flight.
    """
    from bisect import bisect_right
    by = {}
    for i, r in enumerate(rows):
        by.setdefault(r[0], []).append(i)
    out = []
    gap = int(gap_s * 1e9)
    for idxs in by.values():
        idxs.sort(key=lambda i: rows[i][1]["ts"])
        ts = [rows[i][1]["ts"] for i in idxs]
        for i in idxs:
            e = rows[i][1]["t1"]
            for k in range(bisect_right(ts, e), bisect_right(ts, e + gap)):
                j = idxs[k]
                if _iou(rows[i][1]["box"], rows[j][1]["box"]) >= iou_min:
                    out.append((i, j) if i < j else (j, i))
    return out


def calibrate(V, pairs, pos=None, q=CAL_Q, floor=0.5, ceil=0.95):
    """Fit the match cut from proven pairs, on this corpus.

    TWO-SIDED when proven-same pairs are supplied, and that is the
    correction. The one-sided form below reads the q-th percentile of the
    proven-DIFFERENT similarities - "tighter than all but q% of what I
    can show is not the same object" - which is sound only if that sample
    is clean. On fresh_bench it was not: double detections are 0.5% of
    co-existing pairs and q=99.5 reads the top 0.5%, so the cut was very
    nearly a readout of the contamination. It sat at 0.739, where only
    26% of proven-same pairs are accepted, and the store fragmented to
    78% singletons.

    Fixing the contamination alone moves it to 0.692 - still 34% recall.
    A one-sided fit cannot do better, because the tail of the negatives
    says nothing about where the positives are. Hence Youden's J over
    both: the cut that maximises (true accept rate - false merge rate),
    the standard criterion when neither error has a stated price. It is
    fitted, not chosen; no number here is hand-picked.

    Fails safe in stages: no positives, use the one-sided percentile; too
    few negatives to estimate at all, keep MATCH.

    Args:
        V: (N, D) unit-norm track descriptors.
        pairs: proven-DIFFERENT index pairs, from free_negatives.
        pos: proven-SAME index pairs, from free_positives. Optional.
    """
    if len(pairs) < 20:
        return MATCH, len(pairs)
    neg = np.array([float(V[i] @ V[j]) for i, j in pairs])
    if pos is None or len(pos) < 20:
        return float(np.clip(np.percentile(neg, q), floor, ceil)), len(neg)
    p = np.array([float(V[i] @ V[j]) for i, j in pos])
    grid = np.linspace(0.0, 1.0, 201)
    j = ((p[None, :] >= grid[:, None]).mean(1)
         - (neg[None, :] >= grid[:, None]).mean(1))
    return float(np.clip(grid[int(j.argmax())], floor, ceil)), len(neg)


def fit_cut(V, groups, neg, pos=None, grid=None, handoff=None):
    """Fit the cut by MAXIMISING what identity is actually for.

    Both free samples - proven-different and proven-same - are SAME-FRAME
    pairs. The decision the gallery actually makes is not that one. It is
    "is this new track the object I saw in a different episode?", and no
    same-frame pair is evidence about it. So a statistic of those pairs
    can bound the descriptor's quality but cannot choose the operating
    point, and choosing it from one anyway is how the cut ended up where
    it did.

    The target quantity is measurable directly and needs no annotation:
    HOW MANY OBJECTS RECUR ACROSS GROUPS. That is the whole point of a
    persistent id, and it is self-limiting as an objective - too tight
    and every sighting is a new id that recurs nowhere; too loose and
    distinct objects collapse into one attractor, which also recurs
    nowhere because there are no longer distinct objects to recur. It
    therefore has an interior maximum, and that maximum is the fit.

    Measured on fresh_bench (2,097 episodes, 73,521 intervals):

        cut    objects  singletons  RECUR   worst false-merge
        0.75    34,063     80.2%    6,447        0.18%
        0.70    26,159     74.0%    6,475        0.22%   <- peak
        0.60    14,011     57.3%    5,732        0.54%
        0.50     6,320     36.8%    3,876        1.05%
        0.42     2,618     19.4%    2,075        1.72%

    which also disposes of the singleton rate as a target: driving it
    from 80% to 19% costs two thirds of the recurrence and multiplies
    proven-wrong merges by eight. Fewer, bigger, wronger objects.

    Sweeping a cut used to be impossible - Gallery.assign was ~20 minutes
    per pass - so the fit had to be a closed-form statistic. It is now
    seconds, and an objective beats a proxy.

    Args:
        V: (N, D) unit-norm descriptors.
        groups: (N,) group label per descriptor - the episode it sits in.
        neg: proven-different pairs, for the reported false-merge rate.
        pos: proven-same pairs, for the reported recovery rate.
        grid: candidate cuts. Defaults to percentiles of the proven-
              different similarities, so the range adapts to the corpus
              rather than being a hand-written span.
        handoff: track-handoff proven-same pairs (handoff_pairs). When
              supplied in bulk they CONSTRAIN the objective - see below.
    Returns:
        (cut, report) - report lists every candidate that was tried.
    """
    g = np.asarray(groups)
    g = g - g.min() + 1
    span = int(g.max()) + 2
    N = np.asarray(neg, np.int64).reshape(-1, 2)
    Pp = np.asarray(pos if pos is not None else [], np.int64).reshape(-1, 2)
    H = np.asarray(handoff if handoff is not None else [],
                   np.int64).reshape(-1, 2)
    if grid is None:
        s = (np.einsum("ij,ij->i", V[N[:, 0]], V[N[:, 1]]) if len(N)
             else np.array([MATCH]))
        grid = np.unique(np.round(np.percentile(
            s, [90, 95, 97.5, 99, 99.3, 99.5, 99.7, 99.9]), 3))

    def _row(cut):
        ids = Gallery(match=float(cut)).assign(V)
        n = int(ids.max()) + 1
        # objects present in more than one group
        u = np.unique(ids.astype(np.int64) * span + g)
        per = np.bincount((u // span).astype(np.int64), minlength=n)
        row = {"cut": float(cut), "objects": n,
               "recur": int((per > 1).sum()),
               "singleton_pct": round(100.0 * float(
                   (np.bincount(ids) == 1).mean()), 1)}
        if len(N):
            row["false_merge_pct"] = round(100.0 * float(
                (ids[N[:, 0]] == ids[N[:, 1]]).mean()), 3)
        if len(Pp):
            row["recovered_pct"] = round(100.0 * float(
                (ids[Pp[:, 0]] == ids[Pp[:, 1]]).mean()), 1)
        if len(H):
            row["handoff_pct"] = round(100.0 * float(
                (ids[H[:, 0]] == ids[H[:, 1]]).mean()), 1)
        return row

    def _pick(rep):
        # Recurrence alone is GAMEABLE BY FRAGMENTATION on a lookalike
        # corpus: identical twins across episodes mean noise-clusters of
        # one kind still span groups, so the count keeps rising as the
        # cut climbs while real objects shatter (measured on the sim
        # store: 21,492 objects for ~450 true blocks, 2.67 ids per
        # block, 8% same-block event agreement). A cut that cannot
        # re-find an object across a one-second track break identifies
        # nothing - so when handoff pairs exist in bulk, continuity
        # recovery is a CONSTRAINT on the recurrence objective, not a
        # tiebreaker: maximise recurrence among cuts that recover >=70%
        # of handoffs, and if none does, take the best recovery there is.
        if len(H) >= 30:
            ok = [r for r in rep if r.get("handoff_pct", 0.0) >= 70.0]
            if ok:
                return max(ok, key=lambda r: r["recur"])
            return max(rep, key=lambda r: (r.get("handoff_pct", 0.0),
                                           r["recur"]))
        return max(rep, key=lambda r: r["recur"])

    report = [_row(cut) for cut in grid]
    best = _pick(report)
    # The grid is percentiles of the proven-DIFFERENT similarities, and
    # that anchor breaks when the descriptor is much better than the
    # negatives are hard: recurrence keeps rising past the negatives'
    # entire tail, the argmax lands on the grid's top edge, and the
    # "interior maximum" was never actually bracketed. Measured on the
    # DINOv3 rebuild: recur still climbing 7,590 -> 10,541 at the last
    # point. So while the best cut IS the top edge, keep extending
    # upward (midpoint steps toward 0.99) until the maximum is interior
    # or the ceiling is reached - the fit must end bracketed, not
    # truncated by an artifact of where the negatives happened to end.
    while best["cut"] == report[-1]["cut"] and best["cut"] < 0.985:
        report.append(_row(round(best["cut"]
                                 + (0.99 - best["cut"]) / 2, 3)))
        nxt = _pick(report)
        if nxt["cut"] == best["cut"]:
            break
        best = nxt
    return best["cut"], report


class Gallery:
    """Persistent identities, matched by appearance.

    A GALLERY of exemplars per object, not a running mean. The mean is
    what broke the earlier attempt: a centroid that updates as it
    absorbs members lets a group CHAIN - green, to a slightly different
    green, to red - so two objects merge through a path of intermediate
    views. Several fixed exemplars, matched against the best of them,
    has no such path. This is ordinary ReID gallery practice.

    What it is asked about changed: entries are now TRACK descriptors
    pooled over many frames, not single crops, so each question carries
    a whole sighting's worth of evidence.
    """

    def __init__(self, match=MATCH, max_ex=MAX_EXEMPLARS, admit=ADMIT):
        self.match, self.max_ex, self.admit = match, max_ex, admit
        self.ex: list[np.ndarray] = []
        self.n: list[int] = []
        # every exemplar of every object in ONE matrix, plus which object
        # owns each row. self.ex stays the public view of the same thing.
        self._M = np.zeros((0, 0), np.float32)
        self._own = np.zeros(0, np.int32)
        self._m = 0

    def _grow(self, dim):
        if self._M.shape[0] > self._m:
            return
        cap = max(1024, self._M.shape[0] * 2)
        M = np.zeros((cap, dim), np.float32)
        if self._m:
            M[:self._m] = self._M[:self._m]
        own = np.zeros(cap, np.int32)
        own[:self._m] = self._own[:self._m]
        self._M, self._own = M, own

    def assign(self, feats):
        """Which object is each descriptor, in arrival order.

        ONE matrix-vector product per descriptor, not one per object.
        The old form looped over self.ex in Python, so a 73,521-track
        corpus that had grown 32,756 objects did 2.4 billion iterations
        of a two-line body - about 20 of the write's 54 minutes, and slow
        enough that the cut could not be swept to find out it was wrong.

        The reduction is exact, not an approximation: the best object is
        the one owning the best EXEMPLAR, because max over objects of
        (max over that object's exemplars) IS the max over all exemplars.
        So a single argmax replaces the per-object max-then-compare.
        """
        ids = []
        for v in feats:
            v = np.ascontiguousarray(v, np.float32)
            best, bs = -1, -1.0
            if self._m:
                s = self._M[:self._m] @ v
                k = int(s.argmax())
                best, bs = int(self._own[k]), float(s[k])
            if best >= 0 and bs >= self.match:
                ids.append(best)
                self.n[best] += 1
                E = self.ex[best]
                # keep a view only if it ADDS one; near-duplicates of a
                # stored exemplar teach the gallery nothing. Admit it
                # only if it agrees with EVERY held exemplar - min(),
                # not max() - so the acceptance region cannot sprawl.
                if (len(E) < self.max_ex and bs < 0.95
                        and float((E @ v).min()) >= self.admit):
                    self.ex[best] = np.vstack([E, v])
                    self._grow(len(v))
                    self._M[self._m] = v
                    self._own[self._m] = best
                    self._m += 1
            else:
                self.ex.append(v[None, :])
                self.n.append(1)
                best = len(self.ex) - 1
                ids.append(best)
                self._grow(len(v))
                self._M[self._m] = v
                self._own[self._m] = best
                self._m += 1
        return np.asarray(ids, np.int32)

    def centroids(self):
        return np.stack([E.mean(0) / (np.linalg.norm(E.mean(0)) + 1e-8)
                         for E in self.ex]) if self.ex else np.zeros(
                             (0, 512), np.float32)

    def __len__(self):
        return len(self.ex)