From d7a941fc520eb5f1938af50302c100f18555a899 Mon Sep 17 00:00:00 2001 From: Jehan Azad Date: Mon, 7 Sep 2026 06:55:41 +0000 Subject: [PATCH] Adopt pool nodes a dark solve can vouch for MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The association round routinely pairs more nodes for an aircraft than the input the solver is handed uses. pool_n_nodes (the node set of the shared-track component the input was clustered out of) made that measurable, and 21 minutes of live instrumentation put 72% of published dark solves below their pool, mean shortfall 2.27 nodes: published n=2 with pool>=3 was 28 of 45, published n=3 with pool>=4 was 125 of 169. The expensive half is at the bottom — 390 of 481 rejected n=2 candidates had a pool of 3 or more, and an n=3 candidate publishes 81% of the time against 6% for n=2. That is the largest single block of detections this pipeline throws away. Merging harder upstream is not the fix: sweeping the position merge radius to 6 km raised cross-aircraft contamination from 25% to 32%, because at that radius two aircraft are as close together as one aircraft's own pairings. So the solve itself vouches for the extra node instead. _adopt_pool_nodes predicts, through the association stage's own forward model (predict_observation), what each pool node should have measured at the position and velocity just solved, adopts the ones whose measured delay agrees within 6.0 µs and Doppler within 60 Hz, re-solves wider from the first solve's position, and keeps the wider solve only if it passes the usual rms_delay gate and has not walked more than 5 km from the narrow one. With two or more adopted nodes a failed rms gets one retry without the worst-residual adoption, because the rms is a sum over both and a single contaminated adoption should not sink a widening the other node carried; beyond that the narrow solve stands. Cost is bounded at two extra LM solves per eligible candidate. The 6.0 µs default is derived, not tuned: an n=2 dark solve sits ~2 km from truth and 2 km of range error is 2/c ≈ 6.7 µs of bistatic delay, so the gate admits a genuine node at the accuracy this stage actually has while an unrelated target is normally tens of µs away. A node with no registered geometry, or a pool measurement whose node cannot be predicted for, is not adopted rather than adopted unchecked. It runs immediately after the first successful solve and BEFORE trimming and before every gate, so a widened candidate is judged as an n=3 solve throughout — including by the n=2 confirmation gate, which no longer applies to it. That is the point rather than a side effect: the gate exists because two nodes cannot corroborate each other's identity, and a third node whose measured delay matches what the two-node solve predicts for it is exactly the corroboration it was demanding. Only bottom-up dark inputs qualify; anchored and known-lane inputs carry an identity and were never clustered, so they have no pool. Every history record carries adopt_meta (pool size, candidates, adopted node ids, outcome) and n_nodes_pre_adopt, so adoption's true and false positives can be read from the dump rather than inferred from four counters; the counters (solver_adopt_eligible / _widened / _nodes_added / _rejected) are exposed in the solver-stats block. SOLVER_ADOPT_POOL=0 disables the stage outright. Co-Authored-By: Claude Fable 5.1 --- backend/core/state.py | 16 + backend/routes/test.py | 8 + backend/services/tasks/solver.py | 215 ++++++++++++- backend/tests/test_solver_pool_adoption.py | 353 +++++++++++++++++++++ backend/tests/test_solver_stats.py | 8 + libs/retina-analytics | 2 +- 6 files changed, 600 insertions(+), 2 deletions(-) create mode 100644 backend/tests/test_solver_pool_adoption.py diff --git a/backend/core/state.py b/backend/core/state.py index 336930aa..d95c13a2 100644 --- a/backend/core/state.py +++ b/backend/core/state.py @@ -719,6 +719,20 @@ def _adsb_for_seeding() -> dict[str, dict]: # here, not to replace it. solver_resolve_refresh: int = 0 +# Pool adoption (solver.py's _adopt_pool_nodes). A dark candidate solved with +# fewer nodes than the association round paired for it is "eligible"; when the +# extra node's measured delay/Doppler agree with what the narrow solve +# predicts for that node it is adopted and the candidate re-solved wider. +# Live baseline the stage was built against: 72% of published dark solves sit +# below their pool (mean shortfall 2.27 nodes), and 390 of 481 rejected n=2 +# candidates had a pool of 3+. Read widened/eligible as the hit rate and +# nodes_added/widened as the average width bought; rejected counts eligible +# candidates where nothing passed the gates or the wider solve was refused. +solver_adopt_eligible: int = 0 +solver_adopt_widened: int = 0 +solver_adopt_nodes_added: int = 0 +solver_adopt_rejected: int = 0 + # The last few hundred resolve-slot skips, with the claims that blocked them. # Deliberately NOT the solve-history deque: a skip is not a solve outcome, and # writing one record per skip into mlat_solve_history would evict the real @@ -991,6 +1005,7 @@ def _reset_for_tests() -> None: global ws_send_timeouts global solver_pool_timeouts global solver_resolve_skips_dark, solver_resolve_refresh + global solver_adopt_eligible, solver_adopt_widened, solver_adopt_nodes_added, solver_adopt_rejected global mn_superseded, mn_superseded_blocked, mn_superseded_blocked_alt, solver_trimmed global solver_consensus_selected, solver_consensus_filtered global solver_consensus_fallback, solver_consensus_shadow @@ -1093,6 +1108,7 @@ def _reset_for_tests() -> None: tracks_stale_skipped = solver_epoch_align_skipped = 0 solver_stale_drops = 0 solver_resolve_skips = solver_resolve_skips_dark = solver_resolve_refresh = 0 + solver_adopt_eligible = solver_adopt_widened = solver_adopt_nodes_added = solver_adopt_rejected = 0 mn_superseded = mn_superseded_blocked = mn_superseded_blocked_alt = 0 solver_trimmed = 0 solver_consensus_selected = solver_consensus_filtered = 0 diff --git a/backend/routes/test.py b/backend/routes/test.py index 066c7873..cf0a0400 100644 --- a/backend/routes/test.py +++ b/backend/routes/test.py @@ -1520,6 +1520,14 @@ def _solver_window_stats(minutes: float) -> dict: # that an entry nothing else refreshes stops dead-reckoning the # whole 12 s window. "resolve_refresh": state.solver_resolve_refresh, + # Pool adoption (solver.py's _adopt_pool_nodes): dark candidates + # solved narrower than the round's node pool, how many were + # re-solved wider once the narrow solve vouched for the extra + # node's delay/Doppler, and how many node-measurements that added. + "adopt_eligible": state.solver_adopt_eligible, + "adopt_widened": state.solver_adopt_widened, + "adopt_nodes_added": state.solver_adopt_nodes_added, + "adopt_rejected": state.solver_adopt_rejected, "queue_drops": state.solver_queue_drops, # Frames the per-node rate limiter refused before the tracker ever # saw them (tcp_handler's NODE_FRAME_MIN_INTERVAL_S). Not the diff --git a/backend/services/tasks/solver.py b/backend/services/tasks/solver.py index ba7992c4..2ee849c9 100644 --- a/backend/services/tasks/solver.py +++ b/backend/services/tasks/solver.py @@ -11,7 +11,7 @@ from collections import deque from concurrent.futures.process import BrokenProcessPool -from retina_analytics.association import _point_in_beam +from retina_analytics.association import _point_in_beam, predict_observation from config.constants import ( ARC_ONLY_ANOMALY_ALLOWLIST, @@ -684,6 +684,207 @@ def _trim_and_resolve( return result, s_in, trim_meta +# ── Pool adoption (dark bottom-up) ─────────────────────────────────────────── +# The association round often pairs more nodes for an aircraft than the input +# the solver is handed uses: pool_n_nodes is the node set of the shared-track +# component the input was clustered out of (InterNodeAssociator. +# _shared_track_pools), and 21 minutes of live instrumentation put 72% of +# published dark solves below it, mean shortfall 2.27 nodes. The costly half +# is at the bottom: 390 of 481 rejected n=2 candidates had a pool of 3+, and +# an n=3 candidate publishes 81% of the time against 6% for n=2. That is the +# largest single block of detections the pipeline throws away. +# +# The fix is NOT to merge more aggressively upstream — sweeping the position +# merge radius to 6 km raised cross-aircraft contamination from 25% to 32%, +# because at that radius two aircraft are as close as one aircraft's own +# pairings. Instead the solve itself vouches for the extra node: predict what +# that node should have measured at the position and velocity just solved, and +# adopt it only if what it actually measured agrees. A node that agrees on +# delay AND Doppler at an independently solved point is evidence about the +# same aircraft, which is precisely what the n=2 confirmation gate is asking +# for and cannot get from two nodes alone. +# +# Gate defaults. An n=2 dark solve sits ~2 km from truth, and 2 km of range +# error is 2/c ≈ 6.7 µs of bistatic delay, so 6.0 µs admits a genuine node at +# the accuracy this stage actually has without opening the door to an +# unrelated target (a wrong aircraft is normally tens of µs away). 60 Hz on +# Doppler is the same order as the n=2 velocity error against a ~600 MHz +# carrier. Both are env-tunable, and SOLVER_ADOPT_POOL=0 turns the whole +# stage off. +_ADOPT_POOL_ENABLED = os.getenv("SOLVER_ADOPT_POOL", "1").strip().lower() not in ("0", "false", "off") +_ADOPT_DELAY_GATE_US = float(os.getenv("SOLVER_ADOPT_DELAY_GATE_US", "6.0")) +_ADOPT_DOPPLER_GATE_HZ = float(os.getenv("SOLVER_ADOPT_DOPPLER_GATE_HZ", "60")) +# A widened solve that lands more than this from the narrow one is not the +# same aircraft solved better — it is a different geometry the adopted node +# dragged the fit into, and keeping it would publish a position no measurement +# in the original input supports. +_ADOPT_MAX_JUMP_KM = float(os.getenv("SOLVER_ADOPT_MAX_JUMP_KM", "5.0")) + + +def _s_in_with_adopted(s_in: dict, adopted: list[dict], first: dict) -> dict: + """s_in plus the adopted pool measurements, ready to re-solve. + + Pure — s_in is never mutated, so a rejected widening leaves the caller + holding the untouched input. The initial guess becomes the FIRST solve's + position rather than the association grid centroid it came in with: the + narrow solve is the best estimate anything has of where this aircraft is, + and it is also the point the adopted measurements were just gated against, + so starting anywhere else would judge them from a different place than the + one that admitted them. + """ + s_wide = dict(s_in) + meas = [dict(m) for m in (s_in.get("measurements") or [])] + meas.extend({k: m.get(k) for k in ("node_id", "delay_us", "doppler_hz", "snr", "t_s")} for m in adopted) + s_wide["measurements"] = meas + s_wide["n_nodes"] = len({m.get("node_id") for m in meas}) + # Provenance follows the measurement: a published solve names the tracklets + # it was built from, and an adopted node's track is now one of them. + by_node = {nid: list(ids) for nid, ids in (s_in.get("track_ids_by_node") or {}).items()} + for m in adopted: + tid = m.get("track_id") + if tid and tid not in by_node.setdefault(m["node_id"], []): + by_node[m["node_id"]].append(tid) + s_wide["track_ids_by_node"] = {nid: sorted(ids) for nid, ids in by_node.items()} + s_wide["track_ids"] = sorted(set(s_in.get("track_ids") or []) | {t for ids in by_node.values() for t in ids}) + s_wide["initial_guess"] = { + "lat": first.get("lat"), + "lon": first.get("lon"), + "alt_km": float(first.get("alt_m") or 0.0) / 1000.0, + } + if first.get("vel_east") is not None: + s_wide["initial_velocity"] = { + "vel_east_ms": first.get("vel_east"), + "vel_north_ms": first.get("vel_north"), + } + return s_wide + + +def _adopt_pool_nodes( + s_in: dict, + node_cfgs: dict, + result: dict, + solve_fn, + multistart_fn=_pool_solve_multistart, +) -> tuple[dict, dict, dict | None]: + """Widen a narrow dark solve with pool nodes the solve itself vouches for. + + Runs immediately after the first successful solve and BEFORE trimming and + before every gate, so a candidate that adopts a third node is judged as an + n=3 solve throughout — including by the n=2 confirmation gate, which no + longer applies to it. That is the point rather than a side effect: the + gate exists because two nodes cannot corroborate each other's identity, + and a third node whose measured delay matches what the two-node solve + predicts for it is exactly the corroboration it was demanding. + + Only bottom-up dark inputs qualify. Anchored and known-lane inputs have a + transponder identity and were never clustered by + format_track_pairs_for_solver, so they carry no pool to adopt from. + + Returns (result, s_in, meta). meta is None when the stage did not apply + at all; otherwise it records what was tried, so a history record can be + read for adoption's true and false positives rather than just its count. + Cost is bounded at two extra LM solves per eligible candidate (~60-90 ms + each in the pool): the first widening, plus at most one retry with the + worst adopted node dropped. + """ + if not _ADOPT_POOL_ENABLED or not isinstance(s_in, dict) or not isinstance(result, dict): + return result, s_in, None + if not _is_dark_solver_input(s_in) or s_in.get("anchor_key"): + return result, s_in, None + pool_n = s_in.get("pool_n_nodes") + pool_meas = s_in.get("pool_measurements") or [] + if not pool_meas or not pool_n or (result.get("n_nodes") or 0) >= pool_n: + return result, s_in, None + lat, lon = result.get("lat"), result.get("lon") + if lat is None or lon is None: + return result, s_in, None + have = {m.get("node_id") for m in (s_in.get("measurements") or [])} + cands = [m for m in pool_meas if m.get("node_id") not in have and m.get("delay_us") is not None] + if not cands: + return result, s_in, None + + state.bump_counter("solver_adopt_eligible") + meta: dict = {"pool_n": int(pool_n), "candidates": len(cands), "adopted_node_ids": [], "outcome": "none_passed"} + alt_km = float(result.get("alt_m") or 0.0) / 1000.0 + vel_east = float(result.get("vel_east") or 0.0) + vel_north = float(result.get("vel_north") or 0.0) + geometries = state.node_associator.node_geometries if state.node_associator else {} + + adopted: list[dict] = [] + pred_resid: dict[str, float] = {} + for m in cands: + geo = geometries.get(m["node_id"]) + if geo is None: + # Same abstention rule as everywhere else: a node with no + # registered geometry cannot be predicted for, so it is not + # adopted rather than adopted unchecked. + continue + try: + pred_delay, pred_doppler = predict_observation(geo, lat, lon, alt_km, vel_east, vel_north) + except Exception: + logging.exception("Solver pool adoption: prediction failed for node %s", m["node_id"]) + continue + d_delay = abs(float(m["delay_us"]) - pred_delay) + if d_delay > _ADOPT_DELAY_GATE_US: + continue + # Doppler abstains rather than blocks when the pool pairing carried + # none: the delay agreement is the stronger of the two claims and a + # missing measurement is not a disagreement. + if m.get("doppler_hz") is not None and abs(float(m["doppler_hz"]) - pred_doppler) > _ADOPT_DOPPLER_GATE_HZ: + continue + adopted.append(m) + pred_resid[m["node_id"]] = d_delay + + if not adopted: + state.bump_counter("solver_adopt_rejected") + return result, s_in, meta + + for attempt in range(2): + meta["adopted_node_ids"] = sorted(m["node_id"] for m in adopted) + s_wide = _s_in_with_adopted(s_in, adopted, result) + if state.SOLVER_EPOCH_ALIGN: + s_wide, _ = align_measurement_epochs(s_wide, node_cfgs) + try: + wide = _solve_best_altitude(s_wide, node_cfgs, solve_fn, multistart_fn) + except Exception: + logging.exception("Solver pool adoption re-solve failed") + wide = None + if not wide or not wide.get("success"): + meta["outcome"] = "rejected_solve" + break + rms_delay = wide.get("rms_delay") or 0 + if rms_delay > _SOLVER_RMS_DELAY_MAX_US: + # One cheap second chance, and only one: with two or more adopted + # nodes the rms is a sum over both, so a single contaminated + # adoption can sink a widening the other node would have carried. + # Beyond that, keep the narrow solve — this is a bonus path, not a + # search. + if attempt == 0 and len(adopted) >= 2: + residuals = wide.get("per_node_delay_res_us") or {} + worst = max(adopted, key=lambda m: residuals.get(m["node_id"], pred_resid[m["node_id"]])) + meta["dropped_node_id"] = worst["node_id"] + adopted = [m for m in adopted if m["node_id"] != worst["node_id"]] + continue + meta["outcome"] = "rejected_rms" + meta["wide_rms_delay"] = round(float(rms_delay), 3) + break + jump_km = _haversine_km(float(lat), float(lon), float(wide["lat"]), float(wide["lon"])) + if jump_km > _ADOPT_MAX_JUMP_KM: + meta["outcome"] = "rejected_jump" + meta["jump_km"] = round(jump_km, 2) + break + meta["outcome"] = "widened" + meta["jump_km"] = round(jump_km, 2) + meta["wide_rms_delay"] = round(float(rms_delay), 3) + state.bump_counter("solver_adopt_widened") + state.bump_counter("solver_adopt_nodes_added", len(adopted)) + return wide, s_wide, meta + + meta["adopted_node_ids"] = sorted(m["node_id"] for m in adopted) + state.bump_counter("solver_adopt_rejected") + return result, s_in, meta + + # ── Multi-epoch EWMA position smoother (all N) ─────────────────────────────── # This EWMA machinery is now the TRACK_SMOOTHER=ewma fallback; the default # smoother is the Kalman filter in services/track_filter.py. @@ -2523,6 +2724,15 @@ def _process_solver_item( # otherwise — unchanged. A consensus-filtered n=3 input skips this # (below the n≥4 floor) by construction, which is intended: consensus # already chose the subset it trusts. + # Widen before anything judges the solve. An n=2 candidate that + # adopts a pool node the solve itself vouches for reaches every gate + # below as an n=3 solve — including the n=2 confirmation gate, which + # is asking for exactly the corroboration the adopted node supplied. + n_nodes_pre_adopt = result.get("n_nodes") + result, s_in, adopt_meta = _adopt_pool_nodes(s_in, node_cfgs, result, solve_fn, multistart_fn) + if adopt_meta: + n_nodes = result.get("n_nodes", n_nodes) + trim_meta: dict | None = None if ( "initial_guess" in s_in @@ -2537,6 +2747,9 @@ def _process_solver_item( # (published or rejected) so a bad map marker can be traced back to # both what trimming tried and what consensus selected. _extra: dict | None = dict(trim_meta) if trim_meta else {} + if adopt_meta: + _extra["adopt_meta"] = adopt_meta + _extra["n_nodes_pre_adopt"] = n_nodes_pre_adopt if consensus_meta is not None: _extra["consensus_meta"] = consensus_meta # How this solve got its altitude, and — in free mode — what each diff --git a/backend/tests/test_solver_pool_adoption.py b/backend/tests/test_solver_pool_adoption.py new file mode 100644 index 00000000..8d8e1e43 --- /dev/null +++ b/backend/tests/test_solver_pool_adoption.py @@ -0,0 +1,353 @@ +"""Tests for adopting pool nodes a dark solve can vouch for. + +Context (measured live, 21-min window): 72% of published dark solves are +narrower than the node pool their input was clustered out of, mean shortfall +2.27 nodes, and 390 of 481 rejected n=2 candidates had a pool of 3 or more — +a third node the round paired and the position clustering left in a separate +input. An n=3 candidate publishes 81% of the time against 6% for n=2, so +those are the most expensive detections the pipeline discards. + +solver.py's _adopt_pool_nodes runs right after the first successful solve and +before every gate: it predicts what each pool node should have measured at the +just-solved position and velocity, adopts the ones that agree within +SOLVER_ADOPT_DELAY_GATE_US / SOLVER_ADOPT_DOPPLER_GATE_HZ, re-solves wider, +and keeps the wider solve only if it passes the usual rms gate and has not +walked away from the narrow position. + +The forward model is monkeypatched throughout: what is under test is the +adoption decision, not retina_analytics' bistatic geometry (which has its own +tests), and a stubbed prediction is the only way to place a pool measurement a +controlled distance from the gate edge. +""" + +import time + +from core import state +from services.tasks import solver as solver_mod + +LAT, LON = 35.0, -82.0 + +# What the stubbed forward model claims the missing node should have seen. +PRED_DELAY_US, PRED_DOPPLER_HZ = 40.0, 12.0 + +_MISSING = object() + + +def _stub_result(node_ids, rms_delay, lat=LAT, lon=LON, **overrides): + """A solve_multinode-shaped success dict for the given contributing nodes.""" + result = { + "success": True, + "lat": lat, + "lon": lon, + "alt_m": 9000.0, + "timestamp_ms": int(time.time() * 1000), + "vel_east": 150.0, + "vel_north": -60.0, + "rms_delay": rms_delay, + "rms_doppler": 5.0, + "n_nodes": len(node_ids), + "n_measurements": len(node_ids), + "contributing_node_ids": list(node_ids), + } + result.update(overrides) + return result + + +def _s_in(node_ids, pool_measurements=None, pool_n_nodes=None, **overrides): + """A dark bottom-up solver input, optionally carrying a pool. + + Shape matches what InterNodeAssociator._solver_input emits — measurements + with t_s, per-node track ids, and the pool the cluster came out of. + """ + s_in = { + "initial_guess": {"lat": LAT, "lon": LON, "alt_km": 9.0}, + "measurements": [ + {"node_id": nid, "delay_us": 10.0, "doppler_hz": 1.0, "snr": 15.0, "t_s": 1.0} for nid in node_ids + ], + "n_nodes": len(node_ids), + "timestamp_ms": int(time.time() * 1000), + "adsb_hex": None, + "track_ids": [f"t-{nid}" for nid in node_ids], + "track_ids_by_node": {nid: [f"t-{nid}"] for nid in node_ids}, + "chi2_per_dof": 1.0, + "n_epochs": 8, + "pool_n_nodes": pool_n_nodes if pool_n_nodes is not None else len(node_ids), + "pool_node_ids": sorted(node_ids + [m["node_id"] for m in (pool_measurements or [])]), + "pool_measurements": pool_measurements, + "pool_conflicts": 0, + } + s_in.update(overrides) + return s_in + + +def _pool_meas(node_id="pc", delay_us=PRED_DELAY_US, doppler_hz=PRED_DOPPLER_HZ): + return { + "node_id": node_id, + "track_id": f"t-{node_id}", + "delay_us": delay_us, + "doppler_hz": doppler_hz, + "snr": 12.0, + "t_s": 1.0, + } + + +def _stub_solve_fn(table: dict): + """solve_fn keyed on the frozenset of node_ids in s_in["measurements"]. + + Same idiom as test_solver_trimming: _solve_best_altitude calls solve_fn + once per altitude layer, so the answer must depend only on which nodes are + present. An unregistered node set is a solve solve_multinode rejects. + """ + + def fn(s_in, node_cfgs): + nodes = frozenset(m["node_id"] for m in s_in["measurements"]) + base = table.get(nodes, _MISSING) + if base is _MISSING: + return {"success": False} + if base is None: + return None + return dict(base) + + return fn + + +class _AdoptTestBase: + def setup_method(self): + state._reset_for_tests() + solver_mod._reset_for_tests() + + def teardown_method(self): + solver_mod._reset_for_tests() + + def _arm(self, monkeypatch, node_ids=("pc",), delay=PRED_DELAY_US, doppler=PRED_DOPPLER_HZ): + """Register geometry for the pool nodes and stub the forward model. + + A node with no registered geometry is never adopted (the abstention + rule), so the registry entry has to exist even though its contents are + never read once predict_observation is stubbed. + """ + for nid in node_ids: + state.node_associator.node_geometries[nid] = object() + monkeypatch.setattr(solver_mod, "predict_observation", lambda geo, *a, **kw: (delay, doppler)) + + def _run(self, s_in, solve_fn, cfgs=None): + return solver_mod._process_solver_item((dict(s_in), cfgs or {}, time.time()), solve_fn) + + def _last_record(self): + assert state.mlat_solve_history + return state.mlat_solve_history[-1] + + +class TestAdoptWidens(_AdoptTestBase): + """An n=2 candidate whose third node agrees becomes an n=3 solve.""" + + def test_agreeing_pool_node_is_adopted_and_resolved_wider(self, monkeypatch): + self._arm(monkeypatch) + table = { + frozenset({"n1", "n2"}): _stub_result(["n1", "n2"], rms_delay=1.5), + frozenset({"n1", "n2", "pc"}): _stub_result(["n1", "n2", "pc"], rms_delay=1.1), + } + result = self._run( + _s_in(["n1", "n2"], pool_measurements=[_pool_meas()], pool_n_nodes=3), + _stub_solve_fn(table), + ) + + assert result is not None and result["success"] + assert result["n_nodes"] == 3 + assert state.solver_adopt_eligible == 1 + assert state.solver_adopt_widened == 1 + assert state.solver_adopt_nodes_added == 1 + assert state.solver_adopt_rejected == 0 + rec = self._last_record() + assert rec["adopt_meta"]["outcome"] == "widened" + assert rec["adopt_meta"]["adopted_node_ids"] == ["pc"] + assert rec["adopt_meta"]["pool_n"] == 3 + assert rec["n_nodes_pre_adopt"] == 2 + assert rec["n_nodes"] == 3 + + def test_adopted_node_track_joins_the_provenance(self, monkeypatch): + """The published solve must name the tracklet it was widened with. + + Otherwise the third node's measurement is in the fit but its track is + not in track_ids, and nothing downstream (supersession, claiming) can + tell that the tracklet has been consumed. + """ + self._arm(monkeypatch) + table = { + frozenset({"n1", "n2"}): _stub_result(["n1", "n2"], rms_delay=1.5), + frozenset({"n1", "n2", "pc"}): _stub_result(["n1", "n2", "pc"], rms_delay=1.1), + } + self._run(_s_in(["n1", "n2"], pool_measurements=[_pool_meas()], pool_n_nodes=3), _stub_solve_fn(table)) + rec = self._last_record() + assert "t-pc" in rec["track_ids"] + + +class TestAdoptDeclines(_AdoptTestBase): + def test_pool_node_outside_the_delay_gate_is_not_adopted(self, monkeypatch): + """40 µs from prediction is a different aircraft, not a wider solve. + + The gate is 6.0 µs — an n=2 dark solve sits ~2 km from truth and 2 km + of range error is ~6.7 µs of bistatic delay, so anything past that is + not explained by the narrow solve's own error budget. + """ + self._arm(monkeypatch) + table = {frozenset({"n1", "n2"}): _stub_result(["n1", "n2"], rms_delay=1.5)} + result = self._run( + _s_in( + ["n1", "n2"], + pool_measurements=[_pool_meas(delay_us=PRED_DELAY_US + 40.0)], + pool_n_nodes=3, + ), + _stub_solve_fn(table), + ) + + assert result is not None and result["n_nodes"] == 2 + assert state.solver_adopt_eligible == 1 + assert state.solver_adopt_widened == 0 + assert state.solver_adopt_rejected == 1 + rec = self._last_record() + assert rec["adopt_meta"]["outcome"] == "none_passed" + assert rec["adopt_meta"]["candidates"] == 1 + assert rec["adopt_meta"]["adopted_node_ids"] == [] + assert rec["n_nodes"] == 2 + + def test_pool_node_outside_the_doppler_gate_is_not_adopted(self, monkeypatch): + """Delay agreement alone is not enough — a node on the same delay + ellipse but moving wrongly is a different target on that ellipse.""" + self._arm(monkeypatch) + table = {frozenset({"n1", "n2"}): _stub_result(["n1", "n2"], rms_delay=1.5)} + result = self._run( + _s_in( + ["n1", "n2"], + pool_measurements=[_pool_meas(doppler_hz=PRED_DOPPLER_HZ + 400.0)], + pool_n_nodes=3, + ), + _stub_solve_fn(table), + ) + + assert result["n_nodes"] == 2 + assert state.solver_adopt_widened == 0 + assert self._last_record()["adopt_meta"]["outcome"] == "none_passed" + + def test_wide_solve_failing_rms_keeps_the_narrow_one(self, monkeypatch): + """Agreement at the prediction is a claim; the joint fit is the test. + + With a single adopted node there is no cheap second chance to take — + dropping it is just the original solve — so the widening is abandoned + and the narrow result is what the gates below see. + """ + self._arm(monkeypatch) + narrow = _stub_result(["n1", "n2"], rms_delay=1.5) + table = { + frozenset({"n1", "n2"}): narrow, + frozenset({"n1", "n2", "pc"}): _stub_result(["n1", "n2", "pc"], rms_delay=11.0), + } + result = self._run( + _s_in(["n1", "n2"], pool_measurements=[_pool_meas()], pool_n_nodes=3), + _stub_solve_fn(table), + ) + + assert result is not None and result["n_nodes"] == 2 + assert result["rms_delay"] == 1.5 + assert state.solver_adopt_widened == 0 + assert state.solver_adopt_rejected == 1 + rec = self._last_record() + assert rec["adopt_meta"]["outcome"] == "rejected_rms" + assert rec["adopt_meta"]["adopted_node_ids"] == ["pc"] + assert rec["n_nodes_pre_adopt"] == 2 + + def test_wide_solve_that_walks_away_is_refused(self, monkeypatch): + """A widened solve 60 km from the narrow one is a different geometry + the adopted node dragged the fit into, not the same aircraft solved + better.""" + self._arm(monkeypatch) + table = { + frozenset({"n1", "n2"}): _stub_result(["n1", "n2"], rms_delay=1.5), + frozenset({"n1", "n2", "pc"}): _stub_result(["n1", "n2", "pc"], rms_delay=1.0, lat=LAT + 0.55), + } + result = self._run( + _s_in(["n1", "n2"], pool_measurements=[_pool_meas()], pool_n_nodes=3), + _stub_solve_fn(table), + ) + + assert result["n_nodes"] == 2 + assert abs(result["lat"] - LAT) < 1e-9 + assert state.solver_adopt_widened == 0 + assert self._last_record()["adopt_meta"]["outcome"] == "rejected_jump" + + def test_second_chance_drops_the_worst_adopted_node(self, monkeypatch): + """Two adopted nodes, one contaminated: the joint rms is a sum over + both, so retrying once without the worst residual recovers a widening + the blanket rms gate would have thrown away whole.""" + self._arm(monkeypatch, node_ids=("pc", "pd")) + table = { + frozenset({"n1", "n2"}): _stub_result(["n1", "n2"], rms_delay=1.5), + frozenset({"n1", "n2", "pc", "pd"}): _stub_result( + ["n1", "n2", "pc", "pd"], + rms_delay=9.0, + per_node_delay_res_us={"n1": 0.4, "n2": 0.4, "pc": 0.5, "pd": 17.0}, + ), + frozenset({"n1", "n2", "pc"}): _stub_result(["n1", "n2", "pc"], rms_delay=1.2), + } + result = self._run( + _s_in(["n1", "n2"], pool_measurements=[_pool_meas("pc"), _pool_meas("pd")], pool_n_nodes=4), + _stub_solve_fn(table), + ) + + assert result["n_nodes"] == 3 + assert state.solver_adopt_widened == 1 + assert state.solver_adopt_nodes_added == 1 + rec = self._last_record() + assert rec["adopt_meta"]["outcome"] == "widened" + assert rec["adopt_meta"]["adopted_node_ids"] == ["pc"] + assert rec["adopt_meta"]["dropped_node_id"] == "pd" + + +class TestAdoptScope(_AdoptTestBase): + def test_anchored_input_is_skipped(self, monkeypatch): + """An anchored input has a claimed identity and never went through the + clustering the pool describes, so there is nothing to widen from.""" + self._arm(monkeypatch) + table = { + frozenset({"n1", "n2"}): _stub_result(["n1", "n2"], rms_delay=1.5), + frozenset({"n1", "n2", "pc"}): _stub_result(["n1", "n2", "pc"], rms_delay=1.0), + } + result = self._run( + _s_in( + ["n1", "n2"], + pool_measurements=[_pool_meas()], + pool_n_nodes=3, + anchor_key="mn-dark-abc123", + ), + _stub_solve_fn(table), + ) + + assert result["n_nodes"] == 2 + assert state.solver_adopt_eligible == 0 + assert "adopt_meta" not in self._last_record() + + def test_input_already_as_wide_as_its_pool_is_skipped(self, monkeypatch): + """No shortfall, nothing to adopt — and no eligible count either, so + the hit rate is read against candidates that actually had one.""" + self._arm(monkeypatch) + table = {frozenset({"n1", "n2", "n3"}): _stub_result(["n1", "n2", "n3"], rms_delay=1.2)} + result = self._run(_s_in(["n1", "n2", "n3"], pool_measurements=[]), _stub_solve_fn(table)) + + assert result["n_nodes"] == 3 + assert state.solver_adopt_eligible == 0 + assert "adopt_meta" not in self._last_record() + + def test_kill_switch_disables_the_stage(self, monkeypatch): + self._arm(monkeypatch) + monkeypatch.setattr(solver_mod, "_ADOPT_POOL_ENABLED", False) + table = { + frozenset({"n1", "n2"}): _stub_result(["n1", "n2"], rms_delay=1.5), + frozenset({"n1", "n2", "pc"}): _stub_result(["n1", "n2", "pc"], rms_delay=1.0), + } + result = self._run( + _s_in(["n1", "n2"], pool_measurements=[_pool_meas()], pool_n_nodes=3), + _stub_solve_fn(table), + ) + + assert result["n_nodes"] == 2 + assert state.solver_adopt_eligible == 0 diff --git a/backend/tests/test_solver_stats.py b/backend/tests/test_solver_stats.py index 59494513..4183632e 100644 --- a/backend/tests/test_solver_stats.py +++ b/backend/tests/test_solver_stats.py @@ -274,6 +274,10 @@ def test_consensus_and_counters_reflect_state(self): state.solver_resolve_refresh = 3 state.node_frames_rate_limited = 13 state.solver_pool_timeouts = 19 + state.solver_adopt_eligible = 21 + state.solver_adopt_widened = 22 + state.solver_adopt_nodes_added = 23 + state.solver_adopt_rejected = 24 out = _solver_window_stats(10.0) assert out["counters"] == { "successes": 5, @@ -288,6 +292,10 @@ def test_consensus_and_counters_reflect_state(self): "epoch_align_skipped": 14, "resolve_skips_dark": 9, "resolve_refresh": 3, + "adopt_eligible": 21, + "adopt_widened": 22, + "adopt_nodes_added": 23, + "adopt_rejected": 24, "queue_drops": 6, "node_frames_rate_limited": 13, "worker_errors": 0, diff --git a/libs/retina-analytics b/libs/retina-analytics index 0920910b..b3bd3210 160000 --- a/libs/retina-analytics +++ b/libs/retina-analytics @@ -1 +1 @@ -Subproject commit 0920910ba665f4c6392bb6fabf1a46038b38dc0f +Subproject commit b3bd32105dc715e0537aea40731149b96b7bbd0c