Carry the pool's measurements on the solver input

src/retina_analytics/association.py

@@ -2581,7 +2581,7 @@ def _find(x: int) -> int:
         for i, p in enumerate(pairs):
             groups[_find(i)].append(p)
 
-        pool_by_pair = self._shared_track_pools(pairs)
+        pool_by_pair, pool_meas_by_pair = self._shared_track_pools(pairs)
 
         solver_inputs = []
         for merged in groups.values():
@@ -2590,12 +2590,23 @@ def _find(x: int) -> int:
                 self.cluster_splits += 1
             for g in subs:
                 pool: set[str] = set()
+                # node_id → track_id → measurement, so a node the component saw
+                # on two different tracks keeps both candidates until
+                # _solver_input picks between them (highest SNR) and counts the
+                # ambiguity.  Merging by track id also makes this idempotent
+                # across the pairings of one component, which all map to the
+                # same per-component dict.
+                pool_meas: dict[str, dict[str, dict]] = defaultdict(dict)
                 for p in g:
                     pool |= pool_by_pair[id(p)]
-                solver_inputs.append(self._solver_input(g, pool))
+                    for nid, by_track in pool_meas_by_pair[id(p)].items():
+                        pool_meas[nid].update(by_track)
+                solver_inputs.append(self._solver_input(g, pool, pool_meas))
         return solver_inputs
 
-    def _shared_track_pools(self, pairs: list[TrackPairCandidate]) -> dict[int, set[str]]:
+    def _shared_track_pools(
+        self, pairs: list[TrackPairCandidate]
+    ) -> tuple[dict[int, set[str]], dict[int, dict[str, dict[str, dict]]]]:
         """Map each pairing to the node set of its shared-track component.
 
         A second union-find over the same round's pairings, joined not by
@@ -2616,6 +2627,15 @@ def _shared_track_pools(self, pairs: list[TrackPairCandidate]) -> dict[int, set[
         the clustering did not take it" (pool is 3, n_nodes is 2), which is the
         thing no existing counter can tell apart.
 
+        Returns (node_ids_by_pair, measurements_by_pair).  The second map
+        carries the component's actual measurements — node_id → track_id →
+        {node_id, track_id, delay_us, doppler_hz, snr, t_s} — because knowing
+        that a third node was available is only half of what the solver needs:
+        to widen a narrow solve it has to be handed what that node measured,
+        and the pairing that carried it is the only place those numbers exist
+        once the clustering has declined to merge it.  Both maps are per
+        component, so every pairing in a component shares the same dict object.
+
         Keyed by id() rather than by index because _partition_cluster hands
         back the candidate objects, not their positions in `pairs`; the objects
         live for the whole call, so the ids are stable here.
@@ -2640,9 +2660,26 @@ def _find(x: int) -> int:
                     parent[_find(i)] = _find(j)
 
         nodes: dict[int, set[str]] = defaultdict(set)
+        meas: dict[int, dict[str, dict[str, dict]]] = defaultdict(lambda: defaultdict(dict))
         for i, p in enumerate(pairs):
-            nodes[_find(i)].update((p.node_a_id, p.node_b_id))
-        return {id(p): nodes[_find(i)] for i, p in enumerate(pairs)}
+            root = _find(i)
+            nodes[root].update((p.node_a_id, p.node_b_id))
+            for nid, tid, d, f, s, t in (
+                (p.node_a_id, p.track_a_id, p.delay_a, p.doppler_a, p.snr_a, p.t_s_a),
+                (p.node_b_id, p.track_b_id, p.delay_b, p.doppler_b, p.snr_b, p.t_s_b),
+            ):
+                # Same first-writer-wins reasoning as _solver_input: a track's
+                # measurement is its own history[-1], so repeats of the same
+                # (node, track) across the component's pairings are the same
+                # numbers, not rivals.
+                meas[root][nid].setdefault(
+                    tid,
+                    {"node_id": nid, "track_id": tid, "delay_us": d, "doppler_hz": f, "snr": s, "t_s": t},
+                )
+        return (
+            {id(p): nodes[_find(i)] for i, p in enumerate(pairs)},
+            {id(p): meas[_find(i)] for i, p in enumerate(pairs)},
+        )
 
     def _velocity_conflict_matrix(self, pairs: list[TrackPairCandidate]) -> np.ndarray:
         """(n, n) mask: True where two pairings' implied velocities disagree.
@@ -2731,13 +2768,26 @@ def _partition_cluster(self, group: list[TrackPairCandidate]) -> list[list[Track
                 sub_nodes.append(dict(own))
         return subs
 
-    def _solver_input(self, group: list[TrackPairCandidate], pool_node_ids: set[str] | None = None) -> dict:
+    def _solver_input(
+        self,
+        group: list[TrackPairCandidate],
+        pool_node_ids: set[str] | None = None,
+        pool_measurements: dict[str, dict[str, dict]] | None = None,
+    ) -> dict:
         """One node-consistent cluster, in the shape the solver worker takes.
 
         pool_node_ids is the shared-track node pool this cluster came out of
         (see _shared_track_pools); None only for callers that have no round to
         take it from, in which case the input says "not measured" rather than
         claiming the pool equals what was used.
+
+        pool_measurements is that pool's raw measurements, node_id → track_id →
+        measurement.  What lands on the input is only the part the cluster does
+        NOT already carry: one measurement per pool node absent from
+        `measurements`, which is exactly the material the solver worker needs to
+        try widening a solve the clustering left narrow.  Nothing here decides
+        whether widening is right — that judgement needs a solved position, and
+        this stage has none.
         """
         by_node: dict[str, dict] = {}
         for p in group:
@@ -2771,6 +2821,31 @@ def _solver_input(self, group: list[TrackPairCandidate], pool_node_ids: set[str]
         # worker, so this field says "not scored yet" rather than "scored 0".
         worst_chi2 = max((p.chi2_per_dof for p in group if p.chi2_per_dof is not None), default=None)
 
+        # The pool's spare measurements: pool nodes with nothing in this
+        # cluster.  A node the component saw on two tracks is genuinely
+        # ambiguous here — one of them belongs to another aircraft — so take
+        # the strongest and report how many nodes needed that tiebreak, which
+        # is the rate at which a consumer's adoption gate is being handed a
+        # coin flip.  None (not []) when there was no round to take a pool
+        # from, matching pool_n_nodes' "not measured".
+        spare: list[dict] | None = None
+        pool_conflicts: int | None = None
+        if pool_node_ids is not None:
+            spare = []
+            pool_conflicts = 0
+            for nid in sorted(pool_measurements or {}):
+                if nid in by_node:
+                    continue
+                by_track = pool_measurements[nid]
+                if len(by_track) > 1:
+                    pool_conflicts += 1
+                spare.append(
+                    max(
+                        by_track.values(),
+                        key=lambda m: (m["snr"] if m["snr"] is not None else float("-inf"), m["track_id"]),
+                    )
+                )
+
         return {
             "initial_guess": {
                 "lat": sum(p.lat for p in group) / len(group),
@@ -2800,6 +2875,8 @@ def _solver_input(self, group: list[TrackPairCandidate], pool_node_ids: set[str]
             # aircraft instead of against the fleet size.
             "pool_n_nodes": len(pool_node_ids) if pool_node_ids is not None else None,
             "pool_node_ids": sorted(pool_node_ids) if pool_node_ids is not None else None,
+            "pool_measurements": spare,
+            "pool_conflicts": pool_conflicts,
         }
 
     def get_overlap_summary(self) -> list[dict]:

tests/test_track_association.py

@@ -491,6 +491,58 @@ def test_separate_aircraft_do_not_share_a_pool(self):
         assert len(inputs) == 2
         assert all(s_in["n_nodes"] == 2 == s_in["pool_n_nodes"] for s_in in inputs)
 
+    def test_pool_carries_the_missing_node_measurements(self):
+        """The shortfall is only actionable if the numbers travel with it.
+
+        Same scene as the counting test — (A,B),(A,C),(B,C) with (A,C) 55 km
+        off — but read for content.  (A,B) and (B,C) merge into the 3-node
+        input; the far (A,C) pairing is the narrow one, and it must carry
+        site-b's delay/Doppler from the pairing that did observe it, so the
+        solver worker can predict what site-b should have measured at the
+        solved position and decide whether to adopt it.  The 3-node input has
+        nothing spare — every pool node is already in it.
+        """
+        pairs = [
+            _candidate("a1", "b1", delay_b=52.0, doppler_b=-11.0),
+            _candidate("b1", "c1", node_a_id="site-b", node_b_id="site-c", delay_a=52.0, doppler_a=-11.0),
+            _candidate("a1", "c1", node_b_id="site-c", lat=35.38),
+        ]
+        inputs = InterNodeAssociator().format_track_pairs_for_solver(pairs)
+        by_nodes = {frozenset(m["node_id"] for m in s["measurements"]): s for s in inputs}
+        narrow = by_nodes[frozenset({"site-a", "site-c"})]
+        assert [m["node_id"] for m in narrow["pool_measurements"]] == ["site-b"]
+        assert narrow["pool_measurements"][0]["delay_us"] == 52.0
+        assert narrow["pool_measurements"][0]["doppler_hz"] == -11.0
+        assert narrow["pool_measurements"][0]["track_id"] == "b1"
+        assert narrow["pool_conflicts"] == 0
+        assert by_nodes[frozenset({"site-a", "site-b", "site-c"})]["pool_measurements"] == []
+
+    def test_pool_node_seen_on_two_tracks_takes_the_strongest(self):
+        """An ambiguous pool node is a coin flip the input has to declare.
+
+        Three pairings of one shared-track component, each at its own position
+        so none of them merge.  site-c reaches the component on two different
+        tracks (c1 via A–C, c2 via B–C), which means one of them belongs to
+        another aircraft.  Take the higher SNR and count the node in
+        pool_conflicts so the rate is visible rather than silently guessed.
+        """
+        pairs = [
+            _candidate("a1", "b1"),
+            _candidate("b1", "c2", node_a_id="site-b", node_b_id="site-c", lat=36.0, snr_b=4.0, delay_b=71.0),
+            _candidate("a1", "c1", node_b_id="site-c", lat=35.38, snr_b=19.0, delay_b=52.0),
+        ]
+        inputs = InterNodeAssociator().format_track_pairs_for_solver(pairs)
+        narrow = next(s for s in inputs if {m["node_id"] for m in s["measurements"]} == {"site-a", "site-b"})
+        assert [m["track_id"] for m in narrow["pool_measurements"]] == ["c1"]
+        assert narrow["pool_measurements"][0]["delay_us"] == 52.0
+        assert narrow["pool_conflicts"] == 1
+
+    def test_input_with_no_round_reports_no_pool_measurements(self):
+        """None, not [] — the same "not measured" the pool count uses."""
+        s_in = InterNodeAssociator()._solver_input([_candidate("a1", "b1")])
+        assert s_in["pool_measurements"] is None
+        assert s_in["pool_conflicts"] is None
+
 
 class TestClusterPartition:
     """A cluster is one aircraft only if nothing in it says otherwise.