Skip to main content

rust_supervisor/runtime/
control_loop.rs

1//! Runtime control loop.
2//!
3//! This module executes control-plane commands, receives child child_start_count exits,
4//! and applies supervisor restart strategy decisions.
5
6use crate::child_runner::run_exit::TaskExit;
7use crate::child_runner::runner::{ChildRunHandle, ChildRunReport, ChildRunner, wait_for_report};
8use crate::control::command::{CommandMeta, CommandResult, ControlCommand, CurrentState};
9use crate::control::outcome::{
10    ChildAttemptStatus, ChildControlFailure, ChildControlFailurePhase, ChildControlOperation,
11    ChildControlResult, ChildLivenessState, ChildStopState, GenerationFenceDecision,
12    GenerationFenceOutcome, GenerationFencePhase, PendingRestart, RestartLimitState,
13    StaleAttemptReport, StaleReportHandling,
14};
15use crate::error::types::SupervisorError;
16use crate::event::payload::{ProtectionAction, SupervisorEvent, ThrottleGateOwner, What, Where};
17use crate::event::time::{CorrelationId, EventSequenceSource, EventTime, When};
18use crate::id::types::{ChildId, ChildStartCount, Generation, SupervisorPath};
19use crate::observe::fairness::FairnessProbe;
20use crate::observe::pipeline::{ObservabilityPipeline, PipelineStageDiagnostic};
21use crate::policy::backoff::BackoffPolicy;
22use crate::policy::decision::{
23    PolicyEngine, RestartDecision, RestartPolicy, TaskExit as PolicyTaskExit,
24};
25use crate::policy::failure_window::FailureWindow;
26use crate::policy::meltdown::MeltdownTracker;
27use crate::policy::task_role_defaults::{EffectivePolicy, OnSuccessAction};
28use crate::readiness::signal::ReadinessState;
29use crate::registry::entry::{ChildRuntime, ChildRuntimeStatus};
30use crate::registry::store::RegistryStore;
31use crate::runtime::admission::{AdmissionConflict, AdmissionSet};
32use crate::runtime::child_slot::{
33    ChildExitSummary, ChildSlot, DEFAULT_HEARTBEAT_TIMEOUT_SECS, RuntimeTimeBase,
34};
35use crate::runtime::concurrent_gate::GatePermit;
36use crate::runtime::lifecycle::RuntimeExitReport;
37use crate::runtime::message::{ChildStartMessage, ControlPlaneMessage, RuntimeLoopMessage};
38use crate::runtime::pipeline::{ExitClassification, PipelineContext, SupervisionPipeline};
39use crate::runtime::shutdown::{reconcile_shutdown_slots, shutdown_tree_fanout};
40use crate::runtime::shutdown_pipeline::ShutdownPipeline;
41use crate::shutdown::coordinator::{ShutdownCoordinator, ShutdownResult};
42use crate::shutdown::report::{
43    ChildShutdownOutcome, ChildShutdownOutcomeInput, ChildShutdownStatus, ShutdownPipelineReport,
44    ShutdownReconcileReport,
45};
46use crate::shutdown::stage::{ShutdownCause, ShutdownPhase};
47use crate::spec::child::{ChildSpec, RestartPolicy as ChildRestartPolicy};
48use crate::spec::child_declaration::{ChildDeclaration, validate_child_declaration};
49use crate::spec::shutdown::ShutdownBudget;
50use crate::spec::supervisor::{RestartLimit, SupervisorSpec};
51use crate::task::factory_registry::TaskFactoryRegistry;
52use crate::tree::builder::SupervisorTree;
53use crate::tree::order::{restart_execution_plan, shutdown_order, startup_order};
54use std::collections::HashMap;
55use std::sync::{Arc, Mutex};
56use std::time::{Duration, SystemTime, UNIX_EPOCH};
57use tokio::sync::{broadcast, mpsc};
58use tokio::time::{Instant, timeout};
59
60/// Typed event waiting for emission after mutable state borrows end.
61#[derive(Debug)]
62struct PendingRuntimeEvent {
63    /// Child task identifier related to the event.
64    child_id: ChildId,
65    /// Child task path attached to the event location.
66    path: SupervisorPath,
67    /// Generation number attached to event timing.
68    generation: Option<Generation>,
69    /// Attempt attached to event timing.
70    attempt: Option<ChildStartCount>,
71    /// Correlation identifier attached to the event.
72    correlation_id: CorrelationId,
73    /// Typed event payload.
74    what: What,
75}
76
77/// Mutable state owned by the control loop.
78#[derive(Debug)]
79pub struct RuntimeControlState {
80    /// Shutdown state machine used by tree-level shutdown commands.
81    shutdown: ShutdownCoordinator,
82    /// Runtime-owned shutdown pipeline state and cached report.
83    shutdown_pipeline: ShutdownPipeline,
84    /// Runtime slots for declared children.
85    slots: HashMap<ChildId, ChildSlot>,
86    /// Admission set that enforces at-most-one active attempt per child.
87    #[allow(dead_code)]
88    admission_set: AdmissionSet,
89    /// Runtime time base used for public timestamps.
90    time_base: RuntimeTimeBase,
91    /// Event sequence source for typed observability facts.
92    event_sequences: EventSequenceSource,
93    /// Shared typed observability pipeline.
94    observability: Arc<Mutex<ObservabilityPipeline>>,
95    /// Six-stage supervision pipeline for failure processing.
96    supervision_pipeline: SupervisionPipeline,
97    /// Instance-global concurrent restart throttle gate (FR-003).
98    concurrent_gate: crate::runtime::concurrent_gate::SupervisorInstanceGate,
99    /// Fairness probe that detects scheduling starvation (US1).
100    fairness_probe: FairnessProbe,
101    /// Dynamic child manifests accepted after startup.
102    manifests: Vec<String>,
103    /// Registry used to resolve dynamic child task factories.
104    task_factory_registry: TaskFactoryRegistry,
105    /// Registry that owns declared child runtime records.
106    registry: RegistryStore,
107    /// Built supervisor tree used for order and scope planning.
108    tree: SupervisorTree,
109    /// Supervisor specification that owns strategy and dynamic policies.
110    spec: SupervisorSpec,
111    /// Policy engine used to convert task exits into restart decisions.
112    policy_engine: PolicyEngine,
113    /// Sender used by spawned child start_counts to report runtime messages.
114    command_sender: mpsc::Sender<RuntimeLoopMessage>,
115    /// Active count of orphaned child tasks that could not be
116    /// stopped within policy timeouts.
117    pub orphan_count: u64,
118    /// Strategy for process exit, swappable for testing.
119    exit_handler: Arc<dyn crate::exit_handler::ExitHandler>,
120    /// Active gate permits indexed by (child_id, generation, attempt).
121    /// Each permit represents one acquired concurrent restart slot;
122    /// dropping the permit releases the slot automatically (RAII).
123    /// The triple-key ensures that a late-report from an old generation
124    /// or attempt does not release the permit belonging to a newer
125    /// restart attempt of the same child.
126    active_gate_permits: HashMap<(ChildId, Generation, ChildStartCount), GatePermit>,
127}
128
129/// Builds initial [`ChildSlot`] records from the registry.
130fn build_initial_slots(
131    registry: &RegistryStore,
132    spec: &SupervisorSpec,
133) -> HashMap<ChildId, ChildSlot> {
134    registry
135        .declaration_order()
136        .iter()
137        .filter_map(|child_id| {
138            registry.child(child_id).map(|runtime| {
139                let mut slot = ChildSlot::new_placeholder(runtime.id.clone(), runtime.path.clone());
140                // Apply health_policy.stale_after and group from the child spec.
141                if let Some(child_spec) = spec.children.iter().find(|c| &c.id == child_id) {
142                    slot.stale_after = child_spec.health_policy.stale_after;
143                    slot.group = child_spec.group.clone();
144                }
145                (child_id.clone(), slot)
146            })
147        })
148        .collect::<HashMap<_, _>>()
149}
150
151#[allow(dead_code)]
152impl RuntimeControlState {
153    /// Creates control state from a supervisor specification.
154    ///
155    /// # Arguments
156    ///
157    /// - `spec`: Supervisor declaration that owns children and strategy.
158    /// - `command_sender`: Sender used by child start_counts to report exits.
159    ///
160    /// # Returns
161    ///
162    /// Returns a [`RuntimeControlState`] value.
163    pub fn new(
164        spec: SupervisorSpec,
165        command_sender: mpsc::Sender<RuntimeLoopMessage>,
166        observability: Arc<Mutex<ObservabilityPipeline>>,
167    ) -> Result<Self, SupervisorError> {
168        Self::new_with_factory_registry(
169            spec,
170            command_sender,
171            observability,
172            TaskFactoryRegistry::new(),
173        )
174    }
175
176    /// Creates control state with a task factory registry for dynamic children.
177    ///
178    /// # Arguments
179    ///
180    /// - `spec`: Supervisor declaration that owns children and strategy.
181    /// - `command_sender`: Sender used by child start_counts to report exits.
182    /// - `observability`: Shared observability pipeline.
183    /// - `task_factory_registry`: Registry used by `add_child` declarations.
184    ///
185    /// # Returns
186    ///
187    /// Returns a [`RuntimeControlState`] value.
188    pub fn new_with_factory_registry(
189        spec: SupervisorSpec,
190        command_sender: mpsc::Sender<RuntimeLoopMessage>,
191        observability: Arc<Mutex<ObservabilityPipeline>>,
192        task_factory_registry: TaskFactoryRegistry,
193    ) -> Result<Self, SupervisorError> {
194        let tree_shutdown = spec.tree_shutdown;
195        let tree = SupervisorTree::build(&spec)?;
196        let mut registry = RegistryStore::new();
197        registry.register_tree(&tree)?;
198        let time_base = RuntimeTimeBase::new();
199        let slots = build_initial_slots(&registry, &spec);
200
201        let meltdown_tracker = MeltdownTracker::new(spec.meltdown_policy);
202        let failure_window = FailureWindow::new(spec.failure_window_config);
203        let supervision_pipeline = SupervisionPipeline::with_backpressure_config(
204            spec.pipeline_journal_capacity,
205            spec.pipeline_subscriber_capacity,
206            meltdown_tracker,
207            failure_window,
208            spec.restart_budget_config.clone(),
209            spec.group_dependencies.clone(),
210            spec.backpressure_config.clone(),
211        );
212
213        let concurrent_gate = crate::runtime::concurrent_gate::SupervisorInstanceGate::new(
214            spec.concurrent_restart_limit,
215        );
216
217        // Initialize fairness probe with current timestamp
218        let now_unix_nanos = SystemTime::now()
219            .duration_since(UNIX_EPOCH)
220            .unwrap_or_default()
221            .as_nanos();
222        let fairness_probe = FairnessProbe::new(now_unix_nanos);
223
224        Ok(Self {
225            shutdown: ShutdownCoordinator::new(tree_shutdown),
226            shutdown_pipeline: ShutdownPipeline::new(),
227            slots,
228            admission_set: AdmissionSet::new(),
229            time_base,
230            event_sequences: EventSequenceSource::new(),
231            observability,
232            supervision_pipeline,
233            concurrent_gate,
234            fairness_probe,
235            manifests: Vec::new(),
236            task_factory_registry,
237            registry,
238            tree,
239            spec,
240            policy_engine: PolicyEngine::new(),
241            command_sender,
242            orphan_count: 0,
243            exit_handler: Arc::new(crate::exit_handler::DefaultExitHandler),
244            active_gate_permits: HashMap::new(),
245        })
246    }
247
248    /// Returns the shutdown budget for one child, falling back to the tree default.
249    ///
250    /// # Arguments
251    ///
252    /// - `child_id`: Stable child identifier.
253    ///
254    /// # Returns
255    ///
256    /// Returns the effective [`ShutdownBudget`] for shutdown and stop commands.
257    fn child_shutdown_budget(&self, child_id: &ChildId) -> ShutdownBudget {
258        self.spec.shutdown_budget_for(child_id)
259    }
260
261    /// Starts every declared child in supervisor startup order.
262    ///
263    /// Children are spawned in topological order (dependencies first), but
264    /// readiness waiting (a child's dependencies must be ready before it
265    /// starts) is not yet implemented — this is tracked for a future slice.
266    /// As a result, children that declare `dependencies` will start in the
267    /// correct order but dependencies must report ready before the
268    /// dependent child starts. This prevents downstream workers from
269    /// beginning execution before their required services (database,
270    /// cache, message queue) are ready to serve requests.
271    ///
272    /// # Arguments
273    ///
274    /// This function has no arguments.
275    ///
276    /// # Returns
277    ///
278    /// This function does not return a value.
279    pub async fn start_declared_children(&mut self) {
280        let child_ids = startup_order(&self.tree)
281            .into_iter()
282            .map(|node| node.child.id.clone())
283            .collect::<Vec<_>>();
284        for child_id in &child_ids {
285            // Wait for all dependencies to report ready before spawning
286            // this child. This implements true readiness gating on top of
287            // the topological startup order.
288            if let Some(deps) = self
289                .registry
290                .child(child_id)
291                .map(|r| r.spec.dependencies.clone())
292                .filter(|d| !d.is_empty())
293            {
294                self.wait_for_dependencies(&deps, child_id).await;
295            }
296            self.spawn_child_start(child_id.clone(), false, Duration::ZERO);
297        }
298    }
299
300    /// Awaits until every dependency of a child has reported readiness.
301    ///
302    /// Each dependency's readiness is observed through the shared
303    /// readiness receiver on its [`ChildSlot`]. When a dependency has no
304    /// active slot (e.g. already exited), the wait resolves immediately
305    /// to avoid blocking the startup sequence indefinitely.
306    ///
307    /// # Arguments
308    ///
309    /// - `deps`: Child identifiers this child depends on.
310    /// - `child_id`: The child that is waiting (for diagnostics).
311    ///
312    /// # Returns
313    ///
314    /// This function does not return a value.
315    async fn wait_for_dependencies(&mut self, deps: &[ChildId], child_id: &ChildId) {
316        for dep_id in deps {
317            let mut receiver = self
318                .slots
319                .get(dep_id)
320                .and_then(|slot| slot.readiness_receiver.clone());
321
322            let Some(ref mut rx) = receiver else {
323                // Dependency has no active slot — skip wait.
324                continue;
325            };
326
327            // Fast path: already ready.
328            if *rx.borrow_and_update() == ReadinessState::Ready {
329                continue;
330            }
331
332            tracing::info!(
333                child_id = %child_id,
334                dependency = %dep_id,
335                "waiting for dependency readiness",
336            );
337
338            // Slow path: wait for readiness change.
339            loop {
340                let result = rx.changed().await;
341                if result.is_err() {
342                    // Sender dropped — dependency exited or slot cleared.
343                    break;
344                }
345                if *rx.borrow() == ReadinessState::Ready {
346                    break;
347                }
348            }
349
350            tracing::info!(
351                child_id = %child_id,
352                dependency = %dep_id,
353                "dependency ready, proceeding",
354            );
355        }
356    }
357
358    /// Records an active attempt after `spawn_once` so exit routing matches registry identities.
359    ///
360    /// Immediate spawns run this inline; delayed backoff spawns deliver the same handle through
361    /// [`ChildStartMessage::DelayedSpawnAttached`] so `activate_instance` stays on the control loop.
362    ///
363    /// # Arguments
364    ///
365    /// - `child_id`: Stable child owning the spawned attempt.
366    /// - `path`: Supervisor path used when inserting placeholder runtime records.
367    /// - `generation`: Generation pinned from the registry [`ChildRuntime`] passed to `spawn_once`.
368    /// - `attempt`: Attempt counter pinned from the same registry record.
369    /// - `handle`: Runner handle carrying cancellation and completion endpoints.
370    ///
371    /// # Returns
372    ///
373    /// Sets the exit handler strategy for process termination.
374    ///
375    /// The default handler calls `std::process::exit(1)`. Tests can swap in
376    /// a stub that records the exit request without terminating the process.
377    ///
378    /// # Arguments
379    ///
380    /// - `handler`: Exit handler implementation.
381    ///
382    /// # Returns
383    ///
384    /// This function does not return a value.
385    pub fn set_exit_handler(&mut self, handler: Arc<dyn crate::exit_handler::ExitHandler>) {
386        self.exit_handler = handler;
387    }
388
389    /// Activates a spawned child handle in the slots map and spawns a
390    /// watcher that forwards the exit report back to the control loop.
391    ///
392    /// # Arguments
393    ///
394    /// - `child_id`: Stable child owning the spawned attempt.
395    /// - `path`: Supervisor path for the child.
396    /// - `generation`: Generation pinned from the registry runtime record.
397    /// - `attempt`: Attempt counter pinned from the registry runtime record.
398    /// - `handle`: Runner handle carrying cancellation and completion endpoints.
399    ///
400    /// # Returns
401    ///
402    /// This function does not return a value.
403    fn attach_spawned_child_handle(
404        &mut self,
405        child_id: ChildId,
406        path: SupervisorPath,
407        generation: Generation,
408        attempt: ChildStartCount,
409        handle: ChildRunHandle,
410    ) {
411        let mut completion_receiver = handle.completion_receiver.clone();
412        let sender = self.command_sender.clone();
413        self.slots
414            .entry(child_id.clone())
415            .or_insert_with(|| ChildSlot::new_placeholder(child_id.clone(), path))
416            .activate(generation, attempt, ChildAttemptStatus::Running, handle);
417        tokio::spawn(async move {
418            let result = wait_for_report(&mut completion_receiver).await;
419            send_child_result(sender, child_id, result).await;
420        });
421    }
422
423    /// Executes one control command.
424    ///
425    /// # Arguments
426    ///
427    /// - `command`: Command received by the runtime.
428    /// - `event_sender`: Event channel used for lifecycle text.
429    ///
430    /// # Returns
431    ///
432    /// Returns a command result.
433    pub async fn execute_control(
434        &mut self,
435        command: ControlCommand,
436        event_sender: &broadcast::Sender<String>,
437    ) -> Result<CommandResult, SupervisorError> {
438        command.validate_audit_metadata()?;
439        self.reconcile_stop_deadlines();
440        match command {
441            ControlCommand::AddChild {
442                child_manifest,
443                meta: _,
444                target: _,
445            } => {
446                self.ensure_dynamic_child_allowed()?;
447
448                // Reject add_child when shutdown is in progress.
449                if self.shutdown.phase() != ShutdownPhase::Idle {
450                    return Err(SupervisorError::fatal_config(
451                        "Cannot add child: supervisor is shutting down",
452                    ));
453                }
454
455                // Parse manifest as ChildDeclaration.
456                let declaration: ChildDeclaration =
457                    serde_yaml::from_str(&child_manifest).map_err(|e| {
458                        SupervisorError::fatal_config(format!(
459                            "Failed to parse child manifest: {e}"
460                        ))
461                    })?;
462
463                // Validate declaration against existing children.
464                let all_names: std::collections::HashSet<String> =
465                    self.spec.children.iter().map(|c| c.name.clone()).collect();
466                validate_child_declaration(&declaration, &all_names).map_err(|e| {
467                    SupervisorError::fatal_config(format!(
468                        "Child validation failed at {}: {}",
469                        e.field_path, e.reason
470                    ))
471                })?;
472
473                // Convert to ChildSpec and register in registry, slot, and spec.
474                let child_spec =
475                    crate::spec::child::ChildSpec::try_from(declaration).map_err(|e| {
476                        SupervisorError::fatal_config(format!("Child conversion failed: {e:?}"))
477                    })?;
478                let mut child_spec = child_spec;
479                crate::config::factory_binding::bind_child_factory(
480                    &mut child_spec,
481                    &self.task_factory_registry,
482                )?;
483
484                let child_id = child_spec.id.clone();
485                let path = crate::id::types::SupervisorPath::root().join(
486                    crate::id::types::ChildId::new(&child_spec.name)
487                        .value
488                        .clone(),
489                );
490
491                // Register in registry.
492                let runtime = crate::registry::entry::ChildRuntime::new(child_spec.clone(), path);
493                self.registry.register(runtime.clone()).map_err(|e| {
494                    SupervisorError::fatal_config(format!("Registry registration failed: {e}"))
495                })?;
496
497                // Create slot with default restart window.
498                let slot = crate::runtime::child_slot::ChildSlot::new(
499                    child_id.clone(),
500                    runtime.path.clone(),
501                    std::time::Duration::from_secs(60),
502                );
503                self.slots.insert(child_id.clone(), slot);
504
505                // Add to spec and manifests.
506                self.spec.children.push(child_spec);
507                self.manifests.push(child_manifest.clone());
508
509                // Spawn child start.
510                self.spawn_child_start(child_id.clone(), false, Duration::ZERO);
511
512                Ok(CommandResult::ChildAdded { child_manifest })
513            }
514            ControlCommand::RemoveChild { meta, child_id } => Ok(self.execute_stop_child_control(
515                child_id,
516                ChildControlOperation::Removed,
517                "remove_child",
518                &meta,
519                event_sender,
520            )),
521            ControlCommand::RestartChild { meta, child_id } => {
522                Ok(self.execute_restart_child_control(child_id, &meta, event_sender))
523            }
524            ControlCommand::PauseChild { meta, child_id } => Ok(self.execute_stop_child_control(
525                child_id,
526                ChildControlOperation::Paused,
527                "pause_child",
528                &meta,
529                event_sender,
530            )),
531            ControlCommand::ResumeChild { child_id, .. } => {
532                Ok(self.set_child_state(child_id, ChildControlOperation::Active))
533            }
534            ControlCommand::QuarantineChild { meta, child_id } => Ok(self
535                .execute_stop_child_control(
536                    child_id,
537                    ChildControlOperation::Quarantined,
538                    "quarantine_child",
539                    &meta,
540                    event_sender,
541                )),
542            ControlCommand::ShutdownTree { meta } => {
543                let result = self
544                    .execute_shutdown(meta.requested_by, meta.reason, event_sender)
545                    .await?;
546                Ok(CommandResult::Shutdown { result })
547            }
548            ControlCommand::CurrentState { .. } => {
549                self.reconcile_stop_deadlines();
550                Ok(CommandResult::CurrentState {
551                    state: self.build_current_state(),
552                })
553            }
554        }
555    }
556
557    /// Applies policy to a completed child child_start_count.
558    ///
559    /// # Arguments
560    ///
561    /// - `report`: Completed child child_start_count report.
562    /// - `event_sender`: Event channel used for lifecycle text.
563    ///
564    /// # Returns
565    ///
566    /// This function does not return a value.
567    pub fn handle_child_exit(
568        &mut self,
569        report: ChildRunReport,
570        event_sender: &broadcast::Sender<String>,
571    ) {
572        let child_id = report.runtime.id.clone();
573        let generation = report.runtime.generation;
574        let attempt = report.runtime.child_start_count;
575        let exit_kind = report.exit.clone();
576        let mut pending_events = Vec::new();
577        let was_active = self
578            .slots
579            .get(&child_id)
580            .is_some_and(ChildSlot::has_active_attempt);
581        let matches_pending_fence = self
582            .slots
583            .get(&child_id)
584            .and_then(|state| state.generation_fence.pending_restart.as_ref())
585            .is_some_and(|pending_restart| {
586                pending_restart.old_generation == generation
587                    && pending_restart.old_attempt == attempt
588            });
589        let matches_active_attempt = self.slots.get(&child_id).is_some_and(|state| {
590            state.has_active_attempt()
591                && state.generation == Some(generation)
592                && state.attempt == Some(attempt)
593        });
594
595        // FR-003: Release concurrent gate permit only when the exiting
596        // attempt matches the current active or pending-fence attempt.
597        // Using (child_id, generation, attempt) as the key prevents a
598        // stale late-report from an old generation from releasing the
599        // permit of a newer restart of the same child.
600        if matches_active_attempt || matches_pending_fence {
601            self.active_gate_permits
602                .remove(&(child_id.clone(), generation, attempt));
603        }
604        let manual_stop_requested = self
605            .slots
606            .get(&child_id)
607            .is_some_and(|state| state.stop_state == ChildStopState::CancelDelivered);
608        let mut stale_idle_report = false;
609        let count_restart_failure = self.slots.get(&child_id).is_some_and(|state| {
610            state.operation == ChildControlOperation::Active
611                && restart_limit_counts_exit(&exit_kind)
612        });
613        let late_report = !was_active && self.shutdown.phase() == ShutdownPhase::Completed;
614        let mut fence_pending_release = None::<PendingRestart>;
615
616        if let Some(runtime_state) = self.slots.get_mut(&child_id) {
617            if matches_pending_fence {
618                if runtime_state.stop_state == ChildStopState::CancelDelivered {
619                    runtime_state.stop_state = ChildStopState::Completed;
620                    pending_events.push(PendingRuntimeEvent {
621                        child_id: child_id.clone(),
622                        path: runtime_state.path.clone(),
623                        generation: Some(generation),
624                        attempt: Some(attempt),
625                        correlation_id: CorrelationId::from_uuid(
626                            runtime_state
627                                .generation_fence
628                                .pending_restart
629                                .as_ref()
630                                .expect("matches pending implies Some")
631                                .command_id,
632                        ),
633                        what: What::ChildControlStopCompleted {
634                            child_id: child_id.clone(),
635                            generation,
636                            attempt,
637                            exit_kind: exit_kind.clone(),
638                        },
639                    });
640                }
641                fence_pending_release = runtime_state.generation_fence.pending_restart.take();
642                if let Some(pending_release) = fence_pending_release.as_ref() {
643                    let drained_correlation_id =
644                        CorrelationId::from_uuid(pending_release.command_id);
645                    pending_events.push(PendingRuntimeEvent {
646                        child_id: child_id.clone(),
647                        path: runtime_state.path.clone(),
648                        generation: Some(generation),
649                        attempt: Some(attempt),
650                        correlation_id: drained_correlation_id,
651                        what: What::ChildRestartFencePendingDrained {
652                            child_id: child_id.clone(),
653                        },
654                    });
655                }
656                runtime_state.generation_fence.phase = GenerationFencePhase::ReadyToStart;
657                runtime_state.status = ChildAttemptStatus::Stopped;
658                runtime_state.clear_instance();
659            } else if matches_active_attempt
660                || late_report
661                || self.shutdown.phase() != ShutdownPhase::Idle
662            {
663                if runtime_state.stop_state == ChildStopState::CancelDelivered {
664                    runtime_state.stop_state = ChildStopState::Completed;
665                    pending_events.push(PendingRuntimeEvent {
666                        child_id: child_id.clone(),
667                        path: runtime_state.path.clone(),
668                        generation: Some(generation),
669                        attempt: Some(attempt),
670                        correlation_id: CorrelationId::new(),
671                        what: What::ChildControlStopCompleted {
672                            child_id: child_id.clone(),
673                            generation,
674                            attempt,
675                            exit_kind: exit_kind.clone(),
676                        },
677                    });
678                }
679                runtime_state.status = ChildAttemptStatus::Stopped;
680                runtime_state.clear_instance();
681            } else {
682                stale_idle_report = true;
683                let observed_at_unix_nanos = self.time_base.now_unix_nanos();
684                let current_generation = runtime_state.generation;
685                let current_attempt = runtime_state.attempt;
686                let stale_fact = StaleAttemptReport::new(
687                    child_id.clone(),
688                    generation,
689                    attempt,
690                    current_generation,
691                    current_attempt,
692                    exit_kind.clone(),
693                    StaleReportHandling::RecordedForAudit,
694                    observed_at_unix_nanos,
695                );
696                runtime_state.generation_fence.last_stale_report = Some(stale_fact);
697                pending_events.push(PendingRuntimeEvent {
698                    child_id: child_id.clone(),
699                    path: runtime_state.path.clone(),
700                    generation: Some(generation),
701                    attempt: Some(attempt),
702                    correlation_id: CorrelationId::new(),
703                    what: What::ChildAttemptStaleReport {
704                        child_id: child_id.clone(),
705                        reported_generation: generation,
706                        reported_attempt: attempt,
707                        current_generation,
708                        current_attempt,
709                        exit_kind: exit_kind.clone(),
710                        handled_as: StaleReportHandling::RecordedForAudit,
711                    },
712                });
713            }
714        }
715
716        if stale_idle_report {
717            let _ignored = event_sender.send(format!("child_exit:{child_id}"));
718            for event in pending_events {
719                self.emit_pending_event(event);
720            }
721            self.reconcile_stop_deadlines();
722            return;
723        }
724
725        self.record_child_exit(report);
726        let restart_limit_refreshed =
727            self.refresh_restart_limit_for_child(&child_id, count_restart_failure);
728        if let Some((path, restart_limit)) = restart_limit_refreshed.clone() {
729            pending_events.push(PendingRuntimeEvent {
730                child_id: child_id.clone(),
731                path,
732                generation: Some(generation),
733                attempt: Some(attempt),
734                correlation_id: CorrelationId::new(),
735                what: What::ChildRuntimeRestartLimitUpdated {
736                    child_id: child_id.clone(),
737                    restart_limit,
738                },
739            });
740        }
741        let _ignored = event_sender.send(format!("child_exit:{child_id}"));
742        if late_report {
743            let _ignored = event_sender.send(format!("child_shutdown_late_report:{child_id}"));
744        }
745
746        // Execute six-stage supervision pipeline for failure processing.
747        let sequence = self.event_sequences.next().value;
748        // T037: Generate a real CorrelationId to link budget→meltdown→escalation events.
749        let correlation_id_str = uuid::Uuid::new_v4().to_string();
750        let supervisor_path = self
751            .slots
752            .get(&child_id)
753            .map(|state| state.path.clone())
754            .unwrap_or_else(|| SupervisorPath::root().join(child_id.value.clone()));
755
756        let mut pipeline_ctx = PipelineContext::new(
757            child_id.clone(),
758            supervisor_path,
759            sequence,
760            correlation_id_str,
761        );
762        pipeline_ctx.exit_classification = Some(classify_exit_for_pipeline(
763            &exit_kind,
764            manual_stop_requested,
765        ));
766        pipeline_ctx.effective_policy = self
767            .registry
768            .child(&child_id)
769            .map(|runtime| prepare_effective_policy(&runtime.spec));
770
771        // Convert TaskExit to PolicyTaskExit for pipeline.
772        let policy_exit = policy_task_exit(&exit_kind);
773
774        // Execute the complete six-stage pipeline.
775        let pipeline_result = self.supervision_pipeline.execute_pipeline(
776            pipeline_ctx,
777            policy_exit,
778            &self.spec,
779            &self.tree,
780        );
781        self.record_pipeline_stage_diagnostics(&pipeline_result.stage_diagnostics);
782
783        // T019: Record scheduling opportunity for fairness probe after each child exit.
784        self.fairness_probe.record_opportunity(&child_id);
785
786        // T019: Periodically check fairness probe for scheduling starvation.
787        self.check_fairness_probe(event_sender);
788
789        // T029: Reflect group_fuse_active state when meltdown triggers group-level fuse.
790        if let Some(ref budget_eval) = pipeline_result.budget_evaluation
791            && matches!(
792                budget_eval.meltdown_outcome,
793                crate::policy::meltdown::MeltdownOutcome::GroupFuse
794            )
795            && let Some(ref group_id) = pipeline_result.group_id
796        {
797            let _ignored = event_sender.send(format!("group_fuse_active:{group_id}:{child_id}"));
798            // Mark all children in the affected group as non-restartable.
799            for (_cid, slot) in self.slots.iter_mut() {
800                if slot.group.as_deref() == Some(group_id.as_str()) {
801                    slot.last_control_failure = Some(ChildControlFailure::new(
802                        ChildControlFailurePhase::WaitCompletion,
803                        format!("group_fuse_active:{group_id}"),
804                        false,
805                    ));
806                }
807            }
808        }
809
810        if let Some(pending) = fence_pending_release {
811            for event in pending_events {
812                self.emit_pending_event(event);
813            }
814            self.spawn_pending_restart_target(child_id.clone(), pending, exit_kind.clone());
815            self.reconcile_stop_deadlines();
816            return;
817        }
818
819        if !self.should_apply_automatic_policy(&child_id) {
820            if self
821                .slots
822                .get(&child_id)
823                .is_some_and(|state| state.operation == ChildControlOperation::Removed)
824                && let Some(removed) = self.slots.remove(&child_id)
825            {
826                pending_events.push(PendingRuntimeEvent {
827                    child_id: child_id.clone(),
828                    path: removed.path.clone(),
829                    generation: Some(generation),
830                    attempt: Some(attempt),
831                    correlation_id: CorrelationId::new(),
832                    what: What::ChildRuntimeStateRemoved {
833                        child_id: child_id.clone(),
834                        path: removed.path,
835                        final_status: Some(ChildAttemptStatus::Stopped),
836                    },
837                });
838            }
839            for event in pending_events {
840                self.emit_pending_event(event);
841            }
842            self.reconcile_stop_deadlines();
843            return;
844        }
845
846        // Extract action decision from pipeline result.
847        let action_decision = pipeline_result.action_decision.as_ref();
848
849        // Map pipeline protection action to restart decision.
850        let pipeline_driven_decision = if let Some(decision) = action_decision {
851            match decision.action {
852                ProtectionAction::RestartAllowed => {
853                    if role_policy_restarts_success(&pipeline_result) {
854                        Some(RestartDecision::RestartAfter {
855                            delay: Duration::ZERO,
856                        })
857                    } else {
858                        self.restart_decision(&child_id)
859                    }
860                }
861                ProtectionAction::RestartQueued => {
862                    // Queue the restart - for now treat as no immediate restart.
863                    None
864                }
865                ProtectionAction::RestartDenied
866                | ProtectionAction::SupervisionPaused
867                | ProtectionAction::Escalated
868                | ProtectionAction::SupervisedStop => {
869                    // Do not restart - respect pipeline decision.
870                    None
871                }
872            }
873        } else {
874            // Fallback to existing policy engine if pipeline didn't produce a decision.
875            self.restart_decision(&child_id)
876        };
877
878        let Some(decision) = pipeline_driven_decision else {
879            for event in pending_events {
880                self.emit_pending_event(event);
881            }
882            self.reconcile_stop_deadlines();
883            return;
884        };
885
886        if restart_limit_refreshed
887            .as_ref()
888            .is_some_and(|(_path, restart_limit)| {
889                restart_limit.used > restart_limit.limit
890                    && matches!(decision, RestartDecision::RestartAfter { .. })
891            })
892        {
893            let _ignored = event_sender.send(format!("child_restart_limit_exhausted:{child_id}"));
894            for event in pending_events {
895                self.emit_pending_event(event);
896            }
897            self.reconcile_stop_deadlines();
898            return;
899        }
900        self.execute_restart_decision(child_id, decision, event_sender);
901        for event in pending_events {
902            self.emit_pending_event(event);
903        }
904        self.reconcile_stop_deadlines();
905    }
906
907    /// Records a failed child start.
908    ///
909    /// # Arguments
910    ///
911    /// - `child_id`: Child identifier whose child_start_count failed.
912    /// - `message`: Diagnostic error message.
913    /// - `event_sender`: Event channel used for lifecycle text.
914    ///
915    /// # Returns
916    ///
917    /// This function does not return a value.
918    pub fn handle_child_start_failed(
919        &mut self,
920        child_id: ChildId,
921        message: String,
922        event_sender: &broadcast::Sender<String>,
923    ) {
924        let _ignored = event_sender.send(format!("child_start_failed:{child_id}:{message}"));
925
926        let mut fenced_spawn_recovery = Option::<(Generation, ChildStartCount, u64)>::None;
927        let mut repaired_fenced_spawn = false;
928
929        if let Some(runtime_state) = self.slots.get_mut(&child_id)
930            && runtime_state.generation_fence.phase == GenerationFencePhase::ReadyToStart
931        {
932            fenced_spawn_recovery = runtime_state
933                .registry_identity_anchor_for_spawn_attempt
934                .take();
935            repaired_fenced_spawn = true;
936            runtime_state.generation_fence.phase = GenerationFencePhase::Open;
937            runtime_state.last_control_failure = Some(ChildControlFailure::new(
938                ChildControlFailurePhase::WaitCompletion,
939                message,
940                true,
941            ));
942        }
943
944        if repaired_fenced_spawn {
945            if let Some((generation, attempt, restart_count)) = fenced_spawn_recovery
946                && let Some(registry_runtime) = self.registry.child_mut(&child_id)
947            {
948                registry_runtime.generation = generation;
949                registry_runtime.child_start_count = attempt;
950                registry_runtime.restart_count = restart_count;
951            }
952            return;
953        }
954
955        let _result = self.set_child_state(child_id, ChildControlOperation::Quarantined);
956    }
957
958    /// Executes the real shutdown pipeline.
959    ///
960    /// # Arguments
961    ///
962    /// - `requested_by`: Actor that requested shutdown.
963    /// - `reason`: Human-readable shutdown reason.
964    /// - `event_sender`: Event channel used for lifecycle text.
965    ///
966    /// # Returns
967    ///
968    /// Returns a [`ShutdownResult`] with a completed report attached.
969    ///
970    /// This method wraps the core shutdown body in a global hard timeout
971    /// computed from `effective_global_deadline()` (graceful + abort + margin).
972    /// If the timeout fires, the shutdown state machine is forced to
973    /// `Completed` and a `shutdown_global_timeout` event is emitted.
974    ///
975    /// **Note**: Under correct operation this timeout is never reached — the
976    /// body uses `remaining_duration` for every child-level `.await`, so the
977    /// body always completes within `graceful + abort` (< `global_deadline`).
978    /// This timeout is a last-resort guard against programming errors that
979    /// would otherwise cause an infinite hang.  The timeout branch (~5 lines)
980    /// is verified by code review rather than automated test because it cannot
981    /// be reliably triggered without violating Tokio's cooperative scheduling
982    /// guarantees.  Related integration tests (
983    /// `shutdown_tree_aborts_straggler_after_timeout`) confirm that the
984    /// timeout does not fire during normal shutdown.
985    pub(crate) async fn execute_shutdown(
986        &mut self,
987        requested_by: String,
988        reason: String,
989        event_sender: &broadcast::Sender<String>,
990    ) -> Result<ShutdownResult, SupervisorError> {
991        if let Some(report) = self.shutdown_pipeline.cached_report() {
992            return Ok(self
993                .shutdown
994                .result_with_report(report.as_idempotent(), true));
995        }
996
997        let global_deadline = self.shutdown.policy.effective_global_deadline();
998        match tokio::time::timeout(
999            global_deadline,
1000            self.execute_shutdown_body(requested_by, reason, event_sender),
1001        )
1002        .await
1003        {
1004            Ok(result) => result,
1005            Err(_elapsed) => {
1006                // Global hard timeout: the shutdown body did not complete
1007                // within the deadline. Perform emergency force-kill on
1008                // every slot that still holds active handles, then mark
1009                // the shutdown state machine as Completed.
1010                let _ = event_sender.send("shutdown_global_timeout".to_owned());
1011
1012                let wait_order: Vec<ChildId> = self.slots.keys().cloned().collect();
1013                for child_id in &wait_order {
1014                    if let Some(diag) = crate::runtime::shutdown::emergency_force_kill(
1015                        &mut self.slots,
1016                        child_id,
1017                        &mut self.orphan_count,
1018                    ) {
1019                        let _ = event_sender.send(format!(
1020                            "child_orphaned:{diag}:shutdown_phase=global_timeout"
1021                        ));
1022                    }
1023                }
1024
1025                // Advance state machine to Completed. All active handles
1026                // have been orphaned so no further wait is possible.
1027                self.advance_shutdown_phase(event_sender);
1028                self.shutdown.complete();
1029
1030                // Check orphan threshold after forced cleanup.
1031                let threshold = self.shutdown.policy.effective_max_orphan_threshold() as u64;
1032                if self.orphan_count >= threshold {
1033                    let _ = event_sender.send(format!(
1034                        "fatal_orphan_overflow:count={}:threshold={}",
1035                        self.orphan_count, threshold,
1036                    ));
1037                    self.exit_handler.exit(1);
1038                } else if self.orphan_count > 0 {
1039                    let _ = event_sender.send(format!(
1040                        "orphan_count:{}:below_threshold:{}:supervisor_degraded",
1041                        self.orphan_count, threshold,
1042                    ));
1043                }
1044
1045                Err(SupervisorError::FatalConfig {
1046                    message: "shutdown global timeout exceeded".to_owned(),
1047                })
1048            }
1049        }
1050    }
1051
1052    /// Core shutdown body extracted for global timeout wrapping.
1053    ///
1054    /// Performs the four-phase shutdown pipeline and returns the shutdown
1055    /// result. After completion, checks the orphan threshold and triggers
1056    /// a controlled process exit when exceeded.
1057    async fn execute_shutdown_body(
1058        &mut self,
1059        requested_by: String,
1060        reason: String,
1061        event_sender: &broadcast::Sender<String>,
1062    ) -> Result<ShutdownResult, SupervisorError> {
1063        let cause = ShutdownCause::new(requested_by.clone(), reason.clone());
1064        let requested = self.shutdown.request_stop(cause);
1065        let started_at_unix_nanos = unix_epoch_nanos();
1066        let wait_order = self.shutdown_wait_order();
1067        let mut outcomes = HashMap::<ChildId, ChildShutdownOutcome>::new();
1068        let _ignored = event_sender.send(format!(
1069            "shutdown_phase_changed:{:?}:{:?}",
1070            ShutdownPhase::Idle,
1071            self.shutdown.phase()
1072        ));
1073        self.deliver_shutdown_cancellations(&wait_order, event_sender);
1074
1075        self.advance_shutdown_phase(event_sender);
1076        self.drain_graceful_children(&wait_order, &mut outcomes, event_sender)
1077            .await;
1078
1079        self.advance_shutdown_phase(event_sender);
1080        self.abort_remaining_children(&wait_order, &mut outcomes, event_sender)
1081            .await;
1082
1083        self.advance_shutdown_phase(event_sender);
1084        self.reconcile_shutdown_outcomes(&wait_order, &mut outcomes);
1085        // Phase 5: emergency force-kill any slot that still holds handles.
1086        for child_id in &wait_order {
1087            if let Some(diag) = crate::runtime::shutdown::emergency_force_kill(
1088                &mut self.slots,
1089                child_id,
1090                &mut self.orphan_count,
1091            ) {
1092                let _ =
1093                    event_sender.send(format!("child_orphaned:{diag}:shutdown_phase=reconcile"));
1094            }
1095        }
1096        let reconcile = ShutdownReconcileReport::core_runtime_completed();
1097
1098        let from = self.shutdown.phase();
1099        self.shutdown.complete();
1100        let _ignored = event_sender.send(format!(
1101            "shutdown_phase_changed:{from:?}:{:?}",
1102            self.shutdown.phase()
1103        ));
1104        let ordered_outcomes = wait_order
1105            .iter()
1106            .filter_map(|child_id| outcomes.remove(child_id))
1107            .collect::<Vec<_>>();
1108        let report = ShutdownPipelineReport {
1109            cause: requested.cause,
1110            started_at_unix_nanos,
1111            completed_at_unix_nanos: unix_epoch_nanos(),
1112            phase: self.shutdown.phase(),
1113            outcomes: ordered_outcomes,
1114            reconcile,
1115            idempotent: false,
1116        };
1117        let _ignored = event_sender.send(format!(
1118            "shutdown_completed:{len}",
1119            len = report.outcomes.len()
1120        ));
1121        self.shutdown_pipeline.cache_report(report.clone());
1122
1123        // Step 4: active orphan degradation detection.
1124        // After shutdown, check if orphan count exceeds the threshold.
1125        let threshold = self.shutdown.policy.effective_max_orphan_threshold() as u64;
1126        if self.orphan_count >= threshold {
1127            let _ = event_sender.send(format!(
1128                "fatal_orphan_overflow:count={}:threshold={}",
1129                self.orphan_count, threshold,
1130            ));
1131            // All healthy children have completed shutdown at this point.
1132            // A controlled process exit is the safest way to reclaim orphaned
1133            // OS threads that cannot be forcibly terminated.
1134            self.exit_handler.exit(1);
1135        } else if self.orphan_count > 0 {
1136            let _ = event_sender.send(format!(
1137                "orphan_count:{}:below_threshold:{}:supervisor_degraded",
1138                self.orphan_count, threshold,
1139            ));
1140        }
1141
1142        Ok(self.shutdown.result_with_report(report, false))
1143    }
1144
1145    /// Advances the shutdown phase and emits a phase event.
1146    ///
1147    /// # Arguments
1148    ///
1149    /// - `event_sender`: Event channel used for lifecycle text.
1150    ///
1151    /// # Returns
1152    ///
1153    /// Returns the phase after advancing.
1154    fn advance_shutdown_phase(
1155        &mut self,
1156        event_sender: &broadcast::Sender<String>,
1157    ) -> ShutdownPhase {
1158        let from = self.shutdown.phase();
1159        let to = self.shutdown.advance();
1160        let _ignored = event_sender.send(format!("shutdown_phase_changed:{from:?}:{to:?}"));
1161        to
1162    }
1163
1164    /// Returns declared children in shutdown wait order.
1165    ///
1166    /// # Arguments
1167    ///
1168    /// This function has no arguments.
1169    ///
1170    /// # Returns
1171    ///
1172    /// Returns child identifiers in shutdown order.
1173    fn shutdown_wait_order(&self) -> Vec<ChildId> {
1174        shutdown_order(&self.tree)
1175            .into_iter()
1176            .map(|node| node.child.id.clone())
1177            .collect()
1178    }
1179
1180    /// Delivers cancellation to every active child child_start_count.
1181    ///
1182    /// # Arguments
1183    ///
1184    /// - `wait_order`: Stable shutdown order for declared children.
1185    /// - `event_sender`: Event channel used for lifecycle text.
1186    ///
1187    /// # Returns
1188    ///
1189    /// This function does not return a value.
1190    fn deliver_shutdown_cancellations(
1191        &mut self,
1192        wait_order: &[ChildId],
1193        event_sender: &broadcast::Sender<String>,
1194    ) {
1195        for child_id in wait_order {
1196            let Some(runtime_state) = self.slots.get_mut(child_id) else {
1197                continue;
1198            };
1199            if runtime_state.operation == ChildControlOperation::Removed {
1200                continue;
1201            }
1202            if !runtime_state.has_active_attempt() {
1203                continue;
1204            };
1205            runtime_state.cancel();
1206            let _ignored = event_sender.send(format!(
1207                "child_shutdown_cancel_delivered:{}:{}:{}",
1208                runtime_state.child_id,
1209                runtime_state
1210                    .generation
1211                    .map_or(0, |generation| generation.value),
1212                runtime_state.attempt.map_or(0, |attempt| attempt.value)
1213            ));
1214        }
1215    }
1216
1217    /// Drains cooperative child start_counts within the graceful timeout budget.
1218    ///
1219    /// # Arguments
1220    ///
1221    /// - `wait_order`: Stable shutdown order for declared children.
1222    /// - `outcomes`: Output map for completed child outcomes.
1223    /// - `event_sender`: Event channel used for lifecycle text.
1224    ///
1225    /// # Returns
1226    ///
1227    /// This function does not return a value.
1228    async fn drain_graceful_children(
1229        &mut self,
1230        wait_order: &[ChildId],
1231        outcomes: &mut HashMap<ChildId, ChildShutdownOutcome>,
1232        event_sender: &broadcast::Sender<String>,
1233    ) {
1234        for child_id in wait_order {
1235            if outcomes.contains_key(child_id) {
1236                continue;
1237            }
1238            let budget = self.child_shutdown_budget(child_id);
1239            let deadline = Instant::now() + budget.graceful_timeout;
1240            let Some(mut runtime_state) = self.slots.remove(child_id) else {
1241                continue;
1242            };
1243            if runtime_state.operation == ChildControlOperation::Removed {
1244                outcomes.insert(
1245                    child_id.clone(),
1246                    removed_runtime_state_shutdown_outcome(
1247                        &runtime_state,
1248                        ShutdownPhase::GracefulDrain,
1249                    ),
1250                );
1251                self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1252                continue;
1253            }
1254            if !runtime_state.has_active_attempt() {
1255                self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1256                continue;
1257            };
1258            let completed = match remaining_duration(deadline) {
1259                Some(remaining) => timeout(remaining, runtime_state.wait_for_report())
1260                    .await
1261                    .ok(),
1262                None => None,
1263            };
1264            match completed {
1265                Some(Ok(report)) => {
1266                    let outcome = outcome_from_report(
1267                        &runtime_state,
1268                        &report,
1269                        ChildShutdownStatus::Graceful,
1270                        ShutdownPhase::GracefulDrain,
1271                        "child completed during graceful drain",
1272                    );
1273                    self.record_child_exit(report);
1274                    runtime_state.clear_instance();
1275                    self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1276                    let _ignored = event_sender.send(format!("child_shutdown_graceful:{child_id}"));
1277                    outcomes.insert(child_id.clone(), outcome);
1278                }
1279                Some(Err(error)) => {
1280                    outcomes.insert(
1281                        child_id.clone(),
1282                        outcome_from_error(
1283                            &runtime_state,
1284                            ChildShutdownStatus::Graceful,
1285                            ShutdownPhase::GracefulDrain,
1286                            error,
1287                        ),
1288                    );
1289                    self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1290                }
1291                None => {
1292                    self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1293                }
1294            }
1295        }
1296    }
1297
1298    /// Aborts children that did not complete during graceful drain.
1299    ///
1300    /// # Arguments
1301    ///
1302    /// - `wait_order`: Stable shutdown order for declared children.
1303    /// - `outcomes`: Output map for completed child outcomes.
1304    /// - `event_sender`: Event channel used for lifecycle text.
1305    ///
1306    /// # Returns
1307    ///
1308    /// This function does not return a value.
1309    async fn abort_remaining_children(
1310        &mut self,
1311        wait_order: &[ChildId],
1312        outcomes: &mut HashMap<ChildId, ChildShutdownOutcome>,
1313        event_sender: &broadcast::Sender<String>,
1314    ) {
1315        let policy = self.shutdown.policy;
1316        for child_id in wait_order {
1317            if outcomes.contains_key(child_id) {
1318                continue;
1319            }
1320            let Some(mut runtime_state) = self.slots.remove(child_id) else {
1321                continue;
1322            };
1323            if runtime_state.operation == ChildControlOperation::Removed {
1324                outcomes.insert(
1325                    child_id.clone(),
1326                    removed_runtime_state_shutdown_outcome(
1327                        &runtime_state,
1328                        ShutdownPhase::AbortStragglers,
1329                    ),
1330                );
1331                self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1332                continue;
1333            }
1334            if !runtime_state.has_active_attempt() {
1335                self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1336                continue;
1337            };
1338            if !policy.abort_after_timeout {
1339                let budget = self.child_shutdown_budget(child_id);
1340                self.wait_for_late_report(
1341                    child_id,
1342                    runtime_state,
1343                    budget.abort_wait,
1344                    outcomes,
1345                    event_sender,
1346                )
1347                .await;
1348                continue;
1349            }
1350            runtime_state.abort();
1351            let _ignored = event_sender.send(format!(
1352                "child_shutdown_abort_requested:{}",
1353                runtime_state.child_id
1354            ));
1355            let budget = self.child_shutdown_budget(child_id);
1356            match timeout(budget.abort_wait, runtime_state.wait_for_report()).await {
1357                Ok(Ok(report)) => {
1358                    let outcome = outcome_from_report(
1359                        &runtime_state,
1360                        &report,
1361                        ChildShutdownStatus::Aborted,
1362                        ShutdownPhase::AbortStragglers,
1363                        "child completed after abort request",
1364                    );
1365                    self.record_child_exit(report);
1366                    runtime_state.clear_instance();
1367                    self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1368                    let _ignored = event_sender.send(format!("child_shutdown_aborted:{child_id}"));
1369                    outcomes.insert(child_id.clone(), outcome);
1370                }
1371                Ok(Err(error)) => {
1372                    outcomes.insert(
1373                        child_id.clone(),
1374                        outcome_from_error(
1375                            &runtime_state,
1376                            ChildShutdownStatus::AbortFailed,
1377                            ShutdownPhase::AbortStragglers,
1378                            error,
1379                        ),
1380                    );
1381                    self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1382                }
1383                Err(_elapsed) => {
1384                    outcomes.insert(
1385                        child_id.clone(),
1386                        ChildShutdownOutcome::new(ChildShutdownOutcomeInput {
1387                            child_id: runtime_state.child_id.clone(),
1388                            path: runtime_state.path.clone(),
1389                            generation: runtime_state
1390                                .generation
1391                                .unwrap_or_else(Generation::initial),
1392                            child_start_count: runtime_state
1393                                .attempt
1394                                .unwrap_or_else(ChildStartCount::first),
1395                            status: ChildShutdownStatus::AbortFailed,
1396                            cancel_delivered: runtime_state.attempt_cancel_delivered,
1397                            exit: None,
1398                            phase: ShutdownPhase::AbortStragglers,
1399                            reason: "child did not complete after abort request".to_owned(),
1400                        }),
1401                    );
1402                    self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1403                }
1404            }
1405        }
1406    }
1407
1408    /// Waits for a late report when abort is disabled by policy.
1409    ///
1410    /// # Arguments
1411    ///
1412    /// - `child_id`: Child whose child_start_count is being reconciled.
1413    /// - `runtime_state`: Runtime state removed from runtime tracking.
1414    /// - `wait`: Late report wait budget.
1415    /// - `outcomes`: Output map for completed child outcomes.
1416    /// - `event_sender`: Event channel used for lifecycle text.
1417    ///
1418    /// # Returns
1419    ///
1420    /// This function does not return a value.
1421    async fn wait_for_late_report(
1422        &mut self,
1423        child_id: &ChildId,
1424        mut runtime_state: ChildSlot,
1425        wait: Duration,
1426        outcomes: &mut HashMap<ChildId, ChildShutdownOutcome>,
1427        event_sender: &broadcast::Sender<String>,
1428    ) {
1429        match timeout(wait, runtime_state.wait_for_report()).await {
1430            Ok(Ok(report)) => {
1431                let outcome = outcome_from_report(
1432                    &runtime_state,
1433                    &report,
1434                    ChildShutdownStatus::LateReport,
1435                    ShutdownPhase::AbortStragglers,
1436                    "child reported after graceful timeout",
1437                );
1438                self.record_child_exit(report);
1439                runtime_state.clear_instance();
1440                self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1441                let _ignored = event_sender.send(format!("child_shutdown_late_report:{child_id}"));
1442                outcomes.insert(child_id.clone(), outcome);
1443            }
1444            Ok(Err(error)) => {
1445                outcomes.insert(
1446                    child_id.clone(),
1447                    outcome_from_error(
1448                        &runtime_state,
1449                        ChildShutdownStatus::LateReport,
1450                        ShutdownPhase::AbortStragglers,
1451                        error,
1452                    ),
1453                );
1454                self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1455            }
1456            Err(_elapsed) => {
1457                outcomes.insert(
1458                    child_id.clone(),
1459                    ChildShutdownOutcome::new(ChildShutdownOutcomeInput {
1460                        child_id: runtime_state.child_id.clone(),
1461                        path: runtime_state.path.clone(),
1462                        generation: runtime_state.generation.unwrap_or_else(Generation::initial),
1463                        child_start_count: runtime_state
1464                            .attempt
1465                            .unwrap_or_else(ChildStartCount::first),
1466                        status: ChildShutdownStatus::AbortFailed,
1467                        cancel_delivered: runtime_state.attempt_cancel_delivered,
1468                        exit: None,
1469                        phase: ShutdownPhase::AbortStragglers,
1470                        reason: "abort disabled and child did not report before reconcile"
1471                            .to_owned(),
1472                    }),
1473                );
1474                self.reinsert_slot_unless_replaced(child_id.clone(), runtime_state);
1475            }
1476        }
1477    }
1478
1479    /// Reinserts a slot into the slots map unless another entry already exists
1480    /// for the same child_id (inserted by a concurrent command while the slot
1481    /// was removed during drain/abort). This prevents RestartChild or other
1482    /// control commands from being silently overwritten.
1483    ///
1484    /// # Arguments
1485    ///
1486    /// - `child_id`: Child identifier for the slot.
1487    /// - `slot`: The slot to reinsert (or discard if already replaced).
1488    ///
1489    /// # Returns
1490    ///
1491    /// This function does not return a value.
1492    fn reinsert_slot_unless_replaced(&mut self, child_id: ChildId, slot: ChildSlot) {
1493        self.slots.entry(child_id).or_insert(slot);
1494        // If another entry exists, the drain/abort outcome is discarded —
1495        // the concurrent command (e.g. RestartChild) already set a fresh
1496        // placeholder or active state that takes precedence.
1497    }
1498
1499    /// Adds already-exited outcomes for declared children with no active task.
1500    ///
1501    /// # Arguments
1502    ///
1503    /// - `wait_order`: Stable shutdown order for declared children.
1504    /// - `outcomes`: Output map for completed child outcomes.
1505    ///
1506    /// # Returns
1507    ///
1508    /// This function does not return a value.
1509    fn reconcile_shutdown_outcomes(
1510        &self,
1511        wait_order: &[ChildId],
1512        outcomes: &mut HashMap<ChildId, ChildShutdownOutcome>,
1513    ) {
1514        for child_id in wait_order {
1515            if outcomes.contains_key(child_id) {
1516                continue;
1517            }
1518            let Some(runtime) = self.registry.child(child_id) else {
1519                continue;
1520            };
1521            outcomes.insert(
1522                child_id.clone(),
1523                ChildShutdownOutcome::new(ChildShutdownOutcomeInput {
1524                    child_id: runtime.id.clone(),
1525                    path: runtime.path.clone(),
1526                    generation: runtime.generation,
1527                    child_start_count: runtime.child_start_count,
1528                    status: ChildShutdownStatus::AlreadyExited,
1529                    cancel_delivered: false,
1530                    exit: runtime.last_exit.clone(),
1531                    phase: ShutdownPhase::Reconcile,
1532                    reason: "child had no active child_start_count during shutdown".to_owned(),
1533                }),
1534            );
1535        }
1536    }
1537
1538    /// Sets a child state and reports whether the operation was idempotent.
1539    ///
1540    /// # Arguments
1541    ///
1542    /// - `child_id`: Target child identifier.
1543    /// - `next`: Requested managed child state.
1544    ///
1545    /// # Returns
1546    ///
1547    /// Returns a [`CommandResult::ChildControl`] value.
1548    fn set_child_state(
1549        &mut self,
1550        child_id: ChildId,
1551        operation: ChildControlOperation,
1552    ) -> CommandResult {
1553        if !self.slots.contains_key(&child_id) {
1554            let placeholder = self
1555                .registry
1556                .child(&child_id)
1557                .map(|runtime| ChildSlot::new_placeholder(runtime.id.clone(), runtime.path.clone()))
1558                .unwrap_or_else(|| {
1559                    ChildSlot::new_placeholder(
1560                        child_id.clone(),
1561                        crate::id::types::SupervisorPath::root().join(child_id.value.clone()),
1562                    )
1563                });
1564            self.slots.insert(child_id.clone(), placeholder);
1565        }
1566        let runtime_state = self
1567            .slots
1568            .get_mut(&child_id)
1569            .expect("child runtime state should exist after insertion");
1570        let operation_before = runtime_state.operation;
1571        runtime_state.operation = operation;
1572        let outcome = ChildControlResult::new(
1573            child_id,
1574            runtime_state.attempt,
1575            runtime_state.generation,
1576            operation_before,
1577            runtime_state.operation,
1578            Some(runtime_state.status),
1579            false,
1580            if runtime_state.has_active_attempt() {
1581                runtime_state.stop_state
1582            } else {
1583                ChildStopState::NoActiveAttempt
1584            },
1585            runtime_state.restart_limit.clone(),
1586            runtime_state.observe_liveness(self.time_base.now_unix_nanos(), &self.time_base),
1587            operation_before == operation,
1588            runtime_state.last_control_failure.clone(),
1589            None,
1590        );
1591        CommandResult::ChildControl { outcome }
1592    }
1593
1594    /// Executes a stop-style child control command.
1595    ///
1596    /// # Arguments
1597    ///
1598    /// - `child_id`: Target child identifier.
1599    /// - `target_operation`: Operation requested by the command.
1600    /// - `command_name`: Stable command name used in lifecycle text.
1601    /// - `meta`: Audit metadata attached to the command.
1602    /// - `event_sender`: Event channel used for lifecycle text.
1603    ///
1604    /// # Returns
1605    ///
1606    /// Returns a [`CommandResult::ChildControl`] value.
1607    fn execute_stop_child_control(
1608        &mut self,
1609        child_id: ChildId,
1610        target_operation: ChildControlOperation,
1611        command_name: &'static str,
1612        meta: &CommandMeta,
1613        event_sender: &broadcast::Sender<String>,
1614    ) -> CommandResult {
1615        // Unknown child: return an explicit unknown outcome instead of
1616        // creating a placeholder slot. This ensures pause/remove/quarantine
1617        // on non-existent children do not silently produce state changes.
1618        if !self.slots.contains_key(&child_id) && self.registry.child(&child_id).is_none() {
1619            return CommandResult::ChildControl {
1620                outcome: ChildControlResult {
1621                    child_id: child_id.clone(),
1622                    operation_before: ChildControlOperation::Active,
1623                    operation_after: ChildControlOperation::Active,
1624                    cancel_delivered: false,
1625                    idempotent: true,
1626                    attempt: None,
1627                    generation: None,
1628                    stop_state: ChildStopState::Idle,
1629                    restart_limit: RestartLimitState::default(),
1630                    liveness: ChildLivenessState::new(
1631                        None,
1632                        false,
1633                        crate::readiness::signal::ReadinessState::Unreported,
1634                    ),
1635                    failure: None,
1636                    generation_fence: None,
1637                    status: None,
1638                    admission_conflict: None,
1639                },
1640            };
1641        }
1642
1643        if !self.slots.contains_key(&child_id) {
1644            let placeholder = self
1645                .registry
1646                .child(&child_id)
1647                .map(|runtime| ChildSlot::new_placeholder(runtime.id.clone(), runtime.path.clone()))
1648                .unwrap_or_else(|| {
1649                    ChildSlot::new_placeholder(
1650                        child_id.clone(),
1651                        crate::id::types::SupervisorPath::root().join(child_id.value.clone()),
1652                    )
1653                });
1654            self.slots.insert(child_id.clone(), placeholder);
1655        }
1656
1657        let remove_after_outcome;
1658        let correlation_id = CorrelationId::from_uuid(meta.command_id.value);
1659        let mut pending_events = Vec::new();
1660        let graceful_stop_budget = self.child_shutdown_budget(&child_id).graceful_timeout;
1661        let outcome = {
1662            let runtime_state = self
1663                .slots
1664                .get_mut(&child_id)
1665                .expect("child runtime state should exist after insertion");
1666            let stop = apply_stop_control_to_runtime_state(
1667                runtime_state,
1668                target_operation,
1669                command_name,
1670                &meta.command_id.value.to_string(),
1671                correlation_id,
1672                self.time_base
1673                    .now_unix_nanos()
1674                    .saturating_add(graceful_stop_budget.as_nanos()),
1675                event_sender,
1676                &mut pending_events,
1677            );
1678            remove_after_outcome = stop.remove_after_outcome;
1679            build_child_control_outcome(
1680                stop.operation_before,
1681                stop.cancel_delivered,
1682                stop.idempotent,
1683                runtime_state.last_control_failure.clone(),
1684                runtime_state,
1685                &self.time_base,
1686                None,
1687            )
1688        };
1689
1690        for event in pending_events {
1691            self.emit_pending_event(event);
1692        }
1693
1694        let outcome_path = self
1695            .slots
1696            .get(&child_id)
1697            .map(|state| state.path.clone())
1698            .or_else(|| {
1699                self.registry
1700                    .child(&child_id)
1701                    .map(|runtime| runtime.path.clone())
1702            })
1703            .unwrap_or_else(|| SupervisorPath::root().join(child_id.value.clone()));
1704        self.emit_pending_event(PendingRuntimeEvent {
1705            child_id: child_id.clone(),
1706            path: outcome_path,
1707            generation: outcome.generation,
1708            attempt: outcome.attempt,
1709            correlation_id,
1710            what: What::ChildControlCommandCompleted {
1711                child_id: child_id.clone(),
1712                command: command_name.to_owned(),
1713                command_id: meta.command_id.value.to_string(),
1714                requested_by: meta.requested_by.clone(),
1715                reason: meta.reason.clone(),
1716                result: child_control_result_label(&outcome).to_owned(),
1717                outcome: Box::new(outcome.clone()),
1718            },
1719        });
1720
1721        if remove_after_outcome {
1722            if let Some(removed) = self.slots.remove(&child_id) {
1723                self.emit_pending_event(PendingRuntimeEvent {
1724                    child_id: child_id.clone(),
1725                    path: removed.path.clone(),
1726                    generation: removed.generation,
1727                    attempt: removed.attempt,
1728                    correlation_id,
1729                    what: What::ChildRuntimeStateRemoved {
1730                        child_id: child_id.clone(),
1731                        path: removed.path,
1732                        final_status: None,
1733                    },
1734                });
1735            }
1736            let _ignored = event_sender.send(format!("child_runtime_state_removed:{child_id}"));
1737        }
1738
1739        CommandResult::ChildControl { outcome }
1740    }
1741
1742    /// Records the completed child_start_count in the registry.
1743    ///
1744    /// # Arguments
1745    ///
1746    /// - `report`: Completed child child_start_count report.
1747    ///
1748    /// # Returns
1749    ///
1750    /// This function does not return a value.
1751    fn record_child_exit(&mut self, report: ChildRunReport) {
1752        let child_id = report.runtime.id.clone();
1753        if let Some(runtime) = self.registry.child_mut(&child_id) {
1754            runtime.last_exit = Some(report.exit);
1755            runtime.status = ChildRuntimeStatus::Exited;
1756            runtime.generation = report.runtime.generation;
1757            runtime.child_start_count = report.runtime.child_start_count;
1758            runtime.restart_count = report.runtime.restart_count;
1759        }
1760    }
1761
1762    /// Reports whether automatic policy may still act on a child.
1763    ///
1764    /// # Arguments
1765    ///
1766    /// - `child_id`: Child whose latest exit is being evaluated.
1767    ///
1768    /// # Returns
1769    ///
1770    /// Returns `true` when the runtime may restart the child.
1771    fn should_apply_automatic_policy(&self, child_id: &ChildId) -> bool {
1772        if self.shutdown.phase() != crate::shutdown::stage::ShutdownPhase::Idle {
1773            return false;
1774        }
1775        !matches!(
1776            self.slots.get(child_id).map(|state| state.operation),
1777            Some(
1778                ChildControlOperation::Paused
1779                    | ChildControlOperation::Quarantined
1780                    | ChildControlOperation::Removed
1781            )
1782        )
1783    }
1784
1785    /// Calculates a restart decision for the latest child exit.
1786    ///
1787    /// # Arguments
1788    ///
1789    /// - `child_id`: Child whose latest exit is being evaluated.
1790    ///
1791    /// # Returns
1792    ///
1793    /// Returns a restart decision when the child is known.
1794    fn restart_decision(&self, child_id: &ChildId) -> Option<RestartDecision> {
1795        let runtime = self.registry.child(child_id)?;
1796        let exit = runtime.last_exit.as_ref()?;
1797        let policy_exit = policy_task_exit(exit);
1798        let restart_policy = restart_policy(runtime.spec.restart_policy);
1799        let backoff = backoff_policy(runtime.spec.backoff_policy);
1800        Some(self.policy_engine.decide(
1801            restart_policy,
1802            policy_exit,
1803            runtime.child_start_count.value,
1804            &backoff,
1805        ))
1806    }
1807
1808    /// Refreshes restart limit state for one child after an exit.
1809    ///
1810    /// # Arguments
1811    ///
1812    /// - `child_id`: Child whose accounting should be refreshed.
1813    /// - `count_failure`: Whether this exit consumes the restart limit.
1814    ///
1815    /// # Returns
1816    ///
1817    /// Returns the child path and updated restart limit state when the child is tracked.
1818    fn refresh_restart_limit_for_child(
1819        &mut self,
1820        child_id: &ChildId,
1821        count_failure: bool,
1822    ) -> Option<(SupervisorPath, crate::control::outcome::RestartLimitState)> {
1823        let restart_limit = restart_limit_for_child_in_spec(&self.tree, &self.spec, child_id);
1824        let runtime_state = self.slots.get_mut(child_id)?;
1825        let updated = runtime_state.refresh_restart_limit(
1826            restart_limit.window,
1827            restart_limit.max_restarts,
1828            count_failure,
1829            &self.time_base,
1830        );
1831        Some((runtime_state.path.clone(), updated))
1832    }
1833
1834    /// Executes a restart decision after a child exit.
1835    ///
1836    /// # Arguments
1837    ///
1838    /// - `failed_child`: Child whose exit triggered the decision.
1839    /// - `decision`: Restart decision returned by the policy engine.
1840    /// - `event_sender`: Event channel used for lifecycle text.
1841    ///
1842    /// # Returns
1843    ///
1844    /// This function does not return a value.
1845    fn execute_restart_decision(
1846        &mut self,
1847        failed_child: ChildId,
1848        decision: RestartDecision,
1849        event_sender: &broadcast::Sender<String>,
1850    ) {
1851        match decision {
1852            RestartDecision::RestartAfter { delay } => {
1853                self.restart_strategy_scope(failed_child, delay, event_sender);
1854            }
1855            RestartDecision::Quarantine => {
1856                let _result =
1857                    self.set_child_state(failed_child, ChildControlOperation::Quarantined);
1858            }
1859            RestartDecision::ShutdownTree => {
1860                let cause = ShutdownCause::new("runtime", "policy requested tree shutdown");
1861                let _result = self.shutdown.request_stop(cause);
1862            }
1863            RestartDecision::EscalateToParent | RestartDecision::DoNotRestart => {}
1864        }
1865    }
1866
1867    /// Restarts every child selected by the current execution plan.
1868    ///
1869    /// # Arguments
1870    ///
1871    /// - `failed_child`: Child whose exit triggered the restart scope.
1872    /// - `delay`: Delay before every selected child is restarted.
1873    /// - `event_sender`: Event channel used for lifecycle text.
1874    ///
1875    /// # Returns
1876    ///
1877    /// This function does not return a value.
1878    fn restart_strategy_scope(
1879        &mut self,
1880        failed_child: ChildId,
1881        delay: Duration,
1882        event_sender: &broadcast::Sender<String>,
1883    ) {
1884        let plan = restart_execution_plan(&self.tree, &self.spec, &failed_child);
1885        let scope_label = child_scope_label(&plan.scope);
1886        let group_label = plan.group.as_deref().unwrap_or("supervisor");
1887
1888        let _ignored = event_sender.send(format!(
1889            "restart_plan:{:?}:{group_label}:{scope_label}",
1890            plan.strategy
1891        ));
1892
1893        // Acquire one concurrent gate slot per child in the restart scope.
1894        // OneForAll may restart many children at once — each needs its own
1895        // permit so that per-child release (via GatePermit::drop) does not
1896        // drive the gate counter negative.
1897        for child_id in &plan.scope {
1898            // FR-003: Check concurrent restart gate before each spawn.
1899            let Some(permit) = self.concurrent_gate.try_acquire() else {
1900                // Gate saturated — emit throttle event and skip remaining.
1901                let _ignored = event_sender.send(format!(
1902                    "restart_throttled:concurrent_gate_saturated:{group_label}:{scope_label}:{child_id}",
1903                ));
1904                self.emit_throttle_gate_event(
1905                    child_id,
1906                    plan.group.as_deref(),
1907                    ThrottleGateOwner::SupervisorInstance,
1908                );
1909                continue;
1910            };
1911            // Store the permit indexed by (child_id, generation, attempt).
1912            // Using the triple-key ensures handle_child_exit can only
1913            // release a permit that matches the exact generation and
1914            // attempt that acquired it, preventing stale late-reports
1915            // from releasing permits belonging to newer restart attempts.
1916            self.spawn_child_start(child_id.clone(), true, delay);
1917            // Retrieve the generation and attempt assigned by the spawn
1918            // (set inside prepare_child_start + attach_spawned_child_handle).
1919            if let Some(slot) = self.slots.get(child_id)
1920                && let (Some(gen_val), Some(att_val)) = (slot.generation, slot.attempt)
1921            {
1922                self.active_gate_permits
1923                    .insert((child_id.clone(), gen_val, att_val), permit);
1924            }
1925        }
1926    }
1927
1928    /// Emits a typed event for a restart throttle gate hit.
1929    ///
1930    /// # Arguments
1931    ///
1932    /// - `child_id`: Child whose restart was throttled.
1933    /// - `group_id`: Optional restart execution group.
1934    /// - `owner`: Gate owner that limited the restart.
1935    ///
1936    /// # Returns
1937    ///
1938    /// This function does not return a value.
1939    fn emit_throttle_gate_event(
1940        &mut self,
1941        child_id: &ChildId,
1942        group_id: Option<&str>,
1943        owner: ThrottleGateOwner,
1944    ) {
1945        let now = Instant::now();
1946        let uptime = now
1947            .duration_since(self.time_base.base_instant)
1948            .as_millis()
1949            .min(u128::from(u64::MAX)) as u64;
1950        let monotonic_nanos = now.duration_since(self.time_base.base_instant).as_nanos();
1951        let path = self
1952            .slots
1953            .get(child_id)
1954            .map(|state| state.path.clone())
1955            .unwrap_or_else(|| SupervisorPath::root().join(child_id.value.clone()));
1956        let child_name = self
1957            .registry
1958            .child(child_id)
1959            .map(|runtime| runtime.spec.name.clone())
1960            .unwrap_or_else(|| child_id.to_string());
1961        let mut event = SupervisorEvent::new(
1962            When::new(EventTime::from_parts(
1963                monotonic_nanos,
1964                uptime,
1965                Generation::initial(),
1966                ChildStartCount::first(),
1967            )),
1968            Where::new(path).with_child(child_id.clone(), child_name),
1969            What::ChildFailed {
1970                failure: crate::error::types::TaskFailure::new(
1971                    crate::error::types::TaskFailureKind::Error,
1972                    "restart_throttled",
1973                    format!(
1974                        "restart denied by throttle gate {} for group {}",
1975                        owner,
1976                        group_id.unwrap_or("supervisor")
1977                    ),
1978                ),
1979            },
1980            self.event_sequences.next(),
1981            CorrelationId::new(),
1982            1,
1983        );
1984        event.effective_protective_action = Some(ProtectionAction::RestartDenied);
1985        event.throttle_gate_owner = owner;
1986        if let Some(runtime) = self.registry.child(child_id) {
1987            let effective_policy = prepare_effective_policy(&runtime.spec);
1988            event.task_role = Some(effective_policy.task_role);
1989            event.used_fallback_default = effective_policy.used_fallback;
1990            event.effective_policy_source = Some(effective_policy.source);
1991        }
1992        if let Ok(mut observability) = self.observability.lock() {
1993            let _lagged = observability.emit(event);
1994        }
1995    }
1996
1997    /// Records six-stage pipeline diagnostics in shared observability.
1998    ///
1999    /// # Arguments
2000    ///
2001    /// - `diagnostics`: Diagnostics produced by the supervision pipeline.
2002    ///
2003    /// # Returns
2004    ///
2005    /// This function does not return a value.
2006    fn record_pipeline_stage_diagnostics(&self, diagnostics: &[PipelineStageDiagnostic]) {
2007        if let Ok(mut observability) = self.observability.lock() {
2008            observability.record_pipeline_stage_diagnostics(diagnostics);
2009        }
2010    }
2011
2012    /// Checks the fairness probe and emits starvation alerts (T019).
2013    ///
2014    /// # Arguments
2015    ///
2016    /// - `event_sender`: Event channel used for lifecycle text.
2017    ///
2018    /// # Returns
2019    ///
2020    /// This function does not return a value.
2021    fn check_fairness_probe(&mut self, event_sender: &broadcast::Sender<String>) {
2022        let now_unix_nanos = self.time_base.now_unix_nanos();
2023        let all_child_ids: Vec<ChildId> = self.slots.keys().cloned().collect();
2024        if let Some(alert) = self.fairness_probe.check(now_unix_nanos, &all_child_ids) {
2025            // Emit typed event for structured observability (ALIGN-003).
2026            let path = self
2027                .slots
2028                .get(&alert.starved_child_id)
2029                .map(|slot| slot.path.clone())
2030                .unwrap_or_else(|| {
2031                    SupervisorPath::root().join(alert.starved_child_id.value.clone())
2032                });
2033            let generation = self
2034                .slots
2035                .get(&alert.starved_child_id)
2036                .and_then(|slot| slot.generation);
2037            let attempt = self
2038                .slots
2039                .get(&alert.starved_child_id)
2040                .and_then(|slot| slot.attempt);
2041            let pending = PendingRuntimeEvent {
2042                child_id: alert.starved_child_id.clone(),
2043                path,
2044                generation,
2045                attempt,
2046                correlation_id: CorrelationId::new(),
2047                what: What::FairnessProbeStarvation {
2048                    starved_child_id: alert.starved_child_id.clone(),
2049                    skip_count: alert.skip_count,
2050                    probe_start_unix_nanos: alert.probe_start_unix_nanos,
2051                    probe_end_unix_nanos: alert.probe_end_unix_nanos,
2052                },
2053            };
2054            self.emit_pending_event(pending);
2055
2056            // Text-based log for audit trail.
2057            let _ignored = event_sender.send(format!(
2058                "fairness_starvation:{}:skip_count={}:window_start={}:window_end={}",
2059                alert.starved_child_id,
2060                alert.skip_count,
2061                alert.probe_start_unix_nanos,
2062                alert.probe_end_unix_nanos,
2063            ));
2064        }
2065    }
2066
2067    /// Ensures that the dynamic supervisor accepts another child manifest.
2068    ///
2069    /// # Arguments
2070    ///
2071    /// This function has no arguments.
2072    ///
2073    /// # Returns
2074    ///
2075    /// Returns `Ok(())` when another dynamic child can be added.
2076    fn ensure_dynamic_child_allowed(&self) -> Result<(), SupervisorError> {
2077        let current_child_count = self.dynamic_child_count();
2078        if self
2079            .spec
2080            .dynamic_supervisor_policy
2081            .allows_addition(current_child_count)
2082        {
2083            return Ok(());
2084        }
2085        Err(SupervisorError::InvalidTransition {
2086            message: "dynamic supervisor child limit reached".to_owned(),
2087        })
2088    }
2089
2090    /// Counts currently registered declared and dynamic child records.
2091    ///
2092    /// # Arguments
2093    ///
2094    /// This function has no arguments.
2095    ///
2096    /// # Returns
2097    ///
2098    /// Returns the number of registered child records.
2099    fn dynamic_child_count(&self) -> usize {
2100        self.registry.declaration_order().len()
2101    }
2102
2103    /// Handles `RestartChild` with generation fencing semantics.
2104    ///
2105    /// # Arguments
2106    ///
2107    /// - `child_id`: Stable child targeted by restart.
2108    /// - `meta`: Audit metadata forwarded from the caller.
2109    /// - `event_sender`: Lifecycle text broadcaster.
2110    ///
2111    /// # Returns
2112    ///
2113    /// Returns a structured [`CommandResult`] that always uses [`CommandResult::ChildControl`].
2114    fn execute_restart_child_control(
2115        &mut self,
2116        child_id: ChildId,
2117        meta: &CommandMeta,
2118        event_sender: &broadcast::Sender<String>,
2119    ) -> CommandResult {
2120        let correlation_id = CorrelationId::from_uuid(meta.command_id.value);
2121
2122        if self.registry.child(&child_id).is_none() {
2123            let outcome = restart_child_unknown_outcome(child_id.clone());
2124            self.emit_restart_child_completed(
2125                outcome.clone(),
2126                meta,
2127                correlation_id,
2128                event_sender,
2129                Vec::new(),
2130            );
2131            return CommandResult::ChildControl { outcome };
2132        }
2133
2134        if self.shutdown.phase() != ShutdownPhase::Idle {
2135            return self.restart_child_blocked_by_shutdown(
2136                &child_id,
2137                meta,
2138                correlation_id,
2139                event_sender,
2140            );
2141        }
2142
2143        if !self.slots.contains_key(&child_id) {
2144            let placeholder = self
2145                .registry
2146                .child(&child_id)
2147                .map(|runtime| ChildSlot::new_placeholder(runtime.id.clone(), runtime.path.clone()))
2148                .unwrap_or_else(|| {
2149                    ChildSlot::new_placeholder(
2150                        child_id.clone(),
2151                        SupervisorPath::root().join(child_id.value.clone()),
2152                    )
2153                });
2154            self.slots.insert(child_id.clone(), placeholder);
2155        }
2156
2157        let mut pending_events = Vec::new();
2158
2159        // Records which restart branch matched before optional immediate spawn bookkeeping.
2160        enum RestartPrep {
2161            // Outcome resolved without visiting `spawn_child_start`.
2162            Completed(Box<ChildControlResult>),
2163            // Child had no activity and should restart immediately via the shared spawn helper.
2164            DeferredImmediate {
2165                // Operation captured before spawning.
2166                operation_before: ChildControlOperation,
2167            },
2168        }
2169
2170        let restart_prep = {
2171            let graceful_stop_budget = self.child_shutdown_budget(&child_id).graceful_timeout;
2172            let runtime_state = self
2173                .slots
2174                .get_mut(&child_id)
2175                .expect("runtime state exists after insertion");
2176            if runtime_state.generation_fence.pending_restart.is_some() {
2177                let pending = runtime_state
2178                    .generation_fence
2179                    .pending_restart
2180                    .as_mut()
2181                    .expect("checked pending restart");
2182                pending.duplicate_request_count = pending.duplicate_request_count.saturating_add(1);
2183                let pending_for_conflict = pending.clone();
2184                pending_events.push(PendingRuntimeEvent {
2185                    child_id: child_id.clone(),
2186                    path: runtime_state.path.clone(),
2187                    generation: Some(pending_for_conflict.old_generation),
2188                    attempt: Some(pending_for_conflict.old_attempt),
2189                    correlation_id: CorrelationId::from_uuid(meta.command_id.value),
2190                    what: What::ChildRestartConflict {
2191                        child_id: child_id.clone(),
2192                        current_generation: Some(pending_for_conflict.old_generation),
2193                        current_attempt: Some(pending_for_conflict.old_attempt),
2194                        target_generation: Some(pending_for_conflict.target_generation),
2195                        command_id: meta.command_id.value.to_string(),
2196                        decision: "already_pending".to_owned(),
2197                        reason: "duplicate restart merged into pending restart".to_owned(),
2198                    },
2199                });
2200                let fence = GenerationFenceOutcome::new(
2201                    GenerationFenceDecision::AlreadyPending,
2202                    Some(pending_for_conflict.old_generation),
2203                    Some(pending_for_conflict.old_attempt),
2204                    Some(pending_for_conflict.target_generation),
2205                    false,
2206                    pending_for_conflict.abort_requested,
2207                    None,
2208                );
2209                let operation_before = runtime_state.operation;
2210                RestartPrep::Completed(Box::new(build_child_control_outcome(
2211                    operation_before,
2212                    false,
2213                    false,
2214                    runtime_state.last_control_failure.clone(),
2215                    runtime_state,
2216                    &self.time_base,
2217                    Some(fence),
2218                )))
2219            } else if !runtime_state.has_active_attempt() {
2220                RestartPrep::DeferredImmediate {
2221                    operation_before: runtime_state.operation,
2222                }
2223            } else {
2224                let old_generation = runtime_state
2225                    .generation
2226                    .expect("active attempt owns a generation");
2227                let old_attempt = runtime_state
2228                    .attempt
2229                    .expect("active attempt owns an attempt counter");
2230                let cancel_delivered = runtime_state.cancel();
2231                let deadline = self
2232                    .time_base
2233                    .now_unix_nanos()
2234                    .saturating_add(graceful_stop_budget.as_nanos());
2235                runtime_state.stop_deadline_at_unix_nanos = Some(deadline);
2236                if cancel_delivered {
2237                    runtime_state.stop_state = ChildStopState::CancelDelivered;
2238                }
2239                let target_generation = old_generation.next();
2240                let requested_at = self.time_base.now_unix_nanos();
2241                let pending = PendingRestart::new(
2242                    meta.command_id.value,
2243                    meta.requested_by.clone(),
2244                    meta.reason.clone(),
2245                    old_generation,
2246                    old_attempt,
2247                    target_generation,
2248                    requested_at,
2249                    deadline,
2250                    false,
2251                    0,
2252                );
2253                runtime_state.generation_fence.pending_restart = Some(pending.clone());
2254                runtime_state.generation_fence.phase = GenerationFencePhase::WaitingForOldStop;
2255
2256                if cancel_delivered {
2257                    pending_events.push(PendingRuntimeEvent {
2258                        child_id: child_id.clone(),
2259                        path: runtime_state.path.clone(),
2260                        generation: Some(old_generation),
2261                        attempt: Some(old_attempt),
2262                        correlation_id,
2263                        what: What::ChildControlCancelDelivered {
2264                            child_id: child_id.clone(),
2265                            generation: old_generation,
2266                            attempt: old_attempt,
2267                            command: "restart_child".to_owned(),
2268                            command_id: meta.command_id.value.to_string(),
2269                        },
2270                    });
2271                    let _ignored = event_sender.send(format!(
2272                        "child_control_cancel_delivered:{child_id}:restart_child"
2273                    ));
2274                }
2275
2276                pending_events.push(PendingRuntimeEvent {
2277                    child_id: child_id.clone(),
2278                    path: runtime_state.path.clone(),
2279                    generation: Some(old_generation),
2280                    attempt: Some(old_attempt),
2281                    correlation_id,
2282                    what: What::ChildRestartFenceEntered {
2283                        child_id: child_id.clone(),
2284                        old_generation,
2285                        old_attempt,
2286                        target_generation,
2287                        command_id: meta.command_id.value.to_string(),
2288                        requested_by: meta.requested_by.clone(),
2289                        reason: meta.reason.clone(),
2290                        stop_deadline_at_unix_nanos: deadline,
2291                    },
2292                });
2293
2294                let operation_before = runtime_state.operation;
2295                let fence = GenerationFenceOutcome::new(
2296                    GenerationFenceDecision::QueuedAfterStop,
2297                    Some(old_generation),
2298                    Some(old_attempt),
2299                    Some(target_generation),
2300                    cancel_delivered,
2301                    false,
2302                    None,
2303                );
2304                RestartPrep::Completed(Box::new(build_child_control_outcome(
2305                    operation_before,
2306                    cancel_delivered,
2307                    false,
2308                    None,
2309                    runtime_state,
2310                    &self.time_base,
2311                    Some(fence),
2312                )))
2313            }
2314        };
2315
2316        let outcome = match restart_prep {
2317            RestartPrep::Completed(outcome) => *outcome,
2318            RestartPrep::DeferredImmediate { operation_before } => {
2319                self.spawn_child_start(child_id.clone(), true, Duration::ZERO);
2320                let runtime_state = self.slots.get_mut(&child_id).expect("runtime state exists");
2321                let target_generation = self
2322                    .registry
2323                    .child(&child_id)
2324                    .map(|runtime| runtime.generation);
2325                let fence = GenerationFenceOutcome::new(
2326                    GenerationFenceDecision::StartedImmediately,
2327                    None,
2328                    None,
2329                    target_generation,
2330                    false,
2331                    false,
2332                    None,
2333                );
2334                build_child_control_outcome(
2335                    operation_before,
2336                    false,
2337                    false,
2338                    runtime_state.last_control_failure.clone(),
2339                    runtime_state,
2340                    &self.time_base,
2341                    Some(fence),
2342                )
2343            }
2344        };
2345
2346        self.emit_restart_child_completed(
2347            outcome.clone(),
2348            meta,
2349            correlation_id,
2350            event_sender,
2351            pending_events,
2352        );
2353
2354        CommandResult::ChildControl { outcome }
2355    }
2356
2357    /// Emits [`What::ChildControlCommandCompleted`] for an explicit restart command.
2358    ///
2359    /// # Arguments
2360    ///
2361    /// - `outcome`: Command outcome returned to the caller.
2362    /// - `meta`: Audit metadata carried from the command.
2363    /// - `correlation_id`: Correlation shared with related fence events.
2364    /// - `event_sender`: Legacy text broadcaster.
2365    /// - `pending_events`: Fence or cancellation events that must publish first.
2366    ///
2367    /// # Returns
2368    ///
2369    /// This function does not return a value.
2370    fn emit_restart_child_completed(
2371        &mut self,
2372        outcome: ChildControlResult,
2373        meta: &CommandMeta,
2374        correlation_id: CorrelationId,
2375        event_sender: &broadcast::Sender<String>,
2376        mut pending_events: Vec<PendingRuntimeEvent>,
2377    ) {
2378        for event in pending_events.drain(..) {
2379            self.emit_pending_event(event);
2380        }
2381        let outcome_identifier = outcome.child_id.clone();
2382        let outcome_path = self
2383            .slots
2384            .get(&outcome.child_id)
2385            .map(|state| state.path.clone())
2386            .or_else(|| {
2387                self.registry
2388                    .child(&outcome.child_id)
2389                    .map(|runtime| runtime.path.clone())
2390            })
2391            .unwrap_or_else(|| SupervisorPath::root().join(outcome.child_id.value.clone()));
2392        self.emit_pending_event(PendingRuntimeEvent {
2393            child_id: outcome.child_id.clone(),
2394            path: outcome_path,
2395            generation: outcome.generation,
2396            attempt: outcome.attempt,
2397            correlation_id,
2398            what: What::ChildControlCommandCompleted {
2399                child_id: outcome.child_id.clone(),
2400                command: "restart_child".to_owned(),
2401                command_id: meta.command_id.value.to_string(),
2402                requested_by: meta.requested_by.clone(),
2403                reason: meta.reason.clone(),
2404                result: child_control_result_label(&outcome).to_owned(),
2405                outcome: Box::new(outcome),
2406            },
2407        });
2408        let _ignored = event_sender.send(format!(
2409            "child_control_command_completed:{outcome_identifier}:restart_child"
2410        ));
2411    }
2412
2413    /// Blocks restart while the supervisor tree is not idle.
2414    ///
2415    /// # Arguments
2416    ///
2417    /// - `child_id`: Target child identifier.
2418    /// - `meta`: Audit metadata from the command.
2419    /// - `correlation_id`: Correlation binding typed events.
2420    /// - `event_sender`: Legacy text broadcaster.
2421    ///
2422    /// # Returns
2423    ///
2424    /// Returns [`CommandResult::ChildControl`] with [`GenerationFenceDecision::BlockedByShutdown`].
2425    fn restart_child_blocked_by_shutdown(
2426        &mut self,
2427        child_id: &ChildId,
2428        meta: &CommandMeta,
2429        correlation_id: CorrelationId,
2430        event_sender: &broadcast::Sender<String>,
2431    ) -> CommandResult {
2432        if !self.slots.contains_key(child_id) {
2433            let placeholder = self
2434                .registry
2435                .child(child_id)
2436                .map(|runtime| ChildSlot::new_placeholder(runtime.id.clone(), runtime.path.clone()))
2437                .unwrap_or_else(|| {
2438                    ChildSlot::new_placeholder(
2439                        child_id.clone(),
2440                        SupervisorPath::root().join(child_id.value.clone()),
2441                    )
2442                });
2443            self.slots.insert(child_id.clone(), placeholder);
2444        }
2445
2446        let outcome = {
2447            let runtime_state = self.slots.get_mut(child_id).expect("runtime state exists");
2448            runtime_state.generation_fence.phase = GenerationFencePhase::Closed;
2449            let failure = ChildControlFailure::new(
2450                ChildControlFailurePhase::WaitCompletion,
2451                "supervisor tree is shutting down",
2452                false,
2453            );
2454            let fence = GenerationFenceOutcome::new(
2455                GenerationFenceDecision::BlockedByShutdown,
2456                runtime_state.generation,
2457                runtime_state.attempt,
2458                None,
2459                false,
2460                false,
2461                Some(failure.clone()),
2462            );
2463            let operation_before = runtime_state.operation;
2464            runtime_state.last_control_failure = Some(failure);
2465            build_child_control_outcome(
2466                operation_before,
2467                false,
2468                false,
2469                runtime_state.last_control_failure.clone(),
2470                runtime_state,
2471                &self.time_base,
2472                Some(fence),
2473            )
2474        };
2475
2476        let blocked_events = match self.slots.get(child_id).map(|runtime_state| {
2477            (
2478                runtime_state.path.clone(),
2479                runtime_state.generation,
2480                runtime_state.attempt,
2481            )
2482        }) {
2483            Some((path, current_generation, current_attempt)) => {
2484                vec![PendingRuntimeEvent {
2485                    child_id: child_id.clone(),
2486                    path,
2487                    generation: current_generation,
2488                    attempt: current_attempt,
2489                    correlation_id,
2490                    what: What::ChildRestartConflict {
2491                        child_id: child_id.clone(),
2492                        current_generation,
2493                        current_attempt,
2494                        target_generation: None,
2495                        command_id: meta.command_id.value.to_string(),
2496                        decision: "rejected".to_owned(),
2497                        reason: "restart rejected while supervisor tree is shutting down"
2498                            .to_owned(),
2499                    },
2500                }]
2501            }
2502            None => Vec::new(),
2503        };
2504
2505        self.emit_restart_child_completed(
2506            outcome.clone(),
2507            meta,
2508            correlation_id,
2509            event_sender,
2510            blocked_events,
2511        );
2512
2513        CommandResult::ChildControl { outcome }
2514    }
2515
2516    /// Builds the current runtime state report.
2517    ///
2518    /// # Arguments
2519    ///
2520    /// This function has no arguments.
2521    ///
2522    /// # Returns
2523    ///
2524    /// Returns a [`CurrentState`] value.
2525    fn build_current_state(&mut self) -> CurrentState {
2526        let mut child_runtime_records = Vec::new();
2527        let mut pending_events = Vec::new();
2528        let declaration_order = self.registry.declaration_order().to_vec();
2529        for child_id in declaration_order {
2530            if let Some(runtime_state) = self.slots.get_mut(&child_id) {
2531                let liveness = runtime_state
2532                    .observe_liveness(self.time_base.now_unix_nanos(), &self.time_base);
2533                if let Some(event) = heartbeat_stale_event(runtime_state, &liveness) {
2534                    pending_events.push(event);
2535                }
2536                child_runtime_records.push(runtime_state.to_record(liveness));
2537            }
2538        }
2539        for (child_id, runtime_state) in &mut self.slots {
2540            if self.registry.child(child_id).is_some() {
2541                continue;
2542            }
2543            let liveness =
2544                runtime_state.observe_liveness(self.time_base.now_unix_nanos(), &self.time_base);
2545            if let Some(event) = heartbeat_stale_event(runtime_state, &liveness) {
2546                pending_events.push(event);
2547            }
2548            child_runtime_records.push(runtime_state.to_record(liveness));
2549        }
2550        for event in pending_events {
2551            self.emit_pending_event(event);
2552        }
2553        CurrentState {
2554            child_count: self.dynamic_child_count(),
2555            shutdown_completed: self.shutdown.phase()
2556                == crate::shutdown::stage::ShutdownPhase::Completed,
2557            child_runtime_records,
2558        }
2559    }
2560
2561    /// Spawns the target generation queued by a pending manual restart once the old attempt exits.
2562    ///
2563    /// # Arguments
2564    ///
2565    /// - `child_id`: Stable child undergoing a fenced restart.
2566    /// - `pending`: Accepted restart bookkeeping that pins the identity triple transition.
2567    /// - `old_exit`: Exit classification observed for the old attempt.
2568    ///
2569    /// # Returns
2570    ///
2571    /// This function does not return a value.
2572    fn spawn_pending_restart_target(
2573        &mut self,
2574        child_id: ChildId,
2575        pending: PendingRestart,
2576        old_exit: TaskExit,
2577    ) {
2578        let Some(registry_identity_anchor) = self.registry.child(&child_id).map(|runtime| {
2579            (
2580                runtime.generation,
2581                runtime.child_start_count,
2582                runtime.restart_count,
2583            )
2584        }) else {
2585            return;
2586        };
2587        let path = self
2588            .slots
2589            .get(&child_id)
2590            .map(|state| state.path.clone())
2591            .unwrap_or_else(|| SupervisorPath::root().join(child_id.value.clone()));
2592        let correlation_id = CorrelationId::from_uuid(pending.command_id);
2593
2594        if let Some(runtime_state) = self.slots.get_mut(&child_id) {
2595            runtime_state.registry_identity_anchor_for_spawn_attempt =
2596                Some(registry_identity_anchor);
2597        }
2598
2599        {
2600            let Some(registry_runtime) = self.registry.child_mut(&child_id) else {
2601                return;
2602            };
2603            registry_runtime.generation = pending.target_generation;
2604            registry_runtime.child_start_count = registry_runtime.child_start_count.next();
2605            registry_runtime.restart_count = registry_runtime.restart_count.saturating_add(1);
2606            registry_runtime.status = ChildRuntimeStatus::Starting;
2607        }
2608
2609        let Some(runtime) = self.registry.child(&child_id).cloned() else {
2610            return;
2611        };
2612        let new_generation = runtime.generation;
2613        let new_attempt = runtime.child_start_count;
2614
2615        let path_for_handles = path.clone();
2616
2617        match ChildRunner::new().spawn_once(runtime) {
2618            Ok(handle) => {
2619                let sender = self.command_sender.clone();
2620                self.emit_pending_event(PendingRuntimeEvent {
2621                    child_id: child_id.clone(),
2622                    path,
2623                    generation: Some(new_generation),
2624                    attempt: Some(new_attempt),
2625                    correlation_id,
2626                    what: What::ChildRestartFenceReleased {
2627                        child_id: child_id.clone(),
2628                        old_generation: pending.old_generation,
2629                        old_attempt: pending.old_attempt,
2630                        target_generation: pending.target_generation,
2631                        exit_kind: old_exit.clone(),
2632                    },
2633                });
2634                let mut completion_receiver = handle.completion_receiver.clone();
2635                self.slots
2636                    .entry(child_id.clone())
2637                    .or_insert_with(|| {
2638                        ChildSlot::new_placeholder(child_id.clone(), path_for_handles)
2639                    })
2640                    .activate(
2641                        new_generation,
2642                        new_attempt,
2643                        ChildAttemptStatus::Running,
2644                        handle,
2645                    );
2646                tokio::spawn(async move {
2647                    let result = wait_for_report(&mut completion_receiver).await;
2648                    send_child_result(sender, child_id, result).await;
2649                });
2650            }
2651            Err(error) => {
2652                let message = error.to_string();
2653                if let Some(runtime_state) = self.slots.get_mut(&child_id) {
2654                    let identity_anchor_triple_opt = runtime_state
2655                        .registry_identity_anchor_for_spawn_attempt
2656                        .take();
2657                    if let Some((generation, attempt, restart_count)) = identity_anchor_triple_opt {
2658                        if let Some(registry_runtime) = self.registry.child_mut(&child_id) {
2659                            registry_runtime.generation = generation;
2660                            registry_runtime.child_start_count = attempt;
2661                            registry_runtime.restart_count = restart_count;
2662                        }
2663                        // Keep the superseded `(generation, attempt)` identity visible alongside the queued target spawn failure diagnostics.
2664                        runtime_state.generation = Some(generation);
2665                        runtime_state.attempt = Some(attempt);
2666                        runtime_state.status = ChildAttemptStatus::Stopped;
2667                    }
2668                    runtime_state.generation_fence.phase = GenerationFencePhase::Open;
2669                    runtime_state.last_control_failure = Some(ChildControlFailure::new(
2670                        ChildControlFailurePhase::WaitCompletion,
2671                        message,
2672                        true,
2673                    ));
2674                }
2675                // Avoid enqueueing a second asynchronous start-failure loop message because `last_control_failure` already records the deterministic spawn diagnostics.
2676            }
2677        }
2678    }
2679
2680    /// Reconciles expired stop deadlines without blocking the control loop.
2681    ///
2682    /// # Arguments
2683    ///
2684    /// This function has no arguments.
2685    ///
2686    /// # Returns
2687    ///
2688    /// This function does not return a value.
2689    fn reconcile_stop_deadlines(&mut self) {
2690        let now = self.time_base.now_unix_nanos();
2691        let mut pending_events = Vec::new();
2692        for runtime_state in self.slots.values_mut() {
2693            let fence_escalation = if let Some(pending_restart) =
2694                runtime_state.generation_fence.pending_restart.as_ref()
2695            {
2696                if pending_restart.abort_requested {
2697                    None
2698                } else if runtime_state.generation_fence.phase
2699                    == GenerationFencePhase::WaitingForOldStop
2700                    && runtime_state.stop_state == ChildStopState::CancelDelivered
2701                    && runtime_state.has_active_attempt()
2702                    && now >= pending_restart.stop_deadline_at_unix_nanos
2703                {
2704                    match (runtime_state.generation, runtime_state.attempt) {
2705                        (Some(old_generation), Some(old_attempt)) => Some((
2706                            pending_restart.command_id,
2707                            pending_restart.target_generation,
2708                            pending_restart.stop_deadline_at_unix_nanos,
2709                            runtime_state.child_id.clone(),
2710                            runtime_state.path.clone(),
2711                            old_generation,
2712                            old_attempt,
2713                        )),
2714                        _ => None,
2715                    }
2716                } else {
2717                    None
2718                }
2719            } else {
2720                None
2721            };
2722
2723            if let Some((
2724                command_id,
2725                target_generation,
2726                deadline_ns,
2727                fence_child_id,
2728                fence_path,
2729                old_generation,
2730                old_attempt,
2731            )) = fence_escalation
2732            {
2733                let delivered = runtime_state.abort();
2734                if delivered {
2735                    if let Some(pending_mut) = &mut runtime_state.generation_fence.pending_restart {
2736                        pending_mut.abort_requested = true;
2737                    }
2738                    runtime_state.generation_fence.phase = GenerationFencePhase::AbortingOld;
2739                    pending_events.push(PendingRuntimeEvent {
2740                        child_id: fence_child_id.clone(),
2741                        path: fence_path,
2742                        generation: Some(old_generation),
2743                        attempt: Some(old_attempt),
2744                        correlation_id: CorrelationId::from_uuid(command_id),
2745                        what: What::ChildRestartFenceAbortRequested {
2746                            child_id: fence_child_id,
2747                            old_generation,
2748                            old_attempt,
2749                            target_generation,
2750                            command_id: command_id.to_string(),
2751                            deadline_unix_nanos: deadline_ns,
2752                        },
2753                    });
2754                }
2755            }
2756
2757            if runtime_state.generation_fence.pending_restart.is_some() {
2758                continue;
2759            }
2760
2761            if matches!(
2762                runtime_state.generation_fence.phase,
2763                GenerationFencePhase::WaitingForOldStop | GenerationFencePhase::AbortingOld
2764            ) {
2765                continue;
2766            }
2767
2768            if runtime_state.stop_state != ChildStopState::CancelDelivered {
2769                continue;
2770            }
2771            let Some(deadline) = runtime_state.stop_deadline_at_unix_nanos else {
2772                continue;
2773            };
2774            if deadline > now || !runtime_state.has_active_attempt() {
2775                continue;
2776            }
2777            let Some(generation) = runtime_state.generation else {
2778                continue;
2779            };
2780            let Some(attempt) = runtime_state.attempt else {
2781                continue;
2782            };
2783            let status = runtime_state.status;
2784            let failure = ChildControlFailure::new(
2785                ChildControlFailurePhase::WaitCompletion,
2786                "child did not complete before stop deadline",
2787                true,
2788            );
2789            runtime_state.status = status;
2790            runtime_state.stop_state = ChildStopState::Failed;
2791            runtime_state.last_control_failure = Some(failure.clone());
2792            pending_events.push(PendingRuntimeEvent {
2793                child_id: runtime_state.child_id.clone(),
2794                path: runtime_state.path.clone(),
2795                generation: Some(generation),
2796                attempt: Some(attempt),
2797                correlation_id: CorrelationId::new(),
2798                what: What::ChildControlStopFailed {
2799                    child_id: runtime_state.child_id.clone(),
2800                    generation,
2801                    attempt,
2802                    status,
2803                    stop_state: ChildStopState::Failed,
2804                    phase: failure.phase,
2805                    reason: failure.reason,
2806                    recoverable: failure.recoverable,
2807                },
2808            });
2809        }
2810        for event in pending_events {
2811            self.emit_pending_event(event);
2812        }
2813    }
2814
2815    /// Emits one pending typed runtime event.
2816    ///
2817    /// # Arguments
2818    ///
2819    /// - `pending`: Event data collected while runtime state was borrowed.
2820    ///
2821    /// # Returns
2822    ///
2823    /// This function does not return a value.
2824    fn emit_pending_event(&mut self, pending: PendingRuntimeEvent) {
2825        let now = Instant::now();
2826        let uptime = now
2827            .duration_since(self.time_base.base_instant)
2828            .as_millis()
2829            .min(u128::from(u64::MAX)) as u64;
2830        let monotonic_nanos = now.duration_since(self.time_base.base_instant).as_nanos();
2831        let child_name = self
2832            .registry
2833            .child(&pending.child_id)
2834            .map(|runtime| runtime.spec.name.clone())
2835            .unwrap_or_else(|| pending.child_id.to_string());
2836        let event = SupervisorEvent::new(
2837            When::new(EventTime::from_parts(
2838                monotonic_nanos,
2839                uptime,
2840                pending.generation.unwrap_or_else(Generation::initial),
2841                pending.attempt.unwrap_or_else(ChildStartCount::first),
2842            )),
2843            Where::new(pending.path).with_child(pending.child_id, child_name),
2844            pending.what,
2845            self.event_sequences.next(),
2846            pending.correlation_id,
2847            1,
2848        );
2849        if let Ok(mut observability) = self.observability.lock() {
2850            let _lagged = observability.emit(event);
2851        }
2852    }
2853
2854    /// Spawns one child child_start_count and reports the exit back to this control loop.
2855    ///
2856    /// # Arguments
2857    ///
2858    /// - `child_id`: Child that should run.
2859    /// - `is_restart`: Whether this child_start_count is a restart child_start_count.
2860    /// - `delay`: Delay before the child_start_count starts.
2861    ///
2862    /// # Returns
2863    ///
2864    /// This function does not return a value.
2865    fn spawn_child_start(&mut self, child_id: ChildId, is_restart: bool, delay: Duration) {
2866        if self.shutdown.phase() != ShutdownPhase::Idle {
2867            return;
2868        }
2869        if let Some(runtime_state) = self.slots.get(&child_id) {
2870            if runtime_state.generation_fence.pending_restart.is_some() {
2871                if is_restart {
2872                    let path = runtime_state.path.clone();
2873                    let generation = runtime_state.generation;
2874                    let attempt = runtime_state.attempt;
2875                    let pending_target = runtime_state
2876                        .generation_fence
2877                        .pending_restart
2878                        .as_ref()
2879                        .map(|pending| pending.target_generation);
2880                    self.emit_pending_event(PendingRuntimeEvent {
2881                        child_id: child_id.clone(),
2882                        path,
2883                        generation,
2884                        attempt,
2885                        correlation_id: CorrelationId::new(),
2886                        what: What::ChildRestartConflict {
2887                            child_id: child_id.clone(),
2888                            current_generation: generation,
2889                            current_attempt: attempt,
2890                            target_generation: pending_target,
2891                            command_id: "runtime-policy".to_owned(),
2892                            decision: "rejected".to_owned(),
2893                            reason: "automatic restart suppressed while pending manual restart holds the fence".to_owned(),
2894                        },
2895                    });
2896                }
2897                return;
2898            }
2899            if matches!(
2900                runtime_state.generation_fence.phase,
2901                GenerationFencePhase::WaitingForOldStop
2902                    | GenerationFencePhase::AbortingOld
2903                    | GenerationFencePhase::Closed
2904                    | GenerationFencePhase::ReadyToStart
2905            ) {
2906                return;
2907            }
2908        }
2909        let Some(mut runtime) = self.prepare_child_start(&child_id, is_restart) else {
2910            return;
2911        };
2912
2913        // Override isolation for previously orphaned slots.
2914        // If a child was emergency-force-killed, its future probably never
2915        // yielded — restarting it on the async worker pool would immediately
2916        // starve another worker thread. Force BlockingPool instead.
2917        if let Some(slot) = self.slots.get(&child_id)
2918            && slot.orphaned
2919        {
2920            runtime.spec.isolation = crate::spec::child::Isolation::BlockingPool;
2921        }
2922
2923        // Clean up orphaned file system resources before spawn.
2924        // When a previous instance was emergency-force-killed, its Drop never
2925        // ran — leaving behind sockets, PID files, and temp files. Removing
2926        // them here prevents "Address already in use" errors on restart.
2927        for path in &runtime.spec.cleanup_paths {
2928            if let Err(error) = std::fs::remove_file(path)
2929                && error.kind() != std::io::ErrorKind::NotFound
2930            {
2931                tracing::warn!(
2932                    child_id = %child_id,
2933                    path = %path.display(),
2934                    ?error,
2935                    "failed to clean up orphaned resource before spawn",
2936                );
2937            }
2938        }
2939
2940        let sender = self.command_sender.clone();
2941        if !delay.is_zero() {
2942            tokio::spawn(async move {
2943                tokio::time::sleep(delay).await;
2944                let child_id_for_msg = runtime.id.clone();
2945                let path = runtime.path.clone();
2946                let generation = runtime.generation;
2947                let attempt = runtime.child_start_count;
2948                match ChildRunner::new().spawn_once(runtime) {
2949                    Ok(handle) => {
2950                        let _ignored = sender
2951                            .send(RuntimeLoopMessage::ChildStart(
2952                                ChildStartMessage::DelayedSpawnAttached {
2953                                    child_id: child_id_for_msg,
2954                                    path,
2955                                    generation,
2956                                    attempt,
2957                                    handle,
2958                                },
2959                            ))
2960                            .await;
2961                    }
2962                    Err(error) => {
2963                        tokio::spawn(async move {
2964                            send_child_result(sender, child_id_for_msg, Err(error)).await;
2965                        });
2966                    }
2967                }
2968            });
2969            return;
2970        }
2971
2972        let child_id_cloned = runtime.id.clone();
2973        let path = runtime.path.clone();
2974        let generation = runtime.generation;
2975        let child_start_count = runtime.child_start_count;
2976        match ChildRunner::new().spawn_once(runtime) {
2977            Ok(handle) => {
2978                self.attach_spawned_child_handle(
2979                    child_id_cloned,
2980                    path,
2981                    generation,
2982                    child_start_count,
2983                    handle,
2984                );
2985            }
2986            Err(error) => {
2987                tokio::spawn(async move {
2988                    send_child_result(sender, child_id_cloned, Err(error)).await;
2989                });
2990            }
2991        }
2992    }
2993
2994    /// Prepares registry state for one child child_start_count.
2995    ///
2996    /// # Arguments
2997    ///
2998    /// - `child_id`: Child that should run.
2999    /// - `bump_restart_counters`: Whether this spawn should bump generation accounting like a restart.
3000    ///
3001    /// # Returns
3002    ///
3003    /// Returns a runtime record for the child runner.
3004    fn prepare_child_start(
3005        &mut self,
3006        child_id: &ChildId,
3007        bump_restart_counters: bool,
3008    ) -> Option<ChildRuntime> {
3009        let runtime = self.registry.child_mut(child_id)?;
3010        if bump_restart_counters {
3011            runtime.child_start_count = runtime.child_start_count.next();
3012            runtime.generation = runtime.generation.next();
3013            runtime.restart_count = runtime.restart_count.saturating_add(1);
3014        }
3015        runtime.status = ChildRuntimeStatus::Starting;
3016        if let Some(runtime_state) = self.slots.get_mut(child_id) {
3017            runtime_state.operation = ChildControlOperation::Active;
3018        }
3019        Some(runtime.clone())
3020    }
3021
3022    // ------------------------------------------------------------------
3023    // Slot-based lifecycle operations (slot-based replacement for child_runtime_states)
3024    // ------------------------------------------------------------------
3025
3026    /// Executes a shutdown on all slots using the real cancellation+join
3027    /// pipeline.
3028    ///
3029    /// # Arguments
3030    ///
3031    /// - `event_sender`: Event channel used for lifecycle text.
3032    ///
3033    /// # Returns
3034    ///
3035    /// Returns a shutdown result.
3036    pub(crate) async fn handle_shutdown_tree(
3037        &mut self,
3038        requested_by: String,
3039        reason: String,
3040        event_sender: &broadcast::Sender<String>,
3041    ) -> Result<ShutdownResult, SupervisorError> {
3042        let policy = self.shutdown.policy;
3043        let reason_copy = reason.clone();
3044        let cause = ShutdownCause::new(requested_by, reason);
3045        let _started = self.shutdown.request_stop(cause);
3046        let _ignored = event_sender.send(format!(
3047            "shutdown_phase_changed:{:?}:{:?}",
3048            ShutdownPhase::Idle,
3049            self.shutdown.phase()
3050        ));
3051        self.advance_shutdown_phase(event_sender);
3052        self.advance_shutdown_phase(event_sender);
3053
3054        let outcomes = shutdown_tree_fanout(
3055            &mut self.slots,
3056            &policy,
3057            &mut self.admission_set,
3058            &mut self.orphan_count,
3059        )
3060        .await;
3061        let reconcile = reconcile_shutdown_slots(&self.slots);
3062
3063        // Emit orphan warning when residual handles remain after shutdown.
3064        if !reconcile.verified_clean {
3065            let _ignored = event_sender.send(format!(
3066                "shutdown_reconcile_warning: orphan_slots={:?}",
3067                reconcile.orphan_slots
3068            ));
3069        }
3070
3071        self.advance_shutdown_phase(event_sender);
3072        self.advance_shutdown_phase(event_sender);
3073        let _completed = self.shutdown.complete();
3074
3075        let report = ShutdownPipelineReport {
3076            cause: ShutdownCause::new("slot-shutdown", reason_copy),
3077            started_at_unix_nanos: unix_epoch_nanos(),
3078            completed_at_unix_nanos: unix_epoch_nanos(),
3079            phase: ShutdownPhase::Completed,
3080            outcomes,
3081            reconcile: ShutdownReconcileReport::core_runtime_completed(),
3082            idempotent: false,
3083        };
3084        self.shutdown_pipeline.cache_report(report.clone());
3085        let _ignored = event_sender.send(format!(
3086            "shutdown_completed:{len}",
3087            len = report.outcomes.len()
3088        ));
3089
3090        // Step 4: active orphan degradation detection.
3091        let threshold = self.shutdown.policy.effective_max_orphan_threshold() as u64;
3092        if self.orphan_count >= threshold {
3093            let _ = event_sender.send(format!(
3094                "fatal_orphan_overflow:count={}:threshold={}",
3095                self.orphan_count, threshold,
3096            ));
3097            self.exit_handler.exit(1);
3098        } else if self.orphan_count > 0 {
3099            let _ = event_sender.send(format!(
3100                "orphan_count:{}:below_threshold:{}:supervisor_degraded",
3101                self.orphan_count, threshold,
3102            ));
3103        }
3104
3105        Ok(self.shutdown.result_with_report(report, false))
3106    }
3107
3108    /// Applies a control operation to the slot for the given child.
3109    ///
3110    /// # Arguments
3111    ///
3112    /// - `child_id`: Target child identifier.
3113    /// - `operation`: Desired control operation.
3114    ///
3115    /// # Returns
3116    ///
3117    /// Returns `true` when the slot was found and modified.
3118    pub(crate) fn handle_command_on_slot(
3119        &mut self,
3120        child_id: &ChildId,
3121        operation: ChildControlOperation,
3122    ) -> bool {
3123        let Some(slot) = self.slots.get_mut(child_id) else {
3124            return false;
3125        };
3126        slot.operation = operation;
3127        if matches!(
3128            operation,
3129            ChildControlOperation::Quarantined | ChildControlOperation::Removed
3130        ) && slot.has_active_attempt()
3131        {
3132            slot.cancel();
3133        }
3134        true
3135    }
3136
3137    /// Processes a completed child exit through the slot system.
3138    ///
3139    /// # Arguments
3140    ///
3141    /// - `child_id`: Child that exited.
3142    /// - `report`: Completed child run report.
3143    ///
3144    /// # Returns
3145    ///
3146    /// Returns the exit summary stored in the slot, or `None` when no slot
3147    /// exists for this child.
3148    pub(crate) fn process_child_exit_on_slot(
3149        &mut self,
3150        child_id: &ChildId,
3151        report: &ChildRunReport,
3152    ) -> Option<ChildExitSummary> {
3153        let slot = self.slots.get_mut(child_id)?;
3154        let now_nanos = self.time_base.now_unix_nanos();
3155        let summary = ChildExitSummary::from_report(report, now_nanos);
3156        slot.deactivate(summary.clone());
3157        self.admission_set.release(child_id);
3158        Some(summary)
3159    }
3160
3161    /// Observes liveness for every active slot and emits stale heartbeat
3162    /// events.
3163    ///
3164    /// # Arguments
3165    ///
3166    /// - `event_sender`: Event channel used for lifecycle text.
3167    ///
3168    /// # Returns
3169    ///
3170    /// Returns the count of slots with stale heartbeats.
3171    pub(crate) fn observe_slot_liveness(
3172        &mut self,
3173        event_sender: &broadcast::Sender<String>,
3174    ) -> usize {
3175        let mut stale_count = 0usize;
3176        let threshold_nanos = Duration::from_secs(DEFAULT_HEARTBEAT_TIMEOUT_SECS).as_nanos();
3177        let now_nanos = self.time_base.now_unix_nanos();
3178
3179        for (child_id, slot) in self.slots.iter_mut() {
3180            if !slot.has_active_attempt() {
3181                continue;
3182            }
3183            if let Some(last_hb) = slot.last_heartbeat_at
3184                && now_nanos.saturating_sub(last_hb) >= threshold_nanos
3185            {
3186                stale_count += 1;
3187                let _ignored = event_sender.send(format!(
3188                    "child_liveness_stale: child_id={child_id} last_heartbeat_at={last_hb}"
3189                ));
3190            }
3191        }
3192        stale_count
3193    }
3194
3195    /// Checks whether a child slot is eligible for restart.
3196    ///
3197    /// Returns `Ok(())` when restart may proceed, or `Err(AdmissionConflict)`
3198    /// when the slot has a pending restart or an active attempt.
3199    ///
3200    /// # Arguments
3201    ///
3202    /// - `child_id`: Child to check.
3203    /// - `request_generation`: Generation claimed by the restart request.
3204    /// - `request_attempt`: Attempt number claimed by the restart request.
3205    ///
3206    /// # Returns
3207    ///
3208    /// Returns `Ok(())` or `Err(AdmissionConflict)`.
3209    pub(crate) fn check_slot_restart_eligibility(
3210        &self,
3211        child_id: &ChildId,
3212        request_generation: Generation,
3213        request_attempt: ChildStartCount,
3214    ) -> Result<(), AdmissionConflict> {
3215        let Some(slot) = self.slots.get(child_id) else {
3216            return Ok(());
3217        };
3218        if slot.pending_restart {
3219            return Err(AdmissionConflict::new(
3220                child_id.clone(),
3221                slot.generation.unwrap_or(Generation::initial()),
3222                slot.attempt.unwrap_or(ChildStartCount::first()),
3223                "restart rejected: pending restart already exists",
3224            ));
3225        }
3226        if let (Some(active_gen), Some(active_att)) = (slot.generation, slot.attempt)
3227            && (request_generation != active_gen || request_attempt != active_att)
3228        {
3229            return Err(AdmissionConflict::new(
3230                child_id.clone(),
3231                active_gen,
3232                active_att,
3233                "restart conflicts with active attempt",
3234            ));
3235        }
3236        Ok(())
3237    }
3238
3239    /// Copies a `ChildRuntimeState` entry into a `ChildSlot` when a slot
3240    /// does not yet exist for the child.
3241    ///
3242    /// # Arguments
3243    ///
3244    /// - `child_id`: Child to ensure has a slot.
3245    /// - `path`: Supervisor path for the child.
3246    ///
3247    /// # Returns
3248    ///
3249    /// Returns `true` when a new slot was created.
3250    pub(crate) fn ensure_slot_exists(&mut self, child_id: ChildId, path: SupervisorPath) -> bool {
3251        if self.slots.contains_key(&child_id) {
3252            return false;
3253        }
3254        let slot = ChildSlot::new(
3255            child_id.clone(),
3256            path,
3257            Duration::from_secs(60), // Default restart window.
3258        );
3259        self.slots.insert(child_id, slot);
3260        true
3261    }
3262}
3263
3264/// Builds a child control command outcome from the latest runtime state.
3265///
3266/// # Arguments
3267///
3268/// - `operation_before`: Operation observed before command handling.
3269/// - `cancel_delivered`: Whether this command delivered cancellation.
3270/// - `idempotent`: Whether this command reused existing state.
3271/// - `failure`: Failure observed during command handling.
3272/// - `runtime_state`: Runtime state used as the source of truth.
3273/// - `time_base`: Runtime time base used for liveness timestamps.
3274/// - `generation_fence`: Optional fencing metadata for restart commands only.
3275///
3276/// # Returns
3277///
3278/// Returns a [`ChildControlResult`] value.
3279fn build_child_control_outcome(
3280    operation_before: ChildControlOperation,
3281    cancel_delivered: bool,
3282    idempotent: bool,
3283    failure: Option<ChildControlFailure>,
3284    runtime_state: &mut ChildSlot,
3285    time_base: &RuntimeTimeBase,
3286    generation_fence: Option<GenerationFenceOutcome>,
3287) -> ChildControlResult {
3288    let liveness = runtime_state.observe_liveness(time_base.now_unix_nanos(), time_base);
3289    // Data model: status is None when there is no active attempt.
3290    let status = if runtime_state.attempt.is_some() {
3291        Some(runtime_state.status)
3292    } else {
3293        None
3294    };
3295    ChildControlResult::new(
3296        runtime_state.child_id.clone(),
3297        runtime_state.attempt,
3298        runtime_state.generation,
3299        operation_before,
3300        runtime_state.operation,
3301        status,
3302        cancel_delivered,
3303        runtime_state.stop_state,
3304        runtime_state.restart_limit.clone(),
3305        liveness,
3306        idempotent,
3307        failure,
3308        generation_fence,
3309    )
3310}
3311
3312/// Result of applying a stop-style control command to a runtime state record.
3313#[derive(Debug, Clone, Copy)]
3314struct StopControlApplication {
3315    /// Operation observed before command handling.
3316    operation_before: ChildControlOperation,
3317    /// Whether this command delivered cancellation.
3318    cancel_delivered: bool,
3319    /// Whether this command reused existing state.
3320    idempotent: bool,
3321    /// Whether the caller should remove the record after building the outcome.
3322    remove_after_outcome: bool,
3323}
3324
3325/// Applies a stop-style child control command to one runtime state record.
3326///
3327/// # Arguments
3328///
3329/// - `runtime_state`: Runtime state that owns the target child.
3330/// - `target_operation`: Operation requested by the command.
3331/// - `command_name`: Stable command name.
3332/// - `command_id`: Stable command identifier.
3333/// - `correlation_id`: Correlation identifier for emitted events.
3334/// - `stop_deadline_at_unix_nanos`: Deadline written when cancellation is delivered.
3335/// - `event_sender`: Event channel used for lifecycle text.
3336/// - `pending_events`: Typed events collected until mutable borrows end.
3337///
3338/// # Returns
3339///
3340/// Returns the applied stop control facts.
3341#[allow(clippy::too_many_arguments)]
3342fn apply_stop_control_to_runtime_state(
3343    runtime_state: &mut ChildSlot,
3344    target_operation: ChildControlOperation,
3345    command_name: &'static str,
3346    command_id: &str,
3347    correlation_id: CorrelationId,
3348    stop_deadline_at_unix_nanos: u128,
3349    event_sender: &broadcast::Sender<String>,
3350    pending_events: &mut Vec<PendingRuntimeEvent>,
3351) -> StopControlApplication {
3352    let child_id = runtime_state.child_id.clone();
3353    let operation_before = runtime_state.operation;
3354    let had_active_attempt = runtime_state.has_active_attempt();
3355    let already_cancelled_for_target = had_active_attempt
3356        && operation_before == target_operation
3357        && runtime_state.attempt_cancel_delivered;
3358    let idempotent = if had_active_attempt {
3359        already_cancelled_for_target
3360    } else {
3361        operation_before == target_operation && target_operation != ChildControlOperation::Removed
3362    };
3363
3364    if operation_before != target_operation {
3365        runtime_state.operation = target_operation;
3366        pending_events.push(PendingRuntimeEvent {
3367            child_id: child_id.clone(),
3368            path: runtime_state.path.clone(),
3369            generation: runtime_state.generation,
3370            attempt: runtime_state.attempt,
3371            correlation_id,
3372            what: What::ChildControlOperationChanged {
3373                child_id: child_id.clone(),
3374                from: operation_before,
3375                to: target_operation,
3376                command: command_name.to_owned(),
3377                command_id: command_id.to_owned(),
3378            },
3379        });
3380        let _ignored = event_sender.send(format!(
3381            "child_control_operation_changed:{child_id}:{operation_before:?}:{target_operation:?}"
3382        ));
3383    }
3384
3385    let cancel_delivered = if had_active_attempt && !already_cancelled_for_target {
3386        let delivered = runtime_state.cancel();
3387        if delivered {
3388            runtime_state.stop_state = ChildStopState::CancelDelivered;
3389            runtime_state.stop_deadline_at_unix_nanos = Some(stop_deadline_at_unix_nanos);
3390            if let (Some(generation), Some(attempt)) =
3391                (runtime_state.generation, runtime_state.attempt)
3392            {
3393                pending_events.push(PendingRuntimeEvent {
3394                    child_id: child_id.clone(),
3395                    path: runtime_state.path.clone(),
3396                    generation: Some(generation),
3397                    attempt: Some(attempt),
3398                    correlation_id,
3399                    what: What::ChildControlCancelDelivered {
3400                        child_id: child_id.clone(),
3401                        generation,
3402                        attempt,
3403                        command: command_name.to_owned(),
3404                        command_id: command_id.to_owned(),
3405                    },
3406                });
3407            }
3408            let _ignored = event_sender.send(format!(
3409                "child_control_cancel_delivered:{child_id}:{command_name}"
3410            ));
3411        }
3412        delivered
3413    } else {
3414        if !had_active_attempt {
3415            runtime_state.stop_state = ChildStopState::NoActiveAttempt;
3416        }
3417        false
3418    };
3419
3420    StopControlApplication {
3421        operation_before,
3422        cancel_delivered,
3423        idempotent,
3424        remove_after_outcome: target_operation == ChildControlOperation::Removed
3425            && !had_active_attempt,
3426    }
3427}
3428
3429/// Builds a stale heartbeat event when suppression allows emission.
3430///
3431/// # Arguments
3432///
3433/// - `runtime_state`: Runtime state whose heartbeat was observed.
3434/// - `liveness`: Latest liveness state.
3435///
3436/// # Returns
3437///
3438/// Returns a pending event when the stale heartbeat should be emitted.
3439fn heartbeat_stale_event(
3440    runtime_state: &mut ChildSlot,
3441    liveness: &ChildLivenessState,
3442) -> Option<PendingRuntimeEvent> {
3443    let Some(attempt) = runtime_state.attempt else {
3444        runtime_state.stale_event_attempt = None;
3445        return None;
3446    };
3447    if !liveness.heartbeat_stale {
3448        runtime_state.stale_event_attempt = None;
3449        return None;
3450    }
3451    if runtime_state.stale_event_attempt == Some(attempt) {
3452        return None;
3453    }
3454    let since_unix_nanos = liveness.last_heartbeat_at_unix_nanos?;
3455    runtime_state.stale_event_attempt = Some(attempt);
3456    Some(PendingRuntimeEvent {
3457        child_id: runtime_state.child_id.clone(),
3458        path: runtime_state.path.clone(),
3459        generation: runtime_state.generation,
3460        attempt: Some(attempt),
3461        correlation_id: CorrelationId::new(),
3462        what: What::ChildHeartbeatStale {
3463            child_id: runtime_state.child_id.clone(),
3464            attempt,
3465            since_unix_nanos,
3466        },
3467    })
3468}
3469
3470/// Runs the control loop until all command senders are dropped.
3471///
3472/// # Arguments
3473///
3474/// - `state`: Runtime state initialized from the supervisor specification.
3475/// - `receiver`: Runtime command receiver.
3476/// - `event_sender`: Event channel used for audit text.
3477///
3478/// # Returns
3479///
3480/// Returns a [`RuntimeExitReport`] when the control loop ends.
3481pub async fn run_control_loop(
3482    mut state: RuntimeControlState,
3483    mut receiver: mpsc::Receiver<RuntimeLoopMessage>,
3484    event_sender: broadcast::Sender<String>,
3485) -> RuntimeExitReport {
3486    state.start_declared_children().await;
3487    while let Some(message) = receiver.recv().await {
3488        match message {
3489            RuntimeLoopMessage::Control {
3490                command,
3491                reply_sender,
3492            } => {
3493                let command_name = command_name(&command);
3494                let result = state.execute_control(command, &event_sender).await;
3495                let _ignored = event_sender.send(format!("control_command:{command_name}"));
3496                let _ignored = reply_sender.send(result);
3497            }
3498            RuntimeLoopMessage::ChildStart(ChildStartMessage::Exited { report }) => {
3499                state.handle_child_exit(*report, &event_sender);
3500            }
3501            RuntimeLoopMessage::ChildStart(ChildStartMessage::StartFailed {
3502                child_id,
3503                message,
3504            }) => {
3505                state.handle_child_start_failed(child_id, message, &event_sender);
3506            }
3507            RuntimeLoopMessage::ChildStart(ChildStartMessage::DelayedSpawnAttached {
3508                child_id,
3509                path,
3510                generation,
3511                attempt,
3512                handle,
3513            }) => {
3514                state.attach_spawned_child_handle(child_id, path, generation, attempt, handle);
3515            }
3516            RuntimeLoopMessage::ControlPlane(ControlPlaneMessage::ReplayChildExitForTest {
3517                report,
3518            }) => {
3519                state.handle_child_exit(*report, &event_sender);
3520            }
3521            RuntimeLoopMessage::ControlPlane(ControlPlaneMessage::Shutdown {
3522                meta,
3523                reply_sender,
3524            }) => {
3525                let _ignored = event_sender.send(format!(
3526                    "runtime_control_loop_shutdown_requested:{}:{}",
3527                    meta.requested_by, meta.reason
3528                ));
3529                match meta.validate() {
3530                    Ok(()) => {
3531                        let report = RuntimeExitReport::completed(
3532                            "shutdown",
3533                            format!(
3534                                "runtime control plane shutdown requested: {reason}",
3535                                reason = meta.reason
3536                            ),
3537                        );
3538                        let _ignored = reply_sender.send(Ok(report.clone()));
3539                        return report;
3540                    }
3541                    Err(error) => {
3542                        let _ignored = reply_sender.send(Err(error));
3543                        continue;
3544                    }
3545                }
3546            }
3547        }
3548    }
3549    RuntimeExitReport::completed("message_loop", "runtime command channel closed")
3550}
3551
3552/// Returns a stable command name for audit text.
3553///
3554/// # Arguments
3555///
3556/// - `command`: Command being executed.
3557///
3558/// # Returns
3559///
3560/// Returns a static command name.
3561fn command_name(command: &ControlCommand) -> &'static str {
3562    match command {
3563        ControlCommand::AddChild { .. } => "add_child",
3564        ControlCommand::RemoveChild { .. } => "remove_child",
3565        ControlCommand::RestartChild { .. } => "restart_child",
3566        ControlCommand::PauseChild { .. } => "pause_child",
3567        ControlCommand::ResumeChild { .. } => "resume_child",
3568        ControlCommand::QuarantineChild { .. } => "quarantine_child",
3569        ControlCommand::ShutdownTree { .. } => "shutdown_tree",
3570        ControlCommand::CurrentState { .. } => "current_state",
3571    }
3572}
3573
3574/// Sends a child run result back to the control loop.
3575///
3576/// # Arguments
3577///
3578/// - `sender`: Runtime command sender.
3579/// - `child_id`: Child identifier used when the run fails before reporting.
3580/// - `result`: Child run result.
3581///
3582/// # Returns
3583///
3584/// This function does not return a value.
3585async fn send_child_result(
3586    sender: mpsc::Sender<RuntimeLoopMessage>,
3587    child_id: ChildId,
3588    result: Result<ChildRunReport, SupervisorError>,
3589) {
3590    let message = match result {
3591        Ok(report) => RuntimeLoopMessage::ChildStart(ChildStartMessage::Exited {
3592            report: Box::new(report),
3593        }),
3594        Err(error) => RuntimeLoopMessage::ChildStart(ChildStartMessage::StartFailed {
3595            child_id,
3596            message: error.to_string(),
3597        }),
3598    };
3599    let _ignored = sender.send(message).await;
3600}
3601
3602/// Maps child restart policy into policy-engine restart policy.
3603///
3604/// # Arguments
3605///
3606/// - `policy`: Restart policy stored on the child declaration.
3607///
3608/// # Returns
3609///
3610/// Returns the equivalent policy-engine value.
3611fn restart_policy(policy: ChildRestartPolicy) -> RestartPolicy {
3612    match policy {
3613        ChildRestartPolicy::Permanent => RestartPolicy::Permanent,
3614        ChildRestartPolicy::Transient => RestartPolicy::Transient,
3615        ChildRestartPolicy::Temporary => RestartPolicy::Temporary,
3616    }
3617}
3618
3619/// Maps child backoff policy into policy-engine backoff policy.
3620///
3621/// # Arguments
3622///
3623/// Maps a child spec backoff policy into the policy-engine equivalent.
3624///
3625/// # Arguments
3626///
3627/// - `policy`: Backoff policy stored on the child declaration.
3628///
3629/// # Returns
3630///
3631/// Returns the equivalent policy-engine value with full jitter enabled.
3632fn backoff_policy(policy: crate::spec::child::BackoffPolicy) -> BackoffPolicy {
3633    let jitter_percent = (policy.jitter_ratio * 100.0).round().clamp(0.0, 100.0) as u8;
3634    BackoffPolicy::new(
3635        policy.initial_delay,
3636        policy.max_delay,
3637        jitter_percent,
3638        policy.max_delay,
3639    )
3640    .with_full_jitter(42) // Enable full jitter mode per FR-003
3641}
3642
3643/// Maps a child-runner exit into policy-engine task exit.
3644///
3645/// # Arguments
3646///
3647/// - `exit`: Exit reported by the child runner.
3648///
3649/// # Returns
3650///
3651/// Returns the policy-engine exit value.
3652fn policy_task_exit(exit: &TaskExit) -> PolicyTaskExit {
3653    match exit.failure_kind() {
3654        Some(kind) => PolicyTaskExit::Failed { kind: kind.into() },
3655        None => PolicyTaskExit::Succeeded,
3656    }
3657}
3658
3659/// Classifies a child-runner exit into pipeline exit classification.
3660///
3661/// This function maps all six minimum required exit kinds from the specification:
3662/// success, nonzero_exit, panic, timeout, external_cancel, manual_stop.
3663///
3664/// # Arguments
3665///
3666/// - `exit`: Exit reported by the child runner.
3667///
3668/// # Returns
3669///
3670/// Returns the pipeline exit classification value.
3671fn classify_exit_for_pipeline(exit: &TaskExit, manual_stop_requested: bool) -> ExitClassification {
3672    match exit {
3673        TaskExit::Succeeded => ExitClassification::Success,
3674        TaskExit::Cancelled if manual_stop_requested => ExitClassification::ManualStop,
3675        TaskExit::Cancelled => ExitClassification::ExternalCancel,
3676        TaskExit::Failed(failure) => {
3677            // Check if this is an external cancel or timeout based on failure kind.
3678            match failure.kind {
3679                crate::error::types::TaskFailureKind::Cancelled if manual_stop_requested => {
3680                    ExitClassification::ManualStop
3681                }
3682                crate::error::types::TaskFailureKind::Cancelled => {
3683                    ExitClassification::ExternalCancel
3684                }
3685                crate::error::types::TaskFailureKind::Timeout => ExitClassification::Timeout,
3686                _ => ExitClassification::NonZeroExit { exit_code: -1 },
3687            }
3688        }
3689        TaskExit::Panicked(_) => ExitClassification::Crash {
3690            reason: "panic".to_string(),
3691        },
3692        TaskExit::TimedOut => ExitClassification::Timeout,
3693    }
3694}
3695
3696/// Reports whether the role policy should restart a successful exit.
3697///
3698/// # Arguments
3699///
3700/// - `pipeline_result`: Completed supervision pipeline context.
3701///
3702/// # Returns
3703///
3704/// Returns `true` when the effective role treats success as a restartable exit.
3705fn role_policy_restarts_success(pipeline_result: &PipelineContext) -> bool {
3706    pipeline_result.exit_classification == Some(ExitClassification::Success)
3707        && pipeline_result
3708            .effective_policy
3709            .as_ref()
3710            .is_some_and(|policy| policy.policy_pack.on_success_exit == OnSuccessAction::Restart)
3711}
3712
3713/// Builds the effective policy for a child before budget evaluation.
3714///
3715/// # Arguments
3716///
3717/// - `child_spec`: Child specification whose declared role and overrides should be merged.
3718///
3719/// # Returns
3720///
3721/// Returns an [`EffectivePolicy`] ready for the supervision pipeline.
3722fn prepare_effective_policy(child_spec: &ChildSpec) -> EffectivePolicy {
3723    EffectivePolicy::for_child(child_spec)
3724}
3725
3726/// Reports whether an exit should consume restart limit accounting.
3727///
3728/// # Arguments
3729///
3730/// - `exit`: Exit reported by the child runner.
3731///
3732/// # Returns
3733///
3734/// Returns `true` when the exit is an unplanned failure.
3735fn restart_limit_counts_exit(exit: &TaskExit) -> bool {
3736    matches!(
3737        exit,
3738        TaskExit::Failed(_) | TaskExit::Panicked(_) | TaskExit::TimedOut
3739    )
3740}
3741
3742/// Resolves the restart limit for a child from the supervisor strategy layers.
3743///
3744/// # Arguments
3745///
3746/// - `tree`: Supervisor tree used for child group lookup.
3747/// - `spec`: Supervisor specification that owns restart limit layers.
3748/// - `child_id`: Child whose restart limit should be resolved.
3749///
3750/// # Returns
3751///
3752/// Returns the selected restart limit or the runtime default.
3753fn restart_limit_for_child_in_spec(
3754    tree: &SupervisorTree,
3755    spec: &SupervisorSpec,
3756    child_id: &ChildId,
3757) -> RestartLimit {
3758    restart_execution_plan(tree, spec, child_id)
3759        .restart_limit
3760        .unwrap_or_else(default_restart_limit)
3761}
3762
3763/// Returns the runtime default restart limit.
3764///
3765/// # Arguments
3766///
3767/// This function has no arguments.
3768///
3769/// # Returns
3770///
3771/// Returns a conservative effectively-unbounded restart limit.
3772fn default_restart_limit() -> RestartLimit {
3773    RestartLimit::new(u32::MAX, Duration::from_secs(60))
3774}
3775
3776/// Builds a deterministic restart outcome for unknown identifiers.
3777///
3778/// # Arguments
3779///
3780/// - `child_id`: Stable child referenced by the command.
3781///
3782/// # Returns
3783///
3784/// Returns a rejection [`ChildControlResult`] with structured fencing metadata.
3785fn restart_child_unknown_outcome(child_id: ChildId) -> ChildControlResult {
3786    let conflict = ChildControlFailure::new(
3787        ChildControlFailurePhase::WaitCompletion,
3788        "unknown child",
3789        false,
3790    );
3791    let fence = GenerationFenceOutcome::new(
3792        GenerationFenceDecision::Rejected,
3793        None,
3794        None,
3795        None,
3796        false,
3797        false,
3798        Some(conflict.clone()),
3799    );
3800    ChildControlResult::new(
3801        child_id,
3802        None,
3803        None,
3804        ChildControlOperation::Active,
3805        ChildControlOperation::Active,
3806        None,
3807        false,
3808        ChildStopState::NoActiveAttempt,
3809        RestartLimitState::default(),
3810        ChildLivenessState::new(
3811            None,
3812            false,
3813            crate::readiness::signal::ReadinessState::Unreported,
3814        ),
3815        false,
3816        Some(conflict),
3817        Some(fence),
3818    )
3819}
3820
3821/// Classifies a child control command outcome for metrics.
3822///
3823/// # Arguments
3824///
3825/// - `outcome`: Child control command outcome.
3826///
3827/// # Returns
3828///
3829/// Returns `accepted`, `idempotent`, or `failed`.
3830fn child_control_result_label(outcome: &ChildControlResult) -> &'static str {
3831    if outcome.failure.is_some() || outcome.stop_state == ChildStopState::Failed {
3832        "failed"
3833    } else if outcome.idempotent {
3834        "idempotent"
3835    } else {
3836        "accepted"
3837    }
3838}
3839
3840/// Formats a restart scope for lifecycle events.
3841///
3842/// # Arguments
3843///
3844/// - `scope`: Child identifiers selected by strategy.
3845///
3846/// # Returns
3847///
3848/// Returns a comma-separated child identifier list.
3849fn child_scope_label(scope: &[ChildId]) -> String {
3850    scope
3851        .iter()
3852        .map(|child_id| child_id.value.clone())
3853        .collect::<Vec<_>>()
3854        .join(",")
3855}
3856
3857/// Maps managed child state into a control operation.
3858///
3859/// # Arguments
3860///
3861/// - `state`: Managed child state used by the current control loop.
3862///
3863/// # Returns
3864///
3865/// Returns the equivalent child control operation.
3866/// Builds a child shutdown outcome from a completed run report.
3867///
3868/// # Arguments
3869///
3870/// - `runtime_state`: Runtime state that produced the report.
3871/// - `report`: Completed child run report.
3872/// - `status`: Shutdown status assigned to the report.
3873/// - `phase`: Shutdown phase where the report was consumed.
3874/// - `reason`: Human-readable diagnostic reason.
3875///
3876/// # Returns
3877///
3878/// Returns a [`ChildShutdownOutcome`].
3879fn outcome_from_report(
3880    runtime_state: &ChildSlot,
3881    report: &ChildRunReport,
3882    status: ChildShutdownStatus,
3883    phase: ShutdownPhase,
3884    reason: impl Into<String>,
3885) -> ChildShutdownOutcome {
3886    ChildShutdownOutcome::new(ChildShutdownOutcomeInput {
3887        child_id: runtime_state.child_id.clone(),
3888        path: runtime_state.path.clone(),
3889        generation: runtime_state.generation.unwrap_or_else(Generation::initial),
3890        child_start_count: runtime_state.attempt.unwrap_or_else(ChildStartCount::first),
3891        status,
3892        cancel_delivered: runtime_state.attempt_cancel_delivered,
3893        exit: Some(report.exit.clone()),
3894        phase,
3895        reason: reason.into(),
3896    })
3897}
3898
3899/// Builds a shutdown outcome for a removed runtime state record.
3900///
3901/// # Arguments
3902///
3903/// - `runtime_state`: Removed runtime state skipped by shutdown.
3904/// - `phase`: Shutdown phase that observed the removed state.
3905///
3906/// # Returns
3907///
3908/// Returns a [`ChildShutdownOutcome`] marked as already exited.
3909fn removed_runtime_state_shutdown_outcome(
3910    runtime_state: &ChildSlot,
3911    phase: ShutdownPhase,
3912) -> ChildShutdownOutcome {
3913    ChildShutdownOutcome::new(ChildShutdownOutcomeInput {
3914        child_id: runtime_state.child_id.clone(),
3915        path: runtime_state.path.clone(),
3916        generation: runtime_state.generation.unwrap_or_else(Generation::initial),
3917        child_start_count: runtime_state.attempt.unwrap_or_else(ChildStartCount::first),
3918        status: ChildShutdownStatus::AlreadyExited,
3919        cancel_delivered: false,
3920        exit: None,
3921        phase,
3922        reason: "child runtime state was already removed before shutdown".to_owned(),
3923    })
3924}
3925
3926/// Builds a child shutdown outcome from a run report error.
3927///
3928/// # Arguments
3929///
3930/// - `runtime_state`: Runtime state that produced the error.
3931/// - `status`: Shutdown status assigned to the error.
3932/// - `phase`: Shutdown phase where the error was consumed.
3933/// - `error`: Error returned by the child run observer.
3934///
3935/// # Returns
3936///
3937/// Returns a [`ChildShutdownOutcome`].
3938fn outcome_from_error(
3939    runtime_state: &ChildSlot,
3940    status: ChildShutdownStatus,
3941    phase: ShutdownPhase,
3942    error: SupervisorError,
3943) -> ChildShutdownOutcome {
3944    ChildShutdownOutcome::new(ChildShutdownOutcomeInput {
3945        child_id: runtime_state.child_id.clone(),
3946        path: runtime_state.path.clone(),
3947        generation: runtime_state.generation.unwrap_or_else(Generation::initial),
3948        child_start_count: runtime_state.attempt.unwrap_or_else(ChildStartCount::first),
3949        status,
3950        cancel_delivered: runtime_state.attempt_cancel_delivered,
3951        exit: None,
3952        phase,
3953        reason: error.to_string(),
3954    })
3955}
3956
3957/// Returns the remaining duration before a deadline.
3958///
3959/// # Arguments
3960///
3961/// - `deadline`: Monotonic deadline.
3962///
3963/// # Returns
3964///
3965/// Returns `None` when the deadline has already passed.
3966fn remaining_duration(deadline: Instant) -> Option<Duration> {
3967    deadline.checked_duration_since(Instant::now())
3968}
3969
3970/// Returns the current Unix epoch timestamp in nanoseconds.
3971///
3972/// # Arguments
3973///
3974/// This function has no arguments.
3975///
3976/// # Returns
3977///
3978/// Returns a nanosecond timestamp, or zero if system time is before epoch.
3979fn unix_epoch_nanos() -> u128 {
3980    SystemTime::now()
3981        .duration_since(UNIX_EPOCH)
3982        .map_or(0, |duration| duration.as_nanos())
3983}