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