rch_common/
classifier_drift.rs

1//! Classifier drift detection and bounded refresh
2//! (bd-session-history-remediation-ocv9i.6.3).
3//!
4//! Session history captured a subtle placement-integrity failure: the fleet had
5//! healthy *free* slots, yet RCH ran the build locally — not because of capacity
6//! or pressure, but because **stale classifier state** mis-decided the command
7//! until a refresh corrected it. That must be reported distinctly from "no
8//! capacity" and "critical pressure", or operators chase the wrong cause.
9//!
10//! This module is the pure detection + bounded-refresh contract:
11//! [`assess_drift`] decides whether a local choice was genuine or a drift
12//! (chose-local *with* healthy free slots, where a classifier refresh changes
13//! the placement toward remote), classifies it [`DriftKind::Transient`] (refresh
14//! restored remote placement) vs [`DriftKind::Persistent`] (refresh could not),
15//! and records the before/after classification + selected-worker
16//! [`ClassificationTransition`]. [`refresh_allowed`] bounds refresh attempts.
17//! A persistent drift maps to [`AdmissionRejectionCategory::ClassifierLocalDrift`]
18//! and is the distinct reason proof mode fails on.
19
20use serde::{Deserialize, Serialize};
21
22use crate::admission_rejection::AdmissionRejectionCategory;
23
24/// The before/after picture across a classifier refresh.
25#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
26pub struct ClassificationTransition {
27    /// Whether the command classified as compilation BEFORE the refresh.
28    pub compilation_before: bool,
29    /// Whether it classifies as compilation AFTER the refresh.
30    pub compilation_after: bool,
31    /// Worker selected before the refresh (`None` = ran/would run local).
32    #[serde(default, skip_serializing_if = "Option::is_none")]
33    pub worker_before: Option<String>,
34    /// Worker selected after the refresh (`None` = still local).
35    #[serde(default, skip_serializing_if = "Option::is_none")]
36    pub worker_after: Option<String>,
37}
38
39impl ClassificationTransition {
40    /// Whether the refresh changed the placement picture at all.
41    #[must_use]
42    pub fn changed(&self) -> bool {
43        self.compilation_before != self.compilation_after || self.worker_before != self.worker_after
44    }
45
46    /// Whether remote placement is possible after the refresh.
47    #[must_use]
48    pub fn remote_after(&self) -> bool {
49        self.worker_after.is_some()
50    }
51}
52
53/// Observable inputs for drift detection. Plain data; pure assessment.
54#[derive(Debug, Clone, PartialEq, Eq)]
55pub struct DriftObservation {
56    /// RCH chose (or would choose) local execution.
57    pub chose_local: bool,
58    /// Healthy, free build slots available across the fleet at decision time.
59    /// Non-zero means the local choice was NOT capacity-bound.
60    pub healthy_free_slots: u32,
61    /// The before/after classifier transition across a refresh.
62    pub transition: ClassificationTransition,
63}
64
65/// The drift classification.
66#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
67#[serde(rename_all = "snake_case")]
68pub enum DriftKind {
69    /// No drift — local was legitimate (genuinely non-compilation) or the choice
70    /// was capacity-bound (reported separately as no-capacity/pressure).
71    NoDrift,
72    /// Drift detected and a classifier refresh restored remote placement.
73    Transient,
74    /// Drift detected but a refresh could NOT restore remote placement — proof
75    /// mode must fail with a distinct reason.
76    Persistent,
77}
78
79impl DriftKind {
80    /// Stable lowercase token.
81    #[must_use]
82    pub fn as_str(self) -> &'static str {
83        match self {
84            DriftKind::NoDrift => "no_drift",
85            DriftKind::Transient => "transient",
86            DriftKind::Persistent => "persistent",
87        }
88    }
89}
90
91/// The drift report for admission/proof surfaces.
92#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
93pub struct DriftReport {
94    pub kind: DriftKind,
95    /// Healthy free slots observed (echoed so the report proves the local choice
96    /// was not capacity-bound).
97    pub healthy_free_slots: u32,
98    pub transition: ClassificationTransition,
99    /// Refresh attempts performed to reach this state.
100    pub refresh_attempts: u32,
101    pub detail: String,
102}
103
104impl DriftReport {
105    /// Whether drift was detected (transient or persistent).
106    #[must_use]
107    pub fn is_drift(&self) -> bool {
108        !matches!(self.kind, DriftKind::NoDrift)
109    }
110
111    /// The admission-rejection category for a drift (for aggregation via 6.2).
112    #[must_use]
113    pub fn rejection_category(&self) -> Option<AdmissionRejectionCategory> {
114        self.is_drift()
115            .then_some(AdmissionRejectionCategory::ClassifierLocalDrift)
116    }
117
118    /// Whether this drift must fail proof mode with a distinct reason — true
119    /// only for a persistent drift (refresh could not restore remote placement).
120    #[must_use]
121    pub fn blocks_proof(&self) -> bool {
122        matches!(self.kind, DriftKind::Persistent)
123    }
124}
125
126/// Assess classifier drift. Pure and total.
127///
128/// Drift requires ALL of: the command ran/would run local, healthy free slots
129/// existed (so it was not capacity-bound), and the refresh moved the picture
130/// toward remote (classification flipped to compilation, or a worker became
131/// selectable). If after the refresh a worker is selectable it is `Transient`
132/// (restored); if the picture shifted toward remote but no worker is yet
133/// selectable it is `Persistent`. Everything else is `NoDrift`.
134#[must_use]
135pub fn assess_drift(obs: &DriftObservation, refresh_attempts: u32) -> DriftReport {
136    let t = &obs.transition;
137    let drift = obs.chose_local
138        && obs.healthy_free_slots > 0
139        && t.changed()
140        && (t.compilation_after || t.remote_after());
141
142    let kind = if !drift {
143        DriftKind::NoDrift
144    } else if t.remote_after() {
145        DriftKind::Transient
146    } else {
147        DriftKind::Persistent
148    };
149
150    let detail = match kind {
151        DriftKind::NoDrift if obs.chose_local && obs.healthy_free_slots == 0 => {
152            "local choice was capacity-bound (no free slots), not classifier drift".to_string()
153        }
154        DriftKind::NoDrift => "no classifier drift; local choice was legitimate".to_string(),
155        DriftKind::Transient => format!(
156            "classifier drift: ran local despite {} free slot(s); refresh restored remote placement on {}",
157            obs.healthy_free_slots,
158            t.worker_after.as_deref().unwrap_or("a worker"),
159        ),
160        DriftKind::Persistent => format!(
161            "classifier drift: ran local despite {} free slot(s); refresh could NOT restore remote placement",
162            obs.healthy_free_slots,
163        ),
164    };
165
166    DriftReport {
167        kind,
168        healthy_free_slots: obs.healthy_free_slots,
169        transition: t.clone(),
170        refresh_attempts,
171        detail,
172    }
173}
174
175/// Bounded-refresh budget so a misclassification can trigger at most a fixed
176/// number of refreshes (a refresh storm is itself a failure mode).
177#[derive(Debug, Clone, Copy, PartialEq, Eq)]
178pub struct RefreshBudget {
179    pub max_attempts: u32,
180}
181
182impl Default for RefreshBudget {
183    fn default() -> Self {
184        Self { max_attempts: 2 }
185    }
186}
187
188/// Whether another classifier refresh is permitted given attempts already done.
189#[must_use]
190pub fn refresh_allowed(attempts_done: u32, budget: &RefreshBudget) -> bool {
191    attempts_done < budget.max_attempts
192}
193
194#[cfg(test)]
195mod tests {
196    use super::*;
197
198    fn transition(
199        before: bool,
200        after: bool,
201        worker_before: Option<&str>,
202        worker_after: Option<&str>,
203    ) -> ClassificationTransition {
204        ClassificationTransition {
205            compilation_before: before,
206            compilation_after: after,
207            worker_before: worker_before.map(str::to_string),
208            worker_after: worker_after.map(str::to_string),
209        }
210    }
211
212    #[test]
213    fn the_session_history_case_is_transient_drift() {
214        // Fleet had free slots; ran local; refresh flips classification to
215        // compilation and a worker becomes selectable.
216        let obs = DriftObservation {
217            chose_local: true,
218            healthy_free_slots: 3,
219            transition: transition(false, true, None, Some("css")),
220        };
221        let report = assess_drift(&obs, 1);
222        assert_eq!(report.kind, DriftKind::Transient);
223        assert!(report.is_drift());
224        assert_eq!(
225            report.rejection_category(),
226            Some(AdmissionRejectionCategory::ClassifierLocalDrift)
227        );
228        assert!(!report.blocks_proof(), "transient drift was restored");
229        assert!(report.detail.contains("free slot"));
230    }
231
232    #[test]
233    fn capacity_bound_local_is_not_drift() {
234        // Ran local but NO free slots => capacity-bound, reported separately.
235        let obs = DriftObservation {
236            chose_local: true,
237            healthy_free_slots: 0,
238            transition: transition(true, true, None, Some("css")),
239        };
240        let report = assess_drift(&obs, 0);
241        assert_eq!(report.kind, DriftKind::NoDrift);
242        assert!(!report.is_drift());
243        assert!(report.detail.contains("capacity-bound"));
244    }
245
246    #[test]
247    fn persistent_drift_blocks_proof() {
248        // Free slots; ran local; refresh flips classification to compilation but
249        // STILL no worker selectable => refresh could not restore remote.
250        let obs = DriftObservation {
251            chose_local: true,
252            healthy_free_slots: 2,
253            transition: transition(false, true, None, None),
254        };
255        let report = assess_drift(&obs, 2);
256        assert_eq!(report.kind, DriftKind::Persistent);
257        assert!(report.blocks_proof());
258        assert_eq!(
259            report.rejection_category(),
260            Some(AdmissionRejectionCategory::ClassifierLocalDrift)
261        );
262    }
263
264    #[test]
265    fn going_remote_is_never_drift() {
266        let obs = DriftObservation {
267            chose_local: false,
268            healthy_free_slots: 4,
269            transition: transition(true, true, Some("css"), Some("css")),
270        };
271        assert_eq!(assess_drift(&obs, 0).kind, DriftKind::NoDrift);
272    }
273
274    #[test]
275    fn genuinely_non_compilation_local_is_not_drift() {
276        // Free slots, ran local, and refresh changes NOTHING toward remote =>
277        // local was legitimately correct.
278        let obs = DriftObservation {
279            chose_local: true,
280            healthy_free_slots: 5,
281            transition: transition(false, false, None, None),
282        };
283        let report = assess_drift(&obs, 1);
284        assert_eq!(report.kind, DriftKind::NoDrift);
285        assert!(!report.is_drift());
286        assert!(report.rejection_category().is_none());
287    }
288
289    #[test]
290    fn drift_is_separate_from_capacity_and_pressure() {
291        // Drift uses its own category (ClassifierLocalDrift), never
292        // InsufficientSlots/CriticalPressure.
293        let obs = DriftObservation {
294            chose_local: true,
295            healthy_free_slots: 1,
296            transition: transition(false, true, None, Some("w")),
297        };
298        let cat = assess_drift(&obs, 1).rejection_category().unwrap();
299        assert_eq!(cat, AdmissionRejectionCategory::ClassifierLocalDrift);
300        assert_ne!(cat, AdmissionRejectionCategory::InsufficientSlots);
301        assert_ne!(cat, AdmissionRejectionCategory::CriticalPressure);
302    }
303
304    #[test]
305    fn bounded_refresh_stops_at_budget() {
306        let budget = RefreshBudget { max_attempts: 2 };
307        assert!(refresh_allowed(0, &budget));
308        assert!(refresh_allowed(1, &budget));
309        assert!(!refresh_allowed(2, &budget));
310        assert!(!refresh_allowed(3, &budget));
311        assert_eq!(RefreshBudget::default().max_attempts, 2);
312    }
313
314    #[test]
315    fn transition_records_before_after() {
316        let t = transition(false, true, None, Some("css"));
317        assert!(t.changed());
318        assert!(t.remote_after());
319        assert!(!t.compilation_before);
320        assert!(t.compilation_after);
321    }
322
323    #[test]
324    fn report_serializes_with_stable_tokens() {
325        let obs = DriftObservation {
326            chose_local: true,
327            healthy_free_slots: 3,
328            transition: transition(false, true, None, Some("css")),
329        };
330        let report = assess_drift(&obs, 1);
331        let value = serde_json::to_value(&report).unwrap();
332        assert_eq!(value["kind"], "transient");
333        assert_eq!(value["healthy_free_slots"], 3);
334        assert_eq!(value["transition"]["worker_after"], "css");
335        let back: DriftReport = serde_json::from_value(value).unwrap();
336        assert_eq!(back, report);
337    }
338}
rch_common/classifier_drift.rs

rch_common/
classifier_drift.rs
