Route every reputation penalty through one scale, and require 3 trust samples before acting

src/retina_analytics/__init__.py

@@ -13,7 +13,7 @@
 from retina_analytics.detection_area import DetectionAreaState
 from retina_analytics.manager import NodeAnalyticsManager
 from retina_analytics.metrics import NodeMetrics
-from retina_analytics.reputation import NodeReputation
+from retina_analytics.reputation import NodeReputation, set_penalty_scale
 from retina_analytics.trust import AdsReportEntry, TrustScoreState
 
 __all__ = [
@@ -23,6 +23,7 @@
     "DetectionAreaState",
     "NodeMetrics",
     "NodeReputation",
+    "set_penalty_scale",
     "HistoricalCoverageMap",
     "CoverageMapEntry",
     "InterNodeAssociator",

src/retina_analytics/manager.py

@@ -24,10 +24,29 @@
 )
 from retina_analytics.metrics import NodeMetrics
 from retina_analytics.reputation import NodeReputation
-from retina_analytics.trust import AdsReportEntry, TrustScoreState
+from retina_analytics.trust import TRUST_MIN_SAMPLES, AdsReportEntry, TrustScoreState
 
 _RX_RELOCATE_THRESHOLD_KM = 0.05  # 50 m — above real GPS/reporting jitter
 
+# What a node's trust is worth before it has enough samples to be worth
+# anything: the same neutral prior the backend's node_bias.get_node_trust
+# applies below TRUST_MIN_SAMPLES.
+_NEUTRAL_TRUST_PRIOR = 0.5
+
+
+def _trust_for_reputation(ts: TrustScoreState | None) -> float | None:
+    """Trust score to act on, or None while the node is below the sample bar.
+
+    TrustScoreState.score is a good/total ratio, so with one sample it is
+    either 0.0 or 1.0 — a single out-of-threshold residual is a 0.0 that
+    evaluate_trust reads as "critically low" and penalises every pass until
+    the node is blocked (ndebvzgeoij5t2l, test droplet).  None means "no
+    opinion": the caller either skips, or falls back to the neutral prior.
+    """
+    if ts is None or len(ts.samples) < TRUST_MIN_SAMPLES:
+        return None
+    return ts.score
+
 
 def _resolve_fov_prior(
     config: dict, rx_lat: float, rx_lon: float, tx_lat: float, tx_lon: float
@@ -499,9 +518,14 @@ def record_node_tracks(self, node_id: str, confirmed_ids, geolocated_ids=()):
 
     def evaluate_reputations(self):
         for node_id, rep in self.reputations.items():
-            ts = self.trust_scores.get(node_id)
-            if ts and ts.samples:
-                rep.evaluate_trust(ts.score)
+            # Below TRUST_MIN_SAMPLES there is no score to act on: the neutral
+            # 0.5 prior the backend substitutes there sits between the warn
+            # threshold and the reward threshold, so it would neither penalise
+            # nor reward — skipping is equivalent, and it keeps one claim
+            # residual from starting a block.
+            trust = _trust_for_reputation(self.trust_scores.get(node_id))
+            if trust is not None:
+                rep.evaluate_trust(trust)
 
             metrics = self.metrics.get(node_id)
             if metrics:
@@ -520,15 +544,23 @@ def evaluate_reputations(self):
                     if dist > area_a.footprint_radius_km() + area_b.footprint_radius_km():
                         continue
                     overlap = compute_delay_bin_overlap(area_a, area_b)
-                    ts_a = self.trust_scores.get(a_id)
-                    ts_b = self.trust_scores.get(b_id)
-                    if ts_a and ts_b:
-                        self.reputations[a_id].evaluate_neighbour_consistency(
-                            overlap["overlap_ratio"], ts_b.score, neighbour_id=b_id
-                        )
-                        self.reputations[b_id].evaluate_neighbour_consistency(
-                            overlap["overlap_ratio"], ts_a.score, neighbour_id=a_id
-                        )
+                    # "Disagrees with a TRUSTED neighbour" is only evidence if
+                    # the neighbour is actually established: below the sample
+                    # bar it gets the neutral prior, which is under the 0.7
+                    # trusted threshold, so a one-sample neighbour can no
+                    # longer condemn anyone.
+                    trust_a = _trust_for_reputation(self.trust_scores.get(a_id))
+                    trust_b = _trust_for_reputation(self.trust_scores.get(b_id))
+                    self.reputations[a_id].evaluate_neighbour_consistency(
+                        overlap["overlap_ratio"],
+                        _NEUTRAL_TRUST_PRIOR if trust_b is None else trust_b,
+                        neighbour_id=b_id,
+                    )
+                    self.reputations[b_id].evaluate_neighbour_consistency(
+                        overlap["overlap_ratio"],
+                        _NEUTRAL_TRUST_PRIOR if trust_a is None else trust_a,
+                        neighbour_id=a_id,
+                    )
 
     def unblock_node(self, node_id: str):
         rep = self.reputations.get(node_id)

src/retina_analytics/reputation.py

@@ -1,13 +1,24 @@
 """Bad actor detection and reputation tracking per node."""
 
+import math
 import time
 from dataclasses import dataclass, field
+from typing import ClassVar
 
 
 @dataclass
 class NodeReputation:
     """Tracks reputation and handles bad actor detection/blocking for a node."""
 
+    # Multiplier applied to every penalty, process-wide — see
+    # set_penalty_scale.  A ClassVar, NOT a dataclass field, on purpose: the
+    # backend persists reputations with dataclasses.asdict() and restores them
+    # with NodeReputation(**saved), so a real field would round-trip through
+    # snapshots (freezing an operational stance into saved state) and older
+    # snapshots would fail to construct.  Tests may still override it per
+    # instance (rep.penalty_scale = 1.0).
+    penalty_scale: ClassVar[float] = 1.0
+
     node_id: str
     reputation: float = 1.0
     blocked: bool = False
@@ -34,24 +45,43 @@ def _penalize_condition(self, key: str, active: bool, amount: float, reason: str
 
         Deliberately NOT used for trust and detection-rate penalties: those
         are evidence of bad *data* actively being submitted, and repeated
-        escalation to a block is the intended defence
-        (test_bad_actor_gets_blocked pins it).
+        escalation to a block is the intended defence whenever downrating is
+        switched on at all (test_bad_actor_gets_blocked pins that shape, with
+        penalty_scale at its library default of 1.0).  Orthogonal to this:
+        set_penalty_scale can turn every downrating path off, escalation
+        included — see apply_penalty.
         """
         was_active = self._condition_active.get(key, False)
         self._condition_active[key] = active
         if active and not was_active:
             self.apply_penalty(amount, reason)
 
     def apply_penalty(self, amount: float, reason: str):
-        self.reputation = max(0.0, self.reputation - amount)
-        self.penalties.append(
-            {
-                "time": time.time(),
-                "amount": amount,
-                "reason": reason,
-                "reputation_after": self.reputation,
-            }
-        )
+        """Lower the reputation by ``amount`` scaled by penalty_scale.
+
+        Every downrating source in the estate funnels through here (trust,
+        heartbeat, detection rate, neighbour consistency, and the backend's
+        direct ADS-B cross-validation call), so the scale is the single place
+        that decides whether reputations can fall at all.
+        """
+        effective = amount * self.penalty_scale
+        if effective <= 0:
+            # Scale at zero: the trust/heartbeat/neighbour paths still compute
+            # and report their findings, only the downrating stops.  Nothing is
+            # recorded either — "no new penalties" is the operator's check that
+            # the switch is on, so a log of zero-amount entries would make
+            # n_penalties lie about it.
+            return
+        self.reputation = max(0.0, self.reputation - effective)
+        entry = {
+            "time": time.time(),
+            "amount": effective,
+            "reason": reason,
+            "reputation_after": self.reputation,
+        }
+        if effective != amount:
+            entry["base_amount"] = amount
+        self.penalties.append(entry)
         if len(self.penalties) > self.max_penalties:
             self.penalties = self.penalties[-self.max_penalties :]
         if self.reputation < self.reputation_block_threshold and not self.blocked:
@@ -112,4 +142,29 @@ def summary(self) -> dict:
             "block_reason": self.block_reason,
             "n_penalties": len(self.penalties),
             "recent_penalties": self.penalties[-5:] if self.penalties else [],
+            # Published so /api/radar/analytics shows whether downrating is on
+            # at all: zero here explains a node whose reputation never moves.
+            "penalty_scale": self.penalty_scale,
         }
+
+
+def set_penalty_scale(scale: float) -> None:
+    """Set the one multiplier every reputation-lowering path routes through.
+
+    Process-wide (a ClassVar on NodeReputation).  The retina-server backend
+    sets it from the environment (REPUTATION_PENALTY_SCALE) at startup and
+    currently defaults it to 0: a temporary stance while the trust input is a
+    single claim residual per node, after a real node was permanently blocked
+    off one out-of-threshold sample.  At 0 no path here can lower a reputation
+    or block a node — trust, heartbeat, detection-rate, neighbour-consistency
+    and the backend's ADS-B cross-validation penalty all become no-ops, and
+    nothing is recorded in `penalties`.  Rewards are untouched, so a node can
+    still climb back to 1.0.  The library default stays 1.0.
+    """
+    try:
+        value = float(scale)
+    except (TypeError, ValueError) as exc:
+        raise ValueError(f"penalty_scale must be a finite float >= 0, got {scale!r}") from exc
+    if not math.isfinite(value) or value < 0.0:
+        raise ValueError(f"penalty_scale must be a finite float >= 0, got {scale!r}")
+    NodeReputation.penalty_scale = value

src/retina_analytics/trust.py

@@ -3,6 +3,15 @@
 import math
 from dataclasses import dataclass, field
 
+# Minimum samples before TrustScoreState.score means anything.  Below it the
+# score quantizes to {0, 1/2, 1} — one unlucky residual reads as a flat 0.0,
+# which is indistinguishable from a node that is lying about every aircraft.
+# The retina-server backend's node_bias.get_node_trust already substitutes a
+# neutral 0.5 prior below this bar for solver weighting, and imports this
+# constant rather than keeping its own copy, so the two consumers of the same
+# score cannot drift apart again.
+TRUST_MIN_SAMPLES = 3
+
 
 @dataclass
 class AdsReportEntry:

tests/conftest.py

@@ -0,0 +1,25 @@
+"""Shared test fixtures.
+
+NodeReputation.penalty_scale is a process-wide ClassVar, so a test that moves
+it would leak into every test that runs after it (and the leak is silent: the
+victims just stop being penalised).  Pin it per test and put it back.
+"""
+
+import pytest
+
+from retina_analytics.reputation import NodeReputation, set_penalty_scale
+
+
+@pytest.fixture(autouse=True)
+def penalty_scale_1_0():
+    """Run every test with downrating explicitly switched on, then restore.
+
+    1.0 is the library default and the value the penalty/block assertions in
+    this suite are written against; the deployment stance (0) is set by the
+    backend, not here.  Tests that exercise another scale call
+    set_penalty_scale themselves — this fixture undoes it afterwards.
+    """
+    saved = NodeReputation.penalty_scale
+    set_penalty_scale(1.0)
+    yield
+    NodeReputation.penalty_scale = saved

tests/test_analytics.py

@@ -1,5 +1,6 @@
 """Tests for analytics subsystem — trust, reputation, coverage, manager, suggestions."""
 
+import dataclasses
 import time
 
 import pytest
@@ -16,6 +17,7 @@
     TrustScoreState,
 )
 from retina_analytics.cross_node import coverage_suggestion
+from retina_analytics.reputation import set_penalty_scale
 
 # ── Trust Score & Reputation ─────────────────────────────────────────────────
 
@@ -142,13 +144,15 @@ def test_good_trust_keeps_high(self):
 
     def test_bad_actor_gets_blocked(self):
         rep = NodeReputation(node_id="bad-node")
+        rep.penalty_scale = 1.0  # explicit: escalation only happens with downrating on
         for _ in range(15):
             rep.evaluate_trust(0.05)
         assert rep.blocked
         assert "Reputation" in rep.block_reason or "Trust" in rep.block_reason
 
     def test_unblock(self):
         rep = NodeReputation(node_id="bad-node")
+        rep.penalty_scale = 1.0
         for _ in range(15):
             rep.evaluate_trust(0.05)
         rep.unblock()
@@ -166,6 +170,61 @@ def test_high_detection_rate_penalty(self):
         assert rep.reputation < 1.0
 
 
+class TestPenaltyScale:
+    """The one switch every downrating source routes through."""
+
+    def test_zero_scale_never_downrates_from_any_source(self):
+        # A real node (ndebvzgeoij5t2l) was blocked permanently off a single
+        # out-of-threshold sample; with the scale at 0 no source may move a
+        # reputation, however hard it is pushed.
+        set_penalty_scale(0.0)
+        rep = NodeReputation(node_id="pummelled")
+        for _ in range(50):
+            rep.evaluate_trust(0.0)
+        rep.evaluate_heartbeat(time.time() - 600)
+        rep.evaluate_detection_rate(100.0)
+        rep.evaluate_neighbour_consistency(0.0, 0.9, "neighbour")
+        rep.apply_penalty(0.1, "adsb cross-validation")  # the backend's direct call
+        assert rep.reputation == 1.0
+        assert rep.penalties == []
+        assert not rep.blocked
+        assert rep.summary()["n_penalties"] == 0
+
+    def test_zero_scale_still_rewards(self):
+        set_penalty_scale(0.0)
+        rep = NodeReputation(node_id="recovering", reputation=0.5)
+        rep.evaluate_trust(0.9)
+        assert rep.reputation > 0.5
+
+    def test_half_scale_halves_the_recorded_amount(self):
+        set_penalty_scale(0.5)
+        rep = NodeReputation(node_id="half")
+        rep.apply_penalty(0.2, "test")
+        assert abs(rep.reputation - 0.9) < 1e-9
+        assert len(rep.penalties) == 1
+        assert abs(rep.penalties[0]["amount"] - 0.1) < 1e-9
+        assert rep.penalties[0]["base_amount"] == 0.2
+
+    def test_set_penalty_scale_rejects_negative_and_nan(self):
+        for bad in (-0.1, float("nan"), float("inf"), "x"):
+            with pytest.raises(ValueError):
+                set_penalty_scale(bad)
+        assert NodeReputation.penalty_scale == 1.0  # unchanged by the rejections
+
+    def test_summary_reports_the_scale(self):
+        set_penalty_scale(0.0)
+        assert NodeReputation(node_id="n").summary()["penalty_scale"] == 0.0
+
+    def test_scale_is_not_a_dataclass_field(self):
+        # The backend snapshots reputations with asdict() and restores them
+        # with NodeReputation(**saved): a field would freeze the stance into
+        # saved state and break older snapshots on load.
+        set_penalty_scale(0.0)
+        saved = dataclasses.asdict(NodeReputation(node_id="n"))
+        assert "penalty_scale" not in saved
+        assert NodeReputation(**saved).node_id == "n"
+
+
 # ── Historical Coverage Map ──────────────────────────────────────────────────
 
 
@@ -277,6 +336,39 @@ def test_summary_fields(self, mgr):
         assert "reputation" in s
         assert "coverage_map" in s
 
+    def _bad_sample(self, ts_ms):
+        # predicted 0 vs measured 10 µs — way outside delay_threshold_us.
+        return AdsReportEntry(
+            timestamp_ms=ts_ms,
+            predicted_delay=0.0,
+            predicted_doppler=0.0,
+            measured_delay=10.0,
+            measured_doppler=0.0,
+            adsb_hex="abc123",
+            adsb_lat=34.0,
+            adsb_lon=-84.5,
+        )
+
+    def test_one_bad_sample_never_penalises(self, mgr):
+        # The ndebvzgeoij5t2l regression: one residual makes score 0.0, and
+        # before TRUST_MIN_SAMPLES that ran the node into a block in six
+        # minutes of 60 s passes.  Penalties are on (fixture pins scale 1.0).
+        mgr.record_adsb_correlation("node-A", self._bad_sample(1000))
+        for _ in range(20):
+            mgr.evaluate_reputations()
+        rep = mgr.reputations["node-A"]
+        assert rep.reputation == 1.0
+        assert rep.penalties == []
+        assert not rep.blocked
+
+    def test_three_bad_samples_do_penalise(self, mgr):
+        for i in range(3):
+            mgr.record_adsb_correlation("node-A", self._bad_sample(1000 + i))
+        mgr.evaluate_reputations()
+        rep = mgr.reputations["node-A"]
+        assert rep.reputation < 1.0
+        assert rep.penalties
+
     def test_evaluate_reputations(self, mgr):
         mgr.evaluate_reputations()
         # More than "does not raise": every registered node must still hold a