degenbot-workers 0.6.0-alpha.14

Role-switching worker fleet: WorkerRole FSM, one budget authority, bounded priority dispatch, cordon posture (ADR-042).
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
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
196
197
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
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
//! The `Nominal โ‡„ Cordoned` posture FSM (design doc ยง6) โ€” the first
//! consumer of `degenbot.cgroup.throttled` (`cpu_budget::cgroup_throttle_delta`).
//!
//! Thresholds are typed, runtime-tunable config keys (the sign-off
//! amendment 2026-09-09: enter triggers, exit window, and cordon effects
//! calibrated from soak data via the operator channel โ€” never share
//! arithmetic, which is [`crate::budget`]'s authority).
//!
//! Entering/exiting is LOUD: a transition fires a structured log line plus
//! the posture counters (never a silent degrade). Time is passed in as
//! monotonic milliseconds so the FSM is deterministic under test; callers
//! feed `Instant::now()` deltas from their throttle poller.

use degenbot_core::diag;
use degenbot_core::{op_info, op_warn};
use std::collections::VecDeque;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, OnceLock};

use degenbot_config::FleetConfig;
use parking_lot::{Mutex, RwLock};

use crate::role::CordonClass;

/// Process-level fleet posture. NOT a slot state (design doc ยง3.2): it
/// gates lease transitions, it never sheds a running unit.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum FleetPosture {
    /// Normal operation: every dispatchable role leases freely.
    Nominal,
    /// Under cgroup throttling: no new leases for cordon-deferrable roles,
    /// sim intake floored; in-flight units COMPLETE (T7/T8), pins are never
    /// shed, the merge pin and ambient I/O are never cordoned.
    Cordoned,
}

/// Why the posture entered cordon (exported as the transition's cause).
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum EnterReason {
    /// โ‰ฅ `enter_events` throttle events within the rolling enter window.
    EventBurst {
        /// The events observed in the window.
        events: u64,
    },
    /// Throttled-time duty exceeded `duty_percent` over the duty window.
    DutySpike {
        /// Measured duty percent.
        duty_percent: f64,
    },
    /// A lane died mid-flight (the DECIDED option (a)
    /// input). Entered with its OWN exit discipline: see
    /// [`PostureCause::LaneDeath`].
    LaneDeath,
}

/// A typed non-throttle posture cause (the DECIDED
/// option (a): the input is typed AT the posture owner, not a sample
/// it has to infer from; the failure taxonomy stays CLOSED per
/// ADR-040's per-bucket reactions).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PostureCause {
    /// A lane died mid-flight: enter the cordon IMMEDIATELY (no sample
    /// hysteresis โ€” the fleet is degraded NOW) with its OWN exit
    /// discipline: the clean window NEVER lifts a lane-death cordon
    /// (the lane is still dead) โ€” the cordon is STICKY until a fresh
    /// process. In-flight paths get terminal receipts; the process
    /// stays alive (the cordoned posture holds deferrable intake and
    /// floors sim intake โ€” the ยง6 cordon effects).
    LaneDeath,
}

/// The posture change a [`PostureStateMachine::observe`] tick produced.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum PostureChange {
    /// No transition (the FSM does not re-enter while already cordoned).
    Held,
    /// Nominal โ†’ Cordoned (loud: log at warn + counter).
    Entered(EnterReason),
    /// Cordoned โ†’ Nominal after the clean-window hysteresis (log at info).
    Exited,
}

/// Typed thresholds (Q5 amendment) โ€” [`PosturePolicy::from_config`] is the
/// boot source; the operator channel re-derives from the same schema keys.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct PosturePolicy {
    /// Enter trigger (a): >= this many throttle events in the enter window.
    pub enter_events: usize,
    /// Enter window (a): the rolling burst window.
    pub enter_window_ms: u64,
    /// Enter trigger (b): throttled-time duty percent over the duty window
    /// (`2.0` = >2%).
    pub duty_percent: f64,
    /// Enter trigger (b) window.
    pub duty_window_ms: u64,
    /// Exit: this much continuous clean time required after the last dirty
    /// sample (hysteresis prevents flapping; ยง6: 10 s).
    pub exit_clean_ms: u64,
    /// Cordon effect (b): sim intake cap while cordoned; `None` = half the
    /// slot cap (in-flight sims are never cancelled).
    pub sim_intake_floor_override: Option<usize>,
}

impl PosturePolicy {
    /// The design-doc ยง6 defaults (2 events / 1 s, >2% / 5 s, 10 s clean),
    /// used when a host boots without a typed config (hermetic runs).
    #[must_use]
    pub const fn doc_defaults() -> Self {
        Self {
            enter_events: 2,
            enter_window_ms: 1_000,
            duty_percent: 2.0,
            duty_window_ms: 5_000,
            exit_clean_ms: 10_000,
            sim_intake_floor_override: None,
        }
    }

    /// The typed-config projection โ€” every threshold is a schema key.
    #[must_use]
    pub fn from_config(cfg: &FleetConfig) -> Self {
        Self {
            enter_events: cfg.cordon_enter_events,
            enter_window_ms: cfg.cordon_enter_window_ms,
            duty_percent: cfg.cordon_duty_percent,
            duty_window_ms: cfg.cordon_duty_window_ms,
            exit_clean_ms: cfg.cordon_exit_clean_ms,
            sim_intake_floor_override: cfg.cordon_sim_intake_floor,
        }
    }

    /// The cordon sim-intake cap: override or half the slot cap, floored
    /// at 1, never above the cap (ยง6 effect (b)).
    #[must_use]
    pub fn sim_intake_cap(&self, slot_cap: usize) -> usize {
        self.sim_intake_floor_override
            .unwrap_or(slot_cap / 2)
            .min(slot_cap)
            .max(1)
    }

    /// Apply a validated [`PosturePolicyPatch`] to `self`, producing the
    /// effective policy (the re-tune channel's only write
    /// path: current policy + supplied fields). Pure โ€” the caller feeds the
    /// result to [`PostureOwner::retune`]. Call [`PosturePolicyPatch::validate`]
    /// FIRST; this projection never checks semantics.
    #[must_use]
    pub fn patched_with(self, patch: PosturePolicyPatch) -> Self {
        Self {
            enter_events: patch.enter_events.unwrap_or(self.enter_events),
            enter_window_ms: patch.enter_window_ms.unwrap_or(self.enter_window_ms),
            duty_percent: patch.duty_percent.unwrap_or(self.duty_percent),
            duty_window_ms: patch.duty_window_ms.unwrap_or(self.duty_window_ms),
            exit_clean_ms: patch.exit_clean_ms.unwrap_or(self.exit_clean_ms),
            sim_intake_floor_override: match patch.sim_intake_floor_override {
                // Key absent: keep the current override.
                None => self.sim_intake_floor_override,
                // Key present: set it โ€” `Some(v)` = explicit floor,
                // `None` = cleared (back to half the slot cap).
                Some(floor) => floor,
            },
        }
    }
}

/// A partial re-tune request over the six typed thresholds (the
/// operator channel's wire shape): every field is `None` = "key not
/// supplied โ€” keep the current value". `sim_intake_floor_override` is
/// doubly-`Option`: the OUTER `None` is key-absent, and the inner
/// `Some(None)` is the operator supplying the key's `None` value (clear the
/// override, back to half the slot cap โ€” the typed key itself is
/// `opt usize`).
///
/// The semantic rules (windows > 0, duty percent in range, floors >= 1,
/// non-empty patch) are encoded ONCE, in [`Self::validate`] โ€” callers
/// REJECT, never clamp silently.
#[derive(Debug, Clone, Copy, Default, PartialEq)]
pub struct PosturePolicyPatch {
    /// `cordon_enter_events`: the burst trigger count.
    pub enter_events: Option<usize>,
    /// `cordon_enter_window_ms`: the burst window.
    pub enter_window_ms: Option<u64>,
    /// `cordon_duty_percent`: the duty trigger percent.
    pub duty_percent: Option<f64>,
    /// `cordon_duty_window_ms`: the duty window.
    pub duty_window_ms: Option<u64>,
    /// `cordon_exit_clean_ms`: the exit hysteresis.
    pub exit_clean_ms: Option<u64>,
    /// `cordon_sim_intake_floor`: outer `None` = key absent; inner
    /// `Some(None)` = clear the override; `Some(Some(n))` = explicit floor.
    pub sim_intake_floor_override: Option<Option<usize>>,
}

impl PosturePolicyPatch {
    /// Whether NO key was supplied (an empty patch โ€” rejected by
    /// [`Self::validate`]; a re-tune that changes nothing must never look
    /// like a successful one).
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.enter_events.is_none()
            && self.enter_window_ms.is_none()
            && self.duty_percent.is_none()
            && self.duty_window_ms.is_none()
            && self.exit_clean_ms.is_none()
            && self.sim_intake_floor_override.is_none()
    }

    /// The ONE encoding of the re-tune's semantic rules. Every violation is
    /// a typed [`PostureRetuneError`] naming the offending key and value โ€”
    /// the channel rejects, it never clamps.
    ///
    /// # Errors
    ///
    /// [`PostureRetuneError::EmptyPatch`] when no key was supplied, or the
    /// per-key range error for the first offending value.
    pub fn validate(&self) -> Result<(), PostureRetuneError> {
        if self.is_empty() {
            return Err(PostureRetuneError::EmptyPatch);
        }
        if self.enter_events.is_some_and(|v| v < 1) {
            return Err(PostureRetuneError::EnterEvents(
                self.enter_events.unwrap_or_default(),
            ));
        }
        if self.enter_window_ms.is_some_and(|v| v == 0) {
            return Err(PostureRetuneError::EnterWindow(0));
        }
        if let Some(duty) = self.duty_percent {
            // The sane inclusive range: a measured duty percent is a
            // (0.0, 100.0] quantity โ€” 0 would cordon on ANY throttled
            // microsecond and >100 or non-finite cannot be a duty.
            if !(duty > 0.0 && duty <= 100.0) {
                return Err(PostureRetuneError::DutyPercent(duty));
            }
        }
        if self.duty_window_ms.is_some_and(|v| v == 0) {
            return Err(PostureRetuneError::DutyWindow(0));
        }
        if self.exit_clean_ms.is_some_and(|v| v == 0) {
            return Err(PostureRetuneError::ExitClean(0));
        }
        if let Some(Some(floor)) = self.sim_intake_floor_override {
            if floor < 1 {
                return Err(PostureRetuneError::SimIntakeFloor(floor));
            }
        }
        Ok(())
    }
}

/// Why the re-tune channel refused a [`PosturePolicyPatch`]. The rules
/// live once, in [`PosturePolicyPatch::validate`]; the wire
/// layer maps these to its typed channel error verbatim.
#[derive(Debug, Clone, Copy, PartialEq, thiserror::Error)]
pub enum PostureRetuneError {
    /// No threshold key was supplied โ€” an empty patch is refused.
    #[error("at least one cordon threshold key is required (empty patch)")]
    EmptyPatch,
    /// `cordon_enter_events` below the floor of 1.
    #[error("cordon_enter_events must be >= 1, got {0}")]
    EnterEvents(usize),
    /// `cordon_enter_window_ms` must be a positive window.
    #[error("cordon_enter_window_ms must be > 0 ms, got {0} ms")]
    EnterWindow(u64),
    /// `cordon_duty_percent` outside the sane (0.0, 100.0] range.
    #[error("cordon_duty_percent must be in (0.0, 100.0], got {0}")]
    DutyPercent(f64),
    /// `cordon_duty_window_ms` must be a positive window.
    #[error("cordon_duty_window_ms must be > 0 ms, got {0} ms")]
    DutyWindow(u64),
    /// `cordon_exit_clean_ms` must be a positive window.
    #[error("cordon_exit_clean_ms must be > 0 ms, got {0} ms")]
    ExitClean(u64),
    /// `cordon_sim_intake_floor` below the floor of 1.
    #[error("cordon_sim_intake_floor must be >= 1, got {0}")]
    SimIntakeFloor(usize),
}

/// One throttle-poll delta: `cgroup_throttle_delta()`'s counters plus the
/// elapsed wall time of the poll interval.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ThrottleSample {
    /// `nr_throttled` delta since the last sample.
    pub events: u64,
    /// `throttled_usec` delta since the last sample.
    pub throttled_usec: u64,
    /// Elapsed wall time (ยตs) of the poll interval backing the deltas.
    pub elapsed_usec: u64,
}

impl ThrottleSample {
    /// A clean sample: no throttle events, no throttled time.
    #[must_use]
    pub const fn is_clean(&self) -> bool {
        self.events == 0 && self.throttled_usec == 0
    }
}

/// Loud-transition counters (exported with the posture metrics so
/// thresholds are tuned against measurements, ยง6).
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct PostureCounters {
    /// Cordons entered.
    pub entered: u64,
    /// Cordons exited.
    pub exited: u64,
    /// Lease grants denied while cordoned (deferrable intake held + sim
    /// intake suppression above the floor).
    pub intake_suppressed: u64,
    /// Lane-death causes observed (the sticky cordons, incl.
    /// upgrades of an existing throttle cordon).
    pub lane_deaths: u64,
}

#[derive(Debug, Clone, Copy)]
struct Sample {
    now_ms: u64,
    sample: ThrottleSample,
}

/// The posture state machine. Deterministic: time is monotonic ms fed by
/// the caller.
#[derive(Debug)]
pub struct PostureStateMachine {
    state: FleetPosture,
    policy: PosturePolicy,
    samples: VecDeque<Sample>,
    last_unclean_ms: Option<u64>,
    counters: PostureCounters,
    /// A lane-death cordon is STICKY: once set, the clean-window
    /// exit refuses to lift the cordon โ€” the lane is still dead. Only
    /// a fresh process clears it (the operator restart path); there is
    /// deliberately no in-process clear API (a sticky cordon that
    /// silently un-sticks is exactly the silent-narrow class ยง10
    /// forbids).
    lane_death_hold: bool,
}

impl PostureStateMachine {
    /// A fresh machine in [`FleetPosture::Nominal`].
    #[must_use]
    pub fn new(policy: PosturePolicy) -> Self {
        Self {
            state: FleetPosture::Nominal,
            policy,
            samples: VecDeque::new(),
            last_unclean_ms: None,
            counters: PostureCounters::default(),
            lane_death_hold: false,
        }
    }

    /// Current posture.
    #[must_use]
    pub const fn state(&self) -> FleetPosture {
        self.state
    }

    /// Loud-transition counters (metrics export surface).
    #[must_use]
    pub const fn counters(&self) -> &PostureCounters {
        &self.counters
    }

    /// The sticky lane-death hold: once set, no clean window can
    /// ever lift it โ€” the `FleetHost` Faulted transition keys on
    /// THIS typed latch, never on elapsed cordon time.
    #[must_use]
    pub const fn lane_death_held(&self) -> bool {
        self.lane_death_hold
    }

    /// The active policy (operator-channel re-tune reads it through here).
    #[must_use]
    pub const fn policy(&self) -> &PosturePolicy {
        &self.policy
    }

    /// Swap the policy (the operator channel's re-tune). Pure: the state
    /// and the trailing sample window are KEPT โ€” a retune never fabricates
    /// samples, so the next [`Self::observe`] re-derives the posture under
    /// the new thresholds. Semantic validation of the new policy is the
    /// re-tune caller's job (the re-tune channel).
    pub fn set_policy(&mut self, policy: PosturePolicy) {
        self.policy = policy;
    }

    /// Feed one throttle-poll delta at `now_ms`. Returns whether the
    /// posture transitioned (and why) โ€” the caller surfaces that loudly.
    pub fn observe(&mut self, now_ms: u64, sample: ThrottleSample) -> PostureChange {
        self.samples.push_back(Sample { now_ms, sample });
        self.prune(now_ms);
        if !sample.is_clean() {
            self.last_unclean_ms = Some(now_ms);
        }

        match self.state {
            FleetPosture::Nominal => self.maybe_enter(now_ms),
            FleetPosture::Cordoned => self.maybe_exit(now_ms),
        }
    }

    /// Feed one typed non-throttle cause (the DECIDED
    /// option (a) input). A [`PostureCause::LaneDeath`] enters the
    /// cordon from ANY state with its OWN exit discipline: the cordon
    /// is sticky (the clean window never lifts it โ€” see
    /// [`Self::maybe_exit`]), in-flight paths get terminal receipts at
    /// the detection site, and the process stays alive under the ยง6
    /// cordon effects. Idempotent while the hold is already set (the
    /// counter still counts every cause).
    pub fn observe_cause(&mut self, cause: PostureCause) -> PostureChange {
        match cause {
            PostureCause::LaneDeath => {
                self.counters.lane_deaths += 1;
                if self.lane_death_hold {
                    return PostureChange::Held;
                }
                self.lane_death_hold = true;
                if self.state == FleetPosture::Cordoned {
                    // An existing (throttle) cordon UPGRADES to the
                    // sticky hold: no state transition, but the cause
                    // is new and loud.
                    op_warn!(domain = pump, lane_deaths = self.counters.lane_deaths,
                        "lane-death HOLD upgrades an existing cordon โ€” sticky, clean-window exit disabled"
                    );
                    return PostureChange::Held;
                }
                self.enter(EnterReason::LaneDeath)
            }
        }
    }

    /// A lease grant was denied because of the posture (counter only โ€” the
    /// dispatcher supplies the typed error).
    pub const fn note_intake_suppressed(&mut self) {
        self.counters.intake_suppressed += 1;
    }

    /// Whether the posture admits new lease intake for `class` right now
    /// (ยง6: `Never` and `SimPool` lease freely โ€” sim is only intake-FLOORED;
    /// `Deferrable` is held while cordoned).
    #[must_use]
    pub const fn admits_lease(&self, class: CordonClass) -> bool {
        !matches!(
            (self.state, class),
            (FleetPosture::Cordoned, CordonClass::Deferrable)
        )
    }

    /// The sim intake cap in the current posture (ยง6 effect (b)).
    #[must_use]
    pub fn sim_intake_cap(&self, slot_cap: usize) -> usize {
        match self.state {
            FleetPosture::Nominal => slot_cap,
            FleetPosture::Cordoned => self.policy.sim_intake_cap(slot_cap),
        }
    }

    fn prune(&mut self, now_ms: u64) {
        let window = self.policy.duty_window_ms.max(self.policy.enter_window_ms);
        while let Some(front) = self.samples.front() {
            if now_ms.saturating_sub(front.now_ms) > window {
                self.samples.pop_front();
            } else {
                break;
            }
        }
    }

    fn maybe_enter(&mut self, now_ms: u64) -> PostureChange {
        // Trigger (a): event burst inside the enter window.
        let burst: u64 = self
            .samples
            .iter()
            .filter(|s| now_ms.saturating_sub(s.now_ms) <= self.policy.enter_window_ms)
            .map(|s| s.sample.events)
            .sum();
        if burst >= u64::try_from(self.policy.enter_events.max(1)).unwrap_or(1) {
            return self.enter(EnterReason::EventBurst { events: burst });
        }
        // Trigger (b): duty percent over the trailing duty window.
        let (events, throttled_usec, elapsed_usec) = self.duty_window_totals();
        let _ = events;
        if elapsed_usec > 0 {
            #[expect(
                clippy::cast_precision_loss,
                reason = "duty percent is an f64 metric by definition (ยตs/ยตs ratio)"
            )]
            let duty_percent = throttled_usec as f64 / elapsed_usec as f64 * 100.0;
            if duty_percent > self.policy.duty_percent {
                return self.enter(EnterReason::DutySpike { duty_percent });
            }
        }
        PostureChange::Held
    }

    fn duty_window_totals(&self) -> (u64, u64, u64) {
        self.samples
            .iter()
            .fold((0, 0, 0), |(ev, th, el), Sample { sample, .. }| {
                (
                    ev + sample.events,
                    th + sample.throttled_usec,
                    el + sample.elapsed_usec,
                )
            })
    }

    fn enter(&mut self, reason: EnterReason) -> PostureChange {
        self.state = FleetPosture::Cordoned;
        self.counters.entered += 1;
        op_warn!(domain = pump, reason = ?reason,
            entered = self.counters.entered,
            "cordon ENTER โ€” deferrable intake held, sim intake floored; in-flight units complete"
        );
        PostureChange::Entered(reason)
    }

    fn maybe_exit(&mut self, now_ms: u64) -> PostureChange {
        // The lane-death hold is STICKY: the clean window never
        // lifts it โ€” the lane is still dead. Throttle samples keep
        // feeding the machine (harmless bookkeeping); only a fresh
        // process clears the hold.
        if self.lane_death_hold {
            return PostureChange::Held;
        }
        // Exit: `exit_clean_ms` of clean time since the last dirty sample
        // (hysteresis; ยง6: 10 s of clean windows).
        let dirty_recently = self
            .last_unclean_ms
            .is_some_and(|last| now_ms.saturating_sub(last) < self.policy.exit_clean_ms);
        if dirty_recently {
            return PostureChange::Held;
        }
        self.state = FleetPosture::Nominal;
        self.counters.exited += 1;
        op_info!(
            domain = pump,
            exited = self.counters.exited,
            clean_ms = self.policy.exit_clean_ms,
            "cordon EXIT after clean-window hysteresis"
        );
        PostureChange::Exited
    }
}

// ---- the ONE process-level fleet posture owner ------------

/// Watch-style subscription to the fleet posture feed โ€” the workers-crate
/// equivalent of a `tokio::sync::watch` receiver (the crate carries no
/// tokio; this is the `parking_lot` pattern its other feeds use). One
/// producer (the [`PostureOwner`]), many independent consumers: every
/// `FleetHost` subscribes at boot for the T7 shed trigger, tests subscribe
/// to observe transitions, and the Part B operator channel will drive the
/// owner directly. A consumer sees the latest posture and REAL transitions
/// only โ€” a `Held` tick or a no-effect retune never raises an edge.
#[derive(Debug)]
pub struct PostureWatch {
    shared: Arc<FeedShared>,
    seen: AtomicU64,
}

impl PostureWatch {
    /// The latest posture (reads through โ€” never consumes the edge).
    #[must_use]
    pub fn current(&self) -> FleetPosture {
        self.shared.state.read().posture
    }

    /// Did a transition occur since the last drain? (Does not consume.)
    #[must_use]
    pub fn has_changed(&self) -> bool {
        self.shared.state.read().seq != self.seen.load(Ordering::Relaxed)
    }

    /// Drain: the posture if a transition occurred since the last drain,
    /// else `None` (consuming the edge โ€” exactly-once per transition).
    #[must_use]
    pub fn take_if_changed(&self) -> Option<FleetPosture> {
        let state = self.shared.state.read();
        if state.seq == self.seen.load(Ordering::Relaxed) {
            return None;
        }
        self.seen.store(state.seq, Ordering::Relaxed);
        Some(state.posture)
    }
}

/// The broadcast cell behind the feed: the posture plus a monotonic
/// transition sequence (the edge counter the watches diff against).
#[derive(Debug)]
struct FeedShared {
    state: RwLock<FeedState>,
}

#[derive(Debug, Clone, Copy)]
struct FeedState {
    posture: FleetPosture,
    seq: u64,
}

/// The ONE owner of a fleet posture state machine โ€” a shared, thread-safe
/// shell around the pure [`PostureStateMachine`]. Every consumer (every
/// `FleetHost`, the block pump's throttle feed, tests) consults THE SAME
/// instance, so there is exactly one posture per process (per hermetic
/// test scope) and no host-local mirrors to drift apart.
///
/// Sync: the machine sits behind a `parking_lot::Mutex` โ€” every consult is
/// a short read-through critical section, never held across an await
/// (the crate has none); the transition feed is the [`PostureWatch`]
/// broadcast above.
#[derive(Debug)]
pub struct PostureOwner {
    machine: Mutex<PostureStateMachine>,
    broadcast: Arc<FeedShared>,
}

impl PostureOwner {
    /// A fresh owner in [`FleetPosture::Nominal`]. Hermetic tests build
    /// their own owner and inject it via `FleetBoot::owner` โ€” NEVER the
    /// process global ([`process`]/[`install_process_owner`]); posture
    /// leaking across tests is a failure class.
    #[must_use]
    pub fn new(policy: PosturePolicy) -> Self {
        Self {
            machine: Mutex::new(PostureStateMachine::new(policy)),
            broadcast: Arc::new(FeedShared {
                state: RwLock::new(FeedState {
                    posture: FleetPosture::Nominal,
                    seq: 0,
                }),
            }),
        }
    }

    /// Feed one throttle-poll delta. Publishes to the feed ONLY on a real
    /// transition (`Held` ticks are silent โ€” a subscriber never sees a
    /// spurious edge).
    ///
    /// # Feeder-site contract (ADR-044)
    /// A BOT-side caller of this method MUST also wake the fleet hosts on a
    /// non-`Held` change (the degenbot-bot host waker,
    /// `arb_engine::fleet_wake::wake_hosts`), which emits ONE untrusted,
    /// seq-stamped `PostureEdge` per host. This crate cannot know about host
    /// channels (layering), so the wake is the caller's obligation; the
    /// host's `BackstopTick` bounds the damage if a feeder forgets, and the
    /// hint never carries a posture value โ€” hosts re-read the live owner.
    pub fn observe_throttle(&self, now_ms: u64, sample: ThrottleSample) -> PostureChange {
        let (change, posture) = {
            let mut machine = self.machine.lock();
            let change = machine.observe(now_ms, sample);
            (change, machine.state())
        };
        if !matches!(change, PostureChange::Held) {
            self.publish(posture);
        }
        change
    }

    /// Feed one typed non-throttle cause . Publishes to
    /// the feed on a real transition like [`Self::observe_throttle`]
    /// (an idempotent hold-upgrade returns `Held` and stays silent โ€”
    /// the detection site owns the loud lane-death log).
    ///
    /// # Feeder-site contract (ADR-044)
    /// A BOT-side caller of this method MUST also wake the fleet hosts on a
    /// non-`Held` change; see [`Self::observe_throttle`]. On the lane-death
    /// (`Faulted`) arm the wake still fires, and the host drains its held
    /// receipts terminally instead of parking them.
    pub fn observe_cause(&self, cause: PostureCause) -> PostureChange {
        let (change, posture) = {
            let mut machine = self.machine.lock();
            let change = machine.observe_cause(cause);
            (change, machine.state())
        };
        if !matches!(change, PostureChange::Held) {
            self.publish(posture);
        }
        change
    }

    /// The current posture.
    #[must_use]
    pub fn current(&self) -> FleetPosture {
        self.machine.lock().state()
    }

    /// The active policy (read-through; the Part B operator channel reads
    /// and re-tunes through here).
    #[must_use]
    pub fn policy(&self) -> PosturePolicy {
        *self.machine.lock().policy()
    }

    /// Subscribe a watch: the receiver starts at the CURRENT posture with
    /// no pending edge (it observes only transitions from here on).
    #[must_use]
    pub fn subscribe(&self) -> PostureWatch {
        let seen = self.broadcast.state.read().seq;
        PostureWatch {
            shared: Arc::clone(&self.broadcast),
            seen: AtomicU64::new(seen),
        }
    }

    /// Swap the policy (the Part B operator channel's entry point). The
    /// swap is atomic under the machine lock and keeps the state + the
    /// trailing sample window; the feed re-publishes the current posture
    /// so it mirrors the machine post-swap (a no-op unless the posture
    /// itself changed โ€” the feed carries only real transitions, and the
    /// next `observe_throttle` re-derives the posture under the new
    /// thresholds). Semantic validation of the new policy is the caller's
    /// job.
    pub fn retune(&self, new_policy: PosturePolicy) {
        let posture = {
            let mut machine = self.machine.lock();
            machine.set_policy(new_policy);
            machine.state()
        };
        self.publish(posture);
    }

    /// Whether the posture admits new lease intake for `class` right now
    /// (read-through โ€” the dispatcher's enqueue/T-table gates and the seat
    /// hosts' admission all consult this).
    #[must_use]
    pub fn admits_lease(&self, class: CordonClass) -> bool {
        self.machine.lock().admits_lease(class)
    }

    /// Count a lease grant denied because of the posture (read-through โ€”
    /// the tuning loop's suppression metric).
    pub fn note_intake_suppressed(&self) {
        self.machine.lock().note_intake_suppressed();
    }

    /// The sim intake cap in the current posture (read-through).
    #[must_use]
    pub fn sim_intake_cap(&self, slot_cap: usize) -> usize {
        self.machine.lock().sim_intake_cap(slot_cap)
    }

    /// Loud-transition counters snapshot (the tuning loop's metrics).
    #[must_use]
    pub fn counters(&self) -> PostureCounters {
        *self.machine.lock().counters()
    }

    /// The sticky lane-death hold (read-through). A host keys its Faulted
    /// transition on this TYPED latch, never on elapsed cordon time (a long
    /// recoverable EventBurst/Duty cordon must not fault).
    #[must_use]
    pub fn lane_death_held(&self) -> bool {
        self.machine.lock().lane_death_held()
    }

    /// Mirror the machine's state into the feed โ€” compare-then-publish, so
    /// the sequence (and every subscriber's edge) moves ONLY on a real
    /// posture change.
    fn publish(&self, posture: FleetPosture) {
        let mut state = self.broadcast.state.write();
        if state.posture != posture {
            state.posture = posture;
            state.seq = state.seq.wrapping_add(1);
        }
    }
}

/// The process-level posture owner (the holder pattern): ONE
/// instance per process, installed by the FIRST fleet boot (first-wins โ€”
/// later installs log at debug and return the existing owner).
static PROCESS_OWNER: OnceLock<PostureOwner> = OnceLock::new();

/// Install the process-level owner with `policy` (first-wins: the first
/// fleet boot wins; later calls log at debug and return the existing
/// owner). Production installs happen at `FleetHost::boot` โ€” BEFORE the
/// first throttle feed reaches [`process`]. Hermetic tests never call
/// this: they inject fresh owners via `FleetBoot::owner`.
#[must_use]
pub fn install_process_owner(policy: PosturePolicy) -> &'static PostureOwner {
    if PROCESS_OWNER.set(PostureOwner::new(policy)).is_err() {
        diag!(
            domain = pump,
            "process owner already installed โ€” first-wins, keeping the existing owner"
        );
    }
    process()
}

/// The process-level owner, or a doc-default owner when nothing was
/// installed yet (the holder's default stance: tests and standalone
/// constructions observe schema defaults). `FleetBoot` without an injected
/// owner installs its policy here first-wins at boot time, which is why a
/// production feed never lands on the doc-default stance.
#[must_use]
pub fn process() -> &'static PostureOwner {
    PROCESS_OWNER.get_or_init(|| PostureOwner::new(PosturePolicy::doc_defaults()))
}

#[cfg(test)]
mod tests {
    use super::*;

    fn policy() -> PosturePolicy {
        PosturePolicy {
            enter_events: 2,
            enter_window_ms: 1_000,
            duty_percent: 2.0,
            duty_window_ms: 5_000,
            exit_clean_ms: 10_000,
            sim_intake_floor_override: None,
        }
    }

    fn sm() -> PostureStateMachine {
        PostureStateMachine::new(policy())
    }

    fn sample(events: u64, throttled_usec: u64, elapsed_usec: u64) -> ThrottleSample {
        ThrottleSample {
            events,
            throttled_usec,
            elapsed_usec,
        }
    }

    #[test]
    fn nominal_is_the_boot_state() {
        assert_eq!(sm().state(), FleetPosture::Nominal);
    }

    /// The DECIDED option (a): a typed `PostureCause`
    /// enters the cordon IMMEDIATELY (no sample hysteresis).
    #[test]
    fn a_lane_death_cordons_immediately() {
        let mut m = sm();
        assert_eq!(
            m.observe_cause(PostureCause::LaneDeath),
            PostureChange::Entered(EnterReason::LaneDeath)
        );
        assert_eq!(m.state(), FleetPosture::Cordoned);
        assert_eq!(m.counters().lane_deaths, 1);
        assert_eq!(m.counters().entered, 1);
    }

    /// The lane-death cordon is STICKY: the clean window never lifts it
    /// (the lane is still dead) โ€” its own exit discipline, deliberately
    /// different from the throttle hysteresis.
    #[test]
    fn the_lane_death_cordon_is_sticky_across_clean_windows() {
        let mut m = sm();
        m.observe_cause(PostureCause::LaneDeath);
        // A long run of clean samples past the full exit window: the
        // throttle hysteresis would EXIT here โ€” the lane-death hold
        // refuses (never a silent un-cordon).
        for t in (0..30_u64).map(|i| 10_000 + i * 1_000) {
            assert_eq!(
                m.observe(t, sample(0, 0, 1_000_000)),
                PostureChange::Held,
                "a clean window must never lift a lane-death cordon"
            );
        }
        assert_eq!(m.state(), FleetPosture::Cordoned);
        assert_eq!(m.counters().exited, 0, "the sticky cordon never exits");
    }

    /// A lane death UPGRADES an existing throttle cordon: no state
    /// transition (already Cordoned), but the hold turns sticky and the
    /// cause is counted โ€” loud, never silent.
    #[test]
    fn a_lane_death_upgrades_a_throttle_cordon_to_sticky() {
        let mut m = sm();
        // Enter via the throttle hysteresis first.
        m.observe(100, sample(2, 0, 1_000_000));
        assert_eq!(m.state(), FleetPosture::Cordoned);
        assert_eq!(m.counters().entered, 1);
        // The lane death upgrades the hold.
        assert_eq!(
            m.observe_cause(PostureCause::LaneDeath),
            PostureChange::Held,
            "no state transition โ€” the cordon was already up"
        );
        assert_eq!(m.counters().lane_deaths, 1);
        assert_eq!(m.counters().entered, 1, "no second enter counted");
        // And the upgrade is sticky: clean windows no longer lift it.
        for t in (0..30_u64).map(|i| 10_000 + i * 1_000) {
            m.observe(t, sample(0, 0, 1_000_000));
        }
        assert_eq!(m.state(), FleetPosture::Cordoned);
        assert_eq!(m.counters().exited, 0);
    }

    /// Idempotent while the hold is already set (every cause counts).
    #[test]
    fn repeated_lane_deaths_count_but_do_not_re_enter() {
        let mut m = sm();
        m.observe_cause(PostureCause::LaneDeath);
        for _ in 0..3 {
            assert_eq!(
                m.observe_cause(PostureCause::LaneDeath),
                PostureChange::Held
            );
        }
        assert_eq!(m.counters().lane_deaths, 4);
        assert_eq!(m.counters().entered, 1);
    }

    /// The cordon effects hold under a lane-death cause exactly as under
    /// a throttle cause (the ยง6 vocabulary: deferrable intake held,
    /// sim intake floored, in-flight units complete).
    #[test]
    fn lane_death_cordon_effects_match_the_throttle_vocabulary() {
        let mut m = sm();
        m.observe_cause(PostureCause::LaneDeath);
        assert!(!m.admits_lease(CordonClass::Deferrable));
        assert!(m.admits_lease(CordonClass::Never));
        assert!(m.admits_lease(CordonClass::SimPool));
        // The sim intake floor (a Cordoned cap of 4 slots -> 2, the
        // same shape a throttle cordon produces).
        assert_eq!(m.sim_intake_cap(4), 2);
    }

    /// The owner publishes the lane-death transition to the feed (a
    /// subscriber sees the Cordoned edge exactly once).
    #[test]
    fn the_owner_publishes_the_lane_death_transition() {
        let owner = PostureOwner::new(policy());
        let watch = owner.subscribe();
        assert_eq!(watch.current(), FleetPosture::Nominal);
        owner.observe_cause(PostureCause::LaneDeath);
        assert_eq!(watch.current(), FleetPosture::Cordoned);
        assert_eq!(owner.current(), FleetPosture::Cordoned);
    }

    #[test]
    fn enters_on_event_burst_within_the_window() {
        let mut m = sm();
        // One lone event inside the window: below the threshold.
        assert_eq!(m.observe(100, sample(1, 0, 1_000_000)), PostureChange::Held);
        // The second event inside the same trailing 1 s window -> enter,
        // loudly typed.
        let change = m.observe(600, sample(1, 0, 500_000));
        assert_eq!(
            change,
            PostureChange::Entered(EnterReason::EventBurst { events: 2 })
        );
        assert_eq!(m.state(), FleetPosture::Cordoned);
        assert_eq!(m.counters().entered, 1);
        // Already cordoned: further dirty samples do not re-enter.
        assert!(matches!(
            m.observe(700, sample(1, 0, 100_000)),
            PostureChange::Held
        ));
        assert_eq!(m.counters().entered, 1);
    }

    #[test]
    fn burst_outside_the_enter_window_does_not_cordon() {
        let mut m = sm();
        m.observe(0, sample(1, 0, 1_000_000));
        // The 1 s window expired; two lone events > 1 s apart never burst.
        m.observe(3_000, sample(1, 0, 2_000_000));
        m.observe(3_100, sample(0, 0, 100_000));
        assert_eq!(m.state(), FleetPosture::Nominal);
    }

    #[test]
    fn enters_on_duty_spike_over_the_duty_window() {
        let mut m = sm();
        // A 1 s poll interval whose throttled time was 50% โ€” measured over
        // the trailing 5 s window that is 50% duty, far above the 2%
        // threshold: the duty trigger fires on the first observation.
        let change = m.observe(1_000, sample(0, 500_000, 1_000_000));
        assert!(matches!(
            change,
            PostureChange::Entered(EnterReason::DutySpike { .. })
        ));
        assert_eq!(m.state(), FleetPosture::Cordoned);
        assert_eq!(m.counters().entered, 1);
    }

    #[test]
    fn a_rising_duty_only_crosses_once_the_trailing_total_exceeds_the_threshold() {
        let mut m = sm();
        // Sub-threshold ticks (0.05% each) hold the posture...
        for t in 1..5_u64 {
            assert_eq!(
                m.observe(t * 1_000, sample(0, 500, 1_000_000)),
                PostureChange::Held
            );
        }
        // ...until one more dirty tick crosses the window total above 2%.
        let change = m.observe(5_000, sample(0, 5_000_000, 1_000_000));
        assert!(matches!(
            change,
            PostureChange::Entered(EnterReason::DutySpike { .. })
        ));
    }

    #[test]
    fn sub_threshold_duty_never_cordons() {
        let mut m = sm();
        // 1% duty (design doc: steady state 0.18%) for ten windows.
        for t in 0..10_u64 {
            m.observe((t + 1) * 1_000, sample(0, 10_000, 1_000_000));
        }
        assert_eq!(m.state(), FleetPosture::Nominal);
    }

    #[test]
    fn exits_only_after_the_full_clean_hysteresis() {
        let mut m = sm();
        m.observe(100, sample(2, 0, 100_000)); // burst enter
        assert_eq!(m.state(), FleetPosture::Cordoned);
        // Clean time below the hysteresis keeps the cordon.
        let mut t = 200;
        while t < 10_000 {
            assert_eq!(m.observe(t, sample(0, 0, 1_000_000)), PostureChange::Held);
            t += 1_000;
        }
        assert_eq!(m.state(), FleetPosture::Cordoned);
        // Past 10 s clean: exit.
        let change = m.observe(10_200, sample(0, 0, 100_000));
        assert_eq!(change, PostureChange::Exited);
        assert_eq!(m.state(), FleetPosture::Nominal);
        assert_eq!(m.counters().exited, 1);
    }

    #[test]
    fn a_dirty_window_restarts_the_clean_clock() {
        let mut m = sm();
        m.observe(0, sample(2, 0, 100_000));
        let mut t = 1_000;
        while t < 9_000 {
            m.observe(t, sample(0, 0, 1_000_000));
            t += 1_000;
        }
        // A lone dirty sample resets the clean-window clock.
        m.observe(9_000, sample(1, 0, 100_000));
        assert_eq!(m.state(), FleetPosture::Cordoned);
        let mut t = 10_000;
        while t < 18_000 {
            m.observe(t, sample(0, 0, 1_000_000));
            t += 1_000;
        }
        assert_eq!(m.state(), FleetPosture::Cordoned, "only 9 s clean");
        m.observe(19_100, sample(0, 0, 100_000));
        assert_eq!(
            m.state(),
            FleetPosture::Nominal,
            "10 s clean since the reset"
        );
    }

    #[test]
    fn cordon_effects_match_the_sign_off_table() {
        let mut m = sm();
        // Nominal: full sim cap; deferrable admitted.
        assert_eq!(m.sim_intake_cap(4), 4);
        assert!(m.admits_lease(CordonClass::Never));
        assert!(m.admits_lease(CordonClass::SimPool));
        assert!(m.admits_lease(CordonClass::Deferrable));

        m.observe(0, sample(2, 0, 100_000));
        assert_eq!(m.state(), FleetPosture::Cordoned);
        // Cordon: deferrable held, sim floored at half the cap, Never free.
        assert!(!m.admits_lease(CordonClass::Deferrable));
        assert!(m.admits_lease(CordonClass::SimPool));
        assert!(m.admits_lease(CordonClass::Never));
        assert_eq!(m.sim_intake_cap(4), 2, "floor = half the slot cap");
        assert_eq!(m.sim_intake_cap(1), 1, "floored at one");
    }

    #[test]
    fn override_intake_floor_wins_and_caps_at_the_slot_cap() {
        let p = PosturePolicy {
            sim_intake_floor_override: Some(7),
            ..policy()
        };
        assert_eq!(p.sim_intake_cap(4), 4, "nothing above the slot cap");
        let p = PosturePolicy {
            sim_intake_floor_override: Some(0),
            ..policy()
        };
        assert_eq!(p.sim_intake_cap(4), 1, "floored at one");
    }

    #[test]
    fn suppressed_intake_is_counted_for_the_tuning_loop() {
        let mut m = sm();
        m.note_intake_suppressed();
        m.note_intake_suppressed();
        assert_eq!(m.counters().intake_suppressed, 2);
    }

    #[test]
    fn typed_config_projects_all_the_q5_amendment_thresholds() {
        let cfg = degenbot_config::BotConfig::default();
        let p = PosturePolicy::from_config(&cfg.fleet);
        // Defaults mirror design doc ยง6: 2 events / 1 s, >2% / 5 s, 10 s.
        assert_eq!(p.enter_events, 2);
        assert_eq!(p.enter_window_ms, 1_000);
        assert!((p.duty_percent - 2.0).abs() < 1e-9, "duty default is 2%");
        assert_eq!(p.duty_window_ms, 5_000);
        assert_eq!(p.exit_clean_ms, 10_000);
        assert_eq!(p.sim_intake_floor_override, None);
    }

    // ---- PostureOwner -------------------------------------

    #[test]
    fn owner_transitions_publish_only_on_change() {
        let owner = PostureOwner::new(policy());
        let watch = owner.subscribe();
        // A clean sample: no transition, no publication.
        owner.observe_throttle(0, sample(0, 0, 1_000));
        assert!(!watch.has_changed());
        assert_eq!(watch.current(), FleetPosture::Nominal);
        // One lone event below the threshold: still no edge.
        owner.observe_throttle(100, sample(1, 0, 1_000));
        assert!(!watch.has_changed());
        // The bursting event crosses: exactly one edge, and the drain
        // consumes it (exactly-once per transition).
        owner.observe_throttle(600, sample(1, 0, 500_000));
        assert!(watch.has_changed());
        assert_eq!(watch.take_if_changed(), Some(FleetPosture::Cordoned));
        assert_eq!(watch.take_if_changed(), None, "one edge per transition");
        // Already cordoned: dirty ticks are Held โ€” never re-published.
        owner.observe_throttle(700, sample(1, 0, 100_000));
        assert!(!watch.has_changed());
        assert_eq!(watch.current(), FleetPosture::Cordoned);
    }

    #[test]
    fn owner_current_reflects_the_machine_including_exit_hysteresis() {
        let owner = PostureOwner::new(policy());
        assert_eq!(owner.current(), FleetPosture::Nominal);
        owner.observe_throttle(0, sample(3, 0, 1_000));
        assert_eq!(owner.current(), FleetPosture::Cordoned);
        assert_eq!(owner.counters().entered, 1);
        // The full clean hysteresis lifts the cordon (10 s of clean ticks).
        let mut now = 1_000;
        loop {
            owner.observe_throttle(now, sample(0, 0, 1_000));
            if owner.current() == FleetPosture::Nominal {
                break;
            }
            now += 1_000;
            assert!(now <= 60_000, "the cordon never lifted");
        }
        assert_eq!(owner.counters().exited, 1);
    }

    #[test]
    fn first_wins_process_install_keeps_the_existing_owner() {
        let first = install_process_owner(policy());
        let second = install_process_owner(PosturePolicy {
            enter_events: 99,
            ..policy()
        });
        assert!(
            std::ptr::eq(first, second),
            "first-wins: a later install returns the existing owner"
        );
        assert!(std::ptr::eq(first, process()));
        assert_eq!(
            second.policy().enter_events,
            first.policy().enter_events,
            "the losing install's policy never landed"
        );
    }

    #[test]
    fn retune_swaps_thresholds_and_the_next_observe_rederives() {
        let owner = PostureOwner::new(policy());
        let watch = owner.subscribe();
        // One lone event: sub-threshold under the boot policy.
        owner.observe_throttle(0, sample(1, 0, 1_000_000));
        assert_eq!(owner.current(), FleetPosture::Nominal);
        // Retune to a 1-event trigger: no posture change, no edge โ€” but
        // the next lone (clean) sample cordons under the new thresholds.
        owner.retune(PosturePolicy {
            enter_events: 1,
            ..policy()
        });
        assert_eq!(owner.policy().enter_events, 1, "the retune swapped");
        assert!(!watch.has_changed(), "a no-effect retune is not an edge");
        owner.observe_throttle(2_000, sample(1, 0, 100_000));
        assert_eq!(owner.current(), FleetPosture::Cordoned);
        assert_eq!(watch.take_if_changed(), Some(FleetPosture::Cordoned));
    }

    #[test]
    fn two_owners_are_fully_independent_hermetic_isolation() {
        let a = PostureOwner::new(policy());
        let b = PostureOwner::new(policy());
        let watch_a = a.subscribe();
        let watch_b = b.subscribe();
        // A cordons; b never hears about it.
        a.observe_throttle(0, sample(3, 0, 1_000));
        assert_eq!(a.current(), FleetPosture::Cordoned);
        assert_eq!(b.current(), FleetPosture::Nominal);
        assert_eq!(watch_a.take_if_changed(), Some(FleetPosture::Cordoned));
        assert_eq!(
            watch_b.take_if_changed(),
            None,
            "no posture leaks across owners"
        );
        // b's own feed stays silent on a clean tick, and the feeds stay
        // independent in both directions.
        b.observe_throttle(1_000, sample(0, 0, 1_000));
        assert_eq!(watch_b.take_if_changed(), None);
        assert_eq!(watch_a.take_if_changed(), None);
    }

    #[test]
    fn owner_read_throughs_match_the_machine_semantics() {
        let owner = PostureOwner::new(PosturePolicy {
            sim_intake_floor_override: Some(1),
            ..policy()
        });
        assert!(owner.admits_lease(CordonClass::Deferrable));
        assert_eq!(owner.sim_intake_cap(8), 8, "nominal intake is the cap");
        owner.observe_throttle(0, sample(3, 0, 1_000));
        assert!(!owner.admits_lease(CordonClass::Deferrable));
        assert!(owner.admits_lease(CordonClass::Never));
        assert!(owner.admits_lease(CordonClass::SimPool));
        assert_eq!(owner.sim_intake_cap(8), 1, "the cordon floor override");
        assert_eq!(owner.counters().entered, 1);
    }

    // ---- the operator re-tune patch -----------------------

    #[test]
    fn an_empty_patch_is_rejected() {
        assert!(PosturePolicyPatch::default().validate().is_err());
        assert!(PosturePolicyPatch::default().is_empty());
        assert_eq!(
            PosturePolicyPatch::default().validate(),
            Err(PostureRetuneError::EmptyPatch),
            "the empty-patch refusal is its own typed error"
        );
    }

    #[test]
    fn every_threshold_rule_is_a_typed_rejection() {
        assert_eq!(
            PosturePolicyPatch {
                enter_events: Some(0),
                ..PosturePolicyPatch::default()
            }
            .validate(),
            Err(PostureRetuneError::EnterEvents(0)),
            "enter_events >= 1"
        );
        assert_eq!(
            PosturePolicyPatch {
                enter_window_ms: Some(0),
                ..PosturePolicyPatch::default()
            }
            .validate(),
            Err(PostureRetuneError::EnterWindow(0)),
            "windows must be > 0 ms"
        );
        assert_eq!(
            PosturePolicyPatch {
                duty_window_ms: Some(0),
                ..PosturePolicyPatch::default()
            }
            .validate(),
            Err(PostureRetuneError::DutyWindow(0))
        );
        assert_eq!(
            PosturePolicyPatch {
                exit_clean_ms: Some(0),
                ..PosturePolicyPatch::default()
            }
            .validate(),
            Err(PostureRetuneError::ExitClean(0))
        );
        // The sane inclusive duty range (0.0, 100.0]: 0, negatives, >100,
        // and non-finite values are all refused, never clamped.
        for duty in [0.0, -1.0, 100.5, f64::NAN, f64::INFINITY] {
            let rejected = PosturePolicyPatch {
                duty_percent: Some(duty),
                ..PosturePolicyPatch::default()
            }
            .validate();
            assert!(
                matches!(rejected, Err(PostureRetuneError::DutyPercent(_))),
                "duty {duty} must be refused as DutyPercent, got {rejected:?}"
            );
        }
        assert_eq!(
            PosturePolicyPatch {
                sim_intake_floor_override: Some(Some(0)),
                ..PosturePolicyPatch::default()
            }
            .validate(),
            Err(PostureRetuneError::SimIntakeFloor(0)),
            "an explicit floor must be >= 1"
        );
    }

    #[test]
    fn boundary_values_are_admitted() {
        // The inclusive edges pass: 1 event, 1 ms windows, the 100.0% duty
        // ceiling, a floor of exactly 1, and a CLEARED floor.
        let validated = PosturePolicyPatch {
            enter_events: Some(1),
            enter_window_ms: Some(1),
            duty_percent: Some(100.0),
            duty_window_ms: Some(1),
            exit_clean_ms: Some(1),
            sim_intake_floor_override: Some(Some(1)),
        }
        .validate();
        assert_eq!(
            validated,
            Ok(()),
            "inclusive bounds are legal (1 event, 1 ms windows, 100.0% duty, floor 1)"
        );
        let cleared = PosturePolicyPatch {
            sim_intake_floor_override: Some(None),
            ..PosturePolicyPatch::default()
        }
        .validate();
        assert_eq!(
            cleared,
            Ok(()),
            "clearing the floor is a legal one-key patch"
        );
    }

    #[test]
    fn patched_with_touches_only_supplied_keys() {
        let base = policy();
        let patched = base.patched_with(PosturePolicyPatch {
            enter_events: Some(7),
            ..PosturePolicyPatch::default()
        });
        assert_eq!(patched.enter_events, 7, "the supplied key landed");
        assert_eq!(patched.enter_window_ms, base.enter_window_ms);
        assert!(
            (patched.duty_percent - base.duty_percent).abs() < f64::EPSILON,
            "an absent key keeps the current duty percent"
        );
        assert_eq!(patched.duty_window_ms, base.duty_window_ms);
        assert_eq!(patched.exit_clean_ms, base.exit_clean_ms);
        assert_eq!(
            patched.sim_intake_floor_override, base.sim_intake_floor_override,
            "an absent key keeps the current value"
        );
    }

    #[test]
    fn patched_with_distinguishes_floor_set_clear_and_absent() {
        let base = PosturePolicy {
            sim_intake_floor_override: Some(3),
            ..policy()
        };
        // Key absent: the current override is kept.
        assert_eq!(
            base.patched_with(PosturePolicyPatch::default())
                .sim_intake_floor_override,
            Some(3)
        );
        // Key present with a value: the override is set.
        assert_eq!(
            base.patched_with(PosturePolicyPatch {
                sim_intake_floor_override: Some(Some(1)),
                ..PosturePolicyPatch::default()
            })
            .sim_intake_floor_override,
            Some(1)
        );
        // Key present with the key's None value: the override is CLEARED
        // (back to half the slot cap at read time).
        assert_eq!(
            base.patched_with(PosturePolicyPatch {
                sim_intake_floor_override: Some(None),
                ..PosturePolicyPatch::default()
            })
            .sim_intake_floor_override,
            None
        );
    }

    #[test]
    fn a_validated_patch_retunes_the_owner_end_to_end() {
        let owner = PostureOwner::new(policy());
        let patch = PosturePolicyPatch {
            duty_percent: Some(5.0),
            ..PosturePolicyPatch::default()
        };
        assert_eq!(patch.validate(), Ok(()), "the channel validated the patch");
        let effective = owner.policy().patched_with(patch);
        owner.retune(effective);
        assert!(
            (owner.policy().duty_percent - 5.0).abs() < f64::EPSILON,
            "the retuned duty percent is live"
        );
        assert_eq!(owner.policy().enter_events, policy().enter_events);
    }
}