"""SET RETRIEVAL — the robotics query model, seconds-scale, VLM-free. The product truth (user-stated): a robotics team doesn't want the one best clip, they want ALL clips matching a scenario, clean enough to retrain on. Precision of the DELIVERED SET is king; latency budget is seconds. v2 after the visual calibration (2026-07-24) tore v1 down: - Scene-clustered IVF cells hid 45/46 true "lid" episodes from the opened-6 — actions do not live in scene cells. At this corpus size (thousands of episodes) a flat exact scan is sub-millisecond, so query-time cell pruning bought nothing and cost nearly all recall. Cells remain as the browsable scenario MAP (build_scenarios) and as the pruning layout for a future 100k+ corpus. - Single-encoder scoring cannot rank relational actions deep into a list (PE full-scan R@257 = 9/46 on "lid"). The set path now fuses the channels that each own a dimension: PE (appearance-text), ACT (SSv2 action posteriors — put-in vs take-out AUC 0.889), VID (X-CLIP), MOT (delta-appearance contrast, close/open AUC 0.98). - Direction is a HARD FILTER, not a rerank: an episode both direction-aware channels score negative is dropped, not demoted — junk excluded from the delivery, per the set-purity mandate. - The VLM sample audit is GONE (user directive: no LLM/VLM judges; calibration showed 7B est 1.0 on visually ~10%-pure sets). The audited tier is now GEOMETRY: SAM 3 grounds the query's noun phrases and the containment change over time verifies the relation. Deterministic, explainable, abstains honestly. """ from __future__ import annotations import json import os import time import numpy as np _CELLS = {} def _pool_recordings(store): """(keys, matrix) — one pooled PE vector per recording.""" from .embeddings import _vec_table tbl, vecs = _vec_table(store, "pe_vectors") ss = np.asarray(tbl.column("stream").to_pylist()) sa = np.asarray([int(v) for v in tbl.column("ts").to_pylist()]) sb = np.asarray([int(v) for v in tbl.column("t1").to_pylist()]) order = np.lexsort((sa, ss)) keys, mats, rowsets = [], [], [] i = 0 o = order while i < len(o): j = i k = (str(ss[o[i]]), int(sa[o[i]]), int(sb[o[i]])) while j < len(o) and (str(ss[o[j]]), int(sa[o[j]]), int(sb[o[j]])) == k: j += 1 rows = o[i:j] v = np.asarray(vecs[np.sort(rows)]).mean(0) v /= np.linalg.norm(v) + 1e-8 keys.append(k) mats.append(v.astype(np.float32)) rowsets.append(np.sort(rows)) i = j return keys, np.stack(mats), rowsets def build_scenarios(store, min_cluster_size=8): """HDBSCAN scenario groups (the human-browsable map) + KMeans-IVF cells (the physical layout for 100k+ scale), persisted beside pe_vectors. NOT used for query-time pruning at this corpus size — measured hiding 45/46 relational positives.""" t0 = time.time() keys, M, rowsets = _pool_recordings(store) try: from sklearn.cluster import HDBSCAN groups = HDBSCAN(min_cluster_size=min_cluster_size, metric="euclidean", copy=True).fit_predict(M) except Exception: groups = np.zeros(len(M), np.int32) from sklearn.cluster import KMeans k = int(np.clip(len(M) // 32, 16, 256)) lab = KMeans(n_clusters=k, random_state=0, n_init=4).fit_predict(M) cids = sorted(set(int(c) for c in lab)) cents = np.stack([M[lab == c].mean(0) for c in cids]) cents /= np.linalg.norm(cents, axis=1, keepdims=True) + 1e-8 cell_dir = store.table("pe_vectors").dir / "_cache" cell_dir.mkdir(parents=True, exist_ok=True) np.save(cell_dir / "cell_centroids.npy", cents) np.save(cell_dir / "cell_labels.npy", np.asarray(lab, np.int32)) np.save(cell_dir / "cell_matrix.npy", M) np.save(cell_dir / "scenario_groups.npy", np.asarray(groups, np.int32)) (cell_dir / "cells.json").write_text(json.dumps({ "version": store.table("pe_vectors").state().version, "keys": [[k[0], k[1], k[2]] for k in keys], "cells": int(len(cents)), "scenario_groups": int(len(set(int(g) for g in groups if g >= 0))), })) return {"recordings": len(keys), "cells": int(len(cents)), "scenario_groups": int(len(set(int(g) for g in groups if g >= 0))), "seconds": round(time.time() - t0, 1)} def _episodes(store): ep = store.table("episodes").scan() return list(zip((str(s) for s in ep.column("stream").to_pylist()), (int(v) for v in ep.column("ts").to_pylist()), (int(v) for v in ep.column("t1").to_pylist()))) def _auto_action_support(store, text): """Self-recognized no-match signal: map the query onto the action probe's OWN class vocabulary by embedding similarity (no coded verb list — the vocabulary is a model property, the mapping is computed) and report the corpus's best posterior for the top matched classes. Only gates when the mapping is confident: a query far from every class name (a domain this probe does not cover) must never be silenced by it.""" from .action_probe import query_class_weights, ssv2_classes from .embeddings import _vec_table from .pe import _text_vec names = ssv2_classes() qv = _text_vec(text) sims = np.array([float(_text_vec(c.lower()) @ qv) for c in names]) top = np.argsort(-sims)[:3] # SCALE-FREE mapping confidence: is the best class an outlier of # the similarity distribution, or just the least-bad of a flat # field? (An absolute cosine threshold silently disabled the gate # — ledger-caught: fold gate FAIL.) z = float((sims[top[0]] - sims.mean()) / (sims.std() + 1e-9)) if z < 3.0: return None # vocabulary doesn't cover this # CONJUNCTIVE gate, every term scale-free (single-class posterior # could not separate 'fold' 0.008 from 'red-out' 0.010 — measured; # half the vocabulary is below 0.05 on this corpus): # 1. an outlier class exists (z >= 3, above) # 2. those outlier classes are posterior-DEAD here (bottom third # of the class-max distribution) # 3. EXPECTED support under the full similarity distribution is # below the corpus mean — a query whose sim mass spreads over # living classes (red-out: picking/putting) never gates; one # whose mass concentrates on a dead class (fold) does zs = (sims - sims.mean()) / (sims.std() + 1e-9) top = np.where(zs >= 3.0)[0] _, probs = _vec_table(store, "action_probs") cmax = np.asarray(probs).max(0) mp = float(cmax[top].max()) dead = mp < float(np.percentile(cmax, 33)) w = np.exp((sims - sims.max()) / 0.05) w /= w.sum() ratio = float(w @ cmax) / (float(cmax.mean()) + 1e-9) if not (dead and ratio < 1.0): return None return {"classes": [names[int(i)] for i in top], "z": round(z, 2), "max_p": round(mp, 4), "support_ratio": round(ratio, 2)} def corroboration_weights(ch, floor=0.0): """Per-query channel weights from the scores alone. No labels, no channel names, no query types. WHY THIS EXISTS. Every channel used to vote with weight 1.0 whatever it measured. Against the truthset (diagnostic only, never fitted on) the shipped fusion scored BELOW its own best input on all three high-support queries, because inputs at AUC 0.94 and 0.33 were given the same voice. The worst case was q05, where `act` measured AUC 0.330 - not weak, INVERTED, ranking true episodes below false ones - and still voted at full strength. THE RULE. For each channel, build the pooled opinion of the OTHERS (leave-one-out, so a channel never corroborates itself) and take the correlation to it, clipped at zero. A channel that ranks opposite to everything else is refused a vote; a channel that agrees keeps one. Nothing here knows what a channel measures, so it transfers to a corpus with different channels entirely. Measured effect, unsupervised weights vs uniform: q04 AUC 0.855 -> 0.887 prec@support 0.333 -> 0.394 q05 AUC 0.813 -> 0.877 prec@support 0.235 -> 0.311 The gain is almost entirely from ZEROING inverted channels (act on both, mot on q05), not from the shape of the weighting. KNOWN LIMIT, recorded because it bounds this whole approach: the best single channel is often the one that agrees LEAST. iv2 carries q03 at AUC 0.940 and ranks fourth by corroboration. Agreement can veto a bad channel; it cannot identify the good one. Doing that needs a signal of query-specific responsiveness, which this is not. VETO ONLY, AND THAT IS DELIBERATE. The first version of this returned a graded weight per channel (the corroboration value itself, renormalised). Measured end to end that was a REGRESSION - mean precision 0.29 -> 0.16 across q00-q10 - for two reasons found only by running it: the fusion is RRF over RANKS, which does not respond to weight the way the z-score sums in the diagnostic did, and rescaling everything also stripped prf_q of its fitted voice. Graded weighting is therefore NOT shipped. What survives is the part the evidence actually supports: a channel whose correlation to the pooled others is NEGATIVE is inverted, and inverted channels get zero. Everything else keeps weight 1.0, so prf_q and the fitted alphas are untouched. """ names = [c for c in ch] if len(names) < 3: return {c: 1.0 for c in names} Z = {} for c in names: v = np.asarray(ch[c], float) sd = v.std() Z[c] = (v - v.mean()) / sd if sd > 0 else np.zeros_like(v) w = {} for c in names: others = [Z[o] for o in names if o != c] pool = np.mean(others, axis=0) if pool.std() <= 0 or Z[c].std() <= 0: w[c] = 1.0 continue r = float(np.corrcoef(Z[c], pool)[0, 1]) w[c] = 0.0 if (r == r and r < 0.0) else 1.0 return w _EVK: dict = {} # The right-hand sides of this table used to be put_on / put_into / # take_out / open / close - the hand-authored task vocabulary, so query # parsing hardwired the taxonomy and a corpus with different transitions # could not be asked about at all. # # There is no table now. A query names transition types the way it names # anything else: by SIMILARITY to what the corpus attested, resolved # against the store's own discovered types at query time. The closed- # class prepositions survive as SPATIAL RELATION cues (containment vs # support vs separation), which are geometry and hold in any domain - # a box in a drawer, a car in a lane, a pallet on a shelf. _REL_CUES = {"containment": ("into", "in", "inside", "within"), "support": ("on", "onto", "on top", "above", "over"), "separation": ("out of", "from", "off", "away")} def _relation_cue(tl): """Which spatial relation a query asks for, from prepositions alone. Prepositions are CLOSED-CLASS grammar, not corpus vocabulary: English gains new nouns and verbs constantly and new prepositions almost never. "into" encodes containment whether the thing entered is a drawer, a lane, a shelf or a shipping container. That is why this survives the no-hardwire rule while a verb table does not. Longest match wins so "on top" is not shadowed by "on". """ best, blen = None, 0 for rel, pats in _REL_CUES.items(): for p in pats: if p in tl and len(p) > blen: best, blen = rel, len(p) return best def _query_transitions(tl, store=None): """Transition kinds a query asks about, resolved against the types THIS CORPUS discovered - no table of verbs anywhere. WHAT WAS HERE BEFORE. A dict `_VERBK` mapped hand-authored verbs to hand-authored transition names, plus this: if "from the drawer" in tl or "out of" in tl: need = {"take_out"} | (need - {"put_on", "put_into"}) A literal furniture name and three invented task labels, matched against the raw query. The violations pass deleted _VERBK's and _PREPK's DEFINITIONS and left their USES, so this function has raised NameError on every query since - the events channel has been dead, silently, recorded only as `channel events failed` in a benchmark footer nobody had to read. WHAT IT DOES NOW. The corpus discovers transition types from geometry (transitions.FIELDS: displacement, direction, duration, scale, enclosure change, ...) and names none of them. A query names one the same way it names anything else - by the relation it asks for. The bridge is `enclosure_delta`, a physical measurement of whether the moved thing became more or less enclosed: containment ("into", "inside") -> enclosure rises separation ("out of", "from") -> enclosure falls support ("on", "onto") -> enclosure says nothing, so this gates nothing and the ranking channels decide The split is the corpus's own median, not zero: descriptors are standardised per corpus, so zero is an artefact of the scaler while the median is where this corpus actually divides. On a corpus of forklifts and pallets the same code returns that corpus's types. """ if store is None or "transition_types" not in store.tables(): return set() rel = _relation_cue(tl) if rel is None or rel == "support": return set() from .transitions import FIELDS t = store.table("transition_types").scan().to_pydict() if not t.get("type_id"): return set() ei = FIELDS.index("enclosure_delta") enc = np.array([float(c[ei]) for c in t["centroid"]], np.float32) ids = [int(i) for i in t["type_id"]] mid = float(np.median(enc)) sel = enc > mid if rel == "containment" else enc < mid return {f"t{ids[i]}" for i in range(len(ids)) if bool(sel[i])} def _demo_transitions(store, keys): """demo index -> set of transition kinds, cached per table version.""" ver = store.table("events").state().version ck = (str(store.dir), ver, "kinds") if ck in _EVK: return _EVK[ck] ev = store.table("events").scan().to_pydict() kidx = {(k[0], k[1]): i for i, k in enumerate(keys)} out = {} for r in range(len(ev["ts"])): i = kidx.get((str(ev["stream"][r]), int(ev["ts"][r]))) if i is not None: out.setdefault(i, set()).add(ev["kind"][r]) _EVK.clear() _EVK[ck] = out return out def _keysig(keys): """Cache identity for a key ORDER, not just a table version. The teacher passes (stream, t0, t1) and the student (stream, t0); both resolve to the same rows, but a row-aligned matrix cached under one ordering and served to the other would be a silent wrong answer rather than an error, so the ordering is part of the key.""" return (len(keys), keys[0][0], int(keys[0][1]), int(keys[-1][1])) def _motion_pool(store, keys): """Per-demo unit motion vector, cached per table version.""" ver = store.table("motion_vectors").state().version ck = (str(store.dir), ver, "pool", _keysig(keys)) if ck not in _EVK: from .embeddings import _vec_table tb, V = _vec_table(store, "motion_vectors") V = np.asarray(V, np.float32) em = {} for r, (s_, a_) in enumerate(zip(tb.column("stream").to_pylist(), tb.column("ts").to_pylist())): em.setdefault((str(s_), int(a_)), []).append(r) M = np.stack([V[em[(k[0], k[1])]].mean(0) if (k[0], k[1]) in em else np.zeros(V.shape[1], np.float32) for k in keys]) _EVK[ck] = M / (np.linalg.norm(M, axis=1, keepdims=True) + 1e-8) return _EVK[ck] def _auc(s, y): o = np.argsort(s) r = np.empty(len(s)) r[o] = np.arange(1, len(s) + 1) n1 = int(y.sum()) n0 = len(y) - n1 if n1 == 0 or n0 == 0: return 0.5 return float((r[y == 1].sum() - n1 * (n1 + 1) / 2) / (n1 * n0)) def _transition_anchors(store, keys): """A motion PROTOTYPE per transition kind, with a weight the kind has to earn on corpus statistics alone. WHY THIS EXISTS. The text tower cannot ask for a direction. Measured on bridge4h: cos(pe_text("opens the drawer"), pe_text("closes the drawer")) = 0.957, and after the student's query tower still 0.905, so the two queries return the SAME list - 221 of 248 shared - even though their truth sets are disjoint (0 episodes in common). Yet the direction itself is not missing from the store: held-out open-vs- close accuracy is 0.983 in motion space, against 0.65 in PE, 0.56 in SigLIP2 and 0.57 in IV2. The signal is on the video side and only the QUESTION is blind, so the corpus supplies what the sentence cannot: the query names a transition (closed-class English), and the episodes the store itself flagged with that transition define its direction, Rocchio-style against the complement. EARNING THE WEIGHT. A flat weight is wrong, and measurably so: at w=4 this takes q04 from 0.20 to 0.74, but it also drives q08 from 0.33 to 0.00, because 'close' is attested by 654 episodes and 'put_on' by 5, and a centroid over 5 events is noise. So each kind proves itself WITHOUT LABELS: split the class in half, build the prototype on one half, and measure whether it ranks the held-out half above the complement. Bootstrapped, with a 2-sigma lower bound, so a tiny class collapses to chance on its own variance rather than on a hand-set minimum count. Measured reliabilities: close 0.52, open 0.35, put_into 0.02, contact/release/adjust 0.06-0.07, and put_on / take_out exactly 0.00 - the two that did the damage. No truthset, no metadata, no per-dataset vocabulary: the transitions come from cavity/articulation geometry over pixels, and the only English involved is the same closed-class verb map already used to route the gate.""" if "motion_vectors" not in store.tables() or "events" not in \ store.tables(): return {} ck = (str(store.dir), store.table("events").state().version, store.table("motion_vectors").state().version, "anchor", _keysig(keys)) if ck in _EVK: return _EVK[ck] # kinds FIRST: _demo_transitions evicts the whole cache on a miss, # so building the pool before it would throw the pool away and make # every later query recompute it. kinds = _demo_transitions(store, keys) M = _motion_pool(store, keys) pos_of = {} for i, ks in kinds.items(): for k in ks: pos_of.setdefault(k, set()).add(i) n = len(keys) rng = np.random.default_rng(0) out = {} for kd, pos in pos_of.items(): p = np.array(sorted(pos)) comp = np.array(sorted(set(range(n)) - pos)) if len(p) < 4 or len(comp) < 4: continue a = M[p].mean(0) - M[comp].mean(0) a /= np.linalg.norm(a) + 1e-8 scores = [] for _ in range(16): q = rng.permutation(p) h = len(q) // 2 c = M[q[:h]].mean(0) - M[comp].mean(0) c /= np.linalg.norm(c) + 1e-8 pool = np.concatenate([q[h:], comp]) y = np.concatenate([np.ones(len(q) - h), np.zeros(len(comp))]) scores.append(_auc(M[pool] @ c, y)) lcb = float(np.mean(scores) - 2 * np.std(scores)) out[kd] = (a, max(0.0, 2.0 * (lcb - 0.5))) _EVK[ck] = out return out # How loud a fully-earned anchor is allowed to be, in units of the # current score's own spread. Swept 0-12 on the student pipeline: the # objective is flat across 6-12 (0.356 / 0.356 / 0.353) so 8 sits mid # plateau rather than on a spike. LOQO over the ten queries: 8 of 10 # folds choose it unprompted; honest held-out mean 0.323 against 0.298 # with the anchor off. Reliability already scales each kind, so this is # the only free constant the mechanism has. _ANCHOR_W = float(os.environ.get("ELIDEDB_ANCHOR_W", "8.0")) def _anchor_score(store, keys, need): """Corpus direction term for the transitions a query asks for, already scaled by how much each kind earned. Returns None when no asked-for kind is trustworthy, which is the common case for the thinly-attested ones.""" if not need: return None try: anc = _transition_anchors(store, keys) except Exception: return None M = None acc = None for kd in need: a_rel = anc.get(kd) if a_rel is None or a_rel[1] <= 0.0: continue if M is None: M = _motion_pool(store, keys) term = a_rel[1] * (M @ a_rel[0]) acc = term if acc is None else acc + term return acc # OFF BY DEFAULT — MEASURED NEGATIVE, kept as a reproduction. # Fusing this term cost yield at every weight tried: 0.45/0.32 at w=0, # 0.37/0.30 at 1, 0.28/0.25 at 2, 0.22/0.21 at 4. Only q10 ever gained # (0.00 -> 0.50, one document) while q05 collapsed 0.76 -> 0.02. # # The isolated signal was real - identity alone scores AUC 0.862 on # banana and 0.790 on eggplant - so what failed is the COMBINER, not # the evidence. Two faults, both visible in the numbers above: # 1. the conjunction below takes a min over nouns, so the weakest # noun governs, and "the drawer" appears in nearly every query # while matching nothing in particular - one noisy term drags the # whole demo down; # 2. tail concentration did not suppress the uninformative nouns # hard enough - 'a green object' still scored at 0.531 AUC and # still got a vote. # The fix is a per-noun trust gate of the kind the transition anchor # uses (earn it or score zero), not a bigger or smaller weight; a # sweep cannot rescue a term that is wrong on most queries. _NOUN_W = float(os.environ.get("ELIDEDB_NOUN_W", "0.0")) def _participant_match(store, keys, text): """Do this demo's PARTICIPANTS answer the nouns the query asks for? The transition anchor fixed direction; it cannot touch the queries that are about identity - lid, spoon, eggplant, banana - and those are exactly the ones still failing. Measured: for every failing query most or all of the truth already sits in the top 18% of the corpus (q07 all 12, q09/q10 both, q08 13 of 18), so recall is not the problem and ITM already reranks the top 150. What is missing is a reason to prefer one candidate over another, and identity is it. The store carries 5,079 named participant events with SigLIP2 name vectors, and identity alone scores AUC 0.862 on banana and 0.790 on eggplant - while scoring at or below chance on 'a green object' (0.531) and 'a red object' (0.466), because the namer writes 'black object' / 'white object' and the colour never matches. So the noun has to earn its say, and the corpus statistic that decides is concentration: a noun whose match mass spreads evenly over every demo distinguishes nothing (this is IDF, and it is ordinary IR rather than a prior about drawers), while one with a separated tail is naming something real. No labels, no per-dataset vocabulary - the names came from pixels.""" if "events" not in store.tables(): return None ck = (str(store.dir), store.table("events").state().version, "pnames", _keysig(keys)) if ck not in _EVK: ev = store.table("events").scan().to_pydict() NV = np.asarray(ev["name_vec"], np.float32) NV = NV / (np.linalg.norm(NV, axis=1, keepdims=True) + 1e-8) kidx = {(k[0], k[1]): i for i, k in enumerate(keys)} per = {} for r in range(len(ev["ts"])): if ev["name"][r]: i = kidx.get((str(ev["stream"][r]), int(ev["ts"][r]))) if i is not None: per.setdefault(i, []).append(r) _EVK[ck] = (NV, per) NV, per = _EVK[ck] if not per: return None from .sig2 import atoms_of, _text_vec nouns = list(atoms_of(text.lower())) if not nouns: return None terms = [] for nn in nouns: qv = np.asarray(_text_vec(nn), np.float32) qv /= np.linalg.norm(qv) + 1e-8 s = np.full(len(keys), np.nan, np.float32) for i, rows in per.items(): s[i] = float((NV[rows] @ qv).max()) ok = ~np.isnan(s) if ok.sum() < 8: continue v = s[ok] q01, q50, q99 = np.percentile(v, [1, 50, 99]) spec = float((q99 - q50) / (q99 - q01 + 1e-6)) # tail concentration z = np.zeros(len(keys), np.float32) z[ok] = (v - q50) / (v.std() + 1e-6) terms.append(spec * z) if not terms: return None # every asked-for noun must be answered: the weakest one governs return np.min(np.stack(terms), axis=0) def _participant_boost(store, keys, text, sc): if _NOUN_W <= 0: # see _NOUN_W: measured negative return sc, None t = _participant_match(store, keys, text) if t is None or float(np.std(t)) <= 0: return sc, None return sc + _NOUN_W * float(np.std(sc)) * (t / (np.std(t) + 1e-6)), t def _anchor_boost(store, keys, need, sc): """Add the direction term to a running score. One call site's worth of arithmetic, kept in one place so the teacher path and the student path cannot drift apart on it.""" a = _anchor_score(store, keys, need) if a is None: return sc, None return sc + _ANCHOR_W * float(np.std(sc)) * a, a def _motion_density(store, keys, fused, k_max): """Fraction of a demo's motion-space neighbours that the cheap ranking already ranks highly. Measured: of a true q04 episode's 10 nearest neighbours in motion space, 8.5 are also true (q05: 8.2) - the tightest cluster signal in the store, and unused until now. Appearance neighbourhoods do NOT have this property, which is why diffusing over them was a wash and why IVF cells over them once hid 45 of 46 positives.""" if "motion_vectors" not in store.tables(): return None ver = store.table("motion_vectors").state().version ck = (str(store.dir), ver, "nn") if ck not in _EVK: from .embeddings import _vec_table tb, V = _vec_table(store, "motion_vectors") V = np.asarray(V, np.float32) em = {} for r, (s_, a_) in enumerate(zip(tb.column("stream").to_pylist(), tb.column("ts").to_pylist())): em.setdefault((str(s_), int(a_)), []).append(r) M = np.stack([V[em[(k[0], k[1])]].mean(0) if (k[0], k[1]) in em else np.zeros(V.shape[1], np.float32) for k in keys]) M /= np.linalg.norm(M, axis=1, keepdims=True) + 1e-8 S = M @ M.T np.fill_diagonal(S, -9) _EVK[ck] = np.argsort(-S, axis=1)[:, :15] nn = _EVK[ck] top = set(np.argsort(-fused)[:max(k_max, 20)].tolist()) return np.array([len(top & set(nn[i].tolist())) / nn.shape[1] for i in range(len(keys))]) def _knee(sorted_desc): """Set boundary on the sorted fused curve. The raw largest-drop knee cut 183-episode classes to 4 (RRF consensus gives the top few an outsized gap): the boundary is now the last position whose score keeps a fixed fraction of the top-5 mean — scale-stable under RRF (scores are bounded sums of w/(60+rank)) — with the largest-drop knee only allowed to TIGHTEN it, never to cut inside the top-8.""" s = np.asarray(sorted_desc, float) if len(s) < 2: return len(s) top = float(np.mean(s[:min(5, len(s))])) floor_cut = int(np.searchsorted(-s, -0.62 * top, side="right")) d = s[:-1] - s[1:] knee = int(np.argmax(d[2:])) + 2 + 1 if len(d) > 2 else len(s) # NO minimum set size: the user's contract is "up to K, and # everything returned is true" — a forced floor delivers junk return min(floor_cut, knee) def search_set(store, text, purity="fast", k_max=400, audit_n=12, return_ranking=False, cfg_override=None): """The robotics query: ALL matching clips, purity-first, VLM-free. purity="fast" exact fused scan, direction filter, knee cut purity="audited" + geometric relational audit (SAM 3 boxes over time) on a stratified sample of the set; pruned at the last geometry-positive sample. Only fires when the query parses as a relation; abstains otherwise. """ from .fusion import rrf, variant_max from .grounding import parse_relation from .rerank import directional_swap t0 = time.perf_counter() keys = _episodes(store) # LEXICON ONLY beyond this point (no-hardwire rule): the antonym # swap and the parsed relation are dictionary knowledge; every # dataset-facing decision below is derived from the corpus at # query time. sq = directional_swap(text, store) rel = parse_relation(text) tl = text.lower() # CATEGORY-WORD EXPANSION, corpus-attested: WordNet supplies the # candidate hyponyms (dictionary), the store's SigLIP2 frame # space decides which exist HERE (data) — "vessel" starves the # text channels (measured: sup 247, prec 0.25) and SAM cannot # ground it; the specific attested terms can. try: from .vocab import corpus_variants variants = corpus_variants(store, text) except Exception: variants = [text] ch = {} # CHANNEL DEATH MUST BE LOUD (2026-07-28). Every block below used # to swallow its exception, so a broken dependency degraded search # silently: a transformers upgrade killed iv2 and the frozen bench # fell 0.38 -> 0.13 with no error anywhere, diagnosable only by # bisecting the environment. Failures are now recorded per channel # and returned in the result; the caller decides whether a # degraded answer is acceptable (bench_truth refuses to record a # ledger row, the Desk shows a warning). failed = {} def _fail(name, exc): failed[name] = f"{type(exc).__name__}: {exc}"[:200] try: from .pe import pe_lookup vs = [] for vtext in variants: look, _ = pe_lookup(store, vtext) vs.append(np.array([look(*k) for k in keys])) ch["pe"] = variant_max(vs) except Exception as e: _fail("pe", e) try: # act: text-mapped class weights onto the probe's own # vocabulary (contrast against the swap when one exists) — # embedding-derived, no coded class lists from .action_channel import act_lookup look, _ = act_lookup(store, text) ch["act"] = np.array([look(*k) for k in keys]) except Exception as e: _fail('act', e) try: from .sig2 import conj_lookup # sig2 (SigLIP2 appearance) is NOT scored: leave-one-out on the # frozen truthset measured its contribution at exactly zero # (35 true with it, 35 without), while pe and obj answer the same # "what does it look like" question. conj stays and is worth -5, # and it reads sig2_vectors, so the SigLIP2 pass is still paid at # INGEST - dropping the channel saves query work, not ingest. cl = conj_lookup(store, text) if cl is not None: ch["conj"] = np.array([cl(*k) for k in keys]) except Exception as e: _fail('sig2', e) try: # iv2: VIDEO-native text alignment (InternVideo2-Stage2 1B, # temporal modeling the frame-pooled channels lack) — the # 4 frames pass through the encoder together from .iv2 import iv2_lookup vs = [] for vtext in variants: look, _ = iv2_lookup(store, vtext) vs.append(np.array([look(*k) for k in keys])) ch["iv2"] = variant_max(vs) except Exception as e: _fail('iv2', e) try: from .vid import vid_lookup vs = [] for vtext in variants: look, _ = vid_lookup(store, vtext) vs.append(np.array([look(*k) for k in keys])) ch["vid"] = variant_max(vs) except Exception as e: _fail('vid', e) # OBJ channel — FastSAM crops matched against the query's noun # phrases (SigLIP space): the color/attribute binding the product # bench showed missing from the set path try: from .context import embed_texts from .objects import object_lookup nps = [p for p in ((rel[0], rel[2]) if rel else ()) if p and "object" not in p] if not nps: # ONE decomposition for the whole system: atoms_of's # closed-class boundaries (the raw regex here had the # same swallowed-preposition bug atoms_of was fixed for) from .sig2 import atoms_of nps = atoms_of(tl)[:2] if nps: olook = object_lookup(store, embed_texts(nps)) def _obj(k): osc, omo = olook(*k) return osc * (1.0 + omo) if osc == osc else float("nan") ch["obj"] = np.array([_obj(k) for k in keys]) except Exception as e: _fail('obj', e) # NOT a channel: late interaction over the patch grid. Within a # frame the grid separates cleanly (an eggplant frame scores 0.173 # for "an eggplant" against 0.063 for "a banana", where the pooled # vector manages 0.107 vs 0.071) and across episodes it ranks at # 0.08 standalone yield, last of ten, against pooled SigLIP2's 0.53. # MaxSim over 1,024 patches is an extreme-value draw: the episode # with the widest patch spread wins it whatever it contains. Eight # poolings were measured (max, top-k means, within-episode z) and # the best reached 0.15. scripts/patch_ingest.py + patches.py stay # as the reproduction. # NOT a channel: region identity from crops, measured dead at corpus # scale. The hypothesis was that obj failed only because its crops # were cut out of DOWNSCALED decodes. Recut at native 640x480 along # common-fate tracks (scripts/track_ingest.py, 25,335 crops) it looked # strong on an 80-episode pool — yield@10 of 1.00/1.00/0.75/0.50 on # q10/q09/q00/q08 — and collapsed against the real 1,121: banana's two # true episodes rank 280th and 534th, spoon's 21/38/49, standalone # mean yield 0.08 against obj's 0.14 and iv2's 0.24. The pool was 14x # easier and the separation was its artifact. Resolution was not the # blocker; a crop asks a small patch what it is with the context that # would answer the question cropped away. # CONTRAST channels — direction EVIDENCE, computed only when the # lexicon yields a swap. Architectural principle replacing every # hand routing rule (ledger-derived, now task-free): contrast # channels FILTER, content channels ORDER — a contrast score says # "more like the query than its opposite", never "relevant". contrast_ch = {} if sq is not None: if "act" in ch: contrast_ch["act"] = ch["act"] try: from .context import embed_texts from .motion import motion_lookup qv2 = embed_texts([text, sq]) mlook = motion_lookup(store, qv2[0], qv2[1]) contrast_ch["mot"] = np.array([mlook(*k) for k in keys]) ch["mot"] = contrast_ch["mot"] except Exception as e: _fail('mot', e) try: # SELF-RECOGNIZED contrast: Rocchio anchors in the # domain-general video-native space — the corpus itself # defines what this direction looks like here. No class # names, works unchanged on any domain. from .pe import pe_lookup from .prf import prf_contrast look, _ = pe_lookup(store, sq) pe_swap = np.array([look(*k) for k in keys]) contrast_ch["prf"] = prf_contrast( store, keys, ch.get("pe"), pe_swap) ch["prf"] = contrast_ch["prf"] except Exception as e: _fail('prf', e) # ABLATION HOOK: drop channels by name to measure what each one is # actually worth. Reads the env so the live path is untouched when # unset, and so an ablation runs through the SAME code as production # rather than a reimplementation of it. import os as _os _drop = {c.strip() for c in _os.environ.get("ELIDEDB_DROP_CHANNELS", "").split(",") if c.strip()} if _drop: for c in _drop: ch.pop(c, None) contrast_ch.pop(c, None) directional = sq is not None weights = corroboration_weights(ch) filter_q = 1 / 3 fnames = None # None => legacy: every contrast channel cut_alpha = 0.0 # 0 => fill to k_max (legacy, pre-cut) nms_r = 0 # 0 => no temporal event dedup (legacy) prf_n, prf_w = 25, 3.0 # feedback depth and voice, both fitted from pathlib import Path sw = Path(store.dir) / "_set_weights.json" try: if cfg_override is not None: # honest evaluation: weights fitted WITHOUT this query cfg = dict(cfg_override) elif sw.exists(): # FITTED roles (scripts/fit_set_weights.py): ordering # weights for every channel INCLUDING contrasts, plus the # filter quantile — coordinate ascent on the truthset, # LOQO-validated. Data-derived per store; the no-hardwire # rule's answer to hand role rules (the fit independently # rediscovered mot=0-in-ordering). cfg = json.loads(sw.read_text()) if cfg is not None: wk = ("set_weights_dir" if directional and "set_weights_dir" in cfg else "set_weights") fk = ("filter_quantile_dir" if directional and "filter_quantile_dir" in cfg else "filter_quantile") weights = {c: float(cfg[wk].get(c, 1.0)) for c in ch} filter_q = float(cfg.get(fk, 1 / 3)) ck = ("filter_channels_dir" if directional and "filter_channels_dir" in cfg else "filter_channels") if ck in cfg: fnames = list(cfg[ck]) ak = ("cut_alpha_dir" if directional and "cut_alpha_dir" in cfg else "cut_alpha") cut_alpha = float(cfg.get(ak, 0.0)) rk = ("nms_r_dir" if directional and "nms_r_dir" in cfg else "nms_r") nms_r = int(cfg.get(rk, 0)) else: cfg = json.loads((Path(store.dir) / "_channel_weights.json").read_text()) learned = (cfg.get("weights_dir", cfg.get("weights", {})) if directional else cfg.get("weights", {})) weights = {c: float(learned.get(c, 1.0)) for c in ch} except Exception: pass fused = rrf(ch, weights=weights) # PSEUDO-RELEVANCE FEEDBACK (Rocchio, and it is measured, not # assumed). The top of the first fused list is the best available # description of what the user actually meant; its centroid in # appearance space re-scores the corpus and rejoins the fusion as # one more voter. Two rounds of channel work bought nothing here - # per-query weights from score-distribution shape (0.59 vs 0.60 # global) and the spectral meta-learner's label-free reliability # estimate (0.56) both LOST - while this, the oldest trick in IR, # is the only thing that moved the metric: mean yield 0.60 -> 0.61 # at k=100, carried by the queries with real support (q03 0.60 -> # 0.67, q07 0.83 -> 0.92, q08 0.50 -> 0.56). It targets recall, # which is what true/min(k, support) rewards. try: from .embeddings import _vec_table _tb, _V = _vec_table(store, "pe_vectors") _V = np.asarray(_V, np.float32) _rmap = {} for _i, (_s, _a) in enumerate(zip(_tb.column("stream").to_pylist(), _tb.column("ts").to_pylist())): _rmap.setdefault((str(_s), int(_a)), []).append(_i) _ev = np.stack([_V[_rmap[(s_, a_)]].mean(0) if (s_, a_) in _rmap else np.zeros(_V.shape[1], np.float32) for s_, a_, _b in keys]) _ev /= np.linalg.norm(_ev, axis=1, keepdims=True) + 1e-8 _seed = np.argsort(-fused)[:prf_n] _c = _ev[_seed].mean(0) _c /= np.linalg.norm(_c) + 1e-8 ch["prf_q"] = _ev @ _c weights["prf_q"] = float(weights.get("prf_q", prf_w)) fused = rrf(ch, weights=weights) except Exception as e: _fail('prf_q', e) # ITM CASCADE — the cross-encoder rerank, cost-gated by depth. # Measured at k=1.5xsupport: shipped RRF 0.29/0.23, this store's # cosine ensemble under z-fusion 0.32/0.21, ITM alone 0.38/0.25, # cosine+ITM 0.40/0.27. Reranking the top-N of the cheap ranking # reaches the full-scan number exactly (N=500 -> 0.40/0.27), so ITM # enters as EVIDENCE INSIDE A CANDIDATE SET (L7), never a corpus # scan: its vision pass is 0.4s/episode and its tokens are 3.24 GB # corpus-wide (unpoolable - 4x reduction drops rank correlation to # 0.18), so a scan is neither affordable nor storable. # Off by default: ELIDEDB_ITM=1 enables, because the cost is real. import os as _os2 if _os2.environ.get("ELIDEDB_ITM") == "1": try: from .itm import itm_scores, rerank_depth n_re = rerank_depth(k_max, len(keys)) cand = np.argsort(-fused)[:n_re] sc = itm_scores(store, text, [keys[i] for i in cand]) if np.isfinite(sc).any(): zc = np.zeros(len(keys)) ok = np.isfinite(sc) v = sc[ok] zc[cand[ok]] = (v - v.mean()) / (v.std() + 1e-9) # SCORE-DISTRIBUTION-PRESERVING RESCORE. Writing # z-scores into `fused` broke the confidence cut, which # is fitted against RRF's own scale (bounded sums of # w/(60+rank)): q03 returned 83 of a 371 ceiling, # yield 0.87 -> 0.28. The cut is downstream and must # keep seeing the distribution it was fitted on, so the # rerank PERMUTES the candidates and hands back the # same sorted score values in the new order. Order # changes, scale does not. zf = fused[cand] zf = (zf - zf.mean()) / (zf.std() + 1e-9) new = np.argsort(-(zf + zc[cand])) f = fused.copy() f[cand[new]] = np.sort(fused[cand])[::-1] fused = f ch["itm"] = zc except Exception as e: _fail('itm', e) # EVENT-STRUCTURE STAGE, self-routed. Two mechanisms measured on # the oracle-cut metric min(yield, prec): # event filter q04 0.42 -> 0.57, q05 0.38 -> 0.42, but it # DESTROYS q00 0.33 -> 0.07 and q01 0.33 -> 0.22, # because their required transition (put_into) is # one the extractor assigns rarely and wrongly. # motion density q04 -> 0.53, and together 0.66 - over the 0.60 # bar for the first time on any query. # Applied globally the mean FALLS (0.33 -> 0.32); routed, it only # fires where the structure agrees with the ranking. The gate is # unsupervised: if the top candidates the cheap ranking already # likes mostly carry the required transition, the event evidence # and the ranking corroborate each other and the filter is trusted; # if they disagree, the extractor is wrong about this query type # and its opinion is discarded. No labels, no per-query constants. try: if "events" in store.tables(): from .itm import _S as _itm_unused # noqa: F401 need = _query_transitions(tl, store) # membership first (DOES this episode have the transition), # then direction (which WAY it went). Membership alone is # weak - 'close' covers 58% of this corpus, 'open' 80% - so # the mask can gate but cannot rank the two apart. fused, anc = _anchor_boost(store, keys, need, fused) if anc is not None: ch["anchor"] = anc fused, pmt = _participant_boost(store, keys, text, fused) if pmt is not None: ch["pname"] = pmt if need: kinds = _demo_transitions(store, keys) have = np.array([1.0 if (kinds.get(i) or set()) & need else 0.0 for i in range(len(keys))]) head = np.argsort(-fused)[:max(k_max, 20)] agree = float(have[head].mean()) if agree >= 0.35: fused = fused + 0.5 * have * float(np.std(fused)) ch["evk"] = have dens = _motion_density(store, keys, fused, k_max) if dens is not None: fused = fused + 0.5 * dens * float(np.std(fused)) ch["dens"] = dens except Exception as e: _fail('events', e) # NO-MATCH GATE, self-recognized: map the query onto the action # probe's OWN vocabulary by embedding similarity (no hand verb # list) and ask whether ANY episode in this corpus expresses those # classes above noise. Measured separation on bridge4h: absent # actions max 0.008-0.021 (fold/tear/throw) vs present 0.12-0.93; # threshold 0.05 sits in the gap. (A PE-cosine z-gate could not # separate — fold z 2.2 ranked ABOVE lid z 1.9.) # The gate reads `action_probs` directly, outside the per-channel # try/except above, so on a store without that channel it raised and # killed the whole query - `search_set` on lake/bridge4h died in # _auto_action_support rather than answering with the channels it # did have. The gate is an OPTIONAL refinement: its absence should # cost the no-match check, not the search. # # Recorded, not swallowed, per this function's own rule. `act_gate` # carries no fitted weight, so it lands in `degraded` and # bench_truth still refuses to write a ledger row for the run - a # degraded answer stays usable and stays visibly degraded. try: gate = _auto_action_support(store, text) except Exception as e: _fail("act_gate", e) gate = None if gate is not None and gate["max_p"] < 0.05: ms = (time.perf_counter() - t0) * 1e3 return {"clips": [], "borderline": [], "audit": None, "no_match": True, "reason": gate, "direction_filtered": 0, "channels": sorted(ch), "channels_failed": failed, "degraded": sorted(failed), "scored": len(keys), "ms": round(ms, 1)} # DIRECTION HARD FILTER — QUANTILE, NOT SIGN (AUC-validated # channels have uncalibrated zero points; a sign test executed # 130/183 true closes). Shared with the fitter via setpath.py: the # knee/obj-boost divergences of the acceptance sprint (fit LOQO # 0.21 vs live 0.16) were measured regressions from the live path # reshaping what the fit optimized — one code path kills the class. # FITTED VETO AUTHORITY: which channels filter is a per-store # learned artifact, not code. For binding queries this lets conj # act as a hard constraint (each query atom must find its own # frame evidence) instead of a drowned RRF vote — consensus # fusion structurally outvotes a decisive minority channel # (Cormack et al. 2009), and bag-of-concepts encoders cannot # rank binding (Winoground/ARO), so the constraint must prune. from .setpath import (confidence_cut, event_positions, filter_mask, nms_keep) fsrc = dict(ch) fsrc.update(contrast_ch) if fnames is None: fnames = list(contrast_ch) alive = (filter_mask(fsrc, fnames, filter_q) if fnames else np.ones(len(keys), bool)) dropped = int((~alive).sum()) idx = np.where(alive)[0] order = idx[np.argsort(-fused[idx])] if nms_r > 0: # temporal event dedup (fitted radius, shared with the fit): # duplicates of one event give way to the next-ranked # DISTINCT events — the product wants each true event once sid, pos = event_positions(keys) order = nms_keep(order, sid, pos, nms_r) # when roles are FITTED, the boundary is part of the fitted # configuration: the set ends where fused confidence drops below # the fitted alpha x the query's own top mass (setpath. # confidence_cut, shared with the fit — the hand knee here was a # measured fit-live divergence: fit LOQO 0.21 vs live 0.16). The # product ratio is true/returned -> returned/support; padding to # k_max buys yield with junk, and the fitted alpha prices that # trade on the truthset instead of a fixed count. # ALWAYS the confidence cut. This used to read # confidence_cut(...) if sw.exists() else min(_knee(...), k_max) # and NO STORE HAS A FITTED WEIGHTS FILE - it is produced by # fit_set_weights.py, which fits on the truthset and so cannot ship. # Every query therefore took the `else`: `_knee`, the rule this # module's own docstring records as broken - "the biggest gap on an # RRF curve is at the very top, so it returned 3 clips for a query # with 196 true episodes". Measured on fresh_bench it returned 6, 15 # and 12 clips for supports of 247, 165 and 196, while the # unsupervised cut sitting unreachable behind the branch selected # 194, 156 and 184 - i.e. the right answer was already being # computed and thrown away. # # With cut_alpha 0 (no fitted alpha, the only shippable state) # confidence_cut delegates to the median+3*MAD outlier test, which # needs no evaluation data: matches are the episodes scoring unlike # the corpus background, however many that turns out to be. cut = confidence_cut(fused[order], cut_alpha, k_max) chosen = order[:cut] borderline = order[cut:cut + 20] audit = None # The audit stays TIER-2 (opt-in): running it by default was # measured 2026-07-27 — prec +0.01, yield -0.07, with the damage # concentrated where X is a category word ("vessel") whose # corpus-attested variants ground spurious objects and the # geometry kills then execute true clips. With the quorum rule # and clean relation phrases it is far safer than before, but the # fast tier's fused index is the better default by the ledger. if purity == "audited" and len(chosen) > 0 and rel is not None: try: audit, keep_mask = _binding_audit(store, rel, [keys[i] for i in chosen]) except Exception as e: # deployments without the tracker stack keep the fused set # and SAY so rather than failing the query audit = {"unavailable": f"{type(e).__name__}"} keep_mask = None if keep_mask is not None: killed = chosen[~keep_mask] borderline = np.concatenate([killed, borderline]) chosen = chosen[keep_mask] ms = (time.perf_counter() - t0) * 1e3 return { "clips": [{"stream": keys[i][0], "t0": keys[i][1], "t1": keys[i][2], "score": float(fused[i])} for i in chosen], "borderline": [{"stream": keys[i][0], "t0": keys[i][1], "t1": keys[i][2]} for i in borderline], "audit": audit, "direction_filtered": dropped, "channels": sorted(ch), # a channel the fitted configuration gives ordering or filter # authority to, that failed to compute: the answer is degraded # and the caller must be able to see it (see `failed` above) "channels_failed": failed, "degraded": sorted(c for c in failed if abs(weights.get(c, 1.0)) > 0 or c in (fnames or ())), "scored": len(keys), "ms": round(ms, 1), # full fused ordering, for diagnosis: it separates "the ranking # never found the true episodes" from "it found them and the cut # refused to return them" - two failures with opposite fixes. **({"ranking": [(keys[i][0], keys[i][1], float(fused[i])) for i in order]} if return_ranking else {}), # PER-CHANNEL scores, aligned with `ranking`. "the ranking is # weak" is not an actionable finding - the fusion has seven # inputs and they can fail independently. Measured against the # truthset these give a per-channel AUC, which says WHICH input # to fix instead of leaving the whole ranking as the suspect. # Diagnostic only: same arrays the fusion already computed, no # extra work, and only materialised when asked for. **({"contrast_scores": {c: [float(v[i]) for i in order] for c, v in contrast_ch.items()}} if return_ranking else {}), **({"channel_scores": {c: [float(v[i]) for i in order] for c, v in ch.items()}, "channel_weights": {c: float(weights.get(c, 1.0)) for c in ch}} if return_ranking else {}), } # closed-class color words -> OpenCV hue bands (H in 0..180); S/V # floors exclude gray/white. Deterministic pixel evidence from the # tracked masklet — no model, no metadata, fully explainable. _HUE = {"red": [(0, 10), (170, 180)], "orange": [(10, 20)], "yellow": [(20, 33)], "green": [(35, 85)], "blue": [(95, 130)], "purple": [(130, 165)], "pink": [(150, 175)]} def _mask_color_frac(store, stream, t0, t1, tr_x, color): """Fraction of the tracked X masklet's pixels in the color band, measured on the MOVER identity at its most confident frame (the object that acted — a static, genuinely-green bystander must not vouch for a clip where the yellow cheese did the moving).""" import cv2 import pyarrow.compute as pc from .video import FrameSet if tr_x.get("mover_mask") is not None: best_f = int(tr_x["mover_frame"]) m0 = tr_x["mover_mask"] else: best_f = int(np.argmax(tr_x["presence"])) m0 = tr_x["masks"][best_f] if m0 is None: return None frames_tbl = store.table("frames").scan() sel = frames_tbl.filter(pc.and_( pc.equal(frames_tbl.column("stream"), stream), pc.and_(pc.greater_equal(frames_tbl.column("ts"), t0), pc.less_equal(frames_tbl.column("ts"), t1)))) if len(sel) < 4: return None n = len(tr_x["presence"]) pick = np.linspace(0, len(sel) - 1, n).round().astype(int) dec = FrameSet(store, "frames", sel.take(pick[best_f:best_f + 1])).decode(width=480) if not dec: return None img = sorted(dec)[0][1] m = m0 if m.shape != img.shape[:2]: m = cv2.resize(m.astype(np.uint8), (img.shape[1], img.shape[0])) > 0 if m.sum() < 20: return None hsv = cv2.cvtColor(img, cv2.COLOR_RGB2HSV) h, s, v = hsv[..., 0][m], hsv[..., 1][m], hsv[..., 2][m] ok = np.zeros(len(h), bool) for lo, hi in _HUE[color]: ok |= (h >= lo) & (h <= hi) ok &= (s > 60) & (v > 50) return float(ok.mean()) def _binding_audit(store, rel, clip_keys): """SAM 3.1 tracker BINDING audit on every returned clip: does the queried OBJECT actually appear, and does it engage the LANDMARK? Division of labor fixed by measurement: DIRECTION belongs to the index channels (mot 0.98 open/close, act 0.889 put/take, free); the tracker's occlusion-persistent memory made it a poor direction instrument (AUC 0.643) but a robust IDENTITY one — exactly the binding failures the product bench showed (wrong colors, wrong objects). A clip is killed only on positive evidence of absence: the X masklet never appears, or X and Y masklets never come near each other. Tracker failure on a clip = abstain = keep. """ from .grounding import _ioa from .sam3x import track_concepts def _concrete(p): """None for placeholder-only phrases ('the object'); an ATTRIBUTE-bearing phrase ('a green object') is groundable — the blanket 'object'-in-phrase test silently skipped the whole X audit on every attribute query (debug-caught: any_moved and color fracs were real, the audit just never asked).""" if not p: return None words = [w for w in p.split() if w not in ("a", "an", "the", "object", "objects", "something", "thing")] return p if words else None x, _, y = rel x, y = _concrete(x), _concrete(y) if x is None and y is None: return None, None def _variants(p): if p is None: return [None] try: from .vocab import corpus_variants return corpus_variants(store, p) except Exception: return [p] keep = np.ones(len(clip_keys), bool) checked = killed_absent = killed_disjoint = abstained = 0 killed_static = killed_wrong_color = 0 absent_idx = [] for i, (s, a, b) in enumerate(clip_keys): # try phrase variants until X grounds (category words like # "vessel" ground as pot/pan/bowl); first grounding wins tr = None for xv in _variants(x): phrases = [p for p in (xv, y) if p] try: trv = track_concepts(store, s, a, b, phrases, n_frames=8) except Exception: trv = None if trv is None: continue if tr is None: tr, xg = trv, xv if xv is None or max(trv[xv]["presence"]) >= 0.5: tr, xg = trv, xv break if tr is None: abstained += 1 continue # xk = the phrase key that grounded for THIS clip (x itself # stays loop-invariant — an earlier version mutated it and # corrupted later iterations' variant lists) xk = xg if x is not None else None checked += 1 if xk is not None and max(tr[xk]["presence"]) < 0.5: keep[i] = False killed_absent += 1 absent_idx.append(i) continue # THE MANIPULATED-OBJECT TEST: SOME instance of the queried X # must MOVE. "a green object" grounds on any green thing in # the scene (audit-bench-caught: zero kills on the green/ # yellow sets); the query is about the object being ACTED ON, # and that one travels. any_moved spans ALL tracked identities # so a second, static instance never executes a true clip. if xk is not None and not tr[xk].get("any_moved", True): keep[i] = False killed_static += 1 continue # COLOR CHECK, pixel-level: SAM 3's presence token accepts a # yellow-green cheese for "a green object" (audit-bench-caught, # zero kills on attribute queries) — but the masklet hands us # the object's PIXELS, and color words are closed-class. The # object claimed as must actually be . color = next((c for c in _HUE if xk and c in xk), None) if color is not None: frac = _mask_color_frac(store, s, a, b, tr[xk], color) if frac is not None and frac < 0.25: keep[i] = False killed_wrong_color += 1 continue if xk is not None and y is not None \ and max(tr[y]["presence"]) >= 0.5: near = False for f in range(len(tr[xk]["boxes"])): bx, by = tr[xk]["boxes"][f], tr[y]["boxes"][f] if bx is None or by is None: continue # engagement: overlap, or gap under half of X's size if _ioa(bx, by) > 0.02: near = True break gap = max(by[0] - bx[2], bx[0] - by[2], by[1] - bx[3], bx[1] - by[3]) if gap < 0.5 * max(bx[2] - bx[0], bx[3] - bx[1]): near = True break if not near: keep[i] = False killed_disjoint += 1 # GROUNDING-RELIABILITY QUORUM (generalizes the old 100%-absent # rule): absence is only evidence when the phrase grounds in at # least half the checked clips. A phrase the detector cannot find # ("a vessel": grounded 1/10, one spurious static bottle — the # single grounding defeated the 100% rule and the audit executed # a 9/10-true set, measured) indicts the GROUNDING, not the # clips: revert only the absent kills. Kills where grounding # SUCCEEDED (static/color/disjoint) always stand. grounded = checked - killed_absent ungroundable = (checked > 0 and grounded < killed_absent) if ungroundable: for j in absent_idx: keep[j] = True killed_absent = 0 return ({"checked": checked, "killed_absent": killed_absent, "killed_disjoint": killed_disjoint, "killed_static": killed_static, "killed_wrong_color": killed_wrong_color, "abstained": abstained, "ungroundable": ungroundable, "x": x, "y": y}, keep)