Skip to content

Python reference

The seams worth programming against, not every module. LabeloxAV has roughly 600 Python modules; auto-dumping all of them produces a reference nobody can navigate and that goes stale the moment an internal helper moves. What follows is the surface that is deliberately stable: the domain-pack contract, the engine's bridge to it, and the measurement primitives.

The domain pack contract

A pack is how one engine serves two domains. packs/av is autonomous driving; packs/sec is physical security. The engine never imports either: .importlinter forbids core, services and db from importing packs.av or packs.sec, and everything goes through the registry.

packs.base.DomainPack

Bases: Protocol

The contract a concrete pack satisfies. Attribute-only Protocol: any object exposing these fields is a DomainPack (runtime_checkable checks presence). scene_model / ingestion_adapters are Optional / empty until SEC-M2 wires the scene-model fork; every other surface is real in SEC-M1.

Source code in packs/base.py
@runtime_checkable
class DomainPack(Protocol):
    """The contract a concrete pack satisfies. Attribute-only Protocol: any object exposing these fields is a
    DomainPack (runtime_checkable checks presence). scene_model / ingestion_adapters are Optional / empty until
    SEC-M2 wires the scene-model fork; every other surface is real in SEC-M1.
    """
    manifest: PackManifest
    ontology: OntologySpec
    safety_policy: SafetyPolicy
    autolabel_profile: AutoLabelProfile
    eval_strata: EvalStrataSpec
    quality_profile: QualityProfile
    forge_targets: Sequence[ForgeTarget]
    privacy: PrivacyPlaneSpec
    relations: RelationSpec | None
    cliques: CliqueSpec | None
    class_tree: ClassTree | None
    motion_models: MotionModelSpec | None
    scene_model: SceneModelFactory | None
    ingestion_adapters: Sequence[IngestionAdapter]
    # Optional: the frame-level vocabulary a person may set on Frame.scene. Empty means the domain has
    # nothing to say about the scene as a whole.
    context: ContextSpec | None
    track_events: TrackEventSpec | None
    # Optional because they are genuinely domain-specific rather than merely unfinished: an AV pack has no
    # fixed zones to police and no camera to open. A pack that does not fill them is complete, and the engine
    # refuses the corresponding route rather than pretending the capability exists.
    zone_policy: ZonePolicy | None
    stream_source: StreamSource | None

packs.base.OntologySpec dataclass

The taxonomy plus the ontology facts the audit found in code rather than in the governed YAML.

yaml_path is the governed, versioned artifact (unchanged for AV). stuff_names / stuff_l0 externalise the panoptic split that was a frozenset in services/autolabel/ontology.py:31. supported_core externalises core/config.py:396. superclass_map / size_priors are consumed when the quality reviewer moves behind the pack in SEC-M5; they are None until then (honest incremental fill, not a stub).

Source code in packs/base.py
@dataclass(frozen=True)
class OntologySpec:
    """The taxonomy plus the ontology facts the audit found in code rather than in the governed YAML.

    yaml_path is the governed, versioned artifact (unchanged for AV). stuff_names / stuff_l0 externalise the
    panoptic split that was a frozenset in services/autolabel/ontology.py:31. supported_core externalises
    core/config.py:396. superclass_map / size_priors are consumed when the quality reviewer moves behind the
    pack in SEC-M5; they are None until then (honest incremental fill, not a stub).
    """
    yaml_path: str
    stuff_names: frozenset[str]
    stuff_l0: frozenset[str]
    supported_core: frozenset[str]
    custom_id_base: int
    superclass_map: Mapping[str, str] | None = None       # SEC-M5
    size_priors: Mapping[str, SizeBound] | None = None     # SEC-M5

packs.base.CliqueSpec dataclass

The confusion cliques of a domain, and the cost of confusing classes across them.

Held by the pack because which classes a model confuses is a fact about the world it works in: a scooter and a motorcycle are one decision on an Indian road and two unrelated objects in a warehouse.

Source code in packs/base.py
@dataclass(frozen=True)
class CliqueSpec:
    """The confusion cliques of a domain, and the cost of confusing classes across them.

    Held by the pack because which classes a model confuses is a fact about the world it works in: a
    scooter and a motorcycle are one decision on an Indian road and two unrelated objects in a warehouse.
    """
    cliques: tuple[ConfusionClique, ...]
    # Cost of confusing two classes that are not in a clique together. Higher than any within-clique cost
    # by construction: a confusion the ontology did not anticipate is worse than one it did.
    cross_clique_cost: float = 1.0

    def by_name(self, name: str) -> ConfusionClique | None:
        for c in self.cliques:
            if c.name == name:
                return c
        return None

    def clique_of(self, class_name: str) -> ConfusionClique | None:
        for c in self.cliques:
            if class_name in c.class_names:
                return c
        return None

    def pair_cost(self, a: str, b: str) -> float:
        """Cost of confusing a for b. Zero for a class with itself, the clique cost within one, else cross."""
        if a == b:
            return 0.0
        ca, cb = self.clique_of(a), self.clique_of(b)
        if ca is not None and ca is cb:
            return ca.cost
        return self.cross_clique_cost

pair_cost

pair_cost(a, b)

Cost of confusing a for b. Zero for a class with itself, the clique cost within one, else cross.

Source code in packs/base.py
def pair_cost(self, a: str, b: str) -> float:
    """Cost of confusing a for b. Zero for a class with itself, the clique cost within one, else cross."""
    if a == b:
        return 0.0
    ca, cb = self.clique_of(a), self.clique_of(b)
    if ca is not None and ca is cb:
        return ca.cost
    return self.cross_clique_cost

packs.base.RelationSpec dataclass

The relationship vocabulary, and which ordered class pairs can stand in which relation.

Two disjoint vocabularies are live in the engine today and neither knows about the other: services/api/routers/objects.py validates {rider_of, towed_by, part_of, member_of, occludes} on the editor path, while services/intelligence/scene_graph.py inserts {occluded_by, following, crossing_in_front_of, parked_near} directly and never passes that validation. occludes and occluded_by are the same fact in opposite directions and both are stored, so a query for one silently misses half the corpus.

kinds is the union any writer may use. overlap_pairs maps an ordered (subject l1, object l1) pair to the relation an overlapping pair of those superclasses probably stands in, which is what lets relationship-aware NMS keep a rider and their motorcycle apart and say why. inverse records the pairs that are one fact in two directions, so a reader can normalise rather than guess.

Source code in packs/base.py
@dataclass(frozen=True)
class RelationSpec:
    """The relationship vocabulary, and which ordered class pairs can stand in which relation.

    Two disjoint vocabularies are live in the engine today and neither knows about the other:
    services/api/routers/objects.py validates {rider_of, towed_by, part_of, member_of, occludes} on the
    editor path, while services/intelligence/scene_graph.py inserts {occluded_by, following,
    crossing_in_front_of, parked_near} directly and never passes that validation. `occludes` and
    `occluded_by` are the same fact in opposite directions and both are stored, so a query for one silently
    misses half the corpus.

    `kinds` is the union any writer may use. `overlap_pairs` maps an ordered (subject l1, object l1) pair
    to the relation an overlapping pair of those superclasses probably stands in, which is what lets
    relationship-aware NMS keep a rider and their motorcycle apart and say why. `inverse` records the pairs
    that are one fact in two directions, so a reader can normalise rather than guess.
    """
    kinds: frozenset[str]
    overlap_pairs: Mapping[tuple[str, str], str]
    inverse: Mapping[str, str] = field(default_factory=dict)

    def relation_for_l1(self, subject_l1: str, object_l1: str) -> str | None:
        return self.overlap_pairs.get((subject_l1, object_l1))

    def canonical(self, kind: str) -> str:
        """The direction this engine stores. An inverse maps to its canonical form; anything else is itself."""
        return self.inverse.get(kind, kind)

canonical

canonical(kind)

The direction this engine stores. An inverse maps to its canonical form; anything else is itself.

Source code in packs/base.py
def canonical(self, kind: str) -> str:
    """The direction this engine stores. An inverse maps to its canonical form; anything else is itself."""
    return self.inverse.get(kind, kind)

packs.base.TrackEventSpec dataclass

The track-event vocabulary of a domain.

Source code in packs/base.py
@dataclass(frozen=True)
class TrackEventSpec:
    """The track-event vocabulary of a domain."""

    types: tuple[TrackEventType, ...]

    def names(self) -> frozenset[str]:
        return frozenset(t.name for t in self.types)

    def get(self, name: str) -> TrackEventType | None:
        return next((t for t in self.types if t.name == name), None)

packs.base.ContextSpec dataclass

Frame-level facts a domain wants recorded that are not about any one object.

Kept separate from the object attribute schema because the question is different: an object attribute says what a thing is, a context attribute says what the whole scene was like, and the same frame carries one of each per axis. The engine already stores this on Frame.scene, which the ingest classifier fills with density, weather, road_type and time_of_day; this is the vocabulary a person may add to it.

A domain with nothing to say about the scene leaves it empty and the editor shows no context panel, rather than the engine inventing weather categories for a warehouse camera.

Source code in packs/base.py
@dataclass(frozen=True)
class ContextSpec:
    """Frame-level facts a domain wants recorded that are not about any one object.

    Kept separate from the object attribute schema because the question is different: an object attribute
    says what a thing is, a context attribute says what the whole scene was like, and the same frame carries
    one of each per axis. The engine already stores this on `Frame.scene`, which the ingest classifier fills
    with density, weather, road_type and time_of_day; this is the vocabulary a person may add to it.

    A domain with nothing to say about the scene leaves it empty and the editor shows no context panel,
    rather than the engine inventing weather categories for a warehouse camera.
    """
    attributes: Mapping[str, AttributeSpec] = field(default_factory=dict)

    def validate(self, attrs: Mapping[str, object]) -> list[str]:
        """Errors, empty when valid. Same shape as the ontology's attribute validation so a caller that
        already handles one handles the other."""
        errors: list[str] = []
        for key, val in attrs.items():
            spec = self.attributes.get(key)
            if spec is None:
                errors.append(f"unknown context attribute '{key}'")
                continue
            errors.extend(spec.errors_for(key, val))
        return errors

validate

validate(attrs)

Errors, empty when valid. Same shape as the ontology's attribute validation so a caller that already handles one handles the other.

Source code in packs/base.py
def validate(self, attrs: Mapping[str, object]) -> list[str]:
    """Errors, empty when valid. Same shape as the ontology's attribute validation so a caller that
    already handles one handles the other."""
    errors: list[str] = []
    for key, val in attrs.items():
        spec = self.attributes.get(key)
        if spec is None:
            errors.append(f"unknown context attribute '{key}'")
            continue
        errors.extend(spec.errors_for(key, val))
    return errors

packs.base.RegionSpec dataclass

How a domain turns a recorded place string into a stratum.

A corpus records places the way the capture rig happened to spell them, and the strings are not a taxonomy: this one says BLR on 372 sessions and Bengaluru on one, which is a single city that looks like two strata to anything counting coverage.

resolve maps a lowercased, punctuation-free string to (city, state, urban_class), or None when the string names nowhere this domain models. classes is the urban_class vocabulary, ordered from largest, so a report can render the strata in a meaningful order rather than alphabetically.

Source code in packs/base.py
@dataclass(frozen=True)
class RegionSpec:
    """How a domain turns a recorded place string into a stratum.

    A corpus records places the way the capture rig happened to spell them, and the strings are not a
    taxonomy: this one says `BLR` on 372 sessions and `Bengaluru` on one, which is a single city that looks
    like two strata to anything counting coverage.

    `resolve` maps a lowercased, punctuation-free string to (city, state, urban_class), or None when the
    string names nowhere this domain models. `classes` is the urban_class vocabulary, ordered from largest,
    so a report can render the strata in a meaningful order rather than alphabetically.
    """

    resolve: Callable[[str], tuple[str, str, str] | None]
    classes: tuple[str, ...]
    # Every string `resolve` accepts. Published so a filter can be built by inverting the same table the
    # resolver reads, rather than by adding a denormalised region column that could drift from it.
    keys: Callable[[], frozenset[str]]
    # Places the domain knows are outside its region model, so a resolver can say "outside India" rather
    # than "unknown". The difference matters: unknown is a data-quality problem, outside is a fact.
    outside: Callable[[str], str | None] = lambda _s: None

The engine's bridge

Every per-domain question the engine asks goes through here, so a second domain behaves differently at runtime rather than only existing in the registry.

services.domain

Engine-facing access to the active domain pack.

The single seam the engine uses to consult the pack it belongs to, without statically importing any concrete pack (it goes through packs.registry, the sanctioned bridge). Governance, curation, and eval call these instead of the {"vru","animal"} literals and hardcoded strata the audit found copy-pasted across the tree.

Every helper takes an optional pack_id and defaults to the configured active pack ('av'), so the legacy paths resolve to the AV pack and stay byte-identical. Per-session routing is pack_for_session: a session records the pack it was captured under (Session.pack_id), and passing that id here is what makes the second domain behave differently at runtime rather than only existing in the registry.

active_pack

active_pack(pack_id=None)
Source code in services/domain.py
def active_pack(pack_id: str | None = None) -> DomainPack:
    return get_pack(pack_id or default_pack_id())

pack_for_session async

pack_for_session(db, session_id)

The DomainPack a session belongs to. The routing seam every per-session capability check goes through.

Source code in services/domain.py
async def pack_for_session(db, session_id) -> DomainPack:
    """The DomainPack a session belongs to. The routing seam every per-session capability check goes through."""
    return get_pack(await pack_id_for_session(db, session_id))

safety_l1

safety_l1(pack_id=None)

The l1 superclasses that define a safety-critical class for the active pack (AV: {'vru','animal'}).

Source code in services/domain.py
def safety_l1(pack_id: str | None = None) -> frozenset[str]:
    """The l1 superclasses that define a safety-critical class for the active pack (AV: {'vru','animal'})."""
    return active_pack(pack_id).autolabel_profile.gate_policy.safety_l1

critical_class_names

critical_class_names(pack_id=None)
Source code in services/domain.py
def critical_class_names(pack_id: str | None = None) -> frozenset[str]:
    return active_pack(pack_id).safety_policy.critical_class_names()

critical_class_ids

critical_class_ids(onto, pack_id=None)

The active pack's safety-critical classes resolved to ids against onto. Replaces the hardcoded (and 0-based-buggy) verdyx CRITICAL_CLASSES with names resolved through the ontology.

Source code in services/domain.py
def critical_class_ids(onto, pack_id: str | None = None) -> set[int]:
    """The active pack's safety-critical classes resolved to ids against `onto`. Replaces the hardcoded
    (and 0-based-buggy) verdyx CRITICAL_CLASSES with names resolved through the ontology."""
    return {onto.by_name(n).id for n in critical_class_names(pack_id) if onto.has_name(n)}

context_spec

context_spec(pack_id=None)

The active pack's frame-level context vocabulary, or None when the domain has nothing to say about the scene as a whole. A static-camera pack counting entries is complete without one, and the editor offers no context panel rather than the engine inventing weather categories for a warehouse.

Source code in services/domain.py
def context_spec(pack_id: str | None = None):
    """The active pack's frame-level context vocabulary, or None when the domain has nothing to say about
    the scene as a whole. A static-camera pack counting entries is complete without one, and the editor
    offers no context panel rather than the engine inventing weather categories for a warehouse."""
    return active_pack(pack_id).context

validate_context

validate_context(attrs, pack_id=None)

Errors, empty when valid. A pack with no context spec rejects everything: writing a frame-level fact into a domain that declares none is a caller bug, not an empty vocabulary to be filled in silently.

Source code in services/domain.py
def validate_context(attrs: dict, pack_id: str | None = None) -> list[str]:
    """Errors, empty when valid. A pack with no context spec rejects everything: writing a frame-level fact
    into a domain that declares none is a caller bug, not an empty vocabulary to be filled in silently."""
    spec = context_spec(pack_id)
    if spec is None:
        return ["this domain declares no frame context vocabulary"]
    return spec.validate(attrs)

track_event_spec

track_event_spec(pack_id=None)

The pack's track-event vocabulary, or None when the domain has no behaviour worth spanning.

Source code in services/domain.py
def track_event_spec(pack_id: str | None = None):
    """The pack's track-event vocabulary, or None when the domain has no behaviour worth spanning."""
    return active_pack(pack_id).track_events

validate_track_event_type

validate_track_event_type(
    event_type, class_l1, pack_id=None
)

Empty when this pack admits this event type on a track of this superclass, else the reasons.

Applicability is checked here and not only in the picker, because the picker is not the only writer: the proposers and any importer come through the same door, and a lane_splitting event on a pedestrian track is a wrong statement whichever of them made it.

Source code in services/domain.py
def validate_track_event_type(event_type: str, class_l1: str | None, pack_id: str | None = None) -> list[str]:
    """Empty when this pack admits this event type on a track of this superclass, else the reasons.

    Applicability is checked here and not only in the picker, because the picker is not the only writer: the
    proposers and any importer come through the same door, and a `lane_splitting` event on a pedestrian
    track is a wrong statement whichever of them made it.
    """
    spec = active_pack(pack_id).track_events
    if spec is None:
        return [f"pack '{pack_id or 'default'}' declares no track events"]
    t = spec.get(event_type)
    if t is None:
        return [f"unknown event type '{event_type}'"]
    if t.applies_to == "any" or class_l1 is None:
        return []
    kinds = {"vehicle": {"two_wheeler", "three_wheeler", "four_wheeler", "heavy"}, "vru": {"vru"}}
    if class_l1 not in kinds.get(t.applies_to, set()):
        return [f"event '{event_type}' applies to {t.applies_to} tracks, not '{class_l1}'"]
    return []

class_aliases

class_aliases(class_id, pack_id=None)

Every word that names this class, the display name first. Never empty.

Declared in the pack's own ontology YAML, which OntologySpec.yaml_path is the pack's statement of where its class tree lives, and read back through the loader that already parses it. Not a second mapping on the Pack dataclass: two lists of synonyms for one class tree is how the synonyms drift.

Source code in services/domain.py
def class_aliases(class_id: int, pack_id: str | None = None) -> list[str]:
    """Every word that names this class, the display name first. Never empty.

    Declared in the pack's own ontology YAML, which `OntologySpec.yaml_path` is the pack's statement of
    where its class tree lives, and read back through the loader that already parses it. Not a second
    mapping on the Pack dataclass: two lists of synonyms for one class tree is how the synonyms drift.
    """
    from services.autolabel.ontology import get_ontology

    return get_ontology(pack_id).aliases_for(class_id)

Ontology

services.autolabel.ontology.Ontology dataclass

Source code in services/autolabel/ontology.py
@dataclass
class Ontology:
    version: str
    hierarchy_levels: int
    classes: list[OntologyClassDef]
    attributes: dict[str, AttributeDef] = field(default_factory=dict)
    # Per-subclass (l1) applicable-attribute allowlist. A subclass absent here means all attributes apply.
    attribute_scope: dict[str, list[str]] = field(default_factory=dict)
    # Per-class extras, keyed by class NAME, unioned onto whatever the l1 scope allows.
    #
    # l1 is too coarse for the attributes that matter here. `heavy` holds bus, school_bus, truck, tractor,
    # tipper, ambulance, fire_truck, bullock_cart and harvester together, so a footboard-passenger attribute
    # scoped at l1 is offered on every truck, and an attribute offered is an attribute somebody sets. This
    # layer is additive: no class changes l1, no existing scope entry moves, and a class absent from it
    # behaves exactly as before.
    attribute_scope_class: dict[str, list[str]] = field(default_factory=dict)

    _by_id: dict[int, OntologyClassDef] = field(default_factory=dict, repr=False)
    _by_name: dict[str, OntologyClassDef] = field(default_factory=dict, repr=False)

    def __post_init__(self) -> None:
        self._by_id = {c.id: c for c in self.classes}
        self._by_name = {c.name: c for c in self.classes}

    def by_id(self, class_id: int) -> OntologyClassDef:
        if class_id not in self._by_id:
            raise KeyError(f"class_id {class_id} not in ontology {self.version}")
        return self._by_id[class_id]

    def by_name(self, name: str) -> OntologyClassDef:
        if name not in self._by_name:
            raise KeyError(f"class name '{name}' not in ontology {self.version}")
        return self._by_name[name]

    def has_name(self, name: str) -> bool:
        return name in self._by_name

    def attrs_for_class(self, class_id: int) -> list[str] | None:
        """Attribute names applicable to a class. None means all attributes apply.

        The l1 scope, plus any per-class extras. Union rather than override: a city_bus is still a heavy
        vehicle and still wants the heavy attributes, it just also wants two of its own.
        """
        try:
            c = self.by_id(class_id)
        except KeyError:
            return None
        base = self.attribute_scope.get(c.l1)
        extra = self.attribute_scope_class.get(c.name)
        if base is None:
            # The l1 is unscoped, which already means "everything applies", so extras add nothing.
            return None
        return base if not extra else [*base, *(a for a in extra if a not in base)]

    def derive_attrs(self, attrs: dict, class_id: int | None = None) -> dict:
        """Return `attrs` with every derived attribute recomputed from its source.

        Called after the merge on every write path, so the derived value is a fact about the stored attrs
        rather than a second thing to keep in step. A derived key whose source is absent or out of scope is
        removed: leaving a stale `triple_riding: true` behind after somebody corrected the occupant count to
        one is worse than not having the attribute.
        """
        out = dict(attrs)
        scope = self.attrs_for_class(class_id) if class_id is not None else None
        for name, spec in self.attributes.items():
            if not spec.derived_from:
                continue
            if scope is not None and name not in scope:
                out.pop(name, None)
                continue
            src = out.get(spec.derived_from)
            if src is None:
                out.pop(name, None)
                continue
            fn = _DERIVERS.get(name)
            if fn is None:
                # Declared in the YAML with no implementation here. Refusing to guess: an attribute that
                # silently derives nothing is a field consumers will read and trust.
                raise ValueError(f"attribute '{name}' is declared derived but has no deriver")
            out[name] = fn(src)
        return out

    def aliases_for(self, class_id: int) -> list[str]:
        """What else this class is called, the display name first. Never empty."""
        c = self.by_id(class_id)
        return [c.name, *c.aliases]

    def concept_phrases(self, india_first: bool = True) -> list[str]:
        """Ontology names as open-vocab prompts for SAM 3.1 PCS. India/rare classes first."""
        ordered = sorted(self.classes, key=lambda c: (not c.india, c.id)) if india_first else self.classes
        return [c.name.replace("_", " ") for c in ordered]

    def fallback_ids(self) -> list[int]:
        return [c.id for c in self.classes if c.l1 == "fallback"]

    def is_fallback(self, class_id: int) -> bool:
        return self.by_id(class_id).l1 == "fallback"

    def is_stuff(self, class_id: int) -> bool:
        """True if the class is background stuff (semantic-seg only, never an instance box): any surface or
        ignore-region, plus the curated uncountable structures/vegetation/barriers in STUFF_NAMES."""
        c = self.by_id(class_id)
        return c.l0 in STUFF_L0 or c.name in STUFF_NAMES

    def is_thing(self, class_id: int) -> bool:
        """True if the class is a countable foreground object that legitimately gets one instance box."""
        return not self.is_stuff(class_id)

    def validate_attrs(self, attrs: dict, class_id: int | None = None) -> list[str]:
        """Return a list of validation errors; empty means valid. When class_id is given and its subclass
        declares an attribute scope, an attribute not in that scope is an error (not applicable to class)."""
        allowed = self.attrs_for_class(class_id) if class_id is not None else None
        errors: list[str] = []
        for key, val in attrs.items():
            if key not in self.attributes:
                errors.append(f"unknown attribute '{key}'")
                continue
            if allowed is not None and key not in allowed:
                errors.append(f"attribute '{key}' not applicable to class {class_id}")
                continue
            spec = self.attributes[key]
            if spec.derived_from:
                errors.append(f"attribute '{key}' is computed from '{spec.derived_from}' and cannot be set")
                continue
            if spec.type == "enum":
                if val not in (spec.values or []):
                    errors.append(f"attribute '{key}'={val!r} not in {spec.values}")
            elif spec.type == "float":
                if not isinstance(val, (int, float)):
                    errors.append(f"attribute '{key}' must be float")
                elif spec.range and not (spec.range[0] <= float(val) <= spec.range[1]):
                    errors.append(f"attribute '{key}'={val} out of range {spec.range}")
            elif spec.type == "int":
                if not isinstance(val, int) or isinstance(val, bool):
                    errors.append(f"attribute '{key}' must be int")
                elif spec.range and not (spec.range[0] <= val <= spec.range[1]):
                    # The float branch has always checked this and the int branch never did, so
                    # `occlusion_pct: 400` and `passenger_load: -3` both validated.
                    errors.append(f"attribute '{key}'={val} out of range {spec.range}")
            elif spec.type == "bool":
                if not isinstance(val, bool):
                    errors.append(f"attribute '{key}' must be bool")
            elif spec.type == "bool_array":
                if not (isinstance(val, list) and all(isinstance(x, bool) for x in val)):
                    errors.append(f"attribute '{key}' must be a bool array")
            elif spec.type == "multi_select":
                # A set of values from the vocabulary, not one. `script` is the case it exists for: a
                # Bengaluru signboard routinely carries Kannada and English together, and forcing one
                # records the wrong half.
                if not isinstance(val, list):
                    errors.append(f"attribute '{key}' must be a list")
                elif bad := [v for v in val if v not in (spec.values or [])]:
                    errors.append(f"attribute '{key}' has values not in {spec.values}: {bad}")
                elif len(set(val)) != len(val):
                    errors.append(f"attribute '{key}' has duplicate values")
            else:
                # An unimplemented type used to fall through here and accept anything at all, so adding a
                # type to the YAML silently disabled validation for every attribute using it.
                errors.append(f"attribute '{key}' has unsupported type '{spec.type}'")
        return errors

by_id

by_id(class_id)
Source code in services/autolabel/ontology.py
def by_id(self, class_id: int) -> OntologyClassDef:
    if class_id not in self._by_id:
        raise KeyError(f"class_id {class_id} not in ontology {self.version}")
    return self._by_id[class_id]

by_name

by_name(name)
Source code in services/autolabel/ontology.py
def by_name(self, name: str) -> OntologyClassDef:
    if name not in self._by_name:
        raise KeyError(f"class name '{name}' not in ontology {self.version}")
    return self._by_name[name]

has_name

has_name(name)
Source code in services/autolabel/ontology.py
def has_name(self, name: str) -> bool:
    return name in self._by_name

attrs_for_class

attrs_for_class(class_id)

Attribute names applicable to a class. None means all attributes apply.

The l1 scope, plus any per-class extras. Union rather than override: a city_bus is still a heavy vehicle and still wants the heavy attributes, it just also wants two of its own.

Source code in services/autolabel/ontology.py
def attrs_for_class(self, class_id: int) -> list[str] | None:
    """Attribute names applicable to a class. None means all attributes apply.

    The l1 scope, plus any per-class extras. Union rather than override: a city_bus is still a heavy
    vehicle and still wants the heavy attributes, it just also wants two of its own.
    """
    try:
        c = self.by_id(class_id)
    except KeyError:
        return None
    base = self.attribute_scope.get(c.l1)
    extra = self.attribute_scope_class.get(c.name)
    if base is None:
        # The l1 is unscoped, which already means "everything applies", so extras add nothing.
        return None
    return base if not extra else [*base, *(a for a in extra if a not in base)]

aliases_for

aliases_for(class_id)

What else this class is called, the display name first. Never empty.

Source code in services/autolabel/ontology.py
def aliases_for(self, class_id: int) -> list[str]:
    """What else this class is called, the display name first. Never empty."""
    c = self.by_id(class_id)
    return [c.name, *c.aliases]

validate_attrs

validate_attrs(attrs, class_id=None)

Return a list of validation errors; empty means valid. When class_id is given and its subclass declares an attribute scope, an attribute not in that scope is an error (not applicable to class).

Source code in services/autolabel/ontology.py
def validate_attrs(self, attrs: dict, class_id: int | None = None) -> list[str]:
    """Return a list of validation errors; empty means valid. When class_id is given and its subclass
    declares an attribute scope, an attribute not in that scope is an error (not applicable to class)."""
    allowed = self.attrs_for_class(class_id) if class_id is not None else None
    errors: list[str] = []
    for key, val in attrs.items():
        if key not in self.attributes:
            errors.append(f"unknown attribute '{key}'")
            continue
        if allowed is not None and key not in allowed:
            errors.append(f"attribute '{key}' not applicable to class {class_id}")
            continue
        spec = self.attributes[key]
        if spec.derived_from:
            errors.append(f"attribute '{key}' is computed from '{spec.derived_from}' and cannot be set")
            continue
        if spec.type == "enum":
            if val not in (spec.values or []):
                errors.append(f"attribute '{key}'={val!r} not in {spec.values}")
        elif spec.type == "float":
            if not isinstance(val, (int, float)):
                errors.append(f"attribute '{key}' must be float")
            elif spec.range and not (spec.range[0] <= float(val) <= spec.range[1]):
                errors.append(f"attribute '{key}'={val} out of range {spec.range}")
        elif spec.type == "int":
            if not isinstance(val, int) or isinstance(val, bool):
                errors.append(f"attribute '{key}' must be int")
            elif spec.range and not (spec.range[0] <= val <= spec.range[1]):
                # The float branch has always checked this and the int branch never did, so
                # `occlusion_pct: 400` and `passenger_load: -3` both validated.
                errors.append(f"attribute '{key}'={val} out of range {spec.range}")
        elif spec.type == "bool":
            if not isinstance(val, bool):
                errors.append(f"attribute '{key}' must be bool")
        elif spec.type == "bool_array":
            if not (isinstance(val, list) and all(isinstance(x, bool) for x in val)):
                errors.append(f"attribute '{key}' must be a bool array")
        elif spec.type == "multi_select":
            # A set of values from the vocabulary, not one. `script` is the case it exists for: a
            # Bengaluru signboard routinely carries Kannada and English together, and forcing one
            # records the wrong half.
            if not isinstance(val, list):
                errors.append(f"attribute '{key}' must be a list")
            elif bad := [v for v in val if v not in (spec.values or [])]:
                errors.append(f"attribute '{key}' has values not in {spec.values}: {bad}")
            elif len(set(val)) != len(val):
                errors.append(f"attribute '{key}' has duplicate values")
        else:
            # An unimplemented type used to fall through here and accept anything at all, so adding a
            # type to the YAML silently disabled validation for every attribute using it.
            errors.append(f"attribute '{key}' has unsupported type '{spec.type}'")
    return errors

derive_attrs

derive_attrs(attrs, class_id=None)

Return attrs with every derived attribute recomputed from its source.

Called after the merge on every write path, so the derived value is a fact about the stored attrs rather than a second thing to keep in step. A derived key whose source is absent or out of scope is removed: leaving a stale triple_riding: true behind after somebody corrected the occupant count to one is worse than not having the attribute.

Source code in services/autolabel/ontology.py
def derive_attrs(self, attrs: dict, class_id: int | None = None) -> dict:
    """Return `attrs` with every derived attribute recomputed from its source.

    Called after the merge on every write path, so the derived value is a fact about the stored attrs
    rather than a second thing to keep in step. A derived key whose source is absent or out of scope is
    removed: leaving a stale `triple_riding: true` behind after somebody corrected the occupant count to
    one is worse than not having the attribute.
    """
    out = dict(attrs)
    scope = self.attrs_for_class(class_id) if class_id is not None else None
    for name, spec in self.attributes.items():
        if not spec.derived_from:
            continue
        if scope is not None and name not in scope:
            out.pop(name, None)
            continue
        src = out.get(spec.derived_from)
        if src is None:
            out.pop(name, None)
            continue
        fn = _DERIVERS.get(name)
        if fn is None:
            # Declared in the YAML with no implementation here. Refusing to guess: an attribute that
            # silently derives nothing is a field consumers will read and trust.
            raise ValueError(f"attribute '{name}' is declared derived but has no deriver")
        out[name] = fn(src)
    return out

The confidence gate

Where humans enter. Read the module docstring before quoting any threshold: a threshold is a precision floor only for a class with a fitted operating point, and the configured constants are not that.

services.autolabel.gate

The confidence gate: where humans enter (Principle 04). Calibrated confidence routes each object to auto_accept, review, or annotate.

What a threshold here means, stated once so this file cannot argue with itself. A threshold is a precision floor ONLY for a class with a fitted, measured operating point (ThresholdFit). For every other class the gate falls back to the configured constants - auto_accept: 0.45, safety_auto_accept: 0.47 on a calibrated scale whose ceiling sits near 0.48 - and those are unmeasured constants, not floors. _threshold_for logs exactly that on first use per class. An earlier version of this docstring quoted 0.99/0.95 as the operating point; those numbers never described the running config, and the realized precision of the auto-accepted subset has not been measured against human verdicts. (A VLM judge puts the pooled auto_accept subset at 0.932 strict on 44 decided crops - evidence, not a measurement against humans.)

M-Q.4 hardening: - Per-class calibrated thresholds replace the global constant where a fit exists; safety-critical classes (VRU, animal) use the higher safety constant where it does not. - A rare/fallback class earns auto-accept only with cross-path agreement AND VLM confirmation, never on one model's output. This kills confident-but-wrong rare detections. - The quality reviewer's verdict (geometric/contextual nonsense) demotes an object before it can auto-accept, regardless of score.

gate_object

gate_object(
    obj,
    onto,
    cfg,
    auto_accept_enabled=True,
    quality_ok=True,
    fitted=None,
    joint=None,
    tube=None,
)

Route one object to auto_accept, review or annotate.

joint is a fitted services/oraclyx/joint_calibration.py surface and tube this object's temporal coherence. When both are present the threshold is applied to P(correct | conf, tube) rather than to the raw confidence, which is the point of fitting the surface: a 0.55 detection that is the twentieth frame of a stable track and a 0.55 detection that appears once are not equally likely to be right, and a threshold on confidence alone cannot separate them.

Absent either, the score stays the confidence. Substituting the surface's no-tube fallback silently would change what every existing caller gates on, so the joint path is opt-in at the call site.

Source code in services/autolabel/gate.py
def gate_object(obj: UnifiedObject, onto: Ontology, cfg: GateSettings,
                auto_accept_enabled: bool = True, quality_ok: bool = True,
                fitted: Mapping[int, float] | None = None,
                joint: Any = None, tube: float | None = None) -> GateState:
    """Route one object to auto_accept, review or annotate.

    `joint` is a fitted services/oraclyx/joint_calibration.py surface and `tube` this object's temporal
    coherence. When both are present the threshold is applied to P(correct | conf, tube) rather than to
    the raw confidence, which is the point of fitting the surface: a 0.55 detection that is the twentieth
    frame of a stable track and a 0.55 detection that appears once are not equally likely to be right, and
    a threshold on confidence alone cannot separate them.

    Absent either, the score stays the confidence. Substituting the surface's no-tube fallback silently
    would change what every existing caller gates on, so the joint path is opt-in at the call site.
    """
    conf = obj.conf
    if joint is not None and tube is not None:
        conf = float(joint(obj.conf, tube))
    prov = obj.provenance
    rare = is_rare(obj.class_id, onto)

    # Below the review floor is always a full annotate, whatever else is true.
    if conf < cfg.review_low:
        return GateState.annotate

    # The quality reviewer demoted geometric/contextual nonsense (sky box, impossible size, tyre-as-vehicle,
    # duplicate, pedestrian-in-car). It never auto-accepts; a human confirms or kills it.
    if not quality_ok:
        return GateState.review

    if cfg.force_review_on_mask_box_disagree and prov.mask_box_disagree:
        return GateState.review

    # Two independent segmenters produced materially different masks for this object. Measured on this
    # corpus that happens to about a fifth of objects and concentrates on riders, motorcycles and
    # autorickshaws, where the mask boundary is genuinely ambiguous - which is exactly the clique the pack
    # marks as crossing a safety boundary. `None` means no verifier ran and is not a disagreement.
    if prov.mask_agreement is not None and prov.mask_agreement < cfg.mask_agree_min:
        return GateState.review

    # Strict escape hatch: when set, a rare/fallback class never auto-accepts, whatever else is true. Off by
    # default because M-Q.4's agreement+VLM rule below is the smarter policy; flip on to fully freeze the
    # long tail (e.g. a fresh ontology before any rare class has earned trust).
    if cfg.force_review_on_rare and rare:
        return GateState.review

    # auto_accept_enabled is the governance kill switch: when the loop is paused, nothing auto-accepts.
    if not auto_accept_enabled:
        return GateState.review

    # Per-class calibrated threshold plus cross-path agreement are the baseline for any auto-accept.
    if conf < class_auto_accept(obj.class_id, onto, cfg, fitted) or not prov.agreement:
        return GateState.review

    # A rare/fallback class must also be VLM-confirmed: agreement alone is not enough for the long tail.
    if rare and cfg.rare_needs_agreement_and_vlm and not vlm_confirmed(prov):
        return GateState.review

    return GateState.auto_accept

needs_vlm

needs_vlm(obj, onto, cfg, quality_ok=True, fitted=None)

Path C (VLM) duty-cycle predicate. True only for the uncertain subset: paths disagree, confidence in the (per-class) review band, a rare/fallback class, a mask conflict, or a quality-flagged object that a second look should confirm or kill. Never the full stream.

Source code in services/autolabel/gate.py
def needs_vlm(obj: UnifiedObject, onto: Ontology, cfg: GateSettings, quality_ok: bool = True,
              fitted: Mapping[int, float] | None = None) -> bool:
    """Path C (VLM) duty-cycle predicate. True only for the uncertain subset: paths disagree, confidence in
    the (per-class) review band, a rare/fallback class, a mask conflict, or a quality-flagged object that a
    second look should confirm or kill. Never the full stream."""
    prov = obj.provenance
    class_disagree = any(p.verdict == "overruled" for p in prov.proposals) and len(prov.proposals) > 1
    in_review_band = cfg.review_low <= obj.conf < class_auto_accept(obj.class_id, onto, cfg, fitted)
    mask_disagree = (prov.mask_agreement is not None and prov.mask_agreement < cfg.mask_agree_min)
    return bool(class_disagree or in_review_band or is_rare(obj.class_id, onto)
                or prov.mask_box_disagree or mask_disagree or not quality_ok)

vlm_confirmed

vlm_confirmed(prov)

The VLM saw this object and confirmed (did not overrule) its class.

Source code in services/autolabel/gate.py
def vlm_confirmed(prov: Provenance) -> bool:
    """The VLM saw this object and confirmed (did not overrule) its class."""
    return any(p.path == "path_c_qwen3vl" and p.verdict in ("confirm", "agree") for p in prov.proposals)

Measurement

The judge, its calibration, and the correction that makes a machine-derived precision quotable.

services.labelops.vlm_review

A VLM judging existing labels in bulk, and an honest account of how much that judgement is worth.

The corpus has 253 human verdicts across 570,379 objects. Precision is unmeasurable, the gate cannot be tuned against anything, and the error detector holds 298,529 candidates with one confirmed verdict. Human review does not reach that scale and will not; a machine judge does.

The temptation is to run the judge and report its agreement rate as precision. That is wrong in a way that is hard to see later, because the number looks like a measurement. A judge has its own error rate, so its agreement rate is a blend of how good the labels are and how good the judge is, and nothing downstream can separate them again.

So this module does three things, and the third is the one that matters:

  1. prereview_batch judges every object in a batch and records what it said, in machine_verdict, which is deliberately not the human review table.
  2. judge_agreement compares the judge against humans wherever both have ruled on the same object, which gives the judge's sensitivity and specificity.
  3. judged_precision reports the raw agreement rate AND the rate corrected for the judge's measured error, and refuses to correct when nobody has adjudicated enough to measure the judge.

The asking is deliberately not the autolabel prompt. Path C asks "what is this?", offering a shortlist and taking the answer as a proposal. A judge is asked "the label says X, is that right?", which is a different question with a different failure mode: a model asked to name something will always name something, while a model asked to confirm can decline. unsure is kept as a first-class answer for exactly that reason, and is never folded into either side, because a judge that abstains on the hard crops and is scored only on the easy ones reports a precision that flatters itself.

build_judge_prompt

build_judge_prompt(given_class, alternatives)

Ask whether the existing label is right, rather than asking what the object is.

The difference is not cosmetic. A model asked to name an object always names one, so its answer carries no signal about whether it was sure. Asked to confirm a specific claim it can decline, and the declines are exactly the crops worth a person's time.

The alternatives are offered so that "incorrect" can come with a proposal, which turns a rejected label into a correction instead of just a deletion.

Source code in services/labelops/vlm_review.py
def build_judge_prompt(given_class: str, alternatives: list[str]) -> str:
    """Ask whether the existing label is right, rather than asking what the object is.

    The difference is not cosmetic. A model asked to name an object always names one, so its answer carries
    no signal about whether it was sure. Asked to confirm a specific claim it can decline, and the declines
    are exactly the crops worth a person's time.

    The alternatives are offered so that "incorrect" can come with a proposal, which turns a rejected label
    into a correction instead of just a deletion.
    """
    from services.domain import active_pack

    preamble = active_pack().autolabel_profile.vlm_prompt_template
    return (
        f"{preamble}\n"
        f"This crop has been labelled: {given_class}.\n"
        "Decide whether that label is correct for the main object in the crop.\n"
        f"If it is wrong, name the correct class from this list where possible: {alternatives}.\n"
        "Answer 'unsure' when the crop is too small, blurred, occluded or ambiguous to judge. Do not guess: "
        "an unsure answer is more useful than a wrong confident one.\n"
        'Respond with strict JSON only, no prose: '
        '{"verdict": "correct"|"incorrect"|"unsure", "correct_class": "<class or null>", '
        '"confidence": <0.0-1.0>, "reason": "<short>"}'
    )

parse_judge_reply

parse_judge_reply(data, onto, given_class=None)

Normalise a judge reply, refusing anything that is not one of the three verdicts.

A reply that does not parse becomes unsure rather than being dropped. Dropping it would silently shrink the denominator, which biases the rate upward: the crops a judge garbles are not a random subset.

A rejection that names the asked class as its own correction is not a rejection. It happens most on object_fallback, where "is this the right label?" is a confusing question about a class that means none of the above, and the model answers incorrect while its stated reason confirms the label ("the object is a fire hydrant, which is not in the provided class list"). Counting that as an error would report the fallback class as broken on exactly the crops it is working. unsure is the designed answer for a reply carrying no usable opinion, and this is one.

Source code in services/labelops/vlm_review.py
def parse_judge_reply(data: dict, onto, given_class: str | None = None) -> dict:
    """Normalise a judge reply, refusing anything that is not one of the three verdicts.

    A reply that does not parse becomes `unsure` rather than being dropped. Dropping it would silently
    shrink the denominator, which biases the rate upward: the crops a judge garbles are not a random subset.

    A rejection that names the asked class as its own correction is not a rejection. It happens most on
    `object_fallback`, where "is this the right label?" is a confusing question about a class that means
    none of the above, and the model answers `incorrect` while its stated reason confirms the label
    ("the object is a fire hydrant, which is not in the provided class list"). Counting that as an error
    would report the fallback class as broken on exactly the crops it is working. `unsure` is the designed
    answer for a reply carrying no usable opinion, and this is one.
    """
    verdict = str(data.get("verdict", "")).strip().lower()
    if verdict not in VERDICTS:
        verdict = "unsure"

    proposed = data.get("correct_class")
    proposed_id = None
    if verdict == "incorrect" and proposed and onto.has_name(str(proposed)):
        if given_class and str(proposed).strip() == given_class:
            verdict, proposed_id = "unsure", None
        else:
            proposed_id = onto.by_name(str(proposed)).id

    try:
        conf = float(data.get("confidence"))
        conf = max(0.0, min(1.0, conf))
    except (TypeError, ValueError):
        conf = None

    return {"verdict": verdict, "proposed_class_id": proposed_id,
            "confidence": conf, "reason": str(data.get("reason", ""))[:400]}

judge_objects async

judge_objects(
    db,
    objects,
    batch_id,
    *,
    client=None,
    model_version=None,
    skip_judged=True,
)

Judge a supplied list of objects and record what the judge said.

The loop, separated from how the objects were chosen. prereview_batch picks a flywheel batch; services/labelops/class_precision.py picks a random sample of one class. Both want the same judging, the same idempotent upsert and the same chunked commits, and a second copy of this loop is a second place for the prompt, the abstention handling and the uniqueness key to drift.

Source code in services/labelops/vlm_review.py
async def judge_objects(db: AsyncSession, objects: list[Object], batch_id: str, *,
                        client=None, model_version: str | None = None,
                        skip_judged: bool = True) -> dict:
    """Judge a supplied list of objects and record what the judge said.

    The loop, separated from how the objects were chosen. `prereview_batch` picks a flywheel batch;
    `services/labelops/class_precision.py` picks a random sample of one class. Both want the same judging,
    the same idempotent upsert and the same chunked commits, and a second copy of this loop is a second
    place for the prompt, the abstention handling and the uniqueness key to drift.
    """
    from services.autolabel.ontology import get_ontology
    from services.llm.router import make_vlm_client

    settings = get_settings()
    onto = get_ontology()
    client = client or make_vlm_client(settings)
    provider = getattr(settings.models.vlm, "vision_provider", "ollama")
    model_version = model_version or _model_version_for(settings, provider)

    already: set[_uuid.UUID] = set()
    if skip_judged and objects:
        # Scoped to this batch, matching the uniqueness key. Without the batch filter an object judged in
        # some other batch by the same model reads as already done, and this batch quietly ends up with a
        # hole in it that nothing reports.
        already = set((await db.execute(
            select(MachineVerdict.object_id).where(
                MachineVerdict.judge == JUDGE,
                MachineVerdict.model_version == model_version,
                MachineVerdict.batch_id == batch_id,
                MachineVerdict.object_id.in_([o.object_id for o in objects])))).scalars().all())

    counts = dict.fromkeys(VERDICTS, 0)
    judged = skipped = unreadable = failed = 0
    consecutive_failures = 0
    margin = settings.models.vlm.crop_margin

    # Committed in chunks rather than once at the end. A 300-crop batch against a local model is tens of
    # minutes and against a metered API is real money, and a single commit at the end means no progress is
    # visible while it runs and every verdict is lost if it dies partway. Chunked, a re-run skips what
    # already landed, so an interrupted job resumes instead of restarting.
    commit_every = 10

    for obj in objects:
        if obj.object_id in already:
            skipped += 1
            continue
        crop = await _load_crop(db, obj, margin)
        if crop is None or crop.size == 0:
            unreadable += 1
            continue

        given = onto.by_id(int(obj.class_id)).name
        alts = _alternatives(onto, int(obj.class_id))
        reply = _ask(client, crop, given, alts, model=model_version)
        if reply is None:
            # Counted and not written, the same as an unreadable crop. Writing it would put a verdict in the
            # table that the judge never gave.
            failed += 1
            consecutive_failures += 1
            if consecutive_failures >= MAX_CONSECUTIVE_FAILURES:
                await db.commit()
                raise RuntimeError(
                    f"{MAX_CONSECUTIVE_FAILURES} consecutive judge calls failed on batch {batch_id}; "
                    f"stopping rather than judging the rest of the corpus against a broken judge")
            continue
        consecutive_failures = 0
        parsed = parse_judge_reply(reply, onto, given_class=given)
        counts[parsed["verdict"]] += 1
        judged += 1

        # Upsert on the natural key so a re-run is idempotent rather than additive.
        await db.execute(pg_insert(MachineVerdict).values(
            object_id=obj.object_id, judge=JUDGE, provider=provider, model_version=model_version,
            verdict=parsed["verdict"], proposed_class_id=parsed["proposed_class_id"],
            confidence=parsed["confidence"], agreement=None,
            detail={"given_class": given, "reason": parsed["reason"], "alternatives": alts},
            batch_id=batch_id, ts_ns=now_ns(),
        ).on_conflict_do_update(
            constraint="uq_machine_verdict_object_judge_batch",
            set_={"verdict": parsed["verdict"], "proposed_class_id": parsed["proposed_class_id"],
                  "confidence": parsed["confidence"], "provider": provider,
                  "detail": {"given_class": given, "reason": parsed["reason"], "alternatives": alts},
                  "batch_id": batch_id, "ts_ns": now_ns()}))

        if judged % commit_every == 0:
            await db.commit()
            log.info("vlm_review.progress", batch_id=batch_id, judged=judged,
                     of=len(objects) - len(already), by_verdict=counts)

    await db.commit()
    out = {"batch_id": batch_id, "objects": len(objects), "judged": judged, "skipped": skipped,
           "unreadable": unreadable, "failed": failed, "by_verdict": counts, "judge": JUDGE,
           "provider": provider, "model_version": model_version}
    log.info("vlm_review.prereview", **out)
    return out

prereview_batch async

prereview_batch(
    db,
    batch_id,
    *,
    limit=None,
    client=None,
    model_version=None,
    skip_judged=True,
)

Judge every object in a flywheel batch and record what the judge said.

Idempotent by construction: the verdict table is unique on (object, judge, model_version, batch_id), so re-running the same judge updates in place rather than double-counting, a different model version writes its own row so two judges stay comparable on the same crops, and an object appearing in two batches keeps a verdict in each. skip_judged additionally avoids paying for calls that would only overwrite an identical verdict, which matters when the batch is 300 crops and the judge is a metered API.

The judging itself is judge_objects; this function only decides which objects to judge.

Source code in services/labelops/vlm_review.py
async def prereview_batch(db: AsyncSession, batch_id: str, *, limit: int | None = None,
                          client=None, model_version: str | None = None,
                          skip_judged: bool = True) -> dict:
    """Judge every object in a flywheel batch and record what the judge said.

    Idempotent by construction: the verdict table is unique on (object, judge, model_version, batch_id), so
    re-running the same judge updates in place rather than double-counting, a different model version writes
    its own row so two judges stay comparable on the same crops, and an object appearing in two batches
    keeps a verdict in each. `skip_judged` additionally avoids paying for calls that would only overwrite an
    identical verdict, which matters when the batch is 300 crops and the judge is a metered API.

    The judging itself is `judge_objects`; this function only decides which objects to judge.
    """
    q = (select(Object)
         .where(Object.provenance["flywheel"]["cycle_id"].astext == batch_id)
         .order_by(Object.object_id))
    if limit:
        q = q.limit(limit)
    objects = list((await db.execute(q)).scalars().all())
    return await judge_objects(db, objects, batch_id, client=client, model_version=model_version,
                               skip_judged=skip_judged)

judge_agreement async

judge_agreement(
    db, *, judge=JUDGE, model_version=None, batch_id=None
)

Measure the judge against humans on the objects where both have ruled.

This is the step that makes a machine-derived precision worth quoting. Without it the judge's agreement rate is just a number, because there is no way to tell a corpus with 15% bad labels judged perfectly from a corpus with 5% bad labels judged badly.

Human ground truth is read from the object state a reviewer moved it to: accepted or submitted means the person agreed the label was right, rejected means they did not. Objects a human only reclassified count as incorrect, since reclassifying is disagreeing with the original label.

Source code in services/labelops/vlm_review.py
async def judge_agreement(db: AsyncSession, *, judge: str = JUDGE,
                          model_version: str | None = None, batch_id: str | None = None) -> dict:
    """Measure the judge against humans on the objects where both have ruled.

    This is the step that makes a machine-derived precision worth quoting. Without it the judge's agreement
    rate is just a number, because there is no way to tell a corpus with 15% bad labels judged perfectly
    from a corpus with 5% bad labels judged badly.

    Human ground truth is read from the object state a reviewer moved it to: accepted or submitted means the
    person agreed the label was right, rejected means they did not. Objects a human only reclassified count
    as incorrect, since reclassifying is disagreeing with the original label.
    """
    q = (select(MachineVerdict.verdict, Object.state, func.count(Object.object_id))
         .join(Object, Object.object_id == MachineVerdict.object_id)
         .join(Review, Review.object_id == Object.object_id)
         .where(MachineVerdict.judge == judge,
                Object.state.in_(("accepted", "submitted", "rejected")))
         .group_by(MachineVerdict.verdict, Object.state))
    if model_version:
        q = q.where(MachineVerdict.model_version == model_version)
    if batch_id:
        q = q.where(MachineVerdict.batch_id == batch_id)

    tp = fp = tn = fn = unsure_on_good = unsure_on_bad = 0
    for verdict, state, n in (await db.execute(q)).all():
        human_says_correct = state in ("accepted", "submitted")
        n = int(n)
        if verdict == "unsure":
            if human_says_correct:
                unsure_on_good += n
            else:
                unsure_on_bad += n
        elif verdict == "correct":
            if human_says_correct:
                tp += n
            else:
                fp += n
        else:
            if human_says_correct:
                fn += n
            else:
                tn += n

    # Computed over the crops the judge committed on. The abstentions are reported beside them rather than
    # hidden in the denominator, because "the judge is 96% accurate on the 40% it will answer" is a
    # different claim from "the judge is 96% accurate".
    sens = tp / (tp + fn) if (tp + fn) else None
    spec = tn / (tn + fp) if (tn + fp) else None
    decided = tp + fp + tn + fn
    abstained = unsure_on_good + unsure_on_bad

    return {
        "judge": judge, "model_version": model_version, "batch_id": batch_id,
        "compared_against_human": decided + abstained,
        "decided": decided, "abstained": abstained,
        "sensitivity": round(sens, 4) if sens is not None else None,
        "specificity": round(spec, 4) if spec is not None else None,
        "confusion": {"tp": tp, "fp": fp, "tn": tn, "fn": fn,
                      "unsure_on_good": unsure_on_good, "unsure_on_bad": unsure_on_bad},
        "usable": sens is not None and spec is not None and (sens + spec) > 1.0,
    }

judged_precision async

judged_precision(
    db,
    batch_id,
    *,
    confidence=0.95,
    model_version=None,
    agreement_batch_id=None,
)

Precision from machine verdicts, with the judge's own error corrected for where it can be measured.

Returns both numbers on purpose. raw is what the judge said, which is what a naive implementation would have reported as precision; corrected is that rate inverted through the judge's measured sensitivity and specificity. When nobody has adjudicated enough for those to exist, corrected is null and caveat says why, rather than quietly falling back to the raw number.

The judge is measured on this same batch by default, and that default is a statistical claim rather than a convenience. A judge's sensitivity and specificity are properties of the population it is judging, not constants: one that separates pedestrians from poles cleanly may be much weaker on autorickshaw against e_auto. Measuring it on everything it has ever judged and applying that to one batch silently assumes the batches are alike. Pass agreement_batch_id to measure on a different population when that assumption is the one you actually want, for instance a dedicated adjudication set covering a wider slice than the batch under test.

Source code in services/labelops/vlm_review.py
async def judged_precision(db: AsyncSession, batch_id: str, *, confidence: float = 0.95,
                           model_version: str | None = None,
                           agreement_batch_id: str | None = None) -> dict:
    """Precision from machine verdicts, with the judge's own error corrected for where it can be measured.

    Returns both numbers on purpose. `raw` is what the judge said, which is what a naive implementation
    would have reported as precision; `corrected` is that rate inverted through the judge's measured
    sensitivity and specificity. When nobody has adjudicated enough for those to exist, `corrected` is null
    and `caveat` says why, rather than quietly falling back to the raw number.

    The judge is measured on this same batch by default, and that default is a statistical claim rather
    than a convenience. A judge's sensitivity and specificity are properties of the population it is
    judging, not constants: one that separates pedestrians from poles cleanly may be much weaker on
    autorickshaw against e_auto. Measuring it on everything it has ever judged and applying that to one
    batch silently assumes the batches are alike. Pass `agreement_batch_id` to measure on a different
    population when that assumption is the one you actually want, for instance a dedicated adjudication set
    covering a wider slice than the batch under test.
    """
    from services.labelops.sampling import rogan_gladen_interval, wilson_interval

    q = (select(OntologyClass.name, MachineVerdict.verdict, MachineVerdict.proposed_class_id,
                MachineVerdict.detail)
         .join(Object, Object.object_id == MachineVerdict.object_id)
         .join(OntologyClass, OntologyClass.id == Object.class_id)
         .where(MachineVerdict.batch_id == batch_id, MachineVerdict.judge == JUDGE))
    if model_version:
        q = q.where(MachineVerdict.model_version == model_version)

    from services.autolabel.ontology import get_ontology
    from services.labelops.judge_calibration import _is_refinement

    onto = get_ontology()
    per_class: dict[str, dict] = {}
    refinements = cross_superclass = 0
    for name, verdict, proposed, detail in (await db.execute(q)).all():
        d = per_class.setdefault(name, dict.fromkeys(VERDICTS, 0))
        d[verdict] += 1
        if verdict == "incorrect":
            asked = (detail or {}).get("given_class") or name
            if _is_refinement(onto, asked, proposed):
                refinements += 1
            elif proposed is not None:
                cross_superclass += 1

    total_correct = sum(d["correct"] for d in per_class.values())
    total_decided = sum(d["correct"] + d["incorrect"] for d in per_class.values())
    total_unsure = sum(d["unsure"] for d in per_class.values())

    raw = wilson_interval(total_correct, total_decided, confidence)

    # Prefer the retrospective calibration, which measures the judge against human rulings that already
    # existed and therefore covers both directions. judge_agreement is the fallback and is weaker here for a
    # structural reason: it reads human ground truth off the object's current state, and this corpus has 240
    # accepted objects against essentially no rejected ones, so it can measure sensitivity and has almost no
    # negatives to measure specificity with. Rogan-Gladen needs both.
    from services.labelops.judge_calibration import stored_calibration

    # Which judge produced this batch's verdicts. Derived rather than left None, because a correction has
    # to use the calibration of the judge that actually did the judging: with two judges calibrated, an
    # unscoped lookup would average them into a rate belonging to neither.
    judge_model = model_version
    if judge_model is None:
        judges = [r[0] for r in (await db.execute(
            select(MachineVerdict.model_version)
            .where(MachineVerdict.batch_id == batch_id, MachineVerdict.judge == JUDGE)
            .distinct())).all()]
        judge_model = judges[0] if len(judges) == 1 else None

    calibration = await stored_calibration(db, model_version=judge_model)
    agreement = await judge_agreement(db, model_version=model_version,
                                      batch_id=agreement_batch_id or batch_id)

    corrected = caveat = None
    if calibration:
        corrected = rogan_gladen_interval(raw["p"] or 0.0,
                                          sens_ci=calibration["sensitivity_interval"],
                                          spec_ci=calibration["specificity_interval"])
        if corrected.get("clamped"):
            caveat = corrected["note"]
    elif agreement["usable"]:
        corrected = rogan_gladen_interval(
            raw["p"] or 0.0,
            sens_ci={"p": agreement["sensitivity"], "lo": agreement["sensitivity"],
                     "hi": agreement["sensitivity"]},
            spec_ci={"p": agreement["specificity"], "lo": agreement["specificity"],
                     "hi": agreement["specificity"]})
        caveat = ("corrected against judge agreement on reviewed objects rather than a calibration set, so "
                  "the judge's own uncertainty is not carried through")
    else:
        caveat = (f"no correction applied: only {agreement['decided']} objects have both a machine verdict "
                  f"and a human ruling, and no retrospective calibration exists, so the judge's own error "
                  f"rate is unmeasured. The raw figure is the judge's agreement rate, not the label "
                  f"precision.")

    # The same strict-versus-superclass split the calibration reports, one layer out, and for the same
    # reason. On this corpus the strict rate is 0.509 and reads as "half the labels are wrong"; of the 140
    # rejections behind it, 128 propose another four-wheeler and 2 are genuinely a different kind of thing.
    # Quoting the strict figure alone as corpus precision would be the most misleading number this system
    # could produce.
    superclass_correct = total_correct + refinements
    return {
        "batch_id": batch_id,
        "judged": total_decided, "unsure": total_unsure,
        "raw": raw,
        "raw_superclass": wilson_interval(superclass_correct, total_decided, confidence),
        "rejections": {"refinement_within_superclass": refinements,
                       "cross_superclass": cross_superclass,
                       "note": ("a refinement swapped a class for a sibling under the same L1 (sedan to "
                                "SUV); a cross-superclass rejection says the label named the wrong kind of "
                                "thing entirely, which is the error a safety gate cares about")},
        "corrected": corrected,
        "judge_calibration": calibration,
        "judge_agreement": agreement,
        "caveat": caveat,
        "per_class": {k: {**v, "raw": wilson_interval(v["correct"], v["correct"] + v["incorrect"], confidence)}
                      for k, v in sorted(per_class.items())},
    }

services.labelops.class_precision

Per-class label precision, measured rather than inferred from confidence.

Confidence is the obvious proxy for "which classes are wrong" and it is a poor one. It is the detector's opinion of its own output, so a class the detector is confidently wrong about looks healthy and a class it is diffidently right about looks broken. object_fallback sits at mean confidence 0.806 while meaning "I do not know what this is", and traffic_signal sits at 0.271 across 64,741 objects. Neither number tells you whether the label is right.

So this asks a judge, per class, and reports the answer with its uncertainty. Everything about the judging is services/labelops/vlm_review.py unchanged: the same "the label says X, is that right?" prompt, the same first-class unsure, the same MachineVerdict plane that is deliberately not the human review table, and the same Rogan-Gladen correction through the judge's measured sensitivity and specificity. The only thing that is new here is which crops get judged.

The sample is random within the class, never the highest-scoring. A class's most confident detections are its best case, and measuring those reports how good the detector is when it is surest, which is not what a remediation decision needs to know. This mirrors services/errordetect/judge_detectors.py::sample_candidates.

Human-reviewed objects are excluded. An object a person already ruled on is not evidence about the machine's precision, and including them inflates the rate by exactly the amount of review that has happened.

This is a GPU job and it behaves like one. It takes the core/gpu_slot.py advisory lock for the duration, so it cannot run beside a training run, a corpus relabel or an autolabel pass - any two of those on one card is an out-of-memory part way through a batch, which the caller counts as a failed unit rather than as contention. It yields to a live training job rather than competing with one. And it checks free VRAM between batches and waits rather than pushing the card into swap, because a judge sweep is worth nothing compared to the training run it would take down.

sample_class async

sample_class(
    db, class_id, n, *, seed=None, min_side_px=0.0
)

A pseudo-random sample of one class's machine-labelled objects, stable across runs.

Ordered by a hash of the object id and the seed, not by random(). setseed plus ORDER BY random() looks reproducible and is not: random() is evaluated per row in scan order, so the draw survives a repeat only while the query plan and the table are unchanged. Both change here. A remediation sweep rewrites class_id on exactly the rows this selects, so the follow-up measurement would draw a different sample and the difference between the two numbers would be partly the fix and partly the draw - which is the one thing the before/after comparison exists to avoid. Measured: a re-run with the identical seed took sedan from 80 stored verdicts to 128, meaning 48 fresh crops.

A hash of the id is stable against both. An object keeps its position whatever else happens to the table, so the second sample is the first one plus or minus only the rows that genuinely left the class.

min_side_px drops crops too small to judge. A 6-pixel box is not a label the judge can rule on, and an unsure from an unjudgeable crop tells you nothing about the class while still costing a call.

Source code in services/labelops/class_precision.py
async def sample_class(db: AsyncSession, class_id: int, n: int, *, seed: float | None = None,
                       min_side_px: float = 0.0) -> list[Object]:
    """A pseudo-random sample of one class's machine-labelled objects, stable across runs.

    Ordered by a hash of the object id and the seed, not by `random()`. `setseed` plus `ORDER BY random()`
    looks reproducible and is not: random() is evaluated per row in scan order, so the draw survives a
    repeat only while the query plan and the table are unchanged. Both change here. A remediation sweep
    rewrites `class_id` on exactly the rows this selects, so the follow-up measurement would draw a
    different sample and the difference between the two numbers would be partly the fix and partly the
    draw - which is the one thing the before/after comparison exists to avoid. Measured: a re-run with the
    identical seed took `sedan` from 80 stored verdicts to 128, meaning 48 fresh crops.

    A hash of the id is stable against both. An object keeps its position whatever else happens to the
    table, so the second sample is the first one plus or minus only the rows that genuinely left the class.

    `min_side_px` drops crops too small to judge. A 6-pixel box is not a label the judge can rule on, and an
    `unsure` from an unjudgeable crop tells you nothing about the class while still costing a call.
    """
    q = (select(Object)
         .where(Object.class_id == class_id, Object.source.in_(_MACHINE_SOURCES)))
    if min_side_px > 0:
        q = q.where((Object.bbox[3] - Object.bbox[1]) >= min_side_px,
                    (Object.bbox[4] - Object.bbox[2]) >= min_side_px)
    key = f"{seed if seed is not None else 0.0}"
    q = q.order_by(func.md5(func.concat(func.cast(Object.object_id, Text), key))).limit(n)
    return list((await db.execute(q)).scalars().all())

judge_class async

judge_class(
    db,
    class_name,
    *,
    n=120,
    seed=0.42,
    min_side_px=12.0,
    client=None,
    model_version=None,
    batch=BATCH,
    take_slot=True,
)

Judge a random sample of one class and record the verdicts. Returns the judging summary.

Runs in batches of batch crops, re-checking the card between them, so a training job that starts mid-class waits seconds rather than the length of the class. Verdicts upsert, so a sweep that stops early has still banked everything it judged and a re-run resumes rather than restarting.

Reporting is deliberately not done here: services/labelops/vlm_review.py::judged_precision already turns verdicts into a raw Wilson interval and a corrected one, and already refuses to correct when the judge is unmeasured. A second precision calculation here would be a second thing to keep honest.

Source code in services/labelops/class_precision.py
async def judge_class(db: AsyncSession, class_name: str, *, n: int = 120, seed: float | None = 0.42,
                      min_side_px: float = 12.0, client=None, model_version: str | None = None,
                      batch: int = BATCH, take_slot: bool = True) -> dict:
    """Judge a random sample of one class and record the verdicts. Returns the judging summary.

    Runs in batches of `batch` crops, re-checking the card between them, so a training job that starts
    mid-class waits seconds rather than the length of the class. Verdicts upsert, so a sweep that stops
    early has still banked everything it judged and a re-run resumes rather than restarting.

    Reporting is deliberately not done here: `services/labelops/vlm_review.py::judged_precision` already
    turns verdicts into a raw Wilson interval and a corrected one, and already refuses to correct when the
    judge is unmeasured. A second precision calculation here would be a second thing to keep honest.
    """
    import contextlib

    from core.gpu_slot import gpu_slot
    from services.autolabel.ontology import get_ontology
    from services.labelops.vlm_review import judge_objects

    onto = get_ontology()
    if not onto.has_name(class_name):
        raise ValueError(f"unknown class '{class_name}'")
    cid = onto.by_name(class_name).id

    objects = await sample_class(db, cid, n, seed=seed, min_side_px=min_side_px)
    if not objects:
        return {"class_name": class_name, "class_id": cid, "judged": 0,
                "skipped_reason": "no machine-labelled objects of this class are large enough to judge"}

    holder = f"class_precision:{class_name}"
    # `take_slot=False` is for a caller already holding the slot for a whole sweep, so a fifteen-class run
    # does not release and re-acquire the card between every class and hand it to a waiting job mid-sweep.
    slot = gpu_slot(holder, timeout_s=None) if take_slot else contextlib.nullcontext()

    totals = {"judged": 0, "skipped": 0, "unreadable": 0, "failed": 0}
    counts: dict[str, int] = {}
    stalled = None
    async with slot:
        for i in range(0, len(objects), max(1, batch)):
            head = await wait_for_headroom(db, holder=holder)
            if not head["ok"]:
                stalled = f"gave up after {head['waited_s']}s waiting for {', '.join(head['waited_for'])}"
                log.warning("class_precision.stalled", holder=holder, reason=stalled)
                break
            chunk = objects[i:i + max(1, batch)]
            out = await judge_objects(db, chunk, batch_id_for(class_name), client=client,
                                      model_version=model_version)
            for k in totals:
                totals[k] += out.get(k, 0) or 0
            for k, v in (out.get("by_verdict") or {}).items():
                counts[k] = counts.get(k, 0) + v

    res = {"class_name": class_name, "class_id": cid, "sampled": len(objects),
           "by_verdict": counts, **totals}
    if stalled:
        res["stalled"] = stalled
    log.info("class_precision.judged", **{k: v for k, v in res.items() if k != "by_verdict"})
    return res

class_targets async

class_targets(db, *, min_objects=10000, limit=20)

The classes worth measuring: the biggest ones, because that is where being wrong costs most.

Volume rather than suspicion, on purpose. Picking the classes that already look bad measures a hypothesis instead of testing it, and a class that is large and quietly wrong is the expensive case.

Source code in services/labelops/class_precision.py
async def class_targets(db: AsyncSession, *, min_objects: int = 10_000, limit: int = 20) -> list[dict]:
    """The classes worth measuring: the biggest ones, because that is where being wrong costs most.

    Volume rather than suspicion, on purpose. Picking the classes that already look bad measures a
    hypothesis instead of testing it, and a class that is large and quietly wrong is the expensive case.
    """
    rows = (await db.execute(
        select(Object.class_id, func.count(Object.object_id).label("n"),
               func.avg(Object.conf).label("conf"))
        .where(Object.source.in_(_MACHINE_SOURCES))
        .group_by(Object.class_id)
        .having(func.count(Object.object_id) >= min_objects)
        .order_by(func.count(Object.object_id).desc())
        .limit(limit))).all()

    from services.autolabel.ontology import get_ontology

    onto = get_ontology()
    out = []
    for cid, n, conf in rows:
        try:
            name = onto.by_id(int(cid)).name
        except KeyError:
            continue
        out.append({"class_id": int(cid), "class_name": name, "n": int(n),
                    "mean_conf": round(float(conf), 3)})
    return out

services.labelops.sampling

How many to check, and what the answer is worth once you have.

Every quality number this system reports is a bare point estimate. measured_precision returns a fraction, honeypot_accuracy returns a ratio, the overnight auditor samples a hardcoded 200. None of them says how sure it is, so "precision is 0.87" reads identically whether it came from 12 objects or 12,000, and a customer buying a quality claim is buying the interval as much as the number.

Two things live here.

A Wilson interval, not the textbook normal approximation. At the rates that matter here, a defect rate near 0 or near 1 on a few hundred samples, the normal interval is wrong in the direction that flatters: it produces bounds below zero for a clean batch and is far too narrow when p is extreme. Wilson stays inside [0, 1] and holds its coverage at small n, which is the whole regime this corpus is in.

And a sample size, so "check some" becomes a number somebody can plan around.

Sampling for precision is not sampling for improvement, and the difference matters. The active-learning queue deliberately surfaces the hardest, most uncertain objects, which is right for teaching the model and ruinous for measuring it: judging that batch tells you the accuracy of the worst objects in the corpus, not of the corpus. A precision estimate needs a sample that is random with respect to correctness.

wilson_interval

wilson_interval(successes, n, confidence=0.95)

A proportion with its uncertainty. Returns {p, lo, hi, n, half_width}.

n = 0 gives the whole interval rather than an error, because "we have not checked any" is a real state and reporting it as 0.0 precision would be a lie in the confident direction.

Source code in services/labelops/sampling.py
def wilson_interval(successes: int, n: int, confidence: float = 0.95) -> dict:
    """A proportion with its uncertainty. Returns {p, lo, hi, n, half_width}.

    n = 0 gives the whole interval rather than an error, because "we have not checked any" is a real state
    and reporting it as 0.0 precision would be a lie in the confident direction.
    """
    if n <= 0:
        return {"p": None, "lo": 0.0, "hi": 1.0, "n": 0, "half_width": 1.0,
                "note": "nothing sampled yet, so the rate is unknown rather than zero"}
    z = _Z.get(round(confidence, 2), 1.96)
    p = successes / n
    denom = 1.0 + z * z / n
    centre = (p + z * z / (2 * n)) / denom
    margin = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n)) / denom
    lo, hi = max(0.0, centre - margin), min(1.0, centre + margin)
    return {"p": round(p, 4), "lo": round(lo, 4), "hi": round(hi, 4), "n": n,
            "half_width": round((hi - lo) / 2, 4)}

rogan_gladen

rogan_gladen(observed_p, *, sensitivity, specificity)

Correct a rate measured by an imperfect judge for that judge's own error.

The reason this is needed rather than optional. A VLM can judge 570,379 labels; a person cannot. But the judge is wrong sometimes, and quoting its raw agreement rate as precision embeds its error in every number downstream, in an unknown direction. If the judge is 90% sensitive and calls 85% of labels correct, the true rate is not 85%.

Measure the judge against a human-adjudicated subsample to get its sensitivity (it says correct when the label is correct) and specificity (it says incorrect when the label is wrong), then invert:

p_true = (p_observed + specificity - 1) / (sensitivity + specificity - 1)

the standard Rogan-Gladen prevalence estimator. Returns None when sensitivity + specificity <= 1, which means the judge carries no information (at exactly 1 it is a coin, below it is anti-correlated) and no correction can recover a rate from it. That is a real state and worth refusing to answer for, since the formula happily returns a confident-looking number either side of the singularity.

Clamped to [0, 1]: sampling noise in a small subsample can push the estimate outside the range it is estimating, and a precision of 1.04 is less useful than a precision of 1.0 with a wide interval.

Source code in services/labelops/sampling.py
def rogan_gladen(observed_p: float, *, sensitivity: float, specificity: float) -> float | None:
    """Correct a rate measured by an imperfect judge for that judge's own error.

    The reason this is needed rather than optional. A VLM can judge 570,379 labels; a person cannot. But the
    judge is wrong sometimes, and quoting its raw agreement rate as precision embeds its error in every
    number downstream, in an unknown direction. If the judge is 90% sensitive and calls 85% of labels
    correct, the true rate is not 85%.

    Measure the judge against a human-adjudicated subsample to get its sensitivity (it says correct when the
    label is correct) and specificity (it says incorrect when the label is wrong), then invert:

        p_true = (p_observed + specificity - 1) / (sensitivity + specificity - 1)

    the standard Rogan-Gladen prevalence estimator. Returns None when sensitivity + specificity <= 1, which
    means the judge carries no information (at exactly 1 it is a coin, below it is anti-correlated) and no
    correction can recover a rate from it. That is a real state and worth refusing to answer for, since the
    formula happily returns a confident-looking number either side of the singularity.

    Clamped to [0, 1]: sampling noise in a small subsample can push the estimate outside the range it is
    estimating, and a precision of 1.04 is less useful than a precision of 1.0 with a wide interval.
    """
    denom = sensitivity + specificity - 1.0
    if denom <= 1e-9:
        return None
    return max(0.0, min(1.0, (observed_p + specificity - 1.0) / denom))

rogan_gladen_interval

rogan_gladen_interval(observed_p, *, sens_ci, spec_ci)

Correct a rate for an imperfect judge, carrying the judge's own uncertainty through.

The point version collapses three uncertain quantities into one confident-looking number, and on real data that is worse than useless. Measured on this corpus the judge came out at sensitivity 0.76 (0.65 to 0.84) and specificity 0.80 (0.65 to 0.90), which puts the estimator's denominator anywhere between 0.30 and 0.74. A denominator uncertain by a factor of two makes the corrected rate uncertain by a factor of two, and quoting its midpoint would hide exactly that.

So the correction is evaluated at both ends of the judge's intervals. Note this is the range implied by the judge's uncertainty alone; the sampling error in observed_p is reported separately by the caller and the two are not combined, because combining them would imply a joint interval nobody computed.

clamped is the important flag. Rogan-Gladen is unbounded, so a judge whose measured error cannot explain the observed rate produces an estimate above 1.0 or below 0.0, which then gets clipped into range and reads as a confident 1.0. That is a signal that the model does not fit, not an answer, and it has to be visible.

Source code in services/labelops/sampling.py
def rogan_gladen_interval(observed_p: float, *, sens_ci: dict, spec_ci: dict) -> dict:
    """Correct a rate for an imperfect judge, carrying the judge's own uncertainty through.

    The point version collapses three uncertain quantities into one confident-looking number, and on real
    data that is worse than useless. Measured on this corpus the judge came out at sensitivity 0.76
    (0.65 to 0.84) and specificity 0.80 (0.65 to 0.90), which puts the estimator's denominator anywhere
    between 0.30 and 0.74. A denominator uncertain by a factor of two makes the corrected rate uncertain by
    a factor of two, and quoting its midpoint would hide exactly that.

    So the correction is evaluated at both ends of the judge's intervals. Note this is the range implied by
    the judge's uncertainty alone; the sampling error in `observed_p` is reported separately by the caller
    and the two are not combined, because combining them would imply a joint interval nobody computed.

    `clamped` is the important flag. Rogan-Gladen is unbounded, so a judge whose measured error cannot
    explain the observed rate produces an estimate above 1.0 or below 0.0, which then gets clipped into
    range and reads as a confident 1.0. That is a signal that the model does not fit, not an answer, and it
    has to be visible.
    """
    lo_est = rogan_gladen(observed_p, sensitivity=sens_ci["lo"], specificity=spec_ci["lo"])
    hi_est = rogan_gladen(observed_p, sensitivity=sens_ci["hi"], specificity=spec_ci["hi"])
    mid = rogan_gladen(observed_p, sensitivity=sens_ci["p"], specificity=spec_ci["p"])

    ends = [v for v in (lo_est, hi_est, mid) if v is not None]
    if not ends:
        return {"p": None, "lo": None, "hi": None, "clamped": False,
                "note": "the judge carries no information (sensitivity + specificity <= 1), so no "
                        "correction is possible at any point in its interval"}

    def _raw(sens: float, spec: float) -> float | None:
        d = sens + spec - 1.0
        return None if d <= 1e-9 else (observed_p + spec - 1.0) / d

    raws = [r for r in (_raw(sens_ci["lo"], spec_ci["lo"]), _raw(sens_ci["hi"], spec_ci["hi"]),
                        _raw(sens_ci["p"], spec_ci["p"])) if r is not None]
    clamped = any(r > 1.0 or r < 0.0 for r in raws)

    note = None
    if clamped:
        note = ("the corrected estimate falls outside [0, 1] before clamping, which means the observed rate "
                "is more extreme than this judge's measured error can explain. Treat it as a bound, not a "
                "point estimate: either the judge is better than its calibration suggests, or the "
                "calibration set is not representative of the batch being corrected")
    return {"p": round(mid, 4) if mid is not None else None,
            "lo": round(min(ends), 4), "hi": round(max(ends), 4),
            "clamped": clamped, "note": note}

Temporal

Filling frames between detections, and refusing to when the anchors are not one object.

services.temporal.gap_gate

Whether two detections on a track are plausibly the same object, and therefore whether the hole between them may be filled.

This is the gate that was missing, and its absence is the whole defect. Interpolation arithmetic is not what went wrong in the 137,913-object gap fill: printing a track's boxes in time order shows the fills bridging two real detections smoothly and landing where they should. What went wrong is that most of those pairs were not the same object. 45.9% of track steps have zero box overlap and 59.2% of tracks contain a centre jump of more than a quarter of the frame width, because the tracker's feasibility test admits a match on appearance alone at zero overlap. Interpolating across such a pair draws a smooth path between two unrelated things and puts every box on empty road.

The gate is worth its own module because the same predicate answers two questions: may this hole be filled, and should this track be split here. One definition, so the splitter and the filler cannot disagree about what a discontinuity is.

Measured against the gaps that were actually filled: 64.6% of endpoint pairs stayed within the displacement bound, 70.3% within the scale bound, 34.4% agreed on class, and 20.9% passed all three. Judged precision of the objects produced was 0.209. The correspondence is the evidence that this is the right cut.

GateResult dataclass

Why a hole was or was not filled. reason is None when it passed.

Source code in services/temporal/gap_gate.py
@dataclass(frozen=True)
class GateResult:
    """Why a hole was or was not filled. `reason` is None when it passed."""

    ok: bool
    reason: str | None = None
    # What the gate measured, so a rejection can be argued with rather than only obeyed.
    detail: dict | None = None

same_object

same_object(
    box_a,
    box_b,
    class_a,
    class_b,
    *,
    frame_width,
    gap_frames,
    cliques=None,
    max_travel_frac=MAX_CENTRE_TRAVEL_FRAC,
    max_area_ratio=MAX_AREA_RATIO,
    max_gap_frames=MAX_GAP_FRAMES,
)

Could these two detections be the same object, such that the frames between them may be filled?

Deliberately three cheap geometric and semantic tests rather than an appearance model. Appearance is what the tracker already over-trusted: its gate admits a match when the DINOv3 cosine clears 0.55, which two arbitrary same-class vehicles routinely do. Geometry is the signal appearance was allowed to override.

cliques is the pack's CliqueSpec. Class equality is deliberately not required: the detector renames one object between consecutive frames - a single receding vehicle in this corpus is labelled truck, rider, autorickshaw, suv and motorcycle on five consecutive frames - so exact matching would reject 65.6% of holes, most of them genuine. A shared confusion clique tolerates that instability without letting a pedestrian bridge to a bus.

Source code in services/temporal/gap_gate.py
def same_object(box_a, box_b, class_a: str, class_b: str, *, frame_width: float,
                gap_frames: int, cliques=None,
                max_travel_frac: float = MAX_CENTRE_TRAVEL_FRAC,
                max_area_ratio: float = MAX_AREA_RATIO,
                max_gap_frames: int = MAX_GAP_FRAMES) -> GateResult:
    """Could these two detections be the same object, such that the frames between them may be filled?

    Deliberately three cheap geometric and semantic tests rather than an appearance model. Appearance is what
    the tracker already over-trusted: its gate admits a match when the DINOv3 cosine clears 0.55, which two
    arbitrary same-class vehicles routinely do. Geometry is the signal appearance was allowed to override.

    `cliques` is the pack's `CliqueSpec`. Class equality is deliberately not required: the detector renames
    one object between consecutive frames - a single receding vehicle in this corpus is labelled truck,
    rider, autorickshaw, suv and motorcycle on five consecutive frames - so exact matching would reject
    65.6% of holes, most of them genuine. A shared confusion clique tolerates that instability without
    letting a pedestrian bridge to a bus.
    """
    if gap_frames > max_gap_frames:
        return GateResult(False, "gap_too_long", {"gap_frames": gap_frames, "max": max_gap_frames})

    ax, ay = _centre(box_a)
    bx, by = _centre(box_b)
    travel = ((ax - bx) ** 2 + (ay - by) ** 2) ** 0.5
    limit = max(1.0, float(frame_width) * max_travel_frac)
    if travel > limit:
        return GateResult(False, "endpoints_teleport",
                          {"travel_px": round(travel, 1), "limit_px": round(limit, 1)})

    aa, ab = _area(box_a), _area(box_b)
    ratio = max(aa, ab) / min(aa, ab)
    if ratio > max_area_ratio:
        return GateResult(False, "endpoints_scale_jump",
                          {"area_ratio": round(ratio, 2), "max": max_area_ratio})

    if class_a != class_b:
        # No clique spec means the domain declares no confusions, so the only safe reading of two different
        # class names is two different objects.
        ca = cliques.clique_of(class_a) if cliques is not None else None
        cb = cliques.clique_of(class_b) if cliques is not None else None
        if ca is None or ca is not cb:
            return GateResult(False, "endpoints_class_mismatch",
                              {"class_a": class_a, "class_b": class_b,
                               "clique_a": ca.name if ca else None, "clique_b": cb.name if cb else None})

    return GateResult(True, None, {"travel_px": round(travel, 1), "area_ratio": round(ratio, 2),
                                   "gap_frames": gap_frames})

is_discontinuity

is_discontinuity(
    box_a,
    box_b,
    *,
    frame_width,
    max_travel_frac=MAX_CENTRE_TRAVEL_FRAC,
)

True when a track step cannot be one object continuing, and the track should be cut here.

The geometric half of same_object, without the class test: a track that changes class is the detector being unstable, while a track that teleports is two objects wearing one id. Splitting on class alone would shred correct tracks.

Source code in services/temporal/gap_gate.py
def is_discontinuity(box_a, box_b, *, frame_width: float,
                     max_travel_frac: float = MAX_CENTRE_TRAVEL_FRAC) -> bool:
    """True when a track step cannot be one object continuing, and the track should be cut here.

    The geometric half of `same_object`, without the class test: a track that changes class is the
    detector being unstable, while a track that teleports is two objects wearing one id. Splitting on class
    alone would shred correct tracks.
    """
    ax, ay = _centre(box_a)
    bx, by = _centre(box_b)
    travel = ((ax - bx) ** 2 + (ay - by) ** 2) ** 0.5
    if travel > max(1.0, float(frame_width) * max_travel_frac):
        return True
    # Zero overlap on its own is not enough at 3 fps - a small fast object legitimately clears its own box
    # between samples - so it counts only alongside a large area change, which together read as two objects.
    ix = max(0.0, min(float(box_a[2]), float(box_b[2])) - max(float(box_a[0]), float(box_b[0])))
    iy = max(0.0, min(float(box_a[3]), float(box_b[3])) - max(float(box_a[1]), float(box_b[1])))
    if ix * iy > 0:
        return False
    aa, ab = _area(box_a), _area(box_b)
    return max(aa, ab) / min(aa, ab) > MAX_AREA_RATIO

services.temporal.interpolate

Filling the frames between anchors on a track, and refusing to when the anchors are not one object.

Between anchors, boxes are interpolated in centre-and-size space with a shape-preserving spline and marked source=interpolated with an interp_source, so provenance shows they are machine-filled.

The refusal is the important part. An earlier corpus-wide fill created 137,913 objects that judge at 0.209 precision against 0.603 for real detections. The arithmetic here was never the problem - printing a track's boxes in time order shows the fills landing exactly where they should. The problem is that most anchor pairs were not the same object: only 20.9% of the holes that were filled had endpoints within a plausible displacement, scale and class of each other, and 0.209 of the objects produced were right. So every hole now passes services/temporal/gap_gate.py::same_object before anything is written, and a run reports what it declined and why rather than only what it made.

Two anchor policies, because the two callers want different things and conflating them is how the corpus fill went wrong. keyframe anchors on human-verified or explicitly marked boxes, which is what the editor means by interpolating between keyframes - but only 179 of 11,406 tracks have two of those, so it is useless for a backfill. detection anchors on detector output. Neither ever anchors on interpolated or propagated boxes: the old implementation treated every object as an anchor, so its own output re-anchored the next run and the errors compounded.

interpolate_track_keyframed async

interpolate_track_keyframed(
    track_id,
    method="linear",
    lo_ts=None,
    hi_ts=None,
    *,
    anchor_policy="keyframe",
    run_id=None,
    gate=True,
)

Fill frames between anchors with interpolated boxes, skipping holes whose anchors are not one object.

anchor_policy selects what counts as an anchor - see ANCHOR_SOURCES. run_id makes the fill revertible: every created object is stamped with it and recorded as {"created": True} on the run, which is the shape services/agent/runs.py::revert_run already deletes. gate=False is for the editor's explicit "interpolate between the two boxes I just drew", where a person has already asserted they are one object.

Returns per-hole accounting, including refused keyed by reason. A fill that reports only what it created cannot be distinguished from one that created the wrong thing, which is how the last one passed.

Source code in services/temporal/interpolate.py
async def interpolate_track_keyframed(track_id: UUID, method: str = "linear", lo_ts: int | None = None,
                                      hi_ts: int | None = None, *, anchor_policy: str = "keyframe",
                                      run_id: UUID | None = None, gate: bool = True) -> dict:
    """Fill frames between anchors with interpolated boxes, skipping holes whose anchors are not one object.

    `anchor_policy` selects what counts as an anchor - see ANCHOR_SOURCES. `run_id` makes the fill revertible:
    every created object is stamped with it and recorded as `{"created": True}` on the run, which is the
    shape `services/agent/runs.py::revert_run` already deletes. `gate=False` is for the editor's explicit
    "interpolate between the two boxes I just drew", where a person has already asserted they are one object.

    Returns per-hole accounting, including `refused` keyed by reason. A fill that reports only what it
    created cannot be distinguished from one that created the wrong thing, which is how the last one passed.
    """
    from services.domain import active_pack
    from services.temporal.gap_gate import same_object

    cliques = active_pack().cliques
    maker = get_sessionmaker()
    async with maker() as db:
        tr = await db.get(Track, track_id)
        if tr is None:
            return {"created": 0, "reason": "track not found"}
        anchors = await _anchors(db, track_id, anchor_policy)
        if len(anchors) < 2:
            return {"created": 0, "reason": f"need at least 2 {anchor_policy} anchors"}

        kf_ts = [ts for _, ts in anchors]
        kf_box = np.asarray([list(o.bbox) for o, _ in anchors], dtype=float)
        class_id = anchors[0][0].class_id
        a, b = (lo_ts if lo_ts is not None else kf_ts[0]), (hi_ts if hi_ts is not None else kf_ts[-1])

        # Confine interpolation to the track's own camera. A rig session has frames from several cameras at
        # overlapping timestamps; without this filter the fill would create boxes on every camera's frames,
        # poisoning views the track was never in.
        anchor_frame = await db.get(Frame, anchors[0][0].frame_id)
        cam_id = anchor_frame.cam_id if anchor_frame else None

        fq = (select(Frame.frame_id, Frame.ts_ns)
              .where(Frame.session_id == tr.session_id, Frame.ts_ns > a, Frame.ts_ns < b))
        if cam_id is not None:
            fq = fq.where(Frame.cam_id == cam_id)
        frames = (await db.execute(fq.order_by(Frame.ts_ns))).all()
        # clear existing machine-filled boxes on this track in the segment (idempotent re-interpolation)
        seg_fids = [fid for fid, _ in frames]
        if seg_fids:
            await db.execute(delete(Object).where(
                Object.track_id == track_id, Object.source == "interpolated", Object.frame_id.in_(seg_fids)))

        box_at, src = build_box_interpolator(kf_ts, kf_box, method)

        # Anchor lookup by class name and by index, for the gate.
        onto = get_ontology()
        name_of = {}
        for o, ts in anchors:
            try:
                name_of[ts] = onto.by_id(int(o.class_id)).name
            except KeyError:
                name_of[ts] = str(o.class_id)
        width = float(anchor_frame.width) if anchor_frame and anchor_frame.width else 1920.0

        # Group the holes by the anchor pair that brackets them, so the gate is asked once per hole rather
        # than once per frame and a refusal drops the whole hole rather than half of it.
        kf_set = set(kf_ts)
        holes: dict[tuple[int, int], list] = {}
        for fid, ts in frames:
            if ts in kf_set:
                continue
            lo = max((k for k in kf_ts if k <= ts), default=None)
            hi = min((k for k in kf_ts if k >= ts), default=None)
            if lo is None or hi is None:
                continue
            holes.setdefault((lo, hi), []).append((fid, ts))

        by_ts = {ts: o for o, ts in anchors}
        created = 0
        refused: dict[str, int] = {}
        refused_frames = 0
        for (lo, hi), members in sorted(holes.items()):
            if gate:
                res = same_object(by_ts[lo].bbox, by_ts[hi].bbox, name_of[lo], name_of[hi],
                                  frame_width=width, gap_frames=len(members), cliques=cliques)
                if not res.ok:
                    refused[res.reason] = refused.get(res.reason, 0) + 1
                    refused_frames += len(members)
                    continue
            for fid, ts in members:
                conf = _interp_conf(float(ts), kf_ts)
                prov = {"method": "interpolate", "interp_source": src, "conf_by_gap": conf,
                        "gap_frames": len(members)}
                if run_id is not None:
                    prov["agent_run_id"] = str(run_id)
                db.add(Object(frame_id=fid, track_id=track_id, class_id=class_id, bbox=box_at(float(ts)),
                              conf=conf, source="interpolated", state="annotate", interp_source=src,
                              provenance=prov))
                created += 1

        if run_id is not None and created:
            # Stamped in the same transaction as the objects. Recording the run afterwards would leave a
            # crash between the two with rows nothing can revert, which is the one thing this exists for.
            await db.flush()
            made = (await db.execute(select(Object.object_id).where(
                Object.track_id == track_id, Object.source == "interpolated",
                Object.provenance["agent_run_id"].astext == str(run_id)))).scalars().all()
            run = await db.get(AgentRun, run_id)
            if run is not None:
                run.changes = {**(run.changes or {}), **{str(oid): {"created": True} for oid in made}}
        await db.commit()

    out = {"track_id": str(track_id), "created": created, "method": src, "anchors": len(kf_ts),
           "anchor_policy": anchor_policy, "holes": len(holes), "refused": refused,
           "refused_frames": refused_frames}
    log.info("interpolate.done", **{k: v for k, v in out.items() if k != "refused"})
    return out

build_box_interpolator

build_box_interpolator(kf_ts, kf_box, method)

Return (box_at, src). box_at(ts) gives an [x1,y1,x2,y2] box for any ts inside the keyframe span.

method='cubic' uses a shape-preserving monotone spline (PCHIP) on center and size. Unlike an ordinary cubic it does not overshoot between anchors (no Runge wobble, no box that briefly balloons or inverts), while still curving through the acceleration a straight line would miss. Falls back to linear with <3 keyframes or if SciPy is unavailable.

Source code in services/temporal/interpolate.py
def build_box_interpolator(kf_ts: list[int], kf_box: np.ndarray, method: str):
    """Return (box_at, src). `box_at(ts)` gives an [x1,y1,x2,y2] box for any ts inside the keyframe span.

    method='cubic' uses a shape-preserving monotone spline (PCHIP) on center and size. Unlike an ordinary
    cubic it does not overshoot between anchors (no Runge wobble, no box that briefly balloons or inverts),
    while still curving through the acceleration a straight line would miss. Falls back to linear with <3
    keyframes or if SciPy is unavailable.
    """
    ts = np.asarray(kf_ts, dtype=float)
    cc = _to_cxcywh(np.asarray(kf_box, dtype=float))
    # collapse duplicate timestamps (two keyframes on the same frame) so the spline sees a strictly increasing grid
    uniq_ts, idx = np.unique(ts, return_index=True)
    ts, cc = uniq_ts, cc[idx]

    fns = None
    src = "linear"
    if method in ("cubic", "pchip", "spline") and len(ts) >= 3:
        try:
            from scipy.interpolate import PchipInterpolator

            fns = [PchipInterpolator(ts, cc[:, i], extrapolate=True) for i in range(4)]
            src = "cubic"
        except Exception:  # noqa: BLE001 - SciPy missing/edge case: degrade to linear rather than fail the fill
            fns = None

    def box_at(t: float) -> list[float]:
        if fns is not None:
            cx, cy, w, h = (float(fns[i](t)) for i in range(4))
        else:
            cx, cy, w, h = (float(np.interp(t, ts, cc[:, i])) for i in range(4))
        w, h = max(1.0, w), max(1.0, h)   # a spline must never emit a zero-area or inverted box
        return [cx - w / 2, cy - h / 2, cx + w / 2, cy + h / 2]

    return box_at, src

Reversibility

Every corpus-wide write in this system is undoable through one of these.

services.review_apply

Applying one review decision to a set of objects, in one place.

services/review_policy.py opens by saying the state rule "lives here, the router calls it, and both the single and the bulk review path go through it". That was true of two paths and not of the third: POST /tracks/{id}/relabel wrote o.state = payload.state straight from the request body with a default of accepted, so an annotator could confirm an entire track and skip the QA step the whole two-stage workflow is built on. It also never advanced version, never recorded a revertible batch, and never revalidated attributes, so it was the one bulk write in the repo with no undo and no lock.

Rather than copy four rules into a third router, the per-object loop bulk review had already grown lives here and every caller gets the same guarantees. Bulk review's observable behaviour is unchanged by the move, which tests/test_bulk_review_hardening.py is the proof of.

Two behaviours are opt-in because the two callers genuinely differ, not because one of them is lazy:

skip_human - a track relabel must not overwrite a frame somebody else already ruled on, the same rule services/agent/temporal_repair.py applies and the correction dialog shows as an "already" badge. Bulk review is an explicit list of ids a person just ticked, so there is nothing to protect them from.

guard_class_move - the ontology guard from services/agent/class_move.py refuses a move that changes what kind of thing something is. A track relabel is one decision fanned across ninety frames and is worth guarding. Bulk review deliberately spans classes: the correction dialog exists to gather one systematic error that is usually spread over several source classes, so guarding it there would break the feature commit 38b28dd built.

apply_review_batch async

apply_review_batch(
    db,
    objects,
    *,
    action,
    onto,
    class_id=None,
    attrs=None,
    requested_state=None,
    role=None,
    source="human",
    reviewer="anon",
    uid=None,
    expected_versions=None,
    time_spent_ms=0,
    provenance_extra=None,
    skip_human=False,
    guard_class_move=False,
    revalidate_attrs=False,
)

Apply one decision to every object given, returning what happened to each.

Adds rows to the session and does not commit: the caller lands the edits and the batch stamp in one transaction, for the reason services/review_batch.py records at length.

Source code in services/review_apply.py
async def apply_review_batch(
    db,
    objects: list[Object],
    *,
    action: str,
    onto: Any,
    class_id: int | None = None,
    attrs: dict | None = None,
    requested_state: str | None = None,
    role: str | None = None,
    source: str = "human",
    reviewer: str = "anon",
    uid: Any = None,
    expected_versions: dict[str, int] | None = None,
    time_spent_ms: int = 0,
    provenance_extra: dict | None = None,
    skip_human: bool = False,
    guard_class_move: bool = False,
    revalidate_attrs: bool = False,
) -> ApplyResult:
    """Apply one decision to every object given, returning what happened to each.

    Adds rows to the session and does not commit: the caller lands the edits and the batch stamp in one
    transaction, for the reason `services/review_batch.py` records at length.
    """
    res = ApplyResult()
    res.new_state = state_for(action, requested_state, role, None)
    res.clamped = was_clamped(action, requested_state, role, None)

    expected = expected_versions or {}
    # The batch's time is divided across its members rather than attributed to any one of them, so a grid
    # that triages sixty crops in a minute reports sixty seconds of work and not sixty minutes or zero.
    per_item_ms = int(time_spent_ms / len(objects)) if objects and time_spent_ms > 0 else 0

    for obj in objects:
        oid = str(obj.object_id)

        # Optimistic lock, per object. A stale member is skipped and named rather than failing the whole
        # batch, because one contended object should not discard fifty-nine good verdicts.
        want = expected.get(oid)
        if want is not None and obj.version != want:
            res.stale.append({"object_id": oid, "expected": want, "current": obj.version})
            continue

        # Never overwrite a decision a person already made. `source == "human"` is this repo's universal
        # marker for that, which is why the propagated rows below are not given it.
        if skip_human and obj.source == "human":
            res.skipped_human.append(oid)
            continue

        if guard_class_move and class_id is not None:
            reason = refuse_reason(onto, obj.class_id, class_id)
            if reason is not None:
                res.refused.append({"object_id": oid, "reason": reason})
                continue

        before = {"class_id": obj.class_id, "bbox": list(obj.bbox), "attrs": dict(obj.attrs or {}),
                  "state": obj.state, "source": obj.source, "conf": obj.conf,
                  "provenance": dict(obj.provenance or {})}
        # What the undo needs, captured before the edit. A Review row per object is the audit trail and
        # answers what changed; it is not an undo, because taking back a fifty-object batch through it means
        # fifty manual reversals with the operator remembering each prior value.
        res.changes[oid] = change_record(obj)

        if class_id is not None:
            obj.class_id = class_id

        if attrs:
            errors = onto.validate_attrs(attrs, obj.class_id)   # against the effective (possibly new) class
            if errors:
                raise AttrRejected(oid, errors)
            merged = dict(obj.attrs or {})
            merged.update(attrs)
            obj.attrs = onto.derive_attrs(merged, obj.class_id)

        if revalidate_attrs and obj.attrs:
            # A class change can make a previously valid attribute not applicable, which
            # services/quality/attr_audit.py names as a corpus-corruption source and attributes to this
            # exact path. Dropping the offending keys rather than refusing, because one stale attribute on
            # one frame must not make a ninety-frame track unfixable. change_record already captured the
            # prior attrs, so the drop comes back on revert.
            bad = _inapplicable(onto, obj.attrs, obj.class_id)
            if bad:
                obj.attrs = {k: v for k, v in obj.attrs.items() if k not in bad}
                res.attrs_dropped[oid] = sorted(bad)

        if res.new_state is not None:
            obj.state = res.new_state
        obj.source = source
        if provenance_extra:
            obj.provenance = {**(obj.provenance or {}), **provenance_extra}
        # Advance the lock version, exactly as single review does. Without this a bulk edit was invisible to
        # every other client's optimistic check, so an editor holding the object would overwrite it back.
        obj.version = (obj.version or 1) + 1

        db.add(Review(object_id=obj.object_id, reviewer=reviewer, user_id=uid, action=action,
                      before=before,
                      after={"class_id": obj.class_id, "bbox": list(obj.bbox),
                             "attrs": dict(obj.attrs or {}), "state": obj.state},
                      time_spent_ms=per_item_ms, ts_ns=now_ns()))
        res.n += 1

    return res

services.review_batch

Undo for a bulk review, and why the generic revert cannot do it.

Bulk review writes one Review row per object, which is the audit trail and answers "who changed this and what was it before". What it did not write was anything tying the fifty objects together, so a fifty-object mistake was fifty manual reversals, each one requiring the operator to remember what the value had been. The correction dialog inherits this: it exists to apply one decision widely, which is exactly the operation most worth being able to take back in one move.

AgentRun is already the repo's revertible unit and services/agent/runs.py revert_run already restores from_class, from_state, from_source and from_attrs. It cannot be reused as is, for a reason its own comment states: it refuses to touch anything whose source is human, which is right when undoing an agent's work over a person's, and wrong here because a human review is what set source = "human" in the first place. Every object in the batch would be skipped.

So this kind gets its own revert, the way ontology_merge and cleanup_sweep already do. The ownership check that keeps it safe is not the source column but the run id stamped in provenance: an object is restored only while it still carries THIS run's id, so anything edited afterwards, by a person or by a later batch, is left alone and reported as skipped rather than silently rolled back.

change_record

change_record(obj)

What has to be remembered about one object for the batch to be undoable.

Captured before the edit. from_source matters as much as from_class: a machine label promoted to human by a batch that turns out to be wrong must go back to being a machine label, or the corpus quietly gains human-authored rows nobody authored.

Source code in services/review_batch.py
def change_record(obj: Object) -> dict:
    """What has to be remembered about one object for the batch to be undoable.

    Captured before the edit. `from_source` matters as much as `from_class`: a machine label promoted to
    `human` by a batch that turns out to be wrong must go back to being a machine label, or the corpus quietly
    gains human-authored rows nobody authored.
    """
    return {"from_class": int(obj.class_id), "from_state": obj.state,
            "from_source": obj.source, "from_attrs": dict(obj.attrs or {})}

record_batch async

record_batch(
    db,
    changes,
    *,
    policy=None,
    created_by=None,
    commit=True,
)

Tie an applied batch together so it can be taken back. Returns the run id, or None for an empty batch.

Stamps each object with the run id, which is what the revert checks ownership against.

commit=False lets the caller land the stamp in the same transaction as the edit it describes. That is the correct way to use this: the review router used to commit the edits and only then call this, on the reasoning that a failed edit must not leave a run claiming objects it never changed - true, but it bought that by opening a window where the process could die between the two and leave the batch permanently un-revertible, because revert_batch keys ownership on precisely this stamp. One transaction satisfies both: a failed edit rolls the stamp back with it, and a committed edit is always revertible.

Source code in services/review_batch.py
async def record_batch(db: AsyncSession, changes: dict[str, dict], *, policy: dict | None = None,
                       created_by: str | None = None, commit: bool = True) -> str | None:
    """Tie an applied batch together so it can be taken back. Returns the run id, or None for an empty batch.

    Stamps each object with the run id, which is what the revert checks ownership against.

    `commit=False` lets the caller land the stamp in the same transaction as the edit it describes. That is
    the correct way to use this: the review router used to commit the edits and only then call this, on the
    reasoning that a failed edit must not leave a run claiming objects it never changed - true, but it
    bought that by opening a window where the process could die between the two and leave the batch
    permanently un-revertible, because revert_batch keys ownership on precisely this stamp. One transaction
    satisfies both: a failed edit rolls the stamp back with it, and a committed edit is always revertible.
    """
    if not changes:
        return None
    run_id = uuid.uuid4()
    for oid in changes:
        obj = await db.get(Object, uuid.UUID(oid))
        if obj is None:
            continue
        prov = dict(obj.provenance or {})
        prov["agent_run_id"] = str(run_id)
        prov["review_batch"] = True
        obj.provenance = prov
    db.add(AgentRun(run_id=run_id, kind=KIND, scope={}, status="committed",
                    policy=policy or {}, counts={"objects": len(changes)},
                    changes=changes, critic={}, created_by=created_by or "review"))
    if commit:
        await db.commit()
    log.info("review_batch.recorded", run_id=str(run_id), objects=len(changes))
    return str(run_id)

revert_batch async

revert_batch(db, run)

Put a bulk review back, including the objects it made human-sourced.

Ownership is the stamped run id rather than the source column, so a later edit by anybody wins and is counted as skipped. That is the same protection the generic revert gets from its source == "human" check, expressed in the one way that still works when the run itself is the reason the row says human.

Source code in services/review_batch.py
async def revert_batch(db: AsyncSession, run: AgentRun) -> dict:
    """Put a bulk review back, including the objects it made human-sourced.

    Ownership is the stamped run id rather than the source column, so a later edit by anybody wins and is
    counted as skipped. That is the same protection the generic revert gets from its `source == "human"`
    check, expressed in the one way that still works when the run itself is the reason the row says human.
    """
    reverted = skipped = 0
    for oid, ch in (run.changes or {}).items():
        obj = await db.get(Object, uuid.UUID(oid))
        if obj is None:
            skipped += 1
            continue
        prov = dict(obj.provenance or {})
        if str(prov.get("agent_run_id")) != str(run.run_id):
            skipped += 1
            continue
        if "from_class" in ch:
            obj.class_id = ch["from_class"]
        if "from_state" in ch:
            obj.state = ch["from_state"]
        if "from_source" in ch:
            obj.source = ch["from_source"]
        if "from_attrs" in ch:
            obj.attrs = ch["from_attrs"]
        obj.version = (obj.version or 0) + 1
        prov.pop("agent_run_id", None)
        prov.pop("review_batch", None)
        obj.provenance = prov
        reverted += 1
    # A track relabel also moved the track's own denormalised class, which is not an object and so has no
    # entry in `changes`. Left behind, a revert would put every object back and leave the track claiming the
    # class the operator just took back, which services/intelligence/propagate.py would then write into
    # every interpolated gap box.
    for tid, was in _track_classes(run.policy or {}).items():
        track = await db.get(Track, uuid.UUID(tid))
        if track is not None and was is not None:
            track.class_id = int(was)

    run.status = "reverted"
    run.reverted_at = datetime.now(UTC)
    await db.commit()
    log.info("review_batch.reverted", run_id=str(run.run_id), reverted=reverted, skipped=skipped)
    return {"run_id": str(run.run_id), "reverted": reverted, "skipped": skipped}

services.agent.runs

AgentRun helpers: serialize a run for the API, and revert one exactly. Revert restores each object's prior state/source from the recorded transition, but skips any object a human has touched since the run (source now "human", or the stamped agent_run_id no longer matches) -- the agent never overwrites a person.

revert_run async

revert_run(db, run_id)
Source code in services/agent/runs.py
async def revert_run(db: AsyncSession, run_id: uuid.UUID) -> dict:
    run = await db.get(AgentRun, run_id)
    if run is None:
        raise ValueError("run not found")
    if run.status != "committed":
        raise ValueError(f"run is {run.status}, only a committed run can be reverted")

    # A class merge moved objects wholesale, including human-labelled ones, so undoing it needs its own
    # path: the generic restore below deliberately refuses to touch anything a person owns, which is right
    # for an agent relabel and wrong for reversing an ontology decision.
    if run.kind == "ontology_merge":
        from services.agent.ontology_merge import revert_merge

        return await revert_merge(db, run)

    # A bulk review is a human decision, so the generic restore below cannot undo it: that path refuses to
    # touch anything whose source is `human`, and a human review is precisely what set the column to human.
    # This kind checks ownership by the stamped run id instead.
    if run.kind == review_batch.KIND:
        return await review_batch.revert_batch(db, run)

    # A cleanup sweep removed objects; reverting re-inserts them from the stored snapshots.
    if run.kind == "cleanup_sweep":
        from services.agent.cleanup_sweep import revert_cleanup

        return await revert_cleanup(db, run)

    # A corpus run (e.g. relabel-all) owns no objects itself; it aggregates one child run per frame.
    # Reverting it reverts each child, so 'undo relabel all' is one click.
    child_ids = (run.changes or {}).get("child_runs")
    if child_ids:
        child_reverted = child_children = 0
        for cid in child_ids:
            try:
                r = await revert_run(db, uuid.UUID(cid))
            except ValueError:
                continue
            child_reverted += r["reverted"]
            child_children += 1
        run.status = "reverted"
        run.reverted_at = datetime.now(UTC)
        await db.commit()
        log.info("agent.run.revert_cascade", run_id=str(run_id), children=child_children, reverted=child_reverted)
        return {"run_id": str(run_id), "reverted": child_reverted, "skipped": 0, "children": child_children}

    reverted = 0
    skipped = 0
    for oid, ch in (run.changes or {}).items():
        obj = await db.get(Object, uuid.UUID(oid))
        if obj is None:
            skipped += 1
            continue
        prov = obj.provenance or {}
        # A human took over, or a later agent run owns it now: leave it alone.
        if obj.source == "human" or str(prov.get("agent_run_id")) != str(run_id):
            skipped += 1
            continue
        # Objects the run CREATED (e.g. propagated boxes) are undone by deleting them.
        if ch.get("created"):
            await db.delete(obj)
            reverted += 1
            continue
        # Field-driven restore: put back whatever the run recorded a prior value for.
        if "from_state" in ch:
            obj.state = ch["from_state"]
        if "from_source" in ch:
            obj.source = ch["from_source"]
        if "from_class" in ch:            # a reconcile relabel: restore the original class
            obj.class_id = ch["from_class"]
        if "from_cuboid" in ch:           # an auto-cuboid: clear/restore the 3D box
            obj.cuboid_3d = ch["from_cuboid"]
        if "from_attrs" in ch:            # an auto-attribute fill: restore the prior attrs
            obj.attrs = ch["from_attrs"]
        obj.version = (obj.version or 0) + 1
        prov = dict(prov)
        for k in ("agent_run_id", "agent_critic", "agent_cuboid", "agent_attrs"):
            prov.pop(k, None)
        obj.provenance = prov
        reverted += 1

    run.status = "reverted"
    run.reverted_at = datetime.now(UTC)
    await db.commit()
    log.info("agent.run.revert", run_id=str(run_id), reverted=reverted, skipped=skipped)
    return {"run_id": str(run_id), "reverted": reverted, "skipped": skipped}