1use serde::{Deserialize, Serialize};
35
36use crate::redaction::redact_secrets;
37use crate::schema_versions::{SchemaComponent, current_version};
38
39#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
46#[serde(rename_all = "snake_case")]
47pub enum RemediationActionClass {
48 Healthy,
50 NormalFailOpen,
53 SelfHealingInProgress,
56 OperatorActionRequired,
58}
59
60impl RemediationActionClass {
61 pub const ALL: &'static [RemediationActionClass] = &[
63 Self::Healthy,
64 Self::NormalFailOpen,
65 Self::SelfHealingInProgress,
66 Self::OperatorActionRequired,
67 ];
68
69 #[must_use]
71 pub const fn as_str(self) -> &'static str {
72 match self {
73 Self::Healthy => "healthy",
74 Self::NormalFailOpen => "normal_fail_open",
75 Self::SelfHealingInProgress => "self_healing_in_progress",
76 Self::OperatorActionRequired => "operator_action_required",
77 }
78 }
79
80 #[must_use]
82 pub const fn label(self) -> &'static str {
83 match self {
84 Self::Healthy => "healthy",
85 Self::NormalFailOpen => "normal (fail-open local)",
86 Self::SelfHealingInProgress => "self-healing in progress",
87 Self::OperatorActionRequired => "operator action required",
88 }
89 }
90
91 #[must_use]
93 pub fn max_class(self, other: Self) -> Self {
94 self.max(other)
95 }
96}
97
98#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
102#[serde(rename_all = "snake_case")]
103pub enum BandSeverity {
104 Ok,
106 Info,
108 Warn,
110 Critical,
112}
113
114impl BandSeverity {
115 pub const ALL: &'static [BandSeverity] = &[Self::Ok, Self::Info, Self::Warn, Self::Critical];
117
118 #[must_use]
120 pub const fn as_str(self) -> &'static str {
121 match self {
122 Self::Ok => "ok",
123 Self::Info => "info",
124 Self::Warn => "warn",
125 Self::Critical => "critical",
126 }
127 }
128}
129
130#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
134#[serde(rename_all = "snake_case")]
135pub enum BandId {
136 DesiredFleet,
138 LiveEligibility,
140 AdmissibleWorkers,
142 ProofQueue,
144 ActiveJobs,
146 DiskPressure,
148 TelemetryFreshness,
150 Incidents,
152}
153
154impl BandId {
155 pub const ALL: &'static [BandId] = &[
157 Self::DesiredFleet,
158 Self::LiveEligibility,
159 Self::AdmissibleWorkers,
160 Self::ProofQueue,
161 Self::ActiveJobs,
162 Self::DiskPressure,
163 Self::TelemetryFreshness,
164 Self::Incidents,
165 ];
166
167 #[must_use]
169 pub const fn as_str(self) -> &'static str {
170 match self {
171 Self::DesiredFleet => "desired_fleet",
172 Self::LiveEligibility => "live_eligibility",
173 Self::AdmissibleWorkers => "admissible_workers",
174 Self::ProofQueue => "proof_queue",
175 Self::ActiveJobs => "active_jobs",
176 Self::DiskPressure => "disk_pressure",
177 Self::TelemetryFreshness => "telemetry_freshness",
178 Self::Incidents => "incidents",
179 }
180 }
181
182 #[must_use]
184 pub const fn title(self) -> &'static str {
185 match self {
186 Self::DesiredFleet => "Desired Fleet",
187 Self::LiveEligibility => "Live Eligibility",
188 Self::AdmissibleWorkers => "Admissible Workers",
189 Self::ProofQueue => "Proof Queue",
190 Self::ActiveJobs => "Active Jobs",
191 Self::DiskPressure => "Disk Pressure",
192 Self::TelemetryFreshness => "Telemetry Freshness",
193 Self::Incidents => "Recent Incidents",
194 }
195 }
196}
197
198#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
200pub struct RemediationBand {
201 pub id: BandId,
203 pub title: String,
205 pub severity: BandSeverity,
207 pub action_class: RemediationActionClass,
209 pub headline: String,
211 pub detail_lines: Vec<String>,
213 #[serde(default, skip_serializing_if = "Option::is_none")]
215 pub reason_code: Option<String>,
216}
217
218#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
220pub struct RemediationIncidentLine {
221 pub reason_code: String,
223 pub event_type: String,
225 #[serde(default, skip_serializing_if = "Option::is_none")]
227 pub worker_id: Option<String>,
228 pub age_secs: u64,
230 pub summary: String,
232}
233
234impl RemediationIncidentLine {
235 #[must_use]
237 pub fn new(
238 reason_code: impl Into<String>,
239 event_type: impl Into<String>,
240 worker_id: Option<String>,
241 age_secs: u64,
242 summary: impl AsRef<str>,
243 ) -> Self {
244 Self {
245 reason_code: reason_code.into(),
246 event_type: event_type.into(),
247 worker_id,
248 age_secs,
249 summary: redact_secrets(summary.as_ref()),
250 }
251 }
252}
253
254#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
261pub struct FleetSummaryInput {
262 pub desired: usize,
264 pub eligible: usize,
266 pub bypassed: usize,
268 pub disabled: usize,
270 pub unreachable: usize,
272 pub missing: usize,
274 pub recovered_pending_canary: usize,
276 pub problem_class: String,
278 pub problem_summary: String,
280 pub absence_warnings: usize,
282}
283
284#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
287pub struct AdmissibleWorkersInput {
288 pub admissible: usize,
290 pub total_live: usize,
292 #[serde(default, skip_serializing_if = "Option::is_none")]
294 pub top_blocker_reason: Option<String>,
295}
296
297#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
300pub struct ProofQueueInput {
301 pub queued: usize,
303 pub blocked: usize,
305 pub replaying: usize,
307 pub failed_recent: usize,
309 pub stale_recent: usize,
311}
312
313#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
315pub struct JobsInput {
316 pub active: usize,
318 pub queued: usize,
320 pub stuck: usize,
322}
323
324#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)]
326pub struct DiskPressureInput {
327 pub workers_warning: usize,
329 pub workers_critical: usize,
331 pub reclaim_in_progress: usize,
333 #[serde(default, skip_serializing_if = "Option::is_none")]
335 pub min_free_ratio: Option<f64>,
336}
337
338#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
340pub struct TelemetryFreshnessInput {
341 pub fresh: usize,
343 pub stale: usize,
345 pub unknown: usize,
347 #[serde(default, skip_serializing_if = "Option::is_none")]
349 pub max_age_secs: Option<u64>,
350}
351
352#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
355#[serde(rename_all = "snake_case")]
356pub enum DiskLevel {
357 Ok,
359 Warning,
361 Critical,
363}
364
365impl DiskLevel {
366 #[must_use]
370 pub fn from_pressure_state(token: &str) -> Self {
371 match token {
372 "critical" => Self::Critical,
373 "warning" => Self::Warning,
374 _ => Self::Ok,
375 }
376 }
377}
378
379#[derive(Debug, Clone, PartialEq)]
387pub struct RemediationWorkerRow {
388 pub observation: crate::fleet_diff::WorkerObservation,
390 pub disk_level: DiskLevel,
392 pub reclaiming: bool,
394 pub free_ratio: Option<f64>,
396 pub slots_used: u32,
398 pub slots_total: u32,
400 pub telemetry_known: bool,
402 pub telemetry_fresh: bool,
404 pub telemetry_age_secs: Option<u64>,
406 pub recovered_pending_canary: bool,
408 pub absent_secs: Option<u64>,
410}
411
412#[must_use]
419pub fn build_inputs(
420 rows: &[RemediationWorkerRow],
421 jobs: JobsInput,
422 proof_queue: ProofQueueInput,
423 incidents: Vec<RemediationIncidentLine>,
424 absence_threshold_secs: u64,
425) -> RemediationInputs {
426 use crate::fleet_diff::{WorkerDiffState, derive_worker_diff};
427 use crate::fleet_status::{FleetWorkerSignal, compute_fleet_status};
428
429 let signals: Vec<FleetWorkerSignal> = rows
430 .iter()
431 .map(|r| FleetWorkerSignal {
432 observation: r.observation.clone(),
433 disk_pressure: r.disk_level != DiskLevel::Ok,
434 slots_saturated: r.slots_total > 0 && r.slots_used >= r.slots_total,
435 absent_secs: r.absent_secs,
436 })
437 .collect();
438 let report = compute_fleet_status(&signals, absence_threshold_secs);
439
440 let mut eligible = 0usize;
441 let mut bypassed = 0usize;
442 let mut disabled = 0usize;
443 let mut unreachable = 0usize;
444 let mut missing = 0usize;
445 for r in rows {
446 match derive_worker_diff(&r.observation) {
447 WorkerDiffState::Ready => eligible += 1,
448 WorkerDiffState::TemporarilyBypassed => bypassed += 1,
449 WorkerDiffState::AdminDisabled => disabled += 1,
450 WorkerDiffState::Unreachable => unreachable += 1,
451 WorkerDiffState::MissingFromFleet
452 | WorkerDiffState::RecoveredNotRejoined
453 | WorkerDiffState::Unconfigured => missing += 1,
454 WorkerDiffState::FactsUnknown | WorkerDiffState::CommandIneligible => {}
456 }
457 }
458 let recovered_pending_canary = rows.iter().filter(|r| r.recovered_pending_canary).count();
459 let desired = rows.iter().filter(|r| r.observation.configured).count();
460
461 let fleet = FleetSummaryInput {
462 desired,
463 eligible,
464 bypassed,
465 disabled,
466 unreachable,
467 missing,
468 recovered_pending_canary,
469 problem_class: report.problem_class.as_str().to_string(),
470 problem_summary: report.problem_summary.clone(),
471 absence_warnings: report.absence_alerts.len(),
472 };
473
474 let mut total_live = 0usize;
476 let mut admissible = 0usize;
477 let mut blocker_counts: std::collections::BTreeMap<&'static str, usize> =
478 std::collections::BTreeMap::new();
479 for r in rows {
480 let live = r.observation.in_daemon_pool
481 && r.observation.reachable
482 && !r.observation.admin_disabled;
483 if !live {
484 continue;
485 }
486 total_live += 1;
487 let state = derive_worker_diff(&r.observation);
488 let no_free_slot = r.slots_total == 0 || r.slots_used >= r.slots_total;
493 let at_critical_pressure = r.disk_level == DiskLevel::Critical;
494 if state == WorkerDiffState::Ready && !no_free_slot && !at_critical_pressure {
495 admissible += 1;
496 } else {
497 let reason = match (state, no_free_slot, at_critical_pressure) {
498 (WorkerDiffState::AdminDisabled, _, _) => "administratively disabled",
499 (WorkerDiffState::TemporarilyBypassed, _, _) => "temporarily bypassed",
500 (WorkerDiffState::FactsUnknown, _, _) => "missing capability facts",
501 (WorkerDiffState::Unreachable, _, _) => "unreachable",
502 (_, _, true) => "critical disk pressure",
503 (_, true, _) => "no free slot",
504 (WorkerDiffState::CommandIneligible, _, _) => "not admissible for command",
505 _ => "ineligible",
506 };
507 *blocker_counts.entry(reason).or_default() += 1;
508 }
509 }
510 let top_blocker_reason = if admissible == 0 && total_live > 0 {
511 blocker_counts
512 .iter()
513 .max_by_key(|(_, n)| **n)
514 .map(|(reason, _)| (*reason).to_string())
515 } else {
516 None
517 };
518
519 let mut workers_warning = 0usize;
520 let mut workers_critical = 0usize;
521 let mut reclaim_in_progress = 0usize;
522 let mut min_free_ratio: Option<f64> = None;
523 for r in rows {
524 match r.disk_level {
525 DiskLevel::Warning => workers_warning += 1,
526 DiskLevel::Critical => workers_critical += 1,
527 DiskLevel::Ok => {}
528 }
529 if r.reclaiming && r.disk_level != DiskLevel::Ok {
530 reclaim_in_progress += 1;
531 }
532 if let Some(ratio) = r.free_ratio {
533 min_free_ratio = Some(min_free_ratio.map_or(ratio, |m: f64| m.min(ratio)));
534 }
535 }
536
537 let mut fresh = 0usize;
538 let mut stale = 0usize;
539 let mut unknown = 0usize;
540 let mut max_age_secs: Option<u64> = None;
541 for r in rows {
542 let live = r.observation.in_daemon_pool && r.observation.reachable;
543 if !live {
544 continue;
545 }
546 if !r.telemetry_known {
547 unknown += 1;
548 } else if r.telemetry_fresh {
549 fresh += 1;
550 } else {
551 stale += 1;
552 }
553 if let Some(age) = r.telemetry_age_secs {
554 max_age_secs = Some(max_age_secs.map_or(age, |m| m.max(age)));
555 }
556 }
557
558 RemediationInputs {
559 fleet,
560 admissible: AdmissibleWorkersInput {
561 admissible,
562 total_live,
563 top_blocker_reason,
564 },
565 proof_queue,
566 jobs,
567 disk: DiskPressureInput {
568 workers_warning,
569 workers_critical,
570 reclaim_in_progress,
571 min_free_ratio,
572 },
573 telemetry: TelemetryFreshnessInput {
574 fresh,
575 stale,
576 unknown,
577 max_age_secs,
578 },
579 incidents,
580 }
581}
582
583#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
585pub struct RemediationInputs {
586 pub fleet: FleetSummaryInput,
588 pub admissible: AdmissibleWorkersInput,
590 pub proof_queue: ProofQueueInput,
592 pub jobs: JobsInput,
594 pub disk: DiskPressureInput,
596 pub telemetry: TelemetryFreshnessInput,
598 pub incidents: Vec<RemediationIncidentLine>,
600}
601
602pub const MAX_VIEW_INCIDENTS: usize = 10;
605
606#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
608pub struct RemediationView {
609 pub schema_version: String,
611 pub generated_at_unix_ms: u64,
613 pub overall: RemediationActionClass,
615 pub bands: Vec<RemediationBand>,
617 pub incidents: Vec<RemediationIncidentLine>,
619}
620
621impl RemediationView {
622 #[must_use]
624 pub fn band(&self, id: BandId) -> Option<&RemediationBand> {
625 self.bands.iter().find(|b| b.id == id)
626 }
627
628 #[must_use]
630 pub fn needs_operator_action(&self) -> bool {
631 self.overall == RemediationActionClass::OperatorActionRequired
632 }
633}
634
635#[must_use]
641pub fn assemble(inputs: &RemediationInputs, generated_at_unix_ms: u64) -> RemediationView {
642 let bands = vec![
643 band_desired_fleet(&inputs.fleet),
644 band_live_eligibility(&inputs.fleet),
645 band_admissible(&inputs.admissible, &inputs.fleet),
646 band_proof_queue(&inputs.proof_queue),
647 band_active_jobs(&inputs.jobs),
648 band_disk_pressure(&inputs.disk),
649 band_telemetry(&inputs.telemetry),
650 band_incidents(&inputs.incidents),
651 ];
652
653 let overall = bands.iter().map(|b| b.action_class).fold(
654 RemediationActionClass::Healthy,
655 RemediationActionClass::max_class,
656 );
657
658 let mut incidents = inputs.incidents.clone();
659 incidents.truncate(MAX_VIEW_INCIDENTS);
660 for line in &mut incidents {
662 line.summary = redact_secrets(&line.summary);
663 }
664
665 RemediationView {
666 schema_version: current_version(SchemaComponent::RemediationView).to_string(),
667 generated_at_unix_ms,
668 overall,
669 bands,
670 incidents,
671 }
672}
673
674fn make_band(
679 id: BandId,
680 severity: BandSeverity,
681 action_class: RemediationActionClass,
682 headline: impl AsRef<str>,
683 detail_lines: Vec<String>,
684 reason_code: Option<String>,
685) -> RemediationBand {
686 RemediationBand {
687 id,
688 title: id.title().to_string(),
689 severity,
690 action_class,
691 headline: redact_secrets(headline.as_ref()),
692 detail_lines: detail_lines.iter().map(|l| redact_secrets(l)).collect(),
693 reason_code,
694 }
695}
696
697fn band_desired_fleet(f: &FleetSummaryInput) -> RemediationBand {
698 let details = vec![format!(
699 "desired {} · eligible {} · bypassed {} · disabled {} · unreachable {} · missing {}",
700 f.desired, f.eligible, f.bypassed, f.disabled, f.unreachable, f.missing
701 )];
702
703 if f.desired == 0 {
705 return make_band(
706 BandId::DesiredFleet,
707 BandSeverity::Info,
708 RemediationActionClass::NormalFailOpen,
709 "no workers configured — builds run locally (fail-open)",
710 details,
711 None,
712 );
713 }
714
715 if f.eligible > 0 {
716 let (sev, class) = if f.problem_class == "local_overload" {
719 (BandSeverity::Warn, RemediationActionClass::NormalFailOpen)
720 } else {
721 (BandSeverity::Ok, RemediationActionClass::Healthy)
722 };
723 return make_band(
724 BandId::DesiredFleet,
725 sev,
726 class,
727 &f.problem_summary,
728 details,
729 Some(f.problem_class.clone()),
730 );
731 }
732
733 let self_healing_collapse =
736 f.recovered_pending_canary > 0 || (f.bypassed > 0 && f.disabled == 0 && f.missing == 0);
737 let class = if self_healing_collapse {
738 RemediationActionClass::SelfHealingInProgress
739 } else {
740 RemediationActionClass::OperatorActionRequired
741 };
742 make_band(
743 BandId::DesiredFleet,
744 BandSeverity::Critical,
745 class,
746 &f.problem_summary,
747 details,
748 Some(f.problem_class.clone()),
749 )
750}
751
752fn band_live_eligibility(f: &FleetSummaryInput) -> RemediationBand {
753 let headline = format!("{} of {} desired worker(s) eligible", f.eligible, f.desired);
754 let mut details = Vec::new();
755 if f.bypassed > 0 {
756 details.push(format!("{} temporarily bypassed (auto-rejoin)", f.bypassed));
757 }
758 if f.recovered_pending_canary > 0 {
759 details.push(format!(
760 "{} recovered, canary pending",
761 f.recovered_pending_canary
762 ));
763 }
764 if f.disabled > 0 {
765 details.push(format!("{} admin-disabled", f.disabled));
766 }
767 if f.unreachable > 0 {
768 details.push(format!("{} unreachable", f.unreachable));
769 }
770 if f.missing > 0 {
771 details.push(format!("{} missing from pool", f.missing));
772 }
773 if f.absence_warnings > 0 {
774 details.push(format!("{} absent past policy window", f.absence_warnings));
775 }
776
777 if f.desired == 0 {
778 return make_band(
779 BandId::LiveEligibility,
780 BandSeverity::Info,
781 RemediationActionClass::NormalFailOpen,
782 "no workers configured",
783 details,
784 None,
785 );
786 }
787
788 let operator_gaps = f.disabled + f.unreachable + f.missing;
791 let transient_gaps = f.bypassed + f.recovered_pending_canary;
792
793 let (severity, action_class) = if f.eligible == 0 {
794 if operator_gaps == 0 && transient_gaps > 0 {
795 (
796 BandSeverity::Critical,
797 RemediationActionClass::SelfHealingInProgress,
798 )
799 } else {
800 (
801 BandSeverity::Critical,
802 RemediationActionClass::OperatorActionRequired,
803 )
804 }
805 } else if operator_gaps > 0 {
806 (
807 BandSeverity::Warn,
808 RemediationActionClass::OperatorActionRequired,
809 )
810 } else if transient_gaps > 0 {
811 (
812 BandSeverity::Warn,
813 RemediationActionClass::SelfHealingInProgress,
814 )
815 } else {
816 (BandSeverity::Ok, RemediationActionClass::Healthy)
817 };
818
819 make_band(
820 BandId::LiveEligibility,
821 severity,
822 action_class,
823 headline,
824 details,
825 None,
826 )
827}
828
829fn band_admissible(a: &AdmissibleWorkersInput, f: &FleetSummaryInput) -> RemediationBand {
830 let headline = format!(
831 "{} of {} live worker(s) can run a command now",
832 a.admissible, a.total_live
833 );
834 let mut details = Vec::new();
835 if let Some(reason) = &a.top_blocker_reason {
836 details.push(format!("top blocker: {reason}"));
837 }
838
839 if a.admissible > 0 {
840 return make_band(
841 BandId::AdmissibleWorkers,
842 BandSeverity::Ok,
843 RemediationActionClass::Healthy,
844 headline,
845 details,
846 None,
847 );
848 }
849
850 if a.total_live == 0 {
855 let class = if f.recovered_pending_canary > 0 || f.bypassed > 0 {
856 RemediationActionClass::SelfHealingInProgress
857 } else {
858 RemediationActionClass::NormalFailOpen
859 };
860 return make_band(
861 BandId::AdmissibleWorkers,
862 BandSeverity::Warn,
863 class,
864 "no live workers — commands run locally (fail-open)",
865 details,
866 a.top_blocker_reason.clone(),
867 );
868 }
869
870 make_band(
871 BandId::AdmissibleWorkers,
872 BandSeverity::Critical,
873 RemediationActionClass::OperatorActionRequired,
874 "no live worker is command-admissible",
875 details,
876 a.top_blocker_reason.clone(),
877 )
878}
879
880fn band_proof_queue(p: &ProofQueueInput) -> RemediationBand {
881 let active = p.queued + p.blocked + p.replaying;
882 let headline = format!(
883 "{active} proof(s) pending ({} queued · {} blocked · {} replaying)",
884 p.queued, p.blocked, p.replaying
885 );
886 let mut details = Vec::new();
887 if p.failed_recent > 0 {
888 details.push(format!("{} recently failed (terminal)", p.failed_recent));
889 }
890 if p.stale_recent > 0 {
891 details.push(format!("{} recently stale", p.stale_recent));
892 }
893
894 if active == 0 {
895 let (sev, head) = if p.failed_recent > 0 || p.stale_recent > 0 {
897 (
898 BandSeverity::Info,
899 "no proofs pending; recent terminal outcomes".to_string(),
900 )
901 } else {
902 (BandSeverity::Ok, "no proofs pending".to_string())
903 };
904 let class = if p.failed_recent > 0 || p.stale_recent > 0 {
905 RemediationActionClass::NormalFailOpen
906 } else {
907 RemediationActionClass::Healthy
908 };
909 return make_band(BandId::ProofQueue, sev, class, head, details, None);
910 }
911
912 let (severity, action_class) = if p.replaying > 0 || p.queued > 0 {
916 (
917 BandSeverity::Info,
918 RemediationActionClass::SelfHealingInProgress,
919 )
920 } else {
921 (
923 BandSeverity::Warn,
924 RemediationActionClass::OperatorActionRequired,
925 )
926 };
927
928 make_band(
929 BandId::ProofQueue,
930 severity,
931 action_class,
932 headline,
933 details,
934 None,
935 )
936}
937
938fn band_active_jobs(j: &JobsInput) -> RemediationBand {
939 let headline = format!("{} active · {} queued", j.active, j.queued);
940 let mut details = Vec::new();
941 if j.stuck > 0 {
942 details.push(format!(
943 "{} job(s) stuck (stale heartbeat/wrapper)",
944 j.stuck
945 ));
946 }
947
948 if j.stuck > 0 {
949 return make_band(
950 BandId::ActiveJobs,
951 BandSeverity::Warn,
952 RemediationActionClass::OperatorActionRequired,
953 format!("{} stuck job(s) need attention", j.stuck),
954 details,
955 Some("queue_ambiguity".to_string()),
956 );
957 }
958
959 let (severity, action_class) = if j.active > 0 || j.queued > 0 {
960 (BandSeverity::Info, RemediationActionClass::Healthy)
961 } else {
962 (BandSeverity::Ok, RemediationActionClass::Healthy)
963 };
964 make_band(
965 BandId::ActiveJobs,
966 severity,
967 action_class,
968 if j.active == 0 && j.queued == 0 {
969 "idle".to_string()
970 } else {
971 headline
972 },
973 details,
974 None,
975 )
976}
977
978fn band_disk_pressure(d: &DiskPressureInput) -> RemediationBand {
979 let mut details = Vec::new();
980 if let Some(ratio) = d.min_free_ratio {
981 details.push(format!("tightest free-disk ratio {ratio:.2}"));
982 }
983 if d.reclaim_in_progress > 0 {
984 details.push(format!("{} reclaiming space", d.reclaim_in_progress));
985 }
986
987 if d.workers_critical > 0 {
988 let class = if d.reclaim_in_progress >= d.workers_critical {
991 RemediationActionClass::SelfHealingInProgress
992 } else {
993 RemediationActionClass::OperatorActionRequired
994 };
995 return make_band(
996 BandId::DiskPressure,
997 BandSeverity::Critical,
998 class,
999 format!("{} worker(s) at critical disk pressure", d.workers_critical),
1000 details,
1001 Some("critical_pressure".to_string()),
1002 );
1003 }
1004
1005 if d.workers_warning > 0 {
1006 return make_band(
1007 BandId::DiskPressure,
1008 BandSeverity::Warn,
1009 RemediationActionClass::SelfHealingInProgress,
1010 format!(
1011 "{} worker(s) under disk-pressure warning",
1012 d.workers_warning
1013 ),
1014 details,
1015 Some("disk_full".to_string()),
1016 );
1017 }
1018
1019 make_band(
1020 BandId::DiskPressure,
1021 BandSeverity::Ok,
1022 RemediationActionClass::Healthy,
1023 "no disk pressure",
1024 details,
1025 None,
1026 )
1027}
1028
1029fn band_telemetry(t: &TelemetryFreshnessInput) -> RemediationBand {
1030 let total = t.fresh + t.stale + t.unknown;
1031 let mut details = Vec::new();
1032 if let Some(age) = t.max_age_secs {
1033 details.push(format!("oldest telemetry {age}s"));
1034 }
1035 if t.stale > 0 {
1036 details.push(format!("{} stale", t.stale));
1037 }
1038 if t.unknown > 0 {
1039 details.push(format!("{} unknown age", t.unknown));
1040 }
1041
1042 if t.stale == 0 && t.unknown == 0 {
1043 return make_band(
1044 BandId::TelemetryFreshness,
1045 BandSeverity::Ok,
1046 RemediationActionClass::Healthy,
1047 format!("{} worker(s) reporting fresh telemetry", t.fresh),
1048 details,
1049 None,
1050 );
1051 }
1052
1053 if total > 0 && t.unknown == total {
1056 return make_band(
1057 BandId::TelemetryFreshness,
1058 BandSeverity::Warn,
1059 RemediationActionClass::OperatorActionRequired,
1060 "no worker telemetry available",
1061 details,
1062 Some("telemetry_stale".to_string()),
1063 );
1064 }
1065
1066 make_band(
1067 BandId::TelemetryFreshness,
1068 BandSeverity::Warn,
1069 RemediationActionClass::SelfHealingInProgress,
1070 format!(
1071 "{} of {} worker(s) with stale/unknown telemetry",
1072 t.stale + t.unknown,
1073 total
1074 ),
1075 details,
1076 Some("telemetry_stale".to_string()),
1077 )
1078}
1079
1080fn band_incidents(incidents: &[RemediationIncidentLine]) -> RemediationBand {
1081 if incidents.is_empty() {
1082 return make_band(
1083 BandId::Incidents,
1084 BandSeverity::Ok,
1085 RemediationActionClass::Healthy,
1086 "no recent incidents",
1087 Vec::new(),
1088 None,
1089 );
1090 }
1091
1092 let worst = incidents
1093 .iter()
1094 .min_by_key(|i| i.reason_code.clone())
1095 .map(|i| i.reason_code.clone());
1096 let details: Vec<String> = incidents
1097 .iter()
1098 .take(3)
1099 .map(|i| {
1100 let who = i
1101 .worker_id
1102 .as_deref()
1103 .map_or_else(String::new, |w| format!(" [{w}]"));
1104 format!(
1105 "{} {}{} ({}s ago)",
1106 i.reason_code, i.event_type, who, i.age_secs
1107 )
1108 })
1109 .collect();
1110
1111 make_band(
1114 BandId::Incidents,
1115 BandSeverity::Info,
1116 RemediationActionClass::NormalFailOpen,
1117 format!("{} recent incident(s)", incidents.len()),
1118 details,
1119 worst,
1120 )
1121}
1122
1123#[cfg(test)]
1124mod tests {
1125 use super::*;
1126
1127 fn healthy_fleet() -> FleetSummaryInput {
1128 FleetSummaryInput {
1129 desired: 4,
1130 eligible: 4,
1131 bypassed: 0,
1132 disabled: 0,
1133 unreachable: 0,
1134 missing: 0,
1135 recovered_pending_canary: 0,
1136 problem_class: "healthy".into(),
1137 problem_summary: "4/4 worker(s) ready".into(),
1138 absence_warnings: 0,
1139 }
1140 }
1141
1142 fn healthy_inputs() -> RemediationInputs {
1143 RemediationInputs {
1144 fleet: healthy_fleet(),
1145 admissible: AdmissibleWorkersInput {
1146 admissible: 4,
1147 total_live: 4,
1148 top_blocker_reason: None,
1149 },
1150 proof_queue: ProofQueueInput::default(),
1151 jobs: JobsInput::default(),
1152 disk: DiskPressureInput::default(),
1153 telemetry: TelemetryFreshnessInput {
1154 fresh: 4,
1155 stale: 0,
1156 unknown: 0,
1157 max_age_secs: Some(5),
1158 },
1159 incidents: Vec::new(),
1160 }
1161 }
1162
1163 #[test]
1164 fn healthy_view_is_all_healthy() {
1165 let view = assemble(&healthy_inputs(), 1_000);
1166 assert_eq!(view.overall, RemediationActionClass::Healthy);
1167 assert_eq!(view.bands.len(), BandId::ALL.len());
1168 for band in &view.bands {
1169 assert_eq!(
1170 band.action_class,
1171 RemediationActionClass::Healthy,
1172 "band {} should be healthy",
1173 band.id.as_str()
1174 );
1175 }
1176 assert!(view.incidents.is_empty());
1177 assert!(!view.needs_operator_action());
1178 }
1179
1180 #[test]
1181 fn bands_are_in_canonical_order_and_complete() {
1182 let view = assemble(&healthy_inputs(), 1);
1183 let ids: Vec<BandId> = view.bands.iter().map(|b| b.id).collect();
1184 assert_eq!(ids, BandId::ALL.to_vec());
1185 }
1186
1187 #[test]
1188 fn degraded_but_eligible_is_self_healing_not_collapse() {
1189 let mut inputs = healthy_inputs();
1190 inputs.fleet = FleetSummaryInput {
1191 desired: 4,
1192 eligible: 3,
1193 bypassed: 1,
1194 problem_class: "healthy".into(),
1195 problem_summary: "3/4 worker(s) ready".into(),
1196 ..healthy_fleet()
1197 };
1198 let view = assemble(&inputs, 1);
1199 let live = view.band(BandId::LiveEligibility).unwrap();
1200 assert_eq!(
1201 live.action_class,
1202 RemediationActionClass::SelfHealingInProgress
1203 );
1204 assert_eq!(view.overall, RemediationActionClass::SelfHealingInProgress);
1205 }
1206
1207 #[test]
1208 fn admin_disabled_collapse_requires_operator() {
1209 let mut inputs = healthy_inputs();
1210 inputs.fleet = FleetSummaryInput {
1211 desired: 2,
1212 eligible: 0,
1213 disabled: 2,
1214 problem_class: "admin_intent".into(),
1215 problem_summary: "0/2 ready — operator-disabled".into(),
1216 ..healthy_fleet()
1217 };
1218 inputs.admissible = AdmissibleWorkersInput {
1219 admissible: 0,
1220 total_live: 2,
1221 top_blocker_reason: Some("admin disabled".into()),
1222 };
1223 let view = assemble(&inputs, 1);
1224 assert_eq!(view.overall, RemediationActionClass::OperatorActionRequired);
1225 let fleet = view.band(BandId::DesiredFleet).unwrap();
1226 assert_eq!(
1227 fleet.action_class,
1228 RemediationActionClass::OperatorActionRequired
1229 );
1230 assert_eq!(fleet.severity, BandSeverity::Critical);
1231 }
1232
1233 #[test]
1234 fn bypass_only_collapse_is_self_healing() {
1235 let mut inputs = healthy_inputs();
1236 inputs.fleet = FleetSummaryInput {
1237 desired: 2,
1238 eligible: 0,
1239 bypassed: 2,
1240 recovered_pending_canary: 0,
1241 problem_class: "cloud_disappearance".into(),
1242 problem_summary: "0/2 ready — all temporarily bypassed".into(),
1243 ..healthy_fleet()
1244 };
1245 inputs.admissible = AdmissibleWorkersInput {
1246 admissible: 0,
1247 total_live: 2,
1248 top_blocker_reason: None,
1249 };
1250 let view = assemble(&inputs, 1);
1251 let fleet = view.band(BandId::DesiredFleet).unwrap();
1252 assert_eq!(
1253 fleet.action_class,
1254 RemediationActionClass::SelfHealingInProgress
1255 );
1256 }
1257
1258 #[test]
1259 fn no_admissible_with_live_workers_requires_operator() {
1260 let mut inputs = healthy_inputs();
1261 inputs.admissible = AdmissibleWorkersInput {
1262 admissible: 0,
1263 total_live: 3,
1264 top_blocker_reason: Some("missing rust target".into()),
1265 };
1266 let view = assemble(&inputs, 1);
1267 let band = view.band(BandId::AdmissibleWorkers).unwrap();
1268 assert_eq!(
1269 band.action_class,
1270 RemediationActionClass::OperatorActionRequired
1271 );
1272 assert_eq!(band.reason_code.as_deref(), Some("missing rust target"));
1273 }
1274
1275 #[test]
1276 fn proof_queued_is_self_healing() {
1277 let mut inputs = healthy_inputs();
1278 inputs.proof_queue = ProofQueueInput {
1279 queued: 2,
1280 replaying: 1,
1281 ..ProofQueueInput::default()
1282 };
1283 let view = assemble(&inputs, 1);
1284 let band = view.band(BandId::ProofQueue).unwrap();
1285 assert_eq!(
1286 band.action_class,
1287 RemediationActionClass::SelfHealingInProgress
1288 );
1289 }
1290
1291 #[test]
1292 fn proof_blocked_only_requires_operator() {
1293 let mut inputs = healthy_inputs();
1294 inputs.proof_queue = ProofQueueInput {
1295 blocked: 3,
1296 ..ProofQueueInput::default()
1297 };
1298 let view = assemble(&inputs, 1);
1299 let band = view.band(BandId::ProofQueue).unwrap();
1300 assert_eq!(
1301 band.action_class,
1302 RemediationActionClass::OperatorActionRequired
1303 );
1304 }
1305
1306 #[test]
1307 fn disk_critical_without_reclaim_requires_operator() {
1308 let mut inputs = healthy_inputs();
1309 inputs.disk = DiskPressureInput {
1310 workers_critical: 2,
1311 reclaim_in_progress: 0,
1312 min_free_ratio: Some(0.02),
1313 ..DiskPressureInput::default()
1314 };
1315 let view = assemble(&inputs, 1);
1316 let band = view.band(BandId::DiskPressure).unwrap();
1317 assert_eq!(band.severity, BandSeverity::Critical);
1318 assert_eq!(
1319 band.action_class,
1320 RemediationActionClass::OperatorActionRequired
1321 );
1322 }
1323
1324 #[test]
1325 fn disk_critical_with_full_reclaim_is_self_healing() {
1326 let mut inputs = healthy_inputs();
1327 inputs.disk = DiskPressureInput {
1328 workers_critical: 2,
1329 reclaim_in_progress: 2,
1330 ..DiskPressureInput::default()
1331 };
1332 let view = assemble(&inputs, 1);
1333 let band = view.band(BandId::DiskPressure).unwrap();
1334 assert_eq!(
1335 band.action_class,
1336 RemediationActionClass::SelfHealingInProgress
1337 );
1338 }
1339
1340 #[test]
1341 fn stale_telemetry_is_self_healing_but_all_unknown_needs_operator() {
1342 let mut inputs = healthy_inputs();
1343 inputs.telemetry = TelemetryFreshnessInput {
1344 fresh: 3,
1345 stale: 1,
1346 unknown: 0,
1347 max_age_secs: Some(900),
1348 };
1349 let band = assemble(&inputs, 1)
1350 .band(BandId::TelemetryFreshness)
1351 .cloned()
1352 .unwrap();
1353 assert_eq!(
1354 band.action_class,
1355 RemediationActionClass::SelfHealingInProgress
1356 );
1357
1358 inputs.telemetry = TelemetryFreshnessInput {
1359 fresh: 0,
1360 stale: 0,
1361 unknown: 3,
1362 max_age_secs: None,
1363 };
1364 let band = assemble(&inputs, 1)
1365 .band(BandId::TelemetryFreshness)
1366 .cloned()
1367 .unwrap();
1368 assert_eq!(
1369 band.action_class,
1370 RemediationActionClass::OperatorActionRequired
1371 );
1372 }
1373
1374 #[test]
1375 fn auto_rejoin_pending_is_self_healing() {
1376 let mut inputs = healthy_inputs();
1377 inputs.fleet = FleetSummaryInput {
1378 desired: 4,
1379 eligible: 3,
1380 recovered_pending_canary: 1,
1381 problem_class: "healthy".into(),
1382 problem_summary: "3/4 ready — 1 canary pending".into(),
1383 ..healthy_fleet()
1384 };
1385 let view = assemble(&inputs, 1);
1386 let live = view.band(BandId::LiveEligibility).unwrap();
1387 assert_eq!(
1388 live.action_class,
1389 RemediationActionClass::SelfHealingInProgress
1390 );
1391 assert!(
1392 live.detail_lines
1393 .iter()
1394 .any(|l| l.contains("canary pending"))
1395 );
1396 }
1397
1398 #[test]
1399 fn stuck_jobs_require_operator() {
1400 let mut inputs = healthy_inputs();
1401 inputs.jobs = JobsInput {
1402 active: 2,
1403 queued: 1,
1404 stuck: 1,
1405 };
1406 let band = assemble(&inputs, 1)
1407 .band(BandId::ActiveJobs)
1408 .cloned()
1409 .unwrap();
1410 assert_eq!(
1411 band.action_class,
1412 RemediationActionClass::OperatorActionRequired
1413 );
1414 assert_eq!(band.reason_code.as_deref(), Some("queue_ambiguity"));
1415 }
1416
1417 #[test]
1418 fn incidents_are_informational_and_bounded() {
1419 let mut inputs = healthy_inputs();
1420 inputs.incidents = (0u64..15)
1421 .map(|i| {
1422 RemediationIncidentLine::new(
1423 format!("RCH-I0{:02}", i % 9 + 1),
1424 "selection",
1425 Some("css".into()),
1426 i * 10,
1427 "no admissible workers",
1428 )
1429 })
1430 .collect();
1431 let view = assemble(&inputs, 1);
1432 assert_eq!(view.incidents.len(), MAX_VIEW_INCIDENTS);
1433 let band = view.band(BandId::Incidents).unwrap();
1434 assert_eq!(band.action_class, RemediationActionClass::NormalFailOpen);
1435 assert_eq!(band.severity, BandSeverity::Info);
1436 }
1437
1438 #[test]
1439 fn no_workers_configured_is_normal_fail_open() {
1440 let mut inputs = healthy_inputs();
1441 inputs.fleet = FleetSummaryInput {
1442 desired: 0,
1443 eligible: 0,
1444 problem_class: "healthy".into(),
1445 problem_summary: "no workers configured".into(),
1446 ..healthy_fleet()
1447 };
1448 inputs.admissible = AdmissibleWorkersInput {
1449 admissible: 0,
1450 total_live: 0,
1451 top_blocker_reason: None,
1452 };
1453 inputs.telemetry = TelemetryFreshnessInput::default();
1454 let view = assemble(&inputs, 1);
1455 assert_eq!(view.overall, RemediationActionClass::NormalFailOpen);
1456 }
1457
1458 #[test]
1459 fn free_text_is_redacted_at_construction() {
1460 let mut inputs = healthy_inputs();
1461 inputs.fleet.problem_summary =
1462 "ssh failed for ubuntu@203.0.113.20 key AKIAIOSFODNN7EXAMPLE".into();
1463 inputs.incidents = vec![RemediationIncidentLine::new(
1464 "RCH-I010",
1465 "hook",
1466 None,
1467 3,
1468 "token Bearer abcdef0123456789abcdef0123456789 at /home/ubuntu/.ssh/id_rsa",
1469 )];
1470 let view = assemble(&inputs, 1);
1471 let json = serde_json::to_string(&view).unwrap();
1472 assert!(
1473 !json.contains("AKIAIOSFODNN7EXAMPLE"),
1474 "AWS-shaped key leaked"
1475 );
1476 assert!(
1477 !json.contains("abcdef0123456789abcdef0123456789"),
1478 "bearer leaked"
1479 );
1480 }
1481
1482 #[test]
1483 fn action_class_ordering_is_severity_ordered() {
1484 assert!(
1485 RemediationActionClass::OperatorActionRequired
1486 > RemediationActionClass::SelfHealingInProgress
1487 );
1488 assert!(
1489 RemediationActionClass::SelfHealingInProgress > RemediationActionClass::NormalFailOpen
1490 );
1491 assert!(RemediationActionClass::NormalFailOpen > RemediationActionClass::Healthy);
1492 assert_eq!(
1493 RemediationActionClass::Healthy
1494 .max_class(RemediationActionClass::OperatorActionRequired),
1495 RemediationActionClass::OperatorActionRequired
1496 );
1497 }
1498
1499 #[test]
1500 fn view_round_trips_through_json() {
1501 let view = assemble(&healthy_inputs(), 42);
1502 let json = serde_json::to_string(&view).unwrap();
1503 let back: RemediationView = serde_json::from_str(&json).unwrap();
1504 assert_eq!(view, back);
1505 }
1506
1507 fn ready_row(id: &str) -> RemediationWorkerRow {
1508 use crate::fleet_diff::WorkerObservation;
1509 RemediationWorkerRow {
1510 observation: WorkerObservation {
1511 worker_id: id.into(),
1512 configured: true,
1513 in_daemon_pool: true,
1514 reachable: true,
1515 admin_disabled: false,
1516 temporarily_bypassed: false,
1517 facts_known: true,
1518 command_admissible: true,
1519 },
1520 disk_level: DiskLevel::Ok,
1521 reclaiming: false,
1522 free_ratio: Some(0.8),
1523 slots_used: 0,
1524 slots_total: 8,
1525 telemetry_known: true,
1526 telemetry_fresh: true,
1527 telemetry_age_secs: Some(3),
1528 recovered_pending_canary: false,
1529 absent_secs: None,
1530 }
1531 }
1532
1533 #[test]
1534 fn build_inputs_healthy_fleet_round_trips_to_healthy_view() {
1535 let rows = vec![ready_row("css"), ready_row("ovh-a"), ready_row("ovh-b")];
1536 let inputs = build_inputs(
1537 &rows,
1538 JobsInput::default(),
1539 ProofQueueInput::default(),
1540 Vec::new(),
1541 300,
1542 );
1543 assert_eq!(inputs.fleet.desired, 3);
1544 assert_eq!(inputs.fleet.eligible, 3);
1545 assert_eq!(inputs.admissible.admissible, 3);
1546 assert_eq!(inputs.telemetry.fresh, 3);
1547 let view = assemble(&inputs, 1);
1548 assert_eq!(view.overall, RemediationActionClass::Healthy);
1549 }
1550
1551 #[test]
1552 fn build_inputs_maps_bypass_disk_and_unknown_facts() {
1553 let mut bypassed = ready_row("css");
1554 bypassed.observation.temporarily_bypassed = true;
1555 bypassed.recovered_pending_canary = true;
1556
1557 let mut crit = ready_row("ovh-a");
1558 crit.disk_level = DiskLevel::Critical;
1559 crit.reclaiming = true;
1560 crit.free_ratio = Some(0.01);
1561
1562 let mut no_facts = ready_row("ovh-b");
1563 no_facts.observation.facts_known = false;
1564 no_facts.telemetry_known = false;
1565 no_facts.telemetry_age_secs = None;
1566
1567 let rows = vec![bypassed, crit, no_facts];
1568 let inputs = build_inputs(
1569 &rows,
1570 JobsInput {
1571 active: 1,
1572 queued: 0,
1573 stuck: 0,
1574 },
1575 ProofQueueInput::default(),
1576 Vec::new(),
1577 300,
1578 );
1579 assert_eq!(inputs.fleet.bypassed, 1);
1580 assert_eq!(inputs.fleet.recovered_pending_canary, 1);
1581 assert_eq!(inputs.disk.workers_critical, 1);
1582 assert_eq!(inputs.disk.reclaim_in_progress, 1);
1583 assert_eq!(inputs.disk.min_free_ratio, Some(0.01));
1584 assert_eq!(inputs.telemetry.unknown, 1);
1586 assert!(inputs.admissible.top_blocker_reason.is_some() || inputs.admissible.admissible > 0);
1587 }
1588
1589 #[test]
1590 fn zero_slot_or_critical_pressure_workers_are_not_command_admissible() {
1591 let mut zero_slot = ready_row("hz1");
1595 zero_slot.slots_total = 0;
1596 let mut critical = ready_row("hz2");
1597 critical.disk_level = DiskLevel::Critical;
1598 let mut saturated = ready_row("css");
1599 saturated.slots_used = saturated.slots_total; let rows = vec![zero_slot.clone(), critical.clone(), saturated.clone()];
1602 let inputs = build_inputs(
1603 &rows,
1604 JobsInput::default(),
1605 ProofQueueInput::default(),
1606 Vec::new(),
1607 300,
1608 );
1609 assert_eq!(inputs.admissible.total_live, 3);
1610 assert_eq!(
1611 inputs.admissible.admissible, 0,
1612 "no free slot or critical pressure must not read as admissible"
1613 );
1614 assert!(inputs.admissible.top_blocker_reason.is_some());
1615
1616 let mut free = ready_row("ovh-a");
1618 free.slots_used = 2;
1619 let rows = vec![zero_slot, critical, saturated, free];
1620 let inputs = build_inputs(
1621 &rows,
1622 JobsInput::default(),
1623 ProofQueueInput::default(),
1624 Vec::new(),
1625 300,
1626 );
1627 assert_eq!(inputs.admissible.admissible, 1);
1628 }
1629
1630 #[test]
1631 fn all_enum_tokens_are_stable_and_unique() {
1632 use std::collections::HashSet;
1633 let action: HashSet<_> = RemediationActionClass::ALL
1634 .iter()
1635 .map(|c| c.as_str())
1636 .collect();
1637 assert_eq!(action.len(), RemediationActionClass::ALL.len());
1638 let sev: HashSet<_> = BandSeverity::ALL.iter().map(|s| s.as_str()).collect();
1639 assert_eq!(sev.len(), BandSeverity::ALL.len());
1640 let bands: HashSet<_> = BandId::ALL.iter().map(|b| b.as_str()).collect();
1641 assert_eq!(bands.len(), BandId::ALL.len());
1642 }
1643}