rust_supervisor/config/
state.rs

1//! Immutable configuration state for supervisor runtime values.
2//!
3//! Raw YAML input belongs to [`crate::config::configurable`]. This module owns
4//! semantic validation and conversion into supervisor runtime declarations.
5
6use crate::config::audit::AuditConfig;
7use crate::config::configurable::{
8    DashboardIpcConfig, ObservabilityConfig, PolicyConfig, ShutdownConfig, SupervisorConfig,
9    SupervisorRootConfig,
10};
11use crate::config::ipc_security::IpcSecurityConfig;
12use crate::config::policy::{
13    ChildStrategyOverrideConfig, GroupConfig, GroupDependencyConfig, GroupStrategyConfig,
14    SeverityDefaultConfig,
15};
16use crate::id::types::ChildId;
17use crate::spec::child::ChildSpec;
18use crate::spec::child_declaration::{ChildDeclaration, CompensatingRecord, PendingChild, Phase};
19use crate::spec::supervisor::BackpressureConfig;
20use serde::{Deserialize, Serialize};
21use std::collections::{HashMap, HashSet};
22use std::time::Duration;
23use uuid::Uuid;
24
25/// Supervisor configuration state with add_child transaction support.
26#[derive(Debug, Clone, Serialize, Deserialize)]
27pub struct ConfigState {
28    /// Root supervisor declaration values.
29    pub supervisor: SupervisorRootConfig,
30    /// Runtime policy values.
31    pub policy: PolicyConfig,
32    /// Shutdown budget values.
33    pub shutdown: ShutdownConfig,
34    /// Observability switches and capacities.
35    pub observability: ObservabilityConfig,
36    /// Command audit persistence configuration.
37    pub audit: AuditConfig,
38    /// Backpressure policy for observability event subscribers.
39    pub backpressure: BackpressureConfig,
40    /// Group-level restart budgets and group policy declarations.
41    pub groups: Vec<GroupConfig>,
42    /// Group-level strategy overrides.
43    pub group_strategies: Vec<GroupStrategyConfig>,
44    /// Cross-group failure propagation dependencies.
45    pub group_dependencies: Vec<GroupDependencyConfig>,
46    /// Child-level strategy overrides.
47    pub child_strategy_overrides: Vec<ChildStrategyOverrideConfig>,
48    /// Default severity class per task role.
49    pub severity_defaults: Vec<SeverityDefaultConfig>,
50    /// Optional target-side dashboard IPC configuration.
51    pub dashboard: Option<DashboardIpcConfig>,
52    /// Validated child specifications loaded from YAML declarations.
53    #[serde(default)]
54    pub children: Vec<ChildSpec>,
55    /// SHA-256 hash of the SupervisorSpec for audit reconciliation.
56    #[serde(default)]
57    pub spec_hash: String,
58    /// Pending add_child transactions.
59    #[serde(default)]
60    pub pending_additions: Vec<PendingChild>,
61    /// Compensating records for recovery.
62    #[serde(default)]
63    pub compensating_records: Vec<CompensatingRecord>,
64}
65
66/// Manual partial equality — skips `children` because [`ChildSpec`] contains
67/// `Arc<dyn TaskFactory>` which does not implement `PartialEq`.
68impl PartialEq for ConfigState {
69    /// Compares two ConfigState values, skipping the `children` vector.
70    fn eq(&self, other: &Self) -> bool {
71        self.supervisor == other.supervisor
72            && self.policy == other.policy
73            && self.shutdown == other.shutdown
74            && self.observability == other.observability
75            && self.audit == other.audit
76            && self.backpressure == other.backpressure
77            && self.groups == other.groups
78            && self.group_strategies == other.group_strategies
79            && self.group_dependencies == other.group_dependencies
80            && self.child_strategy_overrides == other.child_strategy_overrides
81            && self.severity_defaults == other.severity_defaults
82            && self.dashboard == other.dashboard
83            && self.spec_hash == other.spec_hash
84            && self.pending_additions == other.pending_additions
85    }
86}
87
88impl TryFrom<SupervisorConfig> for ConfigState {
89    type Error = crate::error::types::SupervisorError;
90
91    /// Converts a deserialized supervisor config into validated state.
92    fn try_from(config: SupervisorConfig) -> Result<Self, Self::Error> {
93        validate_policy(&config.policy)?;
94        validate_shutdown(&config.shutdown)?;
95        validate_observability(&config.observability)?;
96        validate_audit(&config.audit)?;
97        validate_backpressure(&config.backpressure)?;
98        validate_lower_policy(&config.policy)?;
99        validate_supervisor_root(&config.supervisor)?;
100        validate_group_inputs(
101            config.groups.as_slice(),
102            &config.group_strategies,
103            &config.group_dependencies,
104            &config.child_strategy_overrides,
105            &config.severity_defaults,
106            config.children.as_slice(),
107        )?;
108        let dashboard = dashboard_with_default_security(config.dashboard);
109        validate_dashboard(dashboard.as_ref())?;
110
111        // Validate and convert child declarations.
112        use crate::spec::child_declaration::validate_child_declaration;
113        use crate::tree::order::kahn_sort;
114
115        // Collect all child names for validation.
116        let all_names: HashSet<String> = config
117            .children
118            .as_slice()
119            .iter()
120            .map(|c| c.name.clone())
121            .collect();
122
123        // Validate each declaration.
124        for child in config.children.as_slice() {
125            validate_child_declaration(child, &all_names).map_err(|e| {
126                crate::error::types::SupervisorError::fatal_config(format!(
127                    "Child declaration validation failed at {}: {}",
128                    e.field_path, e.reason
129                ))
130            })?;
131        }
132
133        // Convert to ChildSpec list.
134        let declarations: Vec<ChildDeclaration> = config.children.into();
135        let child_specs: Vec<ChildSpec> = declarations
136            .into_iter()
137            .map(ChildSpec::try_from)
138            .collect::<Result<Vec<_>, _>>()
139            .map_err(|e| {
140                crate::error::types::SupervisorError::fatal_config(format!(
141                    "Child declaration conversion failed at {}: {}",
142                    e.field_path, e.reason
143                ))
144            })?;
145
146        // Topological sort.
147        let _sorted = kahn_sort(&child_specs).map_err(|cycle_nodes| {
148            let node_names: Vec<String> = cycle_nodes.iter().map(|id| id.value.clone()).collect();
149            crate::error::types::SupervisorError::fatal_config(format!(
150                "Dependency cycle detected among children: {node_names:?}",
151            ))
152        })?;
153
154        let spec_hash = String::new(); // Will be computed after SupervisorSpec is built.
155
156        Ok(Self {
157            supervisor: config.supervisor,
158            policy: config.policy,
159            shutdown: config.shutdown,
160            observability: config.observability,
161            audit: config.audit,
162            backpressure: config.backpressure,
163            groups: config.groups.into(),
164            group_strategies: config.group_strategies,
165            group_dependencies: config.group_dependencies,
166            child_strategy_overrides: config.child_strategy_overrides,
167            severity_defaults: config.severity_defaults,
168            dashboard,
169            children: child_specs,
170            spec_hash,
171            pending_additions: Vec::new(),
172            compensating_records: Vec::new(),
173        })
174    }
175}
176
177impl ConfigState {
178    /// Converts validated configuration into a supervisor declaration.
179    ///
180    /// # Arguments
181    ///
182    /// This function has no arguments.
183    ///
184    /// # Returns
185    ///
186    /// Returns a [`crate::spec::supervisor::SupervisorSpec`] derived from the
187    /// validated YAML configuration.
188    ///
189    /// # Examples
190    ///
191    /// Begins an add_child transaction by creating a PendingChild entry.
192    ///
193    /// # Arguments
194    ///
195    /// - `declaration`: The child declaration to stage.
196    ///
197    /// # Returns
198    ///
199    /// Returns the generated transaction UUID on success.
200    ///
201    /// # Errors
202    ///
203    /// Returns an error if a transaction is already in progress.
204    pub fn begin_transaction(
205        &mut self,
206        declaration: ChildDeclaration,
207    ) -> Result<Uuid, crate::error::types::SupervisorError> {
208        if self.has_pending_transaction() {
209            return Err(crate::error::types::SupervisorError::fatal_config(
210                "add_child transaction already in progress",
211            ));
212        }
213        let transaction_id = Uuid::new_v4();
214        let child_spec = Box::new(ChildSpec::try_from(declaration.clone()).map_err(|e| {
215            crate::error::types::SupervisorError::fatal_config(format!(
216                "Child declaration conversion failed: {}",
217                e.reason
218            ))
219        })?);
220        let pending = PendingChild {
221            transaction_id,
222            declaration,
223            child_spec,
224            phase: Phase::Parsed,
225            created_at_unix_nanos: std::time::SystemTime::now()
226                .duration_since(std::time::UNIX_EPOCH)
227                .unwrap_or_default()
228                .as_nanos(),
229        };
230        self.pending_additions.push(pending);
231        Ok(transaction_id)
232    }
233
234    /// Commits an add_child transaction, registering the child in the topology.
235    ///
236    /// # Arguments
237    ///
238    /// - `transaction_id`: The transaction UUID to commit.
239    pub fn commit_transaction(
240        &mut self,
241        transaction_id: Uuid,
242    ) -> Result<(), crate::error::types::SupervisorError> {
243        let idx = self
244            .pending_additions
245            .iter()
246            .position(|p| p.transaction_id == transaction_id)
247            .ok_or_else(|| {
248                crate::error::types::SupervisorError::fatal_config(
249                    "transaction not found for commit",
250                )
251            })?;
252
253        let mut pending = self.pending_additions.remove(idx);
254        pending.phase = Phase::Committed;
255
256        // Register child in the topology.
257        let spec = (*pending.child_spec).clone();
258        self.children.push(spec);
259
260        // Compute a deterministic spec hash from the current children list.
261        // Uses JSON serialization for cross-version stability; this is a
262        // content hash for change detection, not a cryptographic digest.
263        self.spec_hash = self.compute_spec_hash();
264
265        Ok(())
266    }
267
268    /// Rolls back an add_child transaction, creating a compensating record.
269    ///
270    /// # Arguments
271    ///
272    /// - `transaction_id`: The transaction UUID to roll back.
273    /// - `error`: Human-readable error description.
274    pub fn rollback_transaction(
275        &mut self,
276        transaction_id: Uuid,
277        error: String,
278    ) -> Result<(), crate::error::types::SupervisorError> {
279        let idx = self
280            .pending_additions
281            .iter()
282            .position(|p| p.transaction_id == transaction_id);
283
284        let pending = if let Some(i) = idx {
285            self.pending_additions.remove(i)
286        } else {
287            return Err(crate::error::types::SupervisorError::fatal_config(
288                "transaction not found for rollback",
289            ));
290        };
291
292        // Create compensating record.
293        let record = CompensatingRecord {
294            transaction_id,
295            operation: "add_child".to_string(),
296            state: "compensated".to_string(),
297            child_name: pending.declaration.name.clone(),
298            declaration_hash: compute_declaration_hash(&pending.declaration),
299            error: Some(error),
300            correlation_id: None,
301            child_id: Some(pending.child_spec.id.value.clone()),
302            created_at_unix_nanos: std::time::SystemTime::now()
303                .duration_since(std::time::UNIX_EPOCH)
304                .unwrap_or_default()
305                .as_nanos(),
306        };
307        self.compensating_records.push(record);
308
309        Ok(())
310    }
311
312    /// Returns true when a pending transaction exists.
313    pub fn has_pending_transaction(&self) -> bool {
314        self.pending_additions
315            .iter()
316            .any(|p| p.phase != Phase::Committed && p.phase != Phase::Compensated)
317    }
318
319    /// Returns the current spec hash for audit reconciliation.
320    pub fn hash(&self) -> &str {
321        &self.spec_hash
322    }
323
324    /// Computes a deterministic spec hash from the current children list.
325    ///
326    /// Uses JSON serialization for cross-version stability. This is a
327    /// content hash for change detection and audit reconciliation, not a
328    /// cryptographic digest. Changes to the serialization format will
329    /// produce different hashes — this is acceptable because the hash is
330    /// always recomputed from the current state after restart.
331    pub fn compute_spec_hash(&self) -> String {
332        let json = serde_json::to_string(&self.children).unwrap_or_default();
333        use std::hash::{Hash, Hasher};
334        let mut hasher = std::collections::hash_map::DefaultHasher::new();
335        json.hash(&mut hasher);
336        format!("v{hash:x}", hash = hasher.finish())
337    }
338
339    /// Recovers pending transactions after a restart.
340    ///
341    /// Iterates compensating records and reconciles them against the
342    /// current spec state. Records with state "pending" that have a
343    /// matching declaration hash are marked as committed.
344    pub fn recover_pending_transactions(&mut self) {
345        let mut recovered = Vec::new();
346        for record in self.compensating_records.iter_mut() {
347            if record.state == "pending" {
348                // Mark as compensated since we cannot truly roll back
349                // runtime state after a restart without full runtime.
350                record.state = "compensated".to_string();
351                recovered.push(record.transaction_id);
352            }
353        }
354        if !recovered.is_empty() {
355            // Log recovery info.
356            #[cfg(debug_assertions)]
357            eprintln!("Recovered {} pending transactions", recovered.len());
358        }
359    }
360
361    /// ```
362    /// let yaml = r#"
363    /// supervisor:
364    ///   strategy: OneForAll
365    ///   control_channel_capacity: 256
366    ///   event_channel_capacity: 256
367    /// policy:
368    ///   child_restart_limit: 10
369    ///   child_restart_window_ms: 60000
370    ///   supervisor_failure_limit: 30
371    ///   supervisor_failure_window_ms: 60000
372    ///   initial_backoff_ms: 10
373    ///   max_backoff_ms: 1000
374    ///   jitter_ratio: 0.0
375    ///   heartbeat_interval_ms: 1000
376    ///   stale_after_ms: 3000
377    /// shutdown:
378    ///   graceful_timeout_ms: 1000
379    ///   abort_wait_ms: 100
380    /// observability:
381    ///   event_journal_capacity: 64
382    ///   metrics_enabled: true
383    ///   audit_enabled: true
384    /// "#;
385    /// let state = rust_supervisor::config::yaml::parse_config_state(yaml).unwrap();
386    /// let spec = state.to_supervisor_spec().unwrap();
387    /// assert_eq!(spec.strategy, rust_supervisor::spec::supervisor::SupervisionStrategy::OneForAll);
388    /// assert_eq!(spec.supervisor_failure_limit, 30);
389    /// ```
390    pub fn to_supervisor_spec(
391        &self,
392    ) -> Result<crate::spec::supervisor::SupervisorSpec, crate::error::types::SupervisorError> {
393        let spec = self.build_supervisor_spec();
394        spec.validate()?;
395        Ok(spec)
396    }
397
398    /// Converts validated configuration into a supervisor declaration and binds factories.
399    ///
400    /// # Arguments
401    ///
402    /// - `registry`: Task factory registry used to resolve worker `factory_key` values.
403    ///
404    /// # Returns
405    ///
406    /// Returns a [`crate::spec::supervisor::SupervisorSpec`] with executable
407    /// task factories assigned to worker children.
408    ///
409    /// # Errors
410    ///
411    /// Returns [`crate::error::types::SupervisorError`] when factory binding or
412    /// supervisor validation fails.
413    pub fn to_supervisor_spec_with_factories(
414        &self,
415        registry: &crate::task::factory_registry::TaskFactoryRegistry,
416    ) -> Result<crate::spec::supervisor::SupervisorSpec, crate::error::types::SupervisorError> {
417        let mut spec = self.build_supervisor_spec();
418        crate::config::factory_binding::bind_task_factories(&mut spec.children, registry)?;
419        spec.validate()?;
420        Ok(spec)
421    }
422
423    /// Builds a supervisor specification before final validation.
424    ///
425    /// # Arguments
426    ///
427    /// This function has no arguments.
428    ///
429    /// # Returns
430    ///
431    /// Returns a supervisor specification assembled from validated config state.
432    fn build_supervisor_spec(&self) -> crate::spec::supervisor::SupervisorSpec {
433        let mut spec = crate::spec::supervisor::SupervisorSpec::root(self.children.clone());
434        spec.strategy = self.supervisor.strategy;
435        spec.config_version = self.config_version();
436        spec.supervisor_failure_limit = self.policy.supervisor_failure_limit;
437        spec.escalation_policy = self.supervisor.escalation_policy;
438        spec.control_channel_capacity = self.supervisor.control_channel_capacity;
439        spec.event_channel_capacity = self.supervisor.event_channel_capacity;
440        spec.backpressure_config = self.backpressure.clone();
441        let group_members = derive_group_members(&self.children);
442        spec.group_configs = self
443            .groups
444            .iter()
445            .map(|group| {
446                let members = group_members
447                    .get(group.name.as_str())
448                    .map(Vec::as_slice)
449                    .unwrap_or(&[]);
450                group.to_runtime(members)
451            })
452            .collect::<Vec<_>>();
453        spec.group_strategies = self
454            .group_strategies
455            .iter()
456            .map(GroupStrategyConfig::to_runtime)
457            .collect::<Vec<_>>();
458        spec.group_dependencies = self
459            .group_dependencies
460            .iter()
461            .map(GroupDependencyConfig::to_runtime)
462            .collect::<Vec<_>>();
463        spec.child_strategy_overrides = self
464            .child_strategy_overrides
465            .iter()
466            .map(ChildStrategyOverrideConfig::to_runtime)
467            .collect::<Vec<_>>();
468        spec.severity_defaults = self
469            .severity_defaults
470            .iter()
471            .map(|default| (default.task_role, default.severity))
472            .collect::<HashMap<_, _>>();
473        spec.dynamic_supervisor_policy = self.supervisor.dynamic_supervisor.to_runtime();
474        spec.meltdown_policy = self.policy.meltdown.to_runtime();
475        spec.failure_window_config = self.policy.failure_window.to_runtime();
476        spec.restart_budget_config = self.policy.restart_budget.to_runtime();
477        spec.pipeline_journal_capacity = self.policy.supervision_pipeline.journal_capacity;
478        spec.pipeline_subscriber_capacity = self.policy.supervision_pipeline.subscriber_capacity;
479        spec.concurrent_restart_limit = self.policy.supervision_pipeline.concurrent_restart_limit;
480        spec.default_backoff_policy = crate::spec::child::BackoffPolicy::new(
481            Duration::from_millis(self.policy.initial_backoff_ms),
482            Duration::from_millis(self.policy.max_backoff_ms),
483            self.policy.jitter_ratio,
484        );
485        spec.default_health_policy = crate::spec::child::HealthPolicy::new(
486            Duration::from_millis(self.policy.heartbeat_interval_ms),
487            Duration::from_millis(self.policy.stale_after_ms),
488        );
489        spec.tree_shutdown = crate::spec::shutdown::TreeShutdownPolicy::with_budget(
490            crate::spec::shutdown::ShutdownBudget::new(
491                Duration::from_millis(self.shutdown.graceful_timeout_ms),
492                Duration::from_millis(self.shutdown.abort_wait_ms),
493            ),
494        );
495        spec.propagate_tree_shutdown_budget();
496        spec.restart_limit = Some(crate::spec::supervisor::RestartLimit::new(
497            self.policy.child_restart_limit,
498            Duration::from_millis(self.policy.child_restart_window_ms),
499        ));
500        spec.metrics_enabled = self.observability.metrics_enabled;
501        spec.audit_enabled = self.observability.audit_enabled;
502        spec
503    }
504
505    /// Builds a stable configuration version string from configured values.
506    ///
507    /// # Arguments
508    ///
509    /// This function has no arguments.
510    ///
511    /// # Returns
512    ///
513    /// Returns a deterministic version string for diagnostics.
514    fn config_version(&self) -> String {
515        format!(
516            "supervisor-{:?}-channels-{}-{}-policy-{}-{}-shutdown-{}-observe-{}-backpressure-{:?}-{}-{}",
517            self.supervisor.strategy,
518            self.supervisor.control_channel_capacity,
519            self.supervisor.event_channel_capacity,
520            self.policy.child_restart_limit,
521            self.policy.supervisor_failure_limit,
522            self.shutdown.graceful_timeout_ms,
523            self.observability.event_journal_capacity,
524            self.backpressure.strategy,
525            self.backpressure.warn_threshold_pct,
526            self.backpressure.critical_threshold_pct
527        )
528    }
529}
530
531/// Validates policy configuration invariants.
532///
533/// # Arguments
534///
535/// - `policy`: Policy configuration loaded from YAML.
536///
537/// # Returns
538///
539/// Returns `Ok(())` when policy values are usable.
540fn validate_policy(policy: &PolicyConfig) -> Result<(), crate::error::types::SupervisorError> {
541    validate_positive(policy.child_restart_limit, "policy.child_restart_limit")?;
542    validate_positive(
543        policy.supervisor_failure_limit,
544        "policy.supervisor_failure_limit",
545    )?;
546    validate_positive(
547        policy.child_restart_window_ms,
548        "policy.child_restart_window_ms",
549    )?;
550    validate_positive(
551        policy.supervisor_failure_window_ms,
552        "policy.supervisor_failure_window_ms",
553    )?;
554    validate_positive(policy.initial_backoff_ms, "policy.initial_backoff_ms")?;
555    validate_positive(policy.max_backoff_ms, "policy.max_backoff_ms")?;
556    validate_positive(policy.heartbeat_interval_ms, "policy.heartbeat_interval_ms")?;
557    validate_positive(policy.stale_after_ms, "policy.stale_after_ms")?;
558    if policy.initial_backoff_ms > policy.max_backoff_ms {
559        return Err(crate::error::types::SupervisorError::fatal_config(
560            "policy.initial_backoff_ms must be less than or equal to policy.max_backoff_ms",
561        ));
562    }
563    if !(0.0..=1.0).contains(&policy.jitter_ratio) {
564        return Err(crate::error::types::SupervisorError::fatal_config(
565            "policy.jitter_ratio must be between 0 and 1",
566        ));
567    }
568    Ok(())
569}
570
571/// Validates lower-level policy sections loaded from YAML.
572///
573/// # Arguments
574///
575/// - `policy`: Policy configuration loaded from YAML.
576///
577/// # Returns
578///
579/// Returns `Ok(())` when lower-level policy values are usable.
580fn validate_lower_policy(
581    policy: &PolicyConfig,
582) -> Result<(), crate::error::types::SupervisorError> {
583    validate_positive(
584        policy.restart_budget.window_secs,
585        "policy.restart_budget.window_secs",
586    )?;
587    validate_positive(
588        policy.restart_budget.max_burst as u64,
589        "policy.restart_budget.max_burst",
590    )?;
591    if !(0.0..=1000.0).contains(&policy.restart_budget.recovery_rate_per_sec)
592        || policy.restart_budget.recovery_rate_per_sec == 0.0
593    {
594        return Err(crate::error::types::SupervisorError::fatal_config(
595            "policy.restart_budget.recovery_rate_per_sec must be greater than 0 and less than or equal to 1000",
596        ));
597    }
598
599    match policy.failure_window.mode {
600        crate::config::policy::FailureWindowMode::TimeSliding => {
601            validate_positive(
602                policy.failure_window.window_secs,
603                "policy.failure_window.window_secs",
604            )?;
605        }
606        crate::config::policy::FailureWindowMode::CountSliding => {
607            validate_positive(
608                policy.failure_window.max_count as u64,
609                "policy.failure_window.max_count",
610            )?;
611        }
612    }
613    validate_positive(
614        policy.failure_window.threshold as u64,
615        "policy.failure_window.threshold",
616    )?;
617
618    validate_positive(
619        policy.meltdown.child_max_restarts as u64,
620        "policy.meltdown.child_max_restarts",
621    )?;
622    validate_positive(
623        policy.meltdown.child_window_secs,
624        "policy.meltdown.child_window_secs",
625    )?;
626    validate_positive(
627        policy.meltdown.group_max_failures as u64,
628        "policy.meltdown.group_max_failures",
629    )?;
630    validate_positive(
631        policy.meltdown.group_window_secs,
632        "policy.meltdown.group_window_secs",
633    )?;
634    validate_positive(
635        policy.meltdown.supervisor_max_failures as u64,
636        "policy.meltdown.supervisor_max_failures",
637    )?;
638    validate_positive(
639        policy.meltdown.supervisor_window_secs,
640        "policy.meltdown.supervisor_window_secs",
641    )?;
642    validate_positive(
643        policy.meltdown.reset_after_secs,
644        "policy.meltdown.reset_after_secs",
645    )?;
646    validate_positive(
647        policy.supervision_pipeline.journal_capacity as u64,
648        "policy.supervision_pipeline.journal_capacity",
649    )?;
650    validate_positive(
651        policy.supervision_pipeline.subscriber_capacity as u64,
652        "policy.supervision_pipeline.subscriber_capacity",
653    )?;
654    validate_positive(
655        policy.supervision_pipeline.concurrent_restart_limit as u64,
656        "policy.supervision_pipeline.concurrent_restart_limit",
657    )
658}
659
660/// Validates root supervisor policy options.
661///
662/// # Arguments
663///
664/// - `supervisor`: Root supervisor configuration loaded from YAML.
665///
666/// # Returns
667///
668/// Returns `Ok(())` when root policy values are usable.
669fn validate_supervisor_root(
670    supervisor: &SupervisorRootConfig,
671) -> Result<(), crate::error::types::SupervisorError> {
672    validate_positive(
673        supervisor.control_channel_capacity as u64,
674        "supervisor.control_channel_capacity",
675    )?;
676    validate_positive(
677        supervisor.event_channel_capacity as u64,
678        "supervisor.event_channel_capacity",
679    )?;
680    if supervisor.dynamic_supervisor.child_limit == Some(0) {
681        return Err(crate::error::types::SupervisorError::fatal_config(
682            "supervisor.dynamic_supervisor.child_limit must be greater than zero",
683        ));
684    }
685    Ok(())
686}
687
688/// Builds group membership rosters from child group assignments.
689///
690/// # Arguments
691///
692/// - `children`: Validated child specifications loaded from YAML.
693///
694/// # Returns
695///
696/// Returns a map from group name to member child identifiers.
697fn derive_group_members(children: &[ChildSpec]) -> HashMap<String, Vec<ChildId>> {
698    let mut members = HashMap::<String, Vec<ChildId>>::new();
699    for child in children {
700        if let Some(group) = child.group.as_deref() {
701            members
702                .entry(group.to_owned())
703                .or_default()
704                .push(child.id.clone());
705        }
706    }
707    members
708}
709
710/// Validates group and override inputs against declared child names.
711///
712/// # Arguments
713///
714/// - `groups`: Group declarations loaded from YAML.
715/// - `group_strategies`: Group strategy overrides loaded from YAML.
716/// - `group_dependencies`: Group dependency edges loaded from YAML.
717/// - `child_strategy_overrides`: Child strategy overrides loaded from YAML.
718/// - `severity_defaults`: Role severity defaults loaded from YAML.
719/// - `children`: Child declarations loaded from YAML.
720///
721/// # Returns
722///
723/// Returns `Ok(())` when group and override declarations are coherent.
724fn validate_group_inputs(
725    groups: &[GroupConfig],
726    group_strategies: &[GroupStrategyConfig],
727    group_dependencies: &[GroupDependencyConfig],
728    child_strategy_overrides: &[ChildStrategyOverrideConfig],
729    severity_defaults: &[SeverityDefaultConfig],
730    children: &[ChildDeclaration],
731) -> Result<(), crate::error::types::SupervisorError> {
732    let child_names = children
733        .iter()
734        .map(|child| child.name.as_str())
735        .collect::<HashSet<_>>();
736    let mut group_names = HashSet::new();
737    for group in groups {
738        if group.name.trim().is_empty() {
739            return Err(crate::error::types::SupervisorError::fatal_config(
740                "groups[].name must not be empty",
741            ));
742        }
743        if !group_names.insert(group.name.as_str()) {
744            return Err(crate::error::types::SupervisorError::fatal_config(format!(
745                "duplicate group name '{}'",
746                group.name
747            )));
748        }
749    }
750
751    for strategy in group_strategies {
752        if !group_names.contains(strategy.group.as_str()) {
753            return Err(crate::error::types::SupervisorError::fatal_config(format!(
754                "group_strategies references unknown group '{}'",
755                strategy.group
756            )));
757        }
758        validate_restart_limit_input(
759            strategy.restart_limit.as_ref(),
760            "group_strategies.restart_limit",
761        )?;
762    }
763
764    for dependency in group_dependencies {
765        if !group_names.contains(dependency.from_group.as_str()) {
766            return Err(crate::error::types::SupervisorError::fatal_config(format!(
767                "group_dependencies references unknown from_group '{}'",
768                dependency.from_group
769            )));
770        }
771        if !group_names.contains(dependency.to_group.as_str()) {
772            return Err(crate::error::types::SupervisorError::fatal_config(format!(
773                "group_dependencies references unknown to_group '{}'",
774                dependency.to_group
775            )));
776        }
777    }
778
779    // Detect cycles in group dependency graph using Kahn's algorithm.
780    if !group_dependencies.is_empty() {
781        let mut in_degree: std::collections::HashMap<&str, usize> =
782            group_names.iter().map(|&n| (n, 0)).collect();
783        let mut adj: std::collections::HashMap<&str, Vec<&str>> =
784            group_names.iter().map(|&n| (n, Vec::new())).collect();
785        for dep in group_dependencies {
786            adj.entry(dep.from_group.as_str())
787                .or_default()
788                .push(dep.to_group.as_str());
789            *in_degree.entry(dep.to_group.as_str()).or_insert(0) += 1;
790        }
791        let mut queue: Vec<&str> = in_degree
792            .iter()
793            .filter(|(_, deg)| **deg == 0)
794            .map(|(n, _)| *n)
795            .collect();
796        let mut visited = 0;
797        while let Some(node) = queue.pop() {
798            visited += 1;
799            if let Some(neighbors) = adj.get(node) {
800                for &next in neighbors {
801                    if let Some(deg) = in_degree.get_mut(next) {
802                        *deg = deg.saturating_sub(1);
803                        if *deg == 0 {
804                            queue.push(next);
805                        }
806                    }
807                }
808            }
809        }
810        if visited != group_names.len() {
811            let cycle_nodes: Vec<&str> = in_degree
812                .iter()
813                .filter(|(_, deg)| **deg > 0)
814                .map(|(n, _)| *n)
815                .collect();
816            return Err(crate::error::types::SupervisorError::fatal_config(format!(
817                "Group dependency cycle detected among groups: {cycle_nodes:?}",
818            )));
819        }
820    }
821
822    for child_override in child_strategy_overrides {
823        if !child_names.contains(child_override.child_id.as_str()) {
824            return Err(crate::error::types::SupervisorError::fatal_config(format!(
825                "child_strategy_overrides references unknown child '{}'",
826                child_override.child_id
827            )));
828        }
829        validate_restart_limit_input(
830            child_override.restart_limit.as_ref(),
831            "child_strategy_overrides.restart_limit",
832        )?;
833    }
834
835    let mut roles = HashSet::new();
836    for default in severity_defaults {
837        if !roles.insert(default.task_role) {
838            return Err(crate::error::types::SupervisorError::fatal_config(format!(
839                "duplicate severity default for task role '{}'",
840                default.task_role
841            )));
842        }
843    }
844    Ok(())
845}
846
847/// Validates an optional YAML restart limit.
848///
849/// # Arguments
850///
851/// - `limit`: Optional restart limit loaded from YAML.
852/// - `path`: Field path used in error messages.
853///
854/// # Returns
855///
856/// Returns `Ok(())` when the optional restart limit is absent or usable.
857fn validate_restart_limit_input(
858    limit: Option<&crate::config::policy::RestartLimitConfig>,
859    path: &str,
860) -> Result<(), crate::error::types::SupervisorError> {
861    let Some(limit) = limit else {
862        return Ok(());
863    };
864    validate_positive(limit.max_restarts as u64, &format!("{path}.max_restarts"))?;
865    validate_positive(limit.window_ms, &format!("{path}.window_ms"))
866}
867
868/// Validates shutdown configuration invariants.
869///
870/// # Arguments
871///
872/// - `shutdown`: Shutdown configuration loaded from YAML.
873///
874/// # Returns
875///
876/// Returns `Ok(())` when shutdown values are usable.
877fn validate_shutdown(
878    shutdown: &ShutdownConfig,
879) -> Result<(), crate::error::types::SupervisorError> {
880    validate_positive(shutdown.graceful_timeout_ms, "shutdown.graceful_timeout_ms")?;
881    validate_positive(shutdown.abort_wait_ms, "shutdown.abort_wait_ms")
882}
883
884/// Validates observability configuration invariants.
885///
886/// # Arguments
887///
888/// - `observability`: Observability configuration loaded from YAML.
889///
890/// # Returns
891///
892/// Returns `Ok(())` when observability values are usable.
893fn validate_observability(
894    observability: &ObservabilityConfig,
895) -> Result<(), crate::error::types::SupervisorError> {
896    validate_positive(
897        observability.event_journal_capacity as u64,
898        "observability.event_journal_capacity",
899    )
900}
901
902/// Validates audit persistence configuration invariants.
903///
904/// # Arguments
905///
906/// - `audit`: Audit persistence configuration loaded from YAML.
907///
908/// # Returns
909///
910/// Returns `Ok(())` when audit persistence values are usable.
911fn validate_audit(audit: &AuditConfig) -> Result<(), crate::error::types::SupervisorError> {
912    match audit.backend.as_str() {
913        "memory" => {}
914        "file" => {
915            let path = audit.file_path.trim();
916            if path.is_empty() {
917                return Err(crate::error::types::SupervisorError::fatal_config(
918                    "audit.file_path is required when audit.backend is file",
919                ));
920            }
921        }
922        backend => {
923            return Err(crate::error::types::SupervisorError::fatal_config(format!(
924                "audit.backend must be memory or file, got {backend}"
925            )));
926        }
927    }
928
929    match audit.failure_strategy.as_str() {
930        "fail_closed" | "defer_bounded" => {}
931        strategy => {
932            return Err(crate::error::types::SupervisorError::fatal_config(format!(
933                "audit.failure_strategy must be fail_closed or defer_bounded, got {strategy}"
934            )));
935        }
936    }
937
938    validate_positive(audit.max_defer_queue as u64, "audit.max_defer_queue")
939}
940
941/// Validates backpressure configuration invariants.
942///
943/// # Arguments
944///
945/// - `backpressure`: Backpressure configuration loaded from YAML.
946///
947/// # Returns
948///
949/// Returns `Ok(())` when thresholds, windows, and capacities are usable.
950fn validate_backpressure(
951    backpressure: &BackpressureConfig,
952) -> Result<(), crate::error::types::SupervisorError> {
953    if backpressure.warn_threshold_pct == 0 || backpressure.warn_threshold_pct > 100 {
954        return Err(crate::error::types::SupervisorError::fatal_config(
955            "backpressure.warn_threshold_pct must be between 1 and 100",
956        ));
957    }
958    if backpressure.critical_threshold_pct == 0 || backpressure.critical_threshold_pct > 100 {
959        return Err(crate::error::types::SupervisorError::fatal_config(
960            "backpressure.critical_threshold_pct must be between 1 and 100",
961        ));
962    }
963    if backpressure.warn_threshold_pct >= backpressure.critical_threshold_pct {
964        return Err(crate::error::types::SupervisorError::fatal_config(
965            "backpressure.warn_threshold_pct must be less than backpressure.critical_threshold_pct",
966        ));
967    }
968    validate_positive(backpressure.window_secs, "backpressure.window_secs")?;
969    validate_positive(
970        backpressure.audit_channel_capacity as u64,
971        "backpressure.audit_channel_capacity",
972    )
973}
974
975/// Applies default security settings to the optional dashboard pipeline.
976///
977/// # Arguments
978///
979/// - `dashboard`: Optional target-side dashboard configuration loaded from YAML.
980///
981/// # Returns
982///
983/// Returns dashboard configuration with a security pipeline when dashboard IPC
984/// is enabled.
985fn dashboard_with_default_security(
986    mut dashboard: Option<DashboardIpcConfig>,
987) -> Option<DashboardIpcConfig> {
988    let Some(dashboard_config) = dashboard.as_mut() else {
989        return dashboard;
990    };
991    if !dashboard_config.enabled {
992        return dashboard;
993    }
994
995    dashboard_config
996        .security_config
997        .get_or_insert_with(IpcSecurityConfig::default);
998    dashboard
999}
1000
1001/// Validates dashboard IPC configuration invariants.
1002///
1003/// # Arguments
1004///
1005/// - `dashboard`: Optional target-side dashboard IPC configuration.
1006///
1007/// # Returns
1008///
1009/// Returns `Ok(())` when IPC is absent, disabled, or semantically valid.
1010/// Validates dashboard IPC configuration when the dashboard module is
1011/// compiled (Unix). On non-Unix platforms, rejects any configuration that
1012/// has `dashboard.enabled = true`.
1013#[cfg(unix)]
1014fn validate_dashboard(
1015    dashboard: Option<&DashboardIpcConfig>,
1016) -> Result<(), crate::error::types::SupervisorError> {
1017    crate::dashboard::config::validate_dashboard_ipc_config(dashboard)
1018        .map(|_| ())
1019        .map_err(|error| crate::error::types::SupervisorError::fatal_config(error.to_string()))
1020}
1021
1022/// On non-Unix platforms the dashboard module is not compiled. If the
1023/// user explicitly set `dashboard.enabled = true`, report a clear error
1024/// instead of silently ignoring the configuration.
1025#[cfg(not(unix))]
1026fn validate_dashboard(
1027    dashboard: Option<&DashboardIpcConfig>,
1028) -> Result<(), crate::error::types::SupervisorError> {
1029    if let Some(config) = dashboard {
1030        if config.enabled {
1031            return Err(crate::error::types::SupervisorError::fatal_config(
1032                "dashboard is enabled but the dashboard IPC module is only available on Unix platforms",
1033            ));
1034        }
1035    }
1036    Ok(())
1037}
1038
1039/// Validates that a runtime configuration number is positive.
1040///
1041/// # Arguments
1042///
1043/// - `value`: Runtime configuration number.
1044/// - `name`: Configuration key name.
1045///
1046/// # Returns
1047///
1048/// Returns `Ok(())` when the value is positive.
1049fn validate_positive(
1050    value: impl Into<u64>,
1051    name: &str,
1052) -> Result<(), crate::error::types::SupervisorError> {
1053    if value.into() == 0 {
1054        Err(crate::error::types::SupervisorError::fatal_config(format!(
1055            "{name} must be greater than zero"
1056        )))
1057    } else {
1058        Ok(())
1059    }
1060}
1061
1062/// Computes a deterministic hash of a ChildDeclaration content.
1063///
1064/// This is used for compensating record identification, not for
1065/// cryptographic security. The hash is always recomputed from the
1066/// declaration content after restart.
1067fn compute_declaration_hash(declaration: &ChildDeclaration) -> String {
1068    let json = serde_json::to_string(declaration).unwrap_or_default();
1069    use std::hash::{Hash, Hasher};
1070    let mut hasher = std::collections::hash_map::DefaultHasher::new();
1071    json.hash(&mut hasher);
1072    format!("v{:x}", hasher.finish())
1073}
rust_supervisor/config/state.rs

rust_supervisor/config/
state.rs
