openlatch-client 0.1.18

//! In-process task supervision — "who restarts this subsystem".
//!
//! Sits beside the OS supervisors in this module ([`super::launchd`],
//! [`super::systemd`], [`super::task_scheduler`]). Both answer the same
//! question at different altitudes: those restart the **process**, this
//! restarts the **tasks inside it**.
//!
//! ## Why this exists
//!
//! The daemon spawns roughly fifteen detached `tokio::spawn` tasks — the cloud
//! worker, the alerts long-poll, the policy poller, the reconciler, the model
//! boundary listener, the log writers. `panic = unwind` is the default, so a
//! panic in any one of them killed that subsystem **silently**: nothing joined
//! the handle, `/health` still answered `ok`, and the only symptom was work
//! that quietly stopped happening. The boundary listener was worse still — a
//! failed bind returned once and never retried, so every agent on the machine
//! got `ECONNREFUSED` while the daemon reported itself healthy.
//!
//! ## Panic capture without a new dependency
//!
//! `futures_util::FutureExt::catch_unwind` is not available here: `futures-util`
//! is a **dev**-dependency. Instead the supervisor spawns the subsystem as its
//! own task and observes the [`tokio::task::JoinHandle`] —
//! [`tokio::task::JoinError::is_panic`] already distinguishes a panic from a
//! cancellation. No `UnwindSafe` bounds, works for any
//! `Future<Output = T> + Send + 'static`, and costs one extra task per
//! supervised subsystem.
//!
//! ## Log discipline
//!
//! A subsystem that fails every backoff window must not flood `daemon.log`.
//! This mirrors the cloud worker's existing idiom (`"…suppressing further
//! warnings until recovery"`, `core/cloud/worker.rs`): the first failure logs
//! at ERROR with its `code`, every consecutive failure logs at DEBUG with the
//! streak length, and one INFO line marks recovery.

use std::future::Future;
use std::sync::atomic::{AtomicI64, AtomicU32, AtomicU8, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use tokio::sync::watch;
use tokio::task::{JoinError, JoinHandle};

use crate::error::{OlError, ERR_TASK_PANICKED, ERR_TASK_RESTART_LIMIT};
use crate::privacy::PrivacyFilter;

/// First restart delay after a failure.
pub const BACKOFF_INITIAL: Duration = Duration::from_secs(1);
/// Ceiling on the restart delay. A subsystem that is broken forever retries
/// forever, once a minute — it is never abandoned.
pub const BACKOFF_MAX: Duration = Duration::from_secs(60);
/// A run that lasted at least this long counts as healthy: the backoff and the
/// consecutive-failure streak both reset. Without it, a task that crashes once
/// an hour would eventually be restarting on the 60 s ceiling.
pub const HEALTHY_AFTER: Duration = Duration::from_secs(60);
/// Consecutive-failure count at which [`ERR_TASK_RESTART_LIMIT`] is emitted —
/// once, at WARN. The supervisor does **not** stop retrying at this point;
/// the streak is surfaced so a permanently broken subsystem is visible without
/// tailing DEBUG logs.
pub const RESTART_LIMIT_WARN_AT: u32 = 5;

/// How long the supervisor waits for its inner task to observe the shutdown
/// signal and drain before aborting it. Deliberately under the daemon's own
/// 5 s bounded join so the supervisor returns first in the normal case and
/// that outer warning keeps meaning "something is genuinely wedged".
const SHUTDOWN_DRAIN: Duration = Duration::from_secs(4);

// ---------------------------------------------------------------------------
// Policy + spec
// ---------------------------------------------------------------------------

/// When a supervised task should be restarted.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RestartPolicy {
    /// Restart on every exit — clean, failed, or panicked. For subsystems that
    /// are supposed to run for the daemon's whole lifetime. These are the tasks
    /// `/health` calls `degraded` when they are not running.
    Always,
    /// Restart only on failure or panic; a clean completion ends the loop. For
    /// one-shot work (a startup check, a channel-drain writer) where finishing
    /// is the expected outcome and looping would spin.
    OnFailure,
    /// Never restart. Observation only — the health entry still records how the
    /// task ended.
    Never,
}

/// Exponentially-growing restart delay with jitter.
///
/// Jitter matters because ~15 subsystems can fail together (a full disk, a
/// dropped network): without it they would all retry in lockstep forever.
#[derive(Debug, Clone)]
pub struct Backoff {
    initial: Duration,
    max: Duration,
    current: Duration,
    rng: u64,
}

impl Default for Backoff {
    fn default() -> Self {
        Self::new(BACKOFF_INITIAL, BACKOFF_MAX)
    }
}

impl Backoff {
    /// Build a backoff that starts at `initial` and doubles up to `max`.
    pub fn new(initial: Duration, max: Duration) -> Self {
        Self {
            initial,
            max,
            current: initial,
            rng: seed(),
        }
    }

    /// The un-jittered delay the next [`Self::next_delay`] call will grow from.
    /// Exposed so tests can assert the growth curve without fighting jitter.
    pub fn base(&self) -> Duration {
        self.current
    }

    /// Return to the initial delay. Called after a healthy run.
    pub fn reset(&mut self) {
        self.current = self.initial;
    }

    /// Return the delay to sleep before the next restart, then grow the base.
    ///
    /// The returned value is the base ±20 %, clamped to `max` — so "caps at
    /// 60 s" stays literally true rather than "caps at 72 s".
    pub fn next_delay(&mut self) -> Duration {
        let base = self.current;
        self.current = (self.current * 2).min(self.max);

        // xorshift64* — a few instructions, no `rand` dependency, and the
        // quality bar here is only "don't align 15 retry loops".
        self.rng ^= self.rng >> 12;
        self.rng ^= self.rng << 25;
        self.rng ^= self.rng >> 27;
        let frac = ((self.rng >> 11) as f64) / ((1u64 << 53) as f64);
        let factor = 0.8 + 0.4 * frac;

        Duration::from_secs_f64(base.as_secs_f64() * factor).min(self.max)
    }
}

/// Seed the jitter PRNG. Mixes the wall clock with a process-global counter so
/// two backoffs built in the same nanosecond still diverge.
fn seed() -> u64 {
    use std::sync::atomic::AtomicU64;
    static COUNTER: AtomicU64 = AtomicU64::new(0x9E37_79B9_7F4A_7C15);
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos() as u64)
        .unwrap_or(0x2545_F491_4F6C_DD1D);
    let n = COUNTER.fetch_add(0x9E37_79B9_7F4A_7C15, Ordering::Relaxed);
    // Never return 0 — xorshift is stuck at zero forever.
    (nanos ^ n) | 1
}

/// What to supervise and how.
#[derive(Debug, Clone)]
pub struct TaskSpec {
    /// Stable identifier. Appears in `daemon.log`, `/health` and
    /// `openlatch status`, so it is a wire-ish name: change it deliberately.
    pub name: &'static str,
    pub restart: RestartPolicy,
    pub backoff: Backoff,
}

impl TaskSpec {
    /// A spec with the default 1 s → 60 s backoff.
    pub fn new(name: &'static str, restart: RestartPolicy) -> Self {
        Self {
            name,
            restart,
            backoff: Backoff::default(),
        }
    }

    /// Override the backoff curve (tests drive this in milliseconds).
    pub fn with_backoff(mut self, backoff: Backoff) -> Self {
        self.backoff = backoff;
        self
    }
}

// ---------------------------------------------------------------------------
// Outcome classification
// ---------------------------------------------------------------------------

/// How one run of a supervised task ended, before panics are folded in.
#[derive(Debug)]
pub enum TaskOutcome {
    /// The future returned normally.
    Completed,
    /// The future returned an error value.
    Failed {
        /// The `OL-XXXX` code when the error carried one.
        code: Option<&'static str>,
        /// Human-readable description. Scrubbed before it is stored.
        error: String,
    },
}

/// Lets the supervisor accept the signatures the daemon's tasks already have,
/// rather than forcing every one of them to change shape.
///
/// Implemented for `()` (the plain background loops) and
/// `Result<(), OlError>` (the boundary listener, the log writers). A blanket
/// `Result<(), E: Display>` impl would overlap with the `OlError` one and lose
/// the error `code` that the log discipline keys off, so it is deliberately
/// not provided.
pub trait IntoOutcome {
    fn into_outcome(self) -> TaskOutcome;
}

impl IntoOutcome for () {
    fn into_outcome(self) -> TaskOutcome {
        TaskOutcome::Completed
    }
}

impl IntoOutcome for Result<(), OlError> {
    fn into_outcome(self) -> TaskOutcome {
        match self {
            Ok(()) => TaskOutcome::Completed,
            Err(e) => TaskOutcome::Failed {
                code: Some(e.code),
                error: e.message,
            },
        }
    }
}

// ---------------------------------------------------------------------------
// Health registry
// ---------------------------------------------------------------------------

/// Lifecycle state of one supervised subsystem.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum TaskState {
    /// Registered, not yet spawned.
    Starting = 0,
    /// The inner task is live.
    Running = 1,
    /// Failed; sleeping out the backoff before the next attempt.
    Restarting = 2,
    /// Failed and will not be restarted (policy said so).
    Failed = 3,
    /// Ended cleanly and will not be restarted.
    Stopped = 4,
}

impl TaskState {
    fn from_u8(v: u8) -> Self {
        match v {
            1 => TaskState::Running,
            2 => TaskState::Restarting,
            3 => TaskState::Failed,
            4 => TaskState::Stopped,
            _ => TaskState::Starting,
        }
    }

    pub fn as_str(self) -> &'static str {
        match self {
            TaskState::Starting => "starting",
            TaskState::Running => "running",
            TaskState::Restarting => "restarting",
            TaskState::Failed => "failed",
            TaskState::Stopped => "stopped",
        }
    }
}

/// Observable state of a single supervised subsystem.
///
/// Every field is either atomic or behind a `Mutex`, so `/health` reads it from
/// an axum handler while the supervisor writes it from its own task.
pub struct TaskHealth {
    name: &'static str,
    policy: RestartPolicy,
    state: AtomicU8,
    restarts: AtomicU32,
    consecutive_failures: AtomicU32,
    last_transition_unix: AtomicI64,
    /// Scrubbed through [`PrivacyFilter`] before it is stored — an error string
    /// can carry a URL with a token in it, and `/health` is unauthenticated.
    last_error: Mutex<Option<String>>,
}

impl TaskHealth {
    pub fn new(name: &'static str, policy: RestartPolicy) -> Self {
        Self {
            name,
            policy,
            state: AtomicU8::new(TaskState::Starting as u8),
            restarts: AtomicU32::new(0),
            consecutive_failures: AtomicU32::new(0),
            last_transition_unix: AtomicI64::new(now_unix()),
            last_error: Mutex::new(None),
        }
    }

    pub fn name(&self) -> &'static str {
        self.name
    }

    pub fn policy(&self) -> RestartPolicy {
        self.policy
    }

    pub fn state(&self) -> TaskState {
        TaskState::from_u8(self.state.load(Ordering::Relaxed))
    }

    pub fn restarts(&self) -> u32 {
        self.restarts.load(Ordering::Relaxed)
    }

    pub fn consecutive_failures(&self) -> u32 {
        self.consecutive_failures.load(Ordering::Relaxed)
    }

    pub fn last_error(&self) -> Option<String> {
        self.last_error.lock().ok().and_then(|g| g.clone())
    }

    /// Unix seconds of the most recent state change.
    pub fn last_transition_unix(&self) -> i64 {
        self.last_transition_unix.load(Ordering::Relaxed)
    }

    /// A `RestartPolicy::Always` subsystem that is not `Running` is what makes
    /// `/health` report `degraded`. `OnFailure` / `Never` tasks legitimately
    /// finish, so they never count.
    pub fn is_degraded(&self) -> bool {
        self.policy == RestartPolicy::Always && self.state() != TaskState::Running
    }

    pub fn set_state(&self, state: TaskState) {
        self.state.store(state as u8, Ordering::Relaxed);
        self.last_transition_unix
            .store(now_unix(), Ordering::Relaxed);
    }

    fn record_restart(&self) {
        self.restarts.fetch_add(1, Ordering::Relaxed);
    }

    /// Record a failed run; returns the new consecutive-failure count.
    fn note_failure(&self, error: &str) -> u32 {
        if let Ok(mut guard) = self.last_error.lock() {
            *guard = Some(scrub(error));
        }
        self.consecutive_failures.fetch_add(1, Ordering::Relaxed) + 1
    }

    /// Clear the failure streak. Returns `true` when there was one to clear —
    /// the caller uses that to emit exactly one recovery line.
    fn note_healthy(&self) -> bool {
        let had = self.consecutive_failures.swap(0, Ordering::Relaxed) > 0;
        if had {
            if let Ok(mut guard) = self.last_error.lock() {
                *guard = None;
            }
        }
        had
    }

    /// The `/health` view of this subsystem.
    pub fn to_json(&self) -> serde_json::Value {
        let mut obj = serde_json::json!({
            "state": self.state().as_str(),
            "restarts": self.restarts(),
        });
        // Omitted rather than zeroed/false while healthy: "never failed" and
        // "recovered after failing" must stay distinguishable, and a reader
        // scanning for trouble should not have to filter out fifteen
        // `"degraded": false` entries to find the one that matters.
        if let Some(map) = obj.as_object_mut() {
            // The daemon applies the "Always and not running" rule here so
            // consumers (`openlatch status`) do not have to re-derive it from a
            // policy the wire format does not carry.
            if self.is_degraded() {
                map.insert("degraded".into(), true.into());
            }
            let failures = self.consecutive_failures();
            if failures > 0 {
                map.insert("consecutive_failures".into(), failures.into());
            }
            if let Some(err) = self.last_error() {
                map.insert("last_error".into(), err.into());
            }
        }
        obj
    }
}

/// Every supervised subsystem in this daemon, in registration order.
///
/// Held as `Arc<HealthRegistry>` on `daemon::AppState` and read by `/health`,
/// `/metrics` and `openlatch status`.
#[derive(Default)]
pub struct HealthRegistry {
    tasks: Mutex<Vec<Arc<TaskHealth>>>,
}

impl HealthRegistry {
    pub fn new() -> Self {
        Self::default()
    }

    /// Register a subsystem and return its health handle. Called by
    /// [`spawn_supervised`]; call it directly only when you need the handle
    /// before the task exists.
    pub fn register(&self, name: &'static str, policy: RestartPolicy) -> Arc<TaskHealth> {
        let health = Arc::new(TaskHealth::new(name, policy));
        if let Ok(mut tasks) = self.tasks.lock() {
            tasks.push(health.clone());
        }
        health
    }

    pub fn tasks(&self) -> Vec<Arc<TaskHealth>> {
        self.tasks
            .lock()
            .map(|t| t.clone())
            .unwrap_or_else(|_| Vec::new())
    }

    /// Names of the `Always` subsystems that are not currently running.
    pub fn degraded_names(&self) -> Vec<&'static str> {
        self.tasks()
            .iter()
            .filter(|t| t.is_degraded())
            .map(|t| t.name())
            .collect()
    }

    pub fn degraded_count(&self) -> u64 {
        self.tasks().iter().filter(|t| t.is_degraded()).count() as u64
    }

    pub fn is_degraded(&self) -> bool {
        self.tasks().iter().any(|t| t.is_degraded())
    }

    /// Total restarts across every subsystem — the `/metrics` counter.
    pub fn total_restarts(&self) -> u64 {
        self.tasks().iter().map(|t| u64::from(t.restarts())).sum()
    }

    /// `{ "<name>": {"state": …, "restarts": N}, … }` for `/health`.
    pub fn subsystems_json(&self) -> serde_json::Value {
        let mut map = serde_json::Map::new();
        for t in self.tasks() {
            map.insert(t.name().to_string(), t.to_json());
        }
        serde_json::Value::Object(map)
    }
}

// ---------------------------------------------------------------------------
// The supervisor
// ---------------------------------------------------------------------------

/// Register `spec.name` on `registry`, then supervise it. The common path.
pub fn spawn_supervised<F, Fut, T>(
    registry: &Arc<HealthRegistry>,
    spec: TaskSpec,
    shutdown: watch::Receiver<bool>,
    factory: F,
) -> JoinHandle<()>
where
    F: FnMut() -> Fut + Send + 'static,
    Fut: Future<Output = T> + Send + 'static,
    T: IntoOutcome + Send + 'static,
{
    let health = registry.register(spec.name, spec.restart);
    supervise(spec, health, shutdown, factory)
}

/// Run `factory`'s future under supervision until `shutdown` flips to `true`.
///
/// `factory` is re-invoked on **every** restart, so it must produce a *fresh*
/// future — capture `Arc`s and clones, never a moved-once resource. A task that
/// owns a channel receiver should hold it behind an
/// `Arc<tokio::sync::Mutex<Receiver<_>>>` and re-lock it on each run, so
/// queued messages survive a panic (see `EventLogger::new_supervised`).
///
/// The returned handle resolves once the supervisor has stopped restarting and
/// its inner task has drained (bounded by [`SHUTDOWN_DRAIN`]).
pub fn supervise<F, Fut, T>(
    spec: TaskSpec,
    health: Arc<TaskHealth>,
    mut shutdown: watch::Receiver<bool>,
    mut factory: F,
) -> JoinHandle<()>
where
    F: FnMut() -> Fut + Send + 'static,
    Fut: Future<Output = T> + Send + 'static,
    T: IntoOutcome + Send + 'static,
{
    tokio::spawn(async move {
        let mut backoff = spec.backoff.clone();

        loop {
            if *shutdown.borrow() {
                health.set_state(TaskState::Stopped);
                return;
            }

            health.set_state(TaskState::Running);
            let started = Instant::now();
            let mut inner = tokio::spawn(factory());

            // Two-step so `inner` is not borrowed by the `select!` any more by
            // the time the shutdown arm needs it for the bounded drain.
            let step = tokio::select! {
                joined = &mut inner => Step::Finished(joined),
                _ = shutdown.wait_for(|stop| *stop) => Step::Shutdown,
            };

            let joined = match step {
                Step::Shutdown => {
                    // The inner task observes the SAME watch channel, so the
                    // normal case is that it is already draining. Abort only if
                    // it is genuinely wedged.
                    if tokio::time::timeout(SHUTDOWN_DRAIN, &mut inner)
                        .await
                        .is_err()
                    {
                        tracing::warn!(
                            task = spec.name,
                            "supervised task did not stop within the drain window — aborting it"
                        );
                        inner.abort();
                    }
                    health.set_state(TaskState::Stopped);
                    return;
                }
                Step::Finished(joined) => joined,
            };

            // A run long enough to count as healthy clears the streak and the
            // backoff, so an hourly crash never creeps up to the 60 s ceiling.
            if started.elapsed() >= HEALTHY_AFTER {
                backoff.reset();
                if health.note_healthy() {
                    tracing::info!(
                        task = spec.name,
                        "supervised task recovered and is running normally again"
                    );
                }
            }

            let outcome = match joined {
                Ok(value) => value.into_outcome(),
                Err(e) if e.is_panic() => TaskOutcome::Failed {
                    code: Some(ERR_TASK_PANICKED),
                    error: panic_message(e),
                },
                // Cancelled: something aborted the inner handle. Only the
                // shutdown arm above does that, so treat it as a clean stop.
                Err(_) => {
                    health.set_state(TaskState::Stopped);
                    return;
                }
            };

            // A task that exits cleanly *because* shutdown was signalled while
            // it was running must not be restarted — re-check before deciding.
            if *shutdown.borrow() {
                health.set_state(TaskState::Stopped);
                return;
            }

            // Never → stop; OnFailure → stop only on a clean completion;
            // Always → come back regardless of how the run ended.
            let restart = !matches!(
                (&outcome, spec.restart),
                (_, RestartPolicy::Never) | (TaskOutcome::Completed, RestartPolicy::OnFailure)
            );

            match &outcome {
                TaskOutcome::Completed => {
                    tracing::debug!(
                        task = spec.name,
                        restart,
                        "supervised task completed normally"
                    );
                }
                TaskOutcome::Failed { code, error } => {
                    let code = code.unwrap_or(ERR_TASK_PANICKED);
                    let failures = health.note_failure(error);
                    if failures == 1 {
                        tracing::error!(
                            code,
                            task = spec.name,
                            error = %error,
                            restart,
                            "supervised task failed — suppressing further warnings until recovery"
                        );
                    } else if failures == RESTART_LIMIT_WARN_AT {
                        tracing::warn!(
                            code = ERR_TASK_RESTART_LIMIT,
                            task = spec.name,
                            consecutive_failures = failures,
                            error = %error,
                            "supervised task is failing persistently; still retrying"
                        );
                    } else {
                        tracing::debug!(
                            code,
                            task = spec.name,
                            consecutive_failures = failures,
                            "supervised task failed again during a degraded streak"
                        );
                    }
                }
            }

            if !restart {
                health.set_state(match outcome {
                    TaskOutcome::Completed => TaskState::Stopped,
                    TaskOutcome::Failed { .. } => TaskState::Failed,
                });
                return;
            }

            health.set_state(TaskState::Restarting);
            health.record_restart();

            let delay = backoff.next_delay();
            tokio::select! {
                _ = tokio::time::sleep(delay) => {}
                _ = shutdown.wait_for(|stop| *stop) => {
                    health.set_state(TaskState::Stopped);
                    return;
                }
            }
        }
    })
}

/// Which arm of the supervisor's `select!` fired. Exists so the shutdown arm
/// can take `&mut inner` for the bounded drain *after* the select releases it.
enum Step<T> {
    Finished(Result<T, JoinError>),
    Shutdown,
}

/// Recover a printable message from a panic payload.
fn panic_message(e: JoinError) -> String {
    let payload = e.into_panic();
    if let Some(s) = payload.downcast_ref::<&'static str>() {
        (*s).to_string()
    } else if let Some(s) = payload.downcast_ref::<String>() {
        s.clone()
    } else {
        "task panicked with a non-string payload".to_string()
    }
}

/// Scrub an error string before it is stored on [`TaskHealth`].
///
/// Uses a module-local filter rather than `privacy::get_filter()`: that global
/// panics when the daemon has not initialised it, and a supervisor that panics
/// while recording someone else's panic is worse than no scrubbing at all.
fn scrub(message: &str) -> String {
    static FILTER: OnceLock<PrivacyFilter> = OnceLock::new();
    let filter = FILTER.get_or_init(|| PrivacyFilter::new(&[]));
    let mut value = serde_json::Value::String(message.to_string());
    crate::privacy::filter_value(&mut value, filter);
    match value {
        serde_json::Value::String(s) => s,
        _ => String::new(),
    }
}

fn now_unix() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::AtomicU32;

    fn fast_spec(name: &'static str, restart: RestartPolicy) -> TaskSpec {
        TaskSpec::new(name, restart).with_backoff(Backoff::new(
            Duration::from_millis(5),
            Duration::from_millis(20),
        ))
    }

    // -- Backoff -------------------------------------------------------------

    #[test]
    fn backoff_grows_and_caps_at_the_ceiling() {
        let mut b = Backoff::default();
        assert_eq!(b.base(), BACKOFF_INITIAL);

        let mut bases = vec![];
        for _ in 0..10 {
            bases.push(b.base());
            let delay = b.next_delay();
            // Jitter is ±20 % of the base, and the result is clamped to the
            // ceiling — so "caps at 60s" is literally true, not "caps at 72s".
            assert!(
                delay <= BACKOFF_MAX,
                "jittered delay {delay:?} exceeded cap"
            );
        }

        assert_eq!(
            bases[..4],
            [
                Duration::from_secs(1),
                Duration::from_secs(2),
                Duration::from_secs(4),
                Duration::from_secs(8)
            ]
        );
        assert_eq!(b.base(), BACKOFF_MAX, "base must saturate at the ceiling");
    }

    #[test]
    fn backoff_reset_returns_to_the_initial_delay() {
        let mut b = Backoff::default();
        for _ in 0..5 {
            b.next_delay();
        }
        assert!(b.base() > BACKOFF_INITIAL);
        b.reset();
        assert_eq!(b.base(), BACKOFF_INITIAL);
    }

    #[test]
    fn backoff_jitter_stays_within_twenty_percent() {
        let mut b = Backoff::new(Duration::from_secs(10), Duration::from_secs(600));
        let d = b.next_delay();
        assert!(
            d >= Duration::from_secs(8) && d <= Duration::from_secs(12),
            "delay {d:?} outside the ±20% band around 10s"
        );
    }

    // -- supervise -----------------------------------------------------------

    /// The headline property: a panicking subsystem does not stay dead.
    #[tokio::test]
    async fn panicking_task_is_restarted() {
        let registry = Arc::new(HealthRegistry::new());
        let (_tx, rx) = watch::channel(false);
        let runs = Arc::new(AtomicU32::new(0));
        let runs_for_task = runs.clone();

        let handle = spawn_supervised(
            &registry,
            fast_spec("panicky", RestartPolicy::OnFailure),
            rx,
            move || {
                let runs = runs_for_task.clone();
                async move {
                    if runs.fetch_add(1, Ordering::SeqCst) == 0 {
                        panic!("first run explodes");
                    }
                }
            },
        );

        // OnFailure: the panic restarts it, the clean second run ends the loop.
        tokio::time::timeout(Duration::from_secs(5), handle)
            .await
            .expect("supervisor did not settle")
            .expect("supervisor task itself must never panic");

        assert_eq!(runs.load(Ordering::SeqCst), 2, "task must have run twice");
        let health = &registry.tasks()[0];
        assert_eq!(health.restarts(), 1);
        assert_eq!(health.state(), TaskState::Stopped);
    }

    /// The panic must be visible on the health entry, not silently swallowed.
    #[tokio::test]
    async fn panic_is_recorded_with_its_message() {
        let registry = Arc::new(HealthRegistry::new());
        let (_tx, rx) = watch::channel(false);
        let runs = Arc::new(AtomicU32::new(0));
        let runs_for_task = runs.clone();

        let handle = spawn_supervised(
            &registry,
            fast_spec("recorded", RestartPolicy::OnFailure),
            rx,
            move || {
                let runs = runs_for_task.clone();
                async move {
                    if runs.fetch_add(1, Ordering::SeqCst) == 0 {
                        panic!("boom in the subsystem");
                    }
                }
            },
        );
        let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;

        let err = registry.tasks()[0]
            .last_error()
            .expect("panic message must be recorded");
        assert!(err.contains("boom in the subsystem"), "got {err}");
    }

    #[tokio::test]
    async fn never_policy_does_not_restart() {
        let registry = Arc::new(HealthRegistry::new());
        let (_tx, rx) = watch::channel(false);
        let runs = Arc::new(AtomicU32::new(0));
        let runs_for_task = runs.clone();

        // Turbofished: the body diverges, so the output type would otherwise
        // fall back to `!` (a hard error from edition 2024 on).
        let handle = spawn_supervised::<_, _, ()>(
            &registry,
            fast_spec("once", RestartPolicy::Never),
            rx,
            move || {
                let runs = runs_for_task.clone();
                async move {
                    runs.fetch_add(1, Ordering::SeqCst);
                    panic!("still not restarted");
                }
            },
        );
        tokio::time::timeout(Duration::from_secs(5), handle)
            .await
            .expect("supervisor did not settle")
            .expect("supervisor task itself must never panic");

        assert_eq!(runs.load(Ordering::SeqCst), 1);
        assert_eq!(registry.tasks()[0].restarts(), 0);
        assert_eq!(registry.tasks()[0].state(), TaskState::Failed);
    }

    /// The shutdown signal must end the loop without one more respawn — this is
    /// what lets `openlatch stop` bound its join instead of racing a restart.
    #[tokio::test]
    async fn shutdown_stops_the_loop_without_another_respawn() {
        let registry = Arc::new(HealthRegistry::new());
        let (tx, rx) = watch::channel(false);
        let runs = Arc::new(AtomicU32::new(0));
        let runs_for_task = runs.clone();

        let handle = spawn_supervised(
            &registry,
            fast_spec("looper", RestartPolicy::Always),
            rx.clone(),
            move || {
                let runs = runs_for_task.clone();
                let mut shutdown = rx.clone();
                async move {
                    runs.fetch_add(1, Ordering::SeqCst);
                    let _ = shutdown.wait_for(|stop| *stop).await;
                }
            },
        );

        tokio::time::sleep(Duration::from_millis(50)).await;
        let before = runs.load(Ordering::SeqCst);
        tx.send(true).expect("shutdown send");

        tokio::time::timeout(Duration::from_secs(5), handle)
            .await
            .expect("supervisor did not stop on shutdown")
            .expect("supervisor task itself must never panic");

        assert_eq!(
            runs.load(Ordering::SeqCst),
            before,
            "no respawn may happen after the shutdown signal"
        );
        assert_eq!(registry.tasks()[0].state(), TaskState::Stopped);
    }

    /// `Always` keeps a repeatedly-failing subsystem coming back forever — this
    /// is the boundary-listener bind-retry property, exercised without a socket.
    #[tokio::test]
    async fn always_policy_keeps_retrying_a_failing_task() {
        let registry = Arc::new(HealthRegistry::new());
        let (tx, rx) = watch::channel(false);
        let runs = Arc::new(AtomicU32::new(0));
        let runs_for_task = runs.clone();

        let handle = spawn_supervised(
            &registry,
            fast_spec("flapper", RestartPolicy::Always),
            rx,
            move || {
                let runs = runs_for_task.clone();
                async move {
                    runs.fetch_add(1, Ordering::SeqCst);
                    Err::<(), OlError>(OlError::new("OL-9999", "still broken"))
                }
            },
        );

        // 5ms→10ms→20ms backoff: several attempts land inside a second.
        tokio::time::sleep(Duration::from_millis(300)).await;
        assert!(
            runs.load(Ordering::SeqCst) >= 3,
            "Always must keep retrying, saw {} runs",
            runs.load(Ordering::SeqCst)
        );
        assert!(registry.is_degraded(), "a flapping Always task is degraded");

        tx.send(true).expect("shutdown send");
        let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
    }

    // -- registry ------------------------------------------------------------

    #[test]
    fn only_always_tasks_count_as_degraded() {
        let registry = HealthRegistry::new();
        let always = registry.register("boundary", RestartPolicy::Always);
        let oneshot = registry.register("update-check", RestartPolicy::OnFailure);

        // Fresh registrations sit in `Starting`, which is not `Running`.
        assert!(registry.is_degraded());
        assert_eq!(registry.degraded_names(), vec!["boundary"]);

        always.set_state(TaskState::Running);
        oneshot.set_state(TaskState::Stopped);
        assert!(!registry.is_degraded());
        assert_eq!(registry.degraded_count(), 0);
    }

    #[test]
    fn subsystems_json_reports_state_and_restarts() {
        let registry = HealthRegistry::new();
        let h = registry.register("cloud-worker", RestartPolicy::Always);
        h.set_state(TaskState::Running);
        h.record_restart();
        h.record_restart();

        let json = registry.subsystems_json();
        assert_eq!(json["cloud-worker"]["state"], "running");
        assert_eq!(json["cloud-worker"]["restarts"], 2);
        assert!(
            json["cloud-worker"].get("last_error").is_none(),
            "a healthy subsystem must not carry a last_error key"
        );
        assert_eq!(registry.total_restarts(), 2);
    }

    #[test]
    fn recorded_errors_are_scrubbed() {
        let health = TaskHealth::new("leaky", RestartPolicy::Always);
        health.note_failure("bind failed for Bearer sk-ant-api03-DEADBEEFDEADBEEFDEADBEEF");
        let stored = health.last_error().expect("error stored");
        assert!(
            !stored.contains("DEADBEEFDEADBEEF"),
            "credential must be masked before storage, got {stored}"
        );
    }

    #[test]
    fn healthy_run_clears_the_failure_streak_once() {
        let health = TaskHealth::new("streaky", RestartPolicy::Always);
        assert_eq!(health.note_failure("a"), 1);
        assert_eq!(health.note_failure("b"), 2);
        assert!(health.note_healthy(), "first clear reports the recovery");
        assert!(!health.note_healthy(), "a second clear reports nothing");
        assert_eq!(health.consecutive_failures(), 0);
        assert!(health.last_error().is_none());
    }
}