Skip to content

Arbiter (per-point routing)

When a reading looks wrong, there are different right answers: a single glitch should be replaced, a real change should be kept. The Arbiter decides, reading by reading, which case it is, and treats it accordingly.

A shield that always replaces deviating samples looks wrong on a real robot: it will smooth out the intentional step change the operator just commanded. A shield that never replaces them looks wrong on a noisy sensor: it lets every glitch through.

The Arbiter decides, point by point, which is which. It labels every sample clean, spike, or regime, and routes each label to the right corrector:

  • clean — passes through whatever the standard shield produced.
  • spike — hard-rejected to the median of the recent causal window.
  • regime — passed through raw, fully trusted.

The examples use a joint velocity at 100 Hz (noise 0.1 around 0.5) with a one-sample collision at sample 1500 and a step to a new level, +1.0, from sample 2000.

1. The three labels

The Arbiter uses the same robust-deviation score as every other detector in the package (see Streaming):

med = median(window)
S   = 1.4826 · median(|w − med|)
z   = |x − med| / S

A sample is deviating when z > n_sigmas (default n_sigmas = 3). Then the run of consecutive deviating samples decides the label:

  • run length ≤ spike_run_max (default 2) → spike
  • longer run, and it settles → regime
  • otherwise (longer run that does not settle) → spike (the whole run is treated as noise)

"Settles" is decided by two conditions on the run:

  1. Internal coherence: the run's own spread is small compared to its distance from the pre-run median: std(run) < 0.5 · |median(run) − med|.
  2. Persistence: the median of a short window after the run is closer to the run's median than to the pre-run median.

The two together are what separates "the signal has moved to a new level" from "the sensor briefly went haywire and came back".

2. Classify the whole series

import numpy as np
rng = np.random.default_rng(42)
v = 0.5 + 0.1 * rng.standard_normal(3000)
v[1500] = 2.5
v[2000:] += 1.0
from dense_armor.utility.protect.arbiter import classify_segments

lab, dev, unc = classify_segments(v, radius=20, ref_mult=3)
print(lab[1500], np.flatnonzero(lab == "regime")[[0, -1]], (lab == "spike").sum())

spike [1999 2026] 26

It returns the label of each sample, its robust deviation z, and an uncertainty in [0, 1] (highest near the n_sigmas boundary). The collision is a spike; samples 1999–2026 are one long, coherent, persistent run, so a regime; afterwards the reference has caught up and samples are clean again. The other 26 spikes are isolated noise values beyond 3 sigma, out of 3,000.

3. Route the labels

import numpy as np
rng = np.random.default_rng(42)
v = 0.5 + 0.1 * rng.standard_normal(3000)
v[1500] = 2.5
v[2000:] += 1.0
from dense_armor.utility.protect.arbiter import route_and_correct

corrected, lab, dev = route_and_correct(v, radius=20, ref_mult=3)
print(round(float(corrected[1500]), 3), np.array_equal(corrected[lab == "regime"], v[lab == "regime"]))

0.489 True

corrected is the same array with each label handled by its rule:

  • spike → replaced by the median of the causal window ending just before the run.
  • regime → unchanged.
  • clean → unchanged.

The collision becomes 0.489, the median before it; regime samples are left exactly as measured.

4. The reference window is causal

The window used to score every sample contains only the samples before it, never after. This matters more than it looks: a symmetric window straddling a real regime change dilutes its own MAD exactly where the detector needs it, because half the window is on the other side of the change. Causal-only means the pre-change scale is used to score the first post-change samples, which is what lets the run be detected as a run.

The same choice was made for every streaming detector in the package.

5. In Orca

Orca.protect_and_forward(..., use_arbiter=True) runs the Arbiter between Stage 1 (the adaptive stabilizer) and Stage 2 (the Collatz gate). Verified on the same 7 scenarios test/testKalman.py uses: never worse than the default (no Arbiter), better on 5 of 7.

from dense_armor.utility.protect.orca import Orca

import numpy as np
rng = np.random.default_rng(0)
data = np.sin(np.linspace(0, 6, 200))[None, :] + 0.01 * rng.standard_normal((4, 200))
data[0, 50] = 999.0
orca = Orca(min_free_ram_percentage=0.05)
out = orca.protect_and_forward(lambda z: 2.0 * z, data, use_arbiter=True)
print(out.shape, orca.etichette_arbitro.shape)

6. Streaming Arbiter (bounded delay)

The batch classify_segments needs to look after a run of deviating samples to decide whether it settled — which means it is not causal. StreamingArbiter gives the same answer with a bounded delay: each sample's final label is committed at most max_delay samples after it arrives, once the run has closed and enough context is available.

import numpy as np
rng = np.random.default_rng(42)
v = 0.5 + 0.1 * rng.standard_normal(3000)
v[1500] = 2.5
v[2000:] += 1.0
from dense_armor.utility.protect.streaming_arbiter import StreamingArbiter

arb = StreamingArbiter(radius=20, ref_mult=3, max_delay=200)
lab = {j: l for x in v for j, l, _ in arb.update(x)}
lab.update({j: l for j, l, _ in arb.flush()})
print(len(lab), lab[1500], lab[2000])

update(x) returns a list of (index, label, corrected_value) for the samples whose final label was decided at this step; flush() finalises whatever is still pending at the end of the stream.

With max_delay at least longest_run + radius * ref_mult, the labels and corrected values equal the batch classify_segments + route_and_correct on the same series. Smaller max_delay trades exactness for latency: labels near run boundaries may change.

7. StreamingHealing

StreamingHealing is the same bounded-delay idea applied to the healing filter (dense_armor.utility.protect.healing.healing_filter): a point is judged against a wide local baseline and replaced when the deviation is not shared by the majority of a narrow window of neighbours. Each sample gets its corrected value as soon as the wide symmetric window around it is fully available.

import numpy as np
rng = np.random.default_rng(42)
v = 0.5 + 0.1 * rng.standard_normal(3000)
v[1500] = 2.5
v[2000:] += 1.0
from dense_armor.utility.protect.streaming_arbiter import StreamingHealing

sh = StreamingHealing(radius=2, wide_mult=3, max_delay=30)
out = [o for x in v for o in sh.update(x)] + sh.flush()
print(len(out), out[0])

8. Hand case

Window of 9 samples [10, 11, 9, 10, 12, 10, 11, 9, 10], then a run of 3 samples [15, 16, 15.5], then 4 samples [15, 15, 15, 15].

  • med of the pre-run window is 10, MAD = 1, S = 1.4826.
  • Score for the first run sample: |15 − 10| / 1.4826 = 3.37 > 3 → deviating.
  • Same for the other two run samples.
  • Run length = 3 > spike_run_max = 2: it is not a spike.
  • std(run) = std([15, 16, 15.5]) = 0.41, |median(run) − med| = |15.5 − 10| = 5.5, 0.5 · 5.5 = 2.75; 0.41 < 2.75, so the run is internally coherent.
  • Median of 4 post-run samples is 15; |15 − 15.5| = 0.5 < |15 − 10| = 5: it persists.
  • Label: regime. The 3 samples pass through unchanged.

Change the three samples to [15, 8, 16]: 15 and 16 deviate, but 8 does not (|8 − 10| / 1.4826 = 1.35 < 3), so there are two runs of length 1, not one run of three. Labels: spike, clean, spike: 15 and 16 are replaced by 10, the 8 is kept.

API reference

arbiter

utility/arbiter.py

Instrada ogni punto verso lo specialista giusto invece di forzarne uno solo su tutto il segnale -- nato dal confronto reale in test/test_pressure_valve_vs_orca_experiment.py: hybrid_shield (phi_ab) vince nettamente sugli impulsi isolati (A, B, C, G) ma perde su un cambio di regime sostenuto (F), pressure_valve fa l'esatto contrario. Nessuno dei due batte l'altro ovunque -- la vera differenza tra i due casi e' la DURATA della deviazione, non la sua ampiezza.

APPROCCIO: per ogni punto, si confronta con una finestra di riferimento LARGA (raggio*ref_mult, la stessa scala usata dalla "molla" JSD di pressure_valve) -- deliberatamente non con la finestra locale stretta, che nel benchmark si e' vista collassare (mediana/MAD/IQR esattamente 0) quando piu' outlier vicini finiscono nella stessa finestra piccola. Questa finestra e' CAUSALE (_window_causal: solo punti PRIMA di i, mai dopo) -- una finestra simmetrica, usata proprio a cavallo di una transizione di livello, mescolerebbe vecchio e nuovo livello nella propria scala, diluendo la deviazione che dovrebbe far scattare la soglia (era cosi' nella prima versione di questo modulo, corretto dopo un caso reale non rilevato -- vedi cronologia in CHANGELOG.md [1.1.10]). Una volta segnati i punti "devianti", la LUNGHEZZA della sequenza consecutiva di punti devianti decide: corta (<= spike_run_max) = impulso isolato, lunga = cambio di regime genuino.

CASO DEGENERE (baseline esattamente piatta, scala robusta = 0): non e' un problema di stimatore -- vedi il LIMITE NOTO nel docstring di pressure_valve, un M-estimator di Huber converge comunque a scala 0 per lo stesso motivo strutturale. Qui non si divide mai per una scala zero: se la dispersione della finestra di riferimento e' sotto epsilon, QUALSIASI scostamento non nullo dalla mediana e' trattato come deviante senza bisogno di una soglia in sigma (su una baseline davvero costante, "diverso da costante" e' gia' la risposta, non serve altro).

CRONOLOGIA (limite trovato e poi risolto nella stessa sessione): la prima versione usava una finestra di riferimento SIMMETRICA, e lo scenario F (rottura strutturale, salto di livello permanente) non veniva rilevato come 'regime' -- deviazione[i] restava 0 su tutta la sequenza attorno alla transizione, perche' la finestra simmetrica, usata proprio a cavallo del salto, conteneva meta' vecchio livello e meta' nuovo, gonfiando la propria scala (MAD) invece di restare piccola. Un primo test aveva scambiato questo per un successo (RMSE 0.0 con route_and_correct standalone), ma era un artefatto: i dati grezzi di quello scenario coincidono GIA' esattamente col target (la "corruzione" e' il salto stesso, non rumore aggiunto), quindi "passa tutto grezzo" da' RMSE 0 indipendentemente da come viene etichettato ogni punto -- il vero banco di prova era la pipeline integrata di Orca (utility/orca.py, _applica_arbitro), dove la mancata rilevazione si traduceva in comportamento SICURO ma non migliorativo (9.4007 identico con o senza l'arbitro). Passando alla finestra CAUSALE (sopra), la rilevazione diventa reale e verificabile: sullo scenario F, gli indici 60-74 (esatti, verificato -- la transizione vera e' a 60) vengono etichettati 'regime', e attraverso la pipeline reale di Orca l'RMSE scende da 9.4007 (default) a 6.0449 (use_arbiter=True) -- un miglioramento reale, non piu' un pareggio. Vedi test/test_arbiter_orca_integration.py per la verifica completa sui 7 scenari con questo fix.

classify_segments 

classify_segments(
    x: ndarray,
    radius: int = 10,
    ref_mult: int = 3,
    n_sigmas: float = 3.0,
    spike_run_max: int = 2,
    eps: float = 1e-09,
) -> Tuple[np.ndarray, np.ndarray, np.ndarray]

Ritorna (etichette, deviazione, incertezza): etichette -- array di stringhe 'clean'/'spike'/'regime', una per punto. deviazione -- |x[i] - riferimento| / scala (o solo |x[i]-riferimento| se la scala e' degenere), stessa unita' per ogni punto. incertezza -- 0..1, quanto il punto e' vicino al confine di decisione (soglia deviante/pulito, o soglia spike/regime sulla lunghezza della sequenza) -- non quanto e' "corretto", ma quanto la classificazione stessa e' precaria.

Source code in dense_armor/utility/protect/arbiter.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def classify_segments(
    x: np.ndarray, radius: int = 10, ref_mult: int = 3,
    n_sigmas: float = 3.0, spike_run_max: int = 2, eps: float = 1e-9,
) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
    """
    Ritorna (etichette, deviazione, incertezza):
      etichette   -- array di stringhe 'clean'/'spike'/'regime', una per punto.
      deviazione  -- |x[i] - riferimento| / scala (o solo |x[i]-riferimento|
                     se la scala e' degenere), stessa unita' per ogni punto.
      incertezza  -- 0..1, quanto il punto e' vicino al confine di decisione
                     (soglia deviante/pulito, o soglia spike/regime sulla
                     lunghezza della sequenza) -- non quanto e' "corretto",
                     ma quanto la classificazione stessa e' precaria.
    """
    x = np.asarray(x, dtype=np.float64).ravel()
    n = x.size
    deviante = np.zeros(n, dtype=bool)
    deviazione = np.zeros(n, dtype=np.float64)
    scarto_normalizzato = np.zeros(n, dtype=np.float64)  # |dev|/soglia, per l'incertezza
    med_rif = np.zeros(n, dtype=np.float64)  # mediana della finestra di riferimento, per punto

    for i in range(n):
        w_rif = _window_causal(x, i, radius * ref_mult)
        if w_rif.size < 4:
            continue
        med, scala = _robust_center_scale(w_rif)
        med_rif[i] = med
        scarto = abs(x[i] - med)
        if scala < eps:
            # Baseline degenere: nessuna divisione, "diverso da costante"
            # e' gia' la risposta. scarto_normalizzato satura a un valore
            # alto/basso netto (mai vicino al confine per costruzione),
            # cosi' l'incertezza su questi punti resta bassa -- il caso
            # degenere e' l'UNICO dove la decisione e' davvero inequivocabile.
            deviazione[i] = scarto
            deviante[i] = scarto > eps
            scarto_normalizzato[i] = 0.0 if deviante[i] else 10.0
        else:
            deviazione[i] = scarto / scala
            deviante[i] = deviazione[i] > n_sigmas
            scarto_normalizzato[i] = deviazione[i] / n_sigmas

    etichette = np.full(n, "clean", dtype=object)
    incertezza = np.zeros(n, dtype=np.float64)

    # Incertezza sulla soglia deviante/pulito: vicino a scarto_normalizzato==1
    # (cioe' esattamente al confine n_sigmas) -> massima ambiguita'.
    incertezza_soglia = np.exp(-4.0 * (scarto_normalizzato - 1.0) ** 2)

    i = 0
    while i < n:
        if not deviante[i]:
            incertezza[i] = incertezza_soglia[i]
            i += 1
            continue
        j = i
        while j < n and deviante[j]:
            j += 1
        run_len = j - i
        if run_len <= spike_run_max:
            label = "spike"
        else:
            # Una sequenza lunga e' un vero cambio di regime solo se i suoi
            # punti si ASSESTANO su un nuovo livello coerente -- non basta
            # che siano "diversi dal vecchio", devono anche essere simili
            # TRA LORO. Verificato: 3 impulsi alternati (+500,-500,+500,
            # scenario A) sono una sequenza di lunghezza 3 (> spike_run_max
            # di default 2), ma la loro dispersione INTERNA e' enorme
            # (std~471) -- senza questo controllo venivano scambiati per un
            # regime e passati grezzi, RMSE 86.6 invece di ~0. Una sequenza
            # che si assesta davvero (scenario F) ha dispersione interna
            # piccola rispetto alla dimensione del salto stesso.
            run_vals = x[i:j]
            run_spread = float(np.std(run_vals))
            run_jump = float(abs(np.median(run_vals) - med_rif[i]))
            coerente = run_spread < 0.5 * max(run_jump, eps)

            # Coerenza interna non basta: un collasso temporaneo che poi
            # TORNA al vecchio livello (scenario C, 3 zeri poi ritorno a
            # 120) e' internamente coerente quanto un vero regime (spread
            # interno zero in entrambi i casi) -- la differenza vera si
            # vede SOLO guardando cosa succede subito dopo la sequenza:
            # un regime vero (scenario F) resta vicino al NUOVO livello,
            # un collasso torna verso quello VECCHIO. Senza questo
            # controllo scenario C restava RMSE 20.8 invece di ~0
            # (scambiato per regime, passato grezzo invece di respinto).
            post_lo, post_hi = j, min(n, j + radius)
            if post_hi > post_lo:
                post_med = float(np.median(x[post_lo:post_hi]))
                run_med = float(np.median(run_vals))
                persiste = abs(post_med - run_med) < abs(post_med - med_rif[i])
            else:
                persiste = True  # fine serie: nessun "dopo" da controllare

            label = "regime" if (coerente and persiste) else "spike"
        etichette[i:j] = label
        # Incertezza sulla soglia spike/regime: vicino a run_len==spike_run_max
        # -> massima ambiguita' sulla lunghezza; combinata (media) con quella
        # sulla soglia deviante/pulito dei singoli punti della sequenza.
        incertezza_run = float(np.exp(-1.0 * (run_len - spike_run_max) ** 2))
        incertezza[i:j] = 0.5 * incertezza_soglia[i:j] + 0.5 * incertezza_run
        i = j

    return etichette, deviazione, incertezza

route_and_correct 

route_and_correct(
    x: ndarray,
    radius: int = 10,
    ref_mult: int = 3,
    n_sigmas: float = 3.0,
    spike_run_max: int = 2,
) -> Tuple[np.ndarray, np.ndarray, np.ndarray]

Applica la classificazione e corregge ogni punto secondo l'etichetta: 'clean'/'regime' -> passa il valore grezzo (un cambio di regime genuino non va respinto, stessa filosofia del gate costante di Orca); 'spike' -> sostituito con la mediana della finestra di riferimento larga (robusta, non contaminata dal singolo impulso breve).

Ritorna (pulito, etichette, incertezza).

Source code in dense_armor/utility/protect/arbiter.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def route_and_correct(
    x: np.ndarray, radius: int = 10, ref_mult: int = 3,
    n_sigmas: float = 3.0, spike_run_max: int = 2,
) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
    """
    Applica la classificazione e corregge ogni punto secondo l'etichetta:
      'clean'/'regime' -> passa il valore grezzo (un cambio di regime
                          genuino non va respinto, stessa filosofia del
                          gate costante di Orca);
      'spike'          -> sostituito con la mediana della finestra di
                          riferimento larga (robusta, non contaminata
                          dal singolo impulso breve).

    Ritorna (pulito, etichette, incertezza)."""
    x = np.asarray(x, dtype=np.float64).ravel()
    n = x.size
    etichette, _, incertezza = classify_segments(x, radius, ref_mult, n_sigmas, spike_run_max)

    pulito = np.copy(x)
    for i in range(n):
        if etichette[i] == "spike":
            w_rif = _window(x, i, radius * ref_mult)
            pulito[i] = float(np.median(w_rif))

    return pulito, etichette, incertezza

streaming_arbiter

utility/streaming_arbiter.py

One-sample-at-a-time version of arbiter.classify_segments + route_and_correct.

The batch function decides spike vs regime by grouping consecutive deviant points into runs and looking at what happens after the run (to distinguish a genuine regime shift from a transient collapse that returns). This is fundamentally non-causal: the decision for a point requires future context. StreamingArbiter therefore decides with a bounded delay: each sample gets a provisional label at once (deviant / normal, from the causal window) and a final label (clean / spike / regime) after at most max_delay samples.

With max_delay large enough to cover both the longest run and the widest window the batch correction uses (radius * ref_mult), the final labels and corrected values equal the batch classify_segments + route_and_correct on the same series. With smaller max_delay, labels near run boundaries may change (documented in the latency table of arbiter.md).

StreamingArbiter 

StreamingArbiter(
    radius: int = 10,
    ref_mult: int = 3,
    n_sigmas: float = 3.0,
    spike_run_max: int = 2,
    max_delay: int = 40,
    eps: float = 1e-09,
)

Bounded-delay streaming version of arbiter.classify_segments.

update(x) adds a sample and returns the list of samples whose final label was decided at this step, as (index, label, corrected_value). flush() finalises everything still pending (call it at the end of a series).

Parameters

radius, ref_mult, n_sigmas, spike_run_max, eps Same meaning and defaults as arbiter.classify_segments. max_delay Maximum number of future samples the streaming version will wait before committing to a final label. With max_delay at least longest_run + radius * ref_mult, the result is identical to the batch function; smaller values trade exactness for lower latency.

Examples

import numpy as np from dense_armor.utility.protect.streaming_arbiter import StreamingArbiter rng = np.random.default_rng(0) x = list(rng.normal(0, 1, 100)) + [50.0] + list(rng.normal(0, 1, 100)) sa = StreamingArbiter(radius=10, max_delay=200) decided = [] for v in x: ... decided.extend(sa.update(v)) decided.extend(sa.flush()) spike_idx = [i for i, lab, _ in decided if lab == "spike"] 100 in spike_idx True

Source code in dense_armor/utility/protect/streaming_arbiter.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def __init__(self, radius: int = 10, ref_mult: int = 3,
             n_sigmas: float = 3.0, spike_run_max: int = 2,
             max_delay: int = 40, eps: float = 1e-9):
    self.radius = radius
    self.ref_mult = ref_mult
    self.n_sigmas = n_sigmas
    self.spike_run_max = spike_run_max
    self.max_delay = int(max_delay)
    self.eps = eps
    self._x: List[float] = []
    self._deviante: List[bool] = []
    self._med_rif: List[float] = []
    self._labels: List[Optional[str]] = []
    self._corrected: List[float] = []
    self._next_undecided: int = 0
    self._emitted: List[Tuple[int, str, float]] = []

update 

update(x: float) -> List[Tuple[int, str, float]]

Add a sample and return the samples finalised at this step.

Source code in dense_armor/utility/protect/streaming_arbiter.py 
160
161
162
163
164
165
166
167
168
def update(self, x: float) -> List[Tuple[int, str, float]]:
    """Add a sample and return the samples finalised at this step."""
    self._emitted = []
    self._x.append(float(x))
    self._labels.append(None)
    self._corrected.append(float(x))
    self._compute_deviante(len(self._x) - 1)
    self._sweep(forced=False)
    return list(self._emitted)

flush 

flush() -> List[Tuple[int, str, float]]

Finalise every sample still pending.

Source code in dense_armor/utility/protect/streaming_arbiter.py 
170
171
172
173
174
def flush(self) -> List[Tuple[int, str, float]]:
    """Finalise every sample still pending."""
    self._emitted = []
    self._sweep(forced=True)
    return list(self._emitted)

StreamingHealing 

StreamingHealing(
    radius: int = 2,
    sustain_threshold: float = 0.7,
    wide_mult: int = 3,
    max_delay: int = 30,
)

One-sample-at-a-time version of healing.healing_filter.

healing_filter judges each point against a wide local baseline and replaces it when the deviation is not shared by the majority of a narrow window of neighbours. It uses a symmetric wide window, so the streaming version decides with the same bounded-delay idea as StreamingArbiter: each sample gets a corrected value as soon as the wide symmetric window around it is fully available.

Parameters

radius, sustain_threshold, wide_mult Same meaning and defaults as healing.healing_filter. max_delay Maximum number of samples to wait before finalising.

Examples

import numpy as np from dense_armor.utility.protect.streaming_arbiter import StreamingHealing rng = np.random.default_rng(0) x = list(rng.normal(0, 0.1, 50)) + [5.0] + list(rng.normal(0, 0.1, 50)) sh = StreamingHealing(radius=2, wide_mult=3, max_delay=100) out = [] for v in x: ... out.extend(sh.update(v)) out.extend(sh.flush()) out = [v for _, v in sorted(out, key=lambda t: t[0])] abs(out[50]) < 1.0 True

Source code in dense_armor/utility/protect/streaming_arbiter.py 
210
211
212
213
214
215
216
217
218
219
def __init__(self, radius: int = 2, sustain_threshold: float = 0.7,
             wide_mult: int = 3, max_delay: int = 30):
    self.radius = radius
    self.sustain_threshold = sustain_threshold
    self.wide_mult = wide_mult
    self.max_delay = int(max_delay)
    self._x: List[float] = []
    self._decided: List[Optional[float]] = []
    self._next: int = 0
    self._emitted: List[Tuple[int, float]] = []

Details

The Arbiter grew out of a real benchmark where hybrid_engine-style rejection won on isolated impulses while pressure_valve's tolerance won on a sustained level change, with neither beating the other everywhere. The obvious answer — pick one per situation — required a classifier for "situation", and the run-length + coherence rule is what the benchmark showed to be the right classifier for the cases that appeared.

The reference window is causal for a specific reason found during validation: a symmetric window straddling a real transition dilutes its own MAD at exactly the positions where the transition is being classified, so a run that starts right at the change is scored against a scale that is half on the old level and half on the new, and neither side wins. Causal-only means the pre-change scale is used for the first post-change samples, which is what lets the run be recognized as a run.