agent-file-tools 0.50.0

//! Dispatch-path metrics and health-report helpers for the subc transport loop.

use std::path::Path;
use std::time::{SystemTime, UNIX_EPOCH};

use super::{
    json, Arc, AtomicBool, AtomicU64, AtomicUsize, Duration, Executor, HashMap, HealthReport,
    HealthStatus, Instant, Ordering, PendingBind, ProjectRootId, RootHealthSnapshot, RouteChannel,
    StdMutex, Value, DISPATCH_PATH_BIND_WARN_AFTER, WRITER_QUEUE_CAPACITY,
};
use crate::context::{App, AppContext, RootHealthState};
use crate::executor::BindBlockerSnapshot;

#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(super) struct ReapBlockerCensus {
    pub(super) deleted_retained: usize,
    pub(super) absence_unconfirmed: usize,
    pub(super) bound_routes: usize,
    pub(super) unbound_quiesced: usize,
    pub(super) bash_waits: usize,
    pub(super) maintenance_pending: usize,
    pub(super) maintenance_queued: usize,
    pub(super) pending_binds: usize,
    pub(super) actor_busy: usize,
    pub(super) actor_state_busy: usize,
    pub(super) artifact_eviction_blocked: usize,
    pub(super) artifact_eviction_failed: usize,
}

impl ReapBlockerCensus {
    pub(super) fn blocker_histogram(&self) -> String {
        format!(
            "absence_unconfirmed={},bound_routes={},unbound_quiesced={},bash_waits={},maintenance_pending={},maintenance_queued={},pending_binds={},actor_busy={},actor_state_busy={},artifact_eviction_blocked={},artifact_eviction_failed={}",
            self.absence_unconfirmed,
            self.bound_routes,
            self.unbound_quiesced,
            self.bash_waits,
            self.maintenance_pending,
            self.maintenance_queued,
            self.pending_binds,
            self.actor_busy,
            self.actor_state_busy,
            self.artifact_eviction_blocked,
            self.artifact_eviction_failed,
        )
    }
}

struct ReapMetrics {
    last_sweep_ms: AtomicU64,
    deleted_retained: AtomicUsize,
    absence_unconfirmed: AtomicUsize,
    bound_routes: AtomicUsize,
    unbound_quiesced: AtomicUsize,
    bash_waits: AtomicUsize,
    maintenance_pending: AtomicUsize,
    maintenance_queued: AtomicUsize,
    pending_binds: AtomicUsize,
    actor_busy: AtomicUsize,
    actor_state_busy: AtomicUsize,
    artifact_eviction_blocked: AtomicUsize,
    artifact_eviction_failed: AtomicUsize,
}

impl ReapMetrics {
    fn new() -> Self {
        Self {
            last_sweep_ms: AtomicU64::new(0),
            deleted_retained: AtomicUsize::new(0),
            absence_unconfirmed: AtomicUsize::new(0),
            bound_routes: AtomicUsize::new(0),
            unbound_quiesced: AtomicUsize::new(0),
            bash_waits: AtomicUsize::new(0),
            maintenance_pending: AtomicUsize::new(0),
            maintenance_queued: AtomicUsize::new(0),
            pending_binds: AtomicUsize::new(0),
            actor_busy: AtomicUsize::new(0),
            actor_state_busy: AtomicUsize::new(0),
            artifact_eviction_blocked: AtomicUsize::new(0),
            artifact_eviction_failed: AtomicUsize::new(0),
        }
    }

    fn record(&self, now_ms: u64, census: ReapBlockerCensus) {
        self.last_sweep_ms.store(now_ms, Ordering::Relaxed);
        self.deleted_retained
            .store(census.deleted_retained, Ordering::Relaxed);
        self.absence_unconfirmed
            .store(census.absence_unconfirmed, Ordering::Relaxed);
        self.bound_routes
            .store(census.bound_routes, Ordering::Relaxed);
        self.unbound_quiesced
            .store(census.unbound_quiesced, Ordering::Relaxed);
        self.bash_waits.store(census.bash_waits, Ordering::Relaxed);
        self.maintenance_pending
            .store(census.maintenance_pending, Ordering::Relaxed);
        self.maintenance_queued
            .store(census.maintenance_queued, Ordering::Relaxed);
        self.pending_binds
            .store(census.pending_binds, Ordering::Relaxed);
        self.actor_busy.store(census.actor_busy, Ordering::Relaxed);
        self.actor_state_busy
            .store(census.actor_state_busy, Ordering::Relaxed);
        self.artifact_eviction_blocked
            .store(census.artifact_eviction_blocked, Ordering::Relaxed);
        self.artifact_eviction_failed
            .store(census.artifact_eviction_failed, Ordering::Relaxed);
    }

    fn snapshot(&self) -> Value {
        json!({
            "deleted_retained": self.deleted_retained.load(Ordering::Relaxed),
            "blockers": {
                "absence_unconfirmed": self.absence_unconfirmed.load(Ordering::Relaxed),
                "bound_routes": self.bound_routes.load(Ordering::Relaxed),
                "unbound_quiesced": self.unbound_quiesced.load(Ordering::Relaxed),
                "bash_waits": self.bash_waits.load(Ordering::Relaxed),
                "maintenance_pending": self.maintenance_pending.load(Ordering::Relaxed),
                "maintenance_queued": self.maintenance_queued.load(Ordering::Relaxed),
                "pending_binds": self.pending_binds.load(Ordering::Relaxed),
                "actor_busy": self.actor_busy.load(Ordering::Relaxed),
                "actor_state_busy": self.actor_state_busy.load(Ordering::Relaxed),
                "artifact_eviction_blocked": self.artifact_eviction_blocked.load(Ordering::Relaxed),
                "artifact_eviction_failed": self.artifact_eviction_failed.load(Ordering::Relaxed),
            },
            "last_sweep_ms": self.last_sweep_ms.load(Ordering::Relaxed),
        })
    }
}

const BG_OBSERVABILITY_INTERVAL: Duration = Duration::from_secs(60);

#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
enum BgEventKind {
    ArmHit,
    ArmMiss,
    NudgeEnqueued,
    SubscriptionInstalled,
    SubscriptionEnded,
}

#[derive(Debug, PartialEq, Eq, Hash)]
struct BgEventKey {
    root: ProjectRootId,
    session: String,
    kind: BgEventKind,
}

struct BgEventRecord {
    window_start: Instant,
    event_count: u64,
    suppressed: u64,
}

#[cfg(test)]
thread_local! {
    static BG_OBSERVABILITY_TEST_LOGS: std::cell::RefCell<Vec<String>> = const { std::cell::RefCell::new(Vec::new()) };
}

fn emit_bg_observability_info(line: String) {
    log::info!("{line}");
    #[cfg(test)]
    BG_OBSERVABILITY_TEST_LOGS.with(|logs| logs.borrow_mut().push(line));
}

#[cfg(test)]
pub(super) fn take_bg_observability_logs_for_test() -> Vec<String> {
    BG_OBSERVABILITY_TEST_LOGS.with(|logs| std::mem::take(&mut *logs.borrow_mut()))
}

pub(super) struct DispatchPathMetrics {
    pub(super) origin: Instant,
    pub(super) frame_loop_last_tick_ms: AtomicU64,
    pub(super) writer_queued: AtomicUsize,
    pub(super) writer_active: AtomicBool,
    pub(super) writer_saturation_count: AtomicU64,
    pub(super) control_completion_queued: AtomicUsize,
    pub(super) maintenance_queued: AtomicUsize,
    pub(super) bash_deferred_queued: AtomicUsize,
    pub(super) bash_poll_touch_queued: AtomicUsize,
    pub(super) reliable_push_budget_deferrals: AtomicU64,
    pub(super) maintenance_budget_deferrals: AtomicU64,
    pub(super) response_tasks_live: AtomicUsize,
    bg_subscriptions: AtomicUsize,
    bg_wake_pending: AtomicUsize,
    bg_events: StdMutex<HashMap<BgEventKey, BgEventRecord>>,
    reap: ReapMetrics,
}

impl DispatchPathMetrics {
    pub(super) fn new() -> Self {
        Self {
            origin: Instant::now(),
            frame_loop_last_tick_ms: AtomicU64::new(0),
            writer_queued: AtomicUsize::new(0),
            writer_active: AtomicBool::new(false),
            writer_saturation_count: AtomicU64::new(0),
            control_completion_queued: AtomicUsize::new(0),
            maintenance_queued: AtomicUsize::new(0),
            bash_deferred_queued: AtomicUsize::new(0),
            bash_poll_touch_queued: AtomicUsize::new(0),
            reliable_push_budget_deferrals: AtomicU64::new(0),
            maintenance_budget_deferrals: AtomicU64::new(0),
            response_tasks_live: AtomicUsize::new(0),
            bg_subscriptions: AtomicUsize::new(0),
            bg_wake_pending: AtomicUsize::new(0),
            bg_events: StdMutex::new(HashMap::new()),
            reap: ReapMetrics::new(),
        }
    }

    fn now_ms(&self) -> u64 {
        duration_millis_u64(self.origin.elapsed())
    }

    pub(super) fn mark_frame_loop_tick(&self) {
        self.frame_loop_last_tick_ms
            .store(self.now_ms(), Ordering::Relaxed);
    }

    pub(super) fn record_reap(&self, census: ReapBlockerCensus) {
        let last_sweep_ms = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(duration_millis_u64)
            .unwrap_or(0);
        self.reap.record(last_sweep_ms, census);
    }

    pub(super) fn record_bg_runtime(&self, subscriptions: usize, wake_pending: usize) {
        self.bg_subscriptions
            .store(subscriptions, Ordering::Relaxed);
        self.bg_wake_pending.store(wake_pending, Ordering::Relaxed);
    }

    fn record_bg_event_at(
        &self,
        root: &ProjectRootId,
        session: &str,
        kind: BgEventKind,
        now: Instant,
    ) -> Option<u64> {
        let Ok(mut events) = self.bg_events.lock() else {
            return None;
        };
        let key = BgEventKey {
            root: root.clone(),
            session: session.to_string(),
            kind,
        };
        if let Some(record) = events.get_mut(&key) {
            if now.saturating_duration_since(record.window_start) < BG_OBSERVABILITY_INTERVAL {
                record.event_count = record.event_count.saturating_add(1);
                record.suppressed = record.suppressed.saturating_add(1);
                return None;
            }
            let suppressed = record.suppressed;
            *record = BgEventRecord {
                window_start: now,
                event_count: 1,
                suppressed: 0,
            };
            return Some(suppressed);
        }
        events.insert(
            key,
            BgEventRecord {
                window_start: now,
                event_count: 1,
                suppressed: 0,
            },
        );
        Some(0)
    }

    fn bg_event_count_60s(&self, kind: BgEventKind) -> u64 {
        let Ok(events) = self.bg_events.try_lock() else {
            return 0;
        };
        let now = Instant::now();
        events
            .iter()
            .filter(|(key, record)| {
                key.kind == kind
                    && now.saturating_duration_since(record.window_start)
                        < BG_OBSERVABILITY_INTERVAL
            })
            .map(|(_, record)| record.event_count)
            .sum()
    }

    pub(super) fn bg_arm_misses_60s_total(&self) -> u64 {
        self.bg_event_count_60s(BgEventKind::ArmMiss)
    }

    fn bg_nudges_enqueued_60s_total(&self) -> u64 {
        self.bg_event_count_60s(BgEventKind::NudgeEnqueued)
    }

    pub(super) fn record_bg_arm_hit(
        &self,
        root: &ProjectRootId,
        session: &str,
        channel: RouteChannel,
    ) {
        if let Some(suppressed) =
            self.record_bg_event_at(root, session, BgEventKind::ArmHit, Instant::now())
        {
            emit_bg_observability_info(format!(
                "subc bg wake: arm HIT root={} session={} channel={} suppressed={suppressed}",
                root.as_path().display(),
                session,
                channel
            ));
        }
    }

    pub(super) fn record_bg_arm_miss(
        &self,
        root: &ProjectRootId,
        session: &str,
        live_root_subscriptions: usize,
    ) {
        if let Some(suppressed) =
            self.record_bg_event_at(root, session, BgEventKind::ArmMiss, Instant::now())
        {
            emit_bg_observability_info(format!(
                "subc bg wake: arm MISS root={} session={} live_root_subscriptions={} suppressed={suppressed}",
                root.as_path().display(),
                session,
                live_root_subscriptions
            ));
        }
    }

    pub(super) fn record_bg_nudge_enqueued(
        &self,
        root: &ProjectRootId,
        session: &str,
        channel: RouteChannel,
    ) {
        if let Some(suppressed) =
            self.record_bg_event_at(root, session, BgEventKind::NudgeEnqueued, Instant::now())
        {
            emit_bg_observability_info(format!(
                "subc bg wake: nudge enqueued root={} session={} channel={} suppressed={suppressed}",
                root.as_path().display(),
                session,
                channel
            ));
        }
    }

    pub(super) fn record_bg_subscription_installed(
        &self,
        root: &ProjectRootId,
        session: &str,
        channel: RouteChannel,
    ) {
        if let Some(suppressed) = self.record_bg_event_at(
            root,
            session,
            BgEventKind::SubscriptionInstalled,
            Instant::now(),
        ) {
            emit_bg_observability_info(format!(
                "subc bg subscription: installed root={} session={} channel={} cause=subscribe suppressed={suppressed}",
                root.as_path().display(),
                session,
                channel
            ));
        }
    }

    pub(super) fn record_bg_subscription_ended(
        &self,
        root: &ProjectRootId,
        session: &str,
        channel: RouteChannel,
        cause: &str,
    ) {
        if let Some(suppressed) = self.record_bg_event_at(
            root,
            session,
            BgEventKind::SubscriptionEnded,
            Instant::now(),
        ) {
            emit_bg_observability_info(format!(
                "subc bg subscription: ended root={} session={} channel={} cause={cause} suppressed={suppressed}",
                root.as_path().display(),
                session,
                channel
            ));
        }
    }

    fn reap_snapshot(&self) -> Value {
        self.reap.snapshot()
    }

    fn snapshot(
        &self,
        pending_binds: &HashMap<RouteChannel, PendingBind>,
        executor: &Executor,
    ) -> Value {
        let now = Instant::now();
        let oldest_pending_age_ms = pending_binds
            .values()
            .map(|bind| duration_millis_u64(now.saturating_duration_since(bind.started_at)))
            .max();
        let last_tick_ms = self.frame_loop_last_tick_ms.load(Ordering::Relaxed);
        json!({
            "frame_loop": {
                "last_tick_age_ms": self.now_ms().saturating_sub(last_tick_ms),
            },
            "pending_binds": {
                "count": pending_binds.len(),
                "oldest_age_ms": oldest_pending_age_ms,
            },
            "completion_channels": {
                "control": self.control_completion_queued.load(Ordering::Relaxed),
                "maintenance": self.maintenance_queued.load(Ordering::Relaxed),
                "bash_deferred": self.bash_deferred_queued.load(Ordering::Relaxed),
                "bash_poll_touch": self.bash_poll_touch_queued.load(Ordering::Relaxed),
            },
            "budget_deferrals": {
                "reliable_push": self.reliable_push_budget_deferrals.load(Ordering::Relaxed),
                "maintenance": self.maintenance_budget_deferrals.load(Ordering::Relaxed),
            },
            "writer": {
                "queued": self.writer_queued.load(Ordering::Relaxed),
                "active": self.writer_active.load(Ordering::Relaxed),
                "capacity": WRITER_QUEUE_CAPACITY,
                "saturation_count": self.writer_saturation_count.load(Ordering::Relaxed),
            },
            "response_tasks": {
                "live": self.response_tasks_live.load(Ordering::Relaxed),
            },
            "mutating_lanes": mutating_lanes_metrics(executor),
        })
    }
}

pub(super) struct ResponseTaskGuard {
    metrics: Arc<DispatchPathMetrics>,
}

impl ResponseTaskGuard {
    pub(super) fn new(metrics: &Arc<DispatchPathMetrics>) -> Self {
        metrics.response_tasks_live.fetch_add(1, Ordering::Relaxed);
        Self {
            metrics: Arc::clone(metrics),
        }
    }
}

impl Drop for ResponseTaskGuard {
    fn drop(&mut self) {
        self.metrics
            .response_tasks_live
            .fetch_sub(1, Ordering::Relaxed);
    }
}

fn duration_millis_u64(duration: Duration) -> u64 {
    duration.as_millis().min(u128::from(u64::MAX)) as u64
}

pub(super) fn warn_slow_pending_binds(
    pending_binds: &mut HashMap<RouteChannel, PendingBind>,
    executor: &Executor,
) {
    let now = Instant::now();
    for (route, pending) in pending_binds.iter_mut() {
        if pending.warned_half_deadline {
            continue;
        }
        let age = now.saturating_duration_since(pending.started_at);
        if age < DISPATCH_PATH_BIND_WARN_AFTER {
            continue;
        }
        pending.warned_half_deadline = true;
        let snapshot = executor
            .try_bind_blocker_snapshot(&pending.bind_root_id, &pending.configure_request_id)
            .unwrap_or_else(|| BindBlockerSnapshot {
                configure_state: "scheduler_busy",
                configure_phase_timings: None,
                blockers: vec!["scheduler_busy".to_string()],
            });
        crate::slog_warn!(
            "{}",
            pending_bind_breadcrumb(
                *route,
                &pending.bind_root_id,
                age,
                &pending.configure_request_id,
                &snapshot,
            )
        );
    }
}

fn pending_bind_breadcrumb(
    route: RouteChannel,
    root_id: &crate::path_identity::ProjectRootId,
    age: Duration,
    configure_request_id: &str,
    snapshot: &BindBlockerSnapshot,
) -> String {
    let blockers = if snapshot.blockers.is_empty() {
        "none".to_string()
    } else {
        snapshot.blockers.join(", ")
    };
    let phase_timings = snapshot
        .configure_phase_timings
        .as_deref()
        .unwrap_or("unavailable");
    format!(
        "subc attach: pending RouteBind route {route} for root {} crossed {}ms (configure_request_id={}, configure_state={}, configure_phase_timings=[{}], blockers=[{}])",
        root_id.as_path().display(),
        duration_millis_u64(age),
        configure_request_id,
        snapshot.configure_state,
        phase_timings,
        blockers,
    )
}

/// Per-root attributed-bytes rollup for health metrics: top roots by size
/// (same cap as the status payload's memory section) plus process RSS and
/// attributed totals. `None` scheduler entries (contended) yield a busy
/// marker instead of waiting.
fn memory_rollup_metrics(
    actor_entries: Option<Vec<(cortexkit_paths::ProjectRootId, Arc<AppContext>)>>,
) -> Value {
    let Some(entries) = actor_entries else {
        return json!({
            "status": "busy",
            "allocator_slack_bytes": 0,
            "allocator_slack_measured": false,
        });
    };
    let mut roots = std::collections::BTreeMap::new();
    for (root_id, ctx) in entries {
        roots.insert(
            root_id.as_path().display().to_string(),
            ctx.memory_root_snapshot(),
        );
    }
    let snapshot = crate::memory::MemorySnapshot::new("ready", roots);
    let per_root: Value = snapshot
        .roots
        .iter()
        .map(|(root, detail)| {
            (
                root.clone(),
                json!({
                    "attributed_bytes": detail.attributed_bytes,
                    "status": detail.status,
                }),
            )
        })
        .collect::<serde_json::Map<String, Value>>()
        .into();
    json!({
        "status": "ready",
        "roots": per_root,
        "roots_total": snapshot.roots_total,
        "roots_omitted": snapshot.roots_omitted,
        "roots_omitted_bytes": snapshot.roots_omitted_bytes,
        "rss_bytes": snapshot.process.rss_bytes,
        // Zero means either measured zero slack or unavailable allocator counters;
        // the sibling boolean disambiguates "no slack" from "unmeasurable".
        "allocator_slack_bytes": snapshot.process.allocator.retained_slack_bytes.unwrap_or(0),
        "allocator_slack_measured": snapshot.process.allocator.retained_slack_bytes.is_some(),
        // Headline number: excludes reclaimable pages RSS still counts.
        "phys_footprint_bytes": snapshot.process.phys_footprint_bytes,
        "total_attributed_bytes": snapshot.process.total_attributed_bytes,
        "sqlite_bytes": snapshot.process.sqlite.memory_used_bytes,
    })
}

fn mutating_lanes_metrics(executor: &Executor) -> Value {
    match executor.try_mutating_lane_snapshots() {
        Some(snapshots) => Value::Array(
            snapshots
                .into_iter()
                .map(|snapshot| {
                    json!({
                        "root": snapshot.root_id.as_path().to_string_lossy(),
                        "request_id": snapshot.request_id,
                        "job": snapshot.command,
                        "started_age_ms": snapshot.started_age_ms,
                    })
                })
                .collect(),
        ),
        None => json!({ "scheduler_busy": true }),
    }
}

fn dispatch_liveness_metrics(executor: &Executor) -> Value {
    match executor.try_dispatch_liveness_snapshot() {
        Some(snapshot) => json!({
            "interactive": {
                "queued": snapshot.interactive.queued,
                "oldest_age_ms": snapshot.interactive.oldest_age_ms,
            },
            "maintenance": {
                "queued": snapshot.maintenance.queued,
                "oldest_age_ms": snapshot.maintenance.oldest_age_ms,
            },
            "running": {
                "interactive": snapshot.running.interactive,
                "maintenance": snapshot.running.maintenance,
            },
            "interactive_reserve": snapshot.interactive_reserve,
            "maintenance_cap": snapshot.maintenance_cap,
        }),
        None => json!({ "scheduler_busy": true }),
    }
}

pub(super) fn build_health_report(
    executor: &Executor,
    pending_binds: &HashMap<RouteChannel, PendingBind>,
    dispatch_path_metrics: &DispatchPathMetrics,
    shared_app: &App,
) -> HealthReport {
    // Health replies must stay cheap under any load (subc-health rule: a
    // probe that queues behind busy state lies about liveness and gets the
    // module health-killed). Every read below is try-lock-only: a contended
    // scheduler lock produces a degraded report with an empty root list instead
    // of waiting.
    let actor_entries = executor.try_actor_entries();
    let scheduler_busy = actor_entries.is_none();
    let actor_entries_for_memory = actor_entries.clone();
    let mut roots: Vec<RootHealthSnapshot> = actor_entries
        .unwrap_or_default()
        .into_iter()
        .map(|(root_id, ctx)| ctx.try_health_snapshot(root_id.as_path()))
        .collect();
    roots.sort_by(|left, right| left.project_root.cmp(&right.project_root));
    let callgraph_repair_entries_60s_total = crate::callgraph_store::repair_entry_rate_total();
    let callgraph_write_metrics_total = crate::callgraph_store::callgraph_write_metrics_total();
    let mut repair_roots_annotated = 0usize;
    for root in &mut roots {
        // Memo-only, never a git probe: artifact_cache_key() spawns a git
        // subprocess per unmemoized root, and this loop runs on the channel-0
        // reply path across every live root. On a host with exec stalls — the
        // exact fault health probes ride through — a spawn loop here pushes
        // the reply past the supervisor deadline and the module is killed as
        // unresponsive (2026-08-08 second outage). An unmemoized root simply
        // loses its per-root repair annotation; the process-wide total above
        // does not depend on cache keys.
        let Some(project_key) =
            crate::search_index::artifact_cache_key_memoized_only(Path::new(&root.project_root))
        else {
            continue;
        };
        repair_roots_annotated += 1;
        if let Some((count, _window_start)) =
            crate::callgraph_store::repair_entry_rate(&project_key)
        {
            if count > 0 {
                root.callgraph_repair_entries_60s = Some(count);
            }
        }
    }
    // Absent-vs-zero disambiguation for the per-root repair annotation: a
    // root without a derived cache key is NOT MEASURED, which must render
    // differently from "measured, zero". The denominator pair makes the
    // coverage visible instead of letting absence read as health.
    let repair_roots_total = roots.len();

    // Compact memory rollup for operator drill-down (`ck health aft`): per-root
    // attributed bytes with the same top-N cap as the status payload, plus
    // process totals. Memory estimates are themselves try-lock-only (busy
    // subsystems report as such), so this stays within the health-path lock
    // doctrine.
    let memory = memory_rollup_metrics(actor_entries_for_memory);

    // Health-verdict rule: DEGRADED means dispatch is impaired (an actor we
    // could not even snapshot without contention), never "a background index
    // is still warming". A serving root with search/callgraph mid-build is
    // healthy — component build states are informational detail, otherwise a
    // module with any active mason worktree reads permanently degraded and
    // the daemon's on-failing policies treat routine warmup as wreckage.
    let busy_roots = roots
        .iter()
        .filter(|root| matches!(root.state, RootHealthState::Busy))
        .count();
    let warming_roots = roots
        .iter()
        .filter(|root| !matches!(root.state, RootHealthState::Busy) && !root.is_fully_ready())
        .count();
    let lsp_children = shared_app.lsp_child_registry().try_health_snapshot();

    let detail = if scheduler_busy {
        Some("executor scheduler state could not be snapshotted without contention".to_string())
    } else if busy_roots > 0 {
        Some(format!(
            "{busy_roots} root actor(s) could not be snapshotted without contention"
        ))
    } else if warming_roots > 0 {
        Some(format!(
            "{warming_roots} root(s) warming background indexes (serving normally)"
        ))
    } else {
        None
    };

    HealthReport {
        // Failing to acquire the scheduler snapshot is itself dispatch contention;
        // reporting Ok with an empty root list would hide the condition being probed.
        status: if scheduler_busy || busy_roots > 0 {
            HealthStatus::Degraded
        } else {
            HealthStatus::Ok
        },
        detail,
        metrics: Some(json!({
            "actor_count": roots.iter().map(|root| root.actor_count).sum::<usize>(),
            "root_count": roots.len(),
            "callgraph_repair_entries_60s_total": callgraph_repair_entries_60s_total,
            "callgraph_repair_roots_annotated": repair_roots_annotated,
            "callgraph_repair_roots_total": repair_roots_total,
            "callgraph_commits_60s_total": callgraph_write_metrics_total.commits_60s,
            "callgraph_pages_or_bytes_written_60s_total": callgraph_write_metrics_total.pages_or_bytes_written_60s,
            // Lifecycle audit counters (fleet leak tracking): process-wide
            // watcher runtimes, registered actor roots, and open routes.
            "runtime": {
                "live_watchers": shared_app.watcher_count(),
                "live_actor_roots": shared_app.actor_root_count(),
                "open_routes": shared_app.open_route_count(),
                "bg_subscriptions": dispatch_path_metrics.bg_subscriptions.load(Ordering::Relaxed),
                "bg_wake_pending": dispatch_path_metrics.bg_wake_pending.load(Ordering::Relaxed),
                "bg_nudges_enqueued_60s_total": dispatch_path_metrics.bg_nudges_enqueued_60s_total(),
                "bg_arm_misses_60s_total": dispatch_path_metrics.bg_arm_misses_60s_total(),
                "spawned_lsp_children": lsp_children.map(|health| health.spawned),
                "lsp_children_with_deleted_cwd": lsp_children.map(|health| health.cwd_gone),
            },
            "memory": memory,
            "roots": roots,
            "reap": dispatch_path_metrics.reap_snapshot(),
            "dispatch_liveness": dispatch_liveness_metrics(executor),
            "dispatch_path": dispatch_path_metrics.snapshot(pending_binds, executor),
        })),
    }
}

#[cfg(test)]
mod tests {
    use super::super::test_support::{test_ctx, test_root};
    use super::super::{Lane, Response};
    use super::*;
    use serde_json::json;

    #[test]
    fn bg_observability_rate_limit_reports_suppressed_count_and_lifecycle_lines() {
        let (_dir, root) = test_root("health-bg-observability-rate");
        let metrics = DispatchPathMetrics::new();
        let now = Instant::now();
        let session = "bg-health-session";

        assert_eq!(
            metrics.record_bg_event_at(&root, session, BgEventKind::ArmHit, now),
            Some(0)
        );
        assert_eq!(
            metrics.record_bg_event_at(
                &root,
                session,
                BgEventKind::ArmHit,
                now + Duration::from_secs(1),
            ),
            None
        );
        assert_eq!(
            metrics.record_bg_event_at(
                &root,
                session,
                BgEventKind::ArmHit,
                now + BG_OBSERVABILITY_INTERVAL,
            ),
            Some(1)
        );

        take_bg_observability_logs_for_test();
        let channel = super::super::route_key(21, 3);
        metrics.record_bg_subscription_installed(&root, session, channel);
        metrics.record_bg_subscription_ended(&root, session, channel, "goodbye");
        assert_eq!(
            take_bg_observability_logs_for_test(),
            vec![
                format!(
                    "subc bg subscription: installed root={} session=bg-health-session channel=21@3 cause=subscribe suppressed=0",
                    root.as_path().display()
                ),
                format!(
                    "subc bg subscription: ended root={} session=bg-health-session channel=21@3 cause=goodbye suppressed=0",
                    root.as_path().display()
                ),
            ]
        );
    }

    #[test]
    fn bg_runtime_health_fields_are_always_present() {
        let executor = Executor::new();
        let metrics = DispatchPathMetrics::new();
        let app = crate::context::App::default_shared();
        let (_dir, root) = test_root("health-bg-runtime-fields");
        let channel = super::super::route_key(22, 4);

        let cold = build_health_report(&executor, &HashMap::new(), &metrics, &app);
        let cold_metrics = cold.metrics.expect("cold health metrics");
        let cold_runtime = &cold_metrics["runtime"];
        assert_eq!(cold_runtime["bg_subscriptions"].as_u64(), Some(0));
        assert_eq!(cold_runtime["bg_wake_pending"].as_u64(), Some(0));
        assert_eq!(
            cold_runtime["bg_nudges_enqueued_60s_total"].as_u64(),
            Some(0)
        );
        assert_eq!(cold_runtime["bg_arm_misses_60s_total"].as_u64(), Some(0));

        metrics.record_bg_runtime(2, 1);
        metrics.record_bg_arm_miss(&root, "missing-session", 2);
        metrics.record_bg_nudge_enqueued(&root, "live-session", channel);

        let hot = build_health_report(&executor, &HashMap::new(), &metrics, &app);
        let hot_metrics = hot.metrics.expect("hot health metrics");
        let hot_runtime = &hot_metrics["runtime"];
        assert_eq!(hot_runtime["bg_subscriptions"].as_u64(), Some(2));
        assert_eq!(hot_runtime["bg_wake_pending"].as_u64(), Some(1));
        assert_eq!(
            hot_runtime["bg_nudges_enqueued_60s_total"].as_u64(),
            Some(1)
        );
        assert_eq!(hot_runtime["bg_arm_misses_60s_total"].as_u64(), Some(1));
    }

    #[test]
    fn pending_bind_breadcrumb_names_every_blocker_class() {
        let (_dir, root) = test_root("breadcrumb-blockers");
        let cases = [
            "queued_behind_configure(2)",
            "queued_behind_maintenance(job=subc-maintenance-drain-watcher lane=Mutating root=/tmp/a age_ms=1)",
            "waiting_on_readers",
            "idle_workers==0(job=subc-bind-other lane=Mutating root=/tmp/b age_ms=2)",
        ];

        for blocker in cases {
            let breadcrumb = pending_bind_breadcrumb(
                RouteChannel {
                    channel: 7,
                    epoch: 1,
                },
                &root,
                Duration::from_secs(6),
                "subc-bind-7",
                &BindBlockerSnapshot {
                    configure_state: "queued",
                    configure_phase_timings: Some("artifact_owner_claim=12ms".to_string()),
                    blockers: vec![blocker.to_string()],
                },
            );
            assert!(
                breadcrumb.contains(blocker),
                "breadcrumb omitted blocker class: {breadcrumb}"
            );
            assert!(
                breadcrumb.contains("configure_phase_timings=[artifact_owner_claim=12ms]"),
                "breadcrumb omitted configure phase timings: {breadcrumb}"
            );
        }
    }

    #[test]
    fn callgraph_repair_rate_is_always_present_and_hot_root_scoped() {
        let executor = Executor::with_config(crate::executor::ExecutorConfig {
            pool_size: 1,
            read_cap: 1,
            actor_cap: 1,
            heavy_permits: 1,
            drr_quantum: 1,
        });
        let (_dir, root) = test_root("health-callgraph-repair-rate");
        assert!(executor.register_actor(root.clone(), test_ctx()));
        let metrics = DispatchPathMetrics::new();
        let app = crate::context::App::default_shared();
        // The production annotation path reads the derivation map only
        // (never spawns git), so derive the key the way configure would —
        // artifact_cache_key records it as a side effect.
        let project_key = crate::search_index::artifact_cache_key(root.as_path());

        let quiet = build_health_report(&executor, &HashMap::new(), &metrics, &app);
        let quiet_metrics = quiet.metrics.expect("quiet health metrics");
        assert_eq!(
            quiet_metrics["callgraph_repair_entries_60s_total"].as_u64(),
            Some(0)
        );
        assert!(quiet_metrics["callgraph_commits_60s_total"].is_u64());
        assert!(quiet_metrics["callgraph_pages_or_bytes_written_60s_total"].is_u64());
        assert!(quiet_metrics["roots"][0]
            .get("callgraph_commits_60s")
            .is_none());
        assert!(quiet_metrics["roots"][0]
            .get("callgraph_repair_entries_60s")
            .is_none());

        for _ in 0..3 {
            crate::callgraph_store::note_repair_entry(&project_key);
        }
        let hot = build_health_report(&executor, &HashMap::new(), &metrics, &app);
        let hot_metrics = hot.metrics.expect("hot health metrics");
        assert_eq!(
            hot_metrics["callgraph_repair_entries_60s_total"].as_u64(),
            Some(3)
        );
        assert_eq!(
            hot_metrics["roots"][0]["callgraph_repair_entries_60s"].as_u64(),
            Some(3)
        );

        crate::callgraph_store::expire_repair_entry_window_for_test(&project_key);
        let decayed = build_health_report(&executor, &HashMap::new(), &metrics, &app);
        let decayed_metrics = decayed.metrics.expect("decayed health metrics");
        assert_eq!(
            decayed_metrics["callgraph_repair_entries_60s_total"].as_u64(),
            Some(0)
        );
        assert!(decayed_metrics["roots"][0]
            .get("callgraph_repair_entries_60s")
            .is_none());
    }

    #[test]
    fn health_report_includes_nonblocking_dispatch_liveness_for_queued_interactive() {
        let executor = Executor::with_config(crate::executor::ExecutorConfig {
            pool_size: 2,
            read_cap: 1,
            actor_cap: 1,
            heavy_permits: 2,
            drr_quantum: 1,
        });
        let (_dir_a, root_a) = test_root("health-liveness-a");
        let (_dir_b, root_b) = test_root("health-liveness-b");
        let (_dir_c, root_c) = test_root("health-liveness-c");
        executor.register_actor(root_a.clone(), test_ctx());
        executor.register_actor(root_b.clone(), test_ctx());
        executor.register_actor(root_c.clone(), test_ctx());

        let (started_tx, started_rx) = crossbeam_channel::bounded(2);
        let (release_tx, release_rx) = crossbeam_channel::bounded(2);
        let mut blockers = Vec::new();
        for (index, root) in [root_a, root_b].into_iter().enumerate() {
            let started_tx = started_tx.clone();
            let release_rx = release_rx.clone();
            blockers.push(executor.submit(
                root,
                Lane::PureRead,
                format!("health-blocker-{index}"),
                Box::new(move |_| {
                    started_tx.send(index).expect("signal blocker start");
                    release_rx
                        .recv_timeout(Duration::from_secs(2))
                        .expect("release blocker");
                    Response::success(format!("blocker-{index}"), json!({ "ok": true }))
                }),
            ));
        }
        for _ in 0..2 {
            started_rx
                .recv_timeout(Duration::from_secs(1))
                .expect("blocker starts");
        }

        let queued = executor.submit(
            root_c,
            Lane::PureRead,
            "queued-interactive".to_string(),
            Box::new(|_| Response::success("queued-interactive", json!({ "ok": true }))),
        );
        std::thread::sleep(Duration::from_millis(75));

        let metrics = DispatchPathMetrics::new();
        let pending_binds = HashMap::new();
        let report = build_health_report(
            &executor,
            &pending_binds,
            &metrics,
            &crate::context::App::default_shared(),
        );
        let dispatch = report
            .metrics
            .as_ref()
            .and_then(|metrics| metrics.get("dispatch_liveness"))
            .expect("dispatch_liveness metric");
        assert_eq!(dispatch.get("scheduler_busy"), None);
        assert_eq!(dispatch["interactive"]["queued"].as_u64(), Some(1));
        assert!(dispatch["interactive"]["oldest_age_ms"].as_u64().is_some());

        for _ in 0..2 {
            release_tx.send(()).expect("release blocker");
        }
        for blocker in blockers {
            blocker
                .recv_timeout(Duration::from_secs(1))
                .expect("blocker completion response");
        }
        queued
            .recv_timeout(Duration::from_secs(1))
            .expect("queued completion response");
    }

    #[test]
    fn health_metrics_memory_rollup_reports_per_root_and_process_totals() {
        let executor = Executor::with_config(crate::executor::ExecutorConfig {
            pool_size: 1,
            read_cap: 1,
            actor_cap: 1,
            heavy_permits: 1,
            drr_quantum: 1,
        });
        let dispatch_path_metrics = Arc::new(DispatchPathMetrics::new());
        let app = crate::context::App::default_shared();
        let registry = app.lsp_child_registry();
        registry.track(std::process::id());
        let report = build_health_report(&executor, &HashMap::new(), &dispatch_path_metrics, &app);
        let metrics = report.metrics.expect("health metrics present");
        let memory = metrics.get("memory").expect("memory rollup present");
        // No actors registered: ready rollup with zero roots and process totals.
        assert_eq!(memory.get("status").and_then(Value::as_str), Some("ready"));
        assert_eq!(memory.get("roots_total").and_then(Value::as_u64), Some(0));
        assert!(memory.get("total_attributed_bytes").is_some());
        assert!(memory.get("rss_bytes").is_some());
        assert!(memory
            .get("allocator_slack_bytes")
            .is_some_and(Value::is_u64));
        assert!(memory
            .get("allocator_slack_measured")
            .is_some_and(Value::is_boolean));
        if memory
            .get("allocator_slack_measured")
            .and_then(Value::as_bool)
            .is_some_and(|measured| !measured)
        {
            assert_eq!(memory["allocator_slack_bytes"], 0);
        }
        // Lifecycle audit counters ride the same probe so operator drill-downs
        // and fleet leak checks read one surface.
        let runtime = metrics.get("runtime").expect("runtime counters present");
        for key in ["live_watchers", "live_actor_roots", "open_routes"] {
            assert!(
                runtime.get(key).and_then(Value::as_u64).is_some(),
                "runtime.{key} must be a number"
            );
        }
        assert_eq!(runtime["spawned_lsp_children"].as_u64(), Some(1));
        assert_eq!(runtime["lsp_children_with_deleted_cwd"].as_u64(), Some(0));

        let busy_memory = memory_rollup_metrics(None);
        assert_eq!(busy_memory["allocator_slack_bytes"].as_u64(), Some(0));
        assert_eq!(
            busy_memory["allocator_slack_measured"].as_bool(),
            Some(false)
        );
        registry.untrack(std::process::id());
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn linux_allocator_relief_smoke_keeps_health_fields_present() {
        let mut allocation = vec![0u8; 32 * 1024 * 1024];
        for byte in allocation.iter_mut().step_by(4096) {
            *byte = 1;
        }
        std::hint::black_box(&allocation);
        drop(allocation);
        let _relief = crate::memory::relieve_allocator_pressure();

        let executor = Executor::with_config(crate::executor::ExecutorConfig {
            pool_size: 1,
            read_cap: 1,
            actor_cap: 1,
            heavy_permits: 1,
            drr_quantum: 1,
        });
        let app = crate::context::App::default_shared();
        let report = build_health_report(
            &executor,
            &HashMap::new(),
            &DispatchPathMetrics::new(),
            &app,
        );
        let memory = report
            .metrics
            .expect("health metrics present")
            .get("memory")
            .cloned()
            .expect("memory rollup present");
        assert!(memory["allocator_slack_bytes"].is_u64());
        assert!(memory["allocator_slack_measured"].is_boolean());
    }

    #[test]
    fn health_snapshot_fast_fails_while_mutating_job_holds_component_lock() {
        let executor = Executor::with_config(crate::executor::ExecutorConfig {
            pool_size: 1,
            read_cap: 1,
            actor_cap: 1,
            heavy_permits: 1,
            drr_quantum: 1,
        });
        let (_dir, root) = test_root("health-mutating-lock");
        let ctx = test_ctx();
        executor.register_actor(root.clone(), Arc::clone(&ctx));
        let (started_tx, started_rx) = crossbeam_channel::bounded(1);
        let (release_tx, release_rx) = crossbeam_channel::bounded(1);
        let blocker = executor.submit(
            root,
            Lane::Mutating,
            "health-lock-blocker".to_string(),
            Box::new(move |ctx| {
                let _index = ctx
                    .search_index()
                    .write()
                    .unwrap_or_else(std::sync::PoisonError::into_inner);
                started_tx.send(()).expect("signal held health lock");
                release_rx.recv().expect("release held health lock");
                Response::success("health-lock-blocker", json!({ "ok": true }))
            }),
        );
        started_rx
            .recv_timeout(Duration::from_secs(1))
            .expect("mutating lock holder starts");

        let report = build_health_report(
            &executor,
            &HashMap::new(),
            &DispatchPathMetrics::new(),
            &crate::context::App::default_shared(),
        );

        // Settle the holder before asserting so a verdict failure cannot drop the release
        // sender and make the worker panic while the test unwinds.
        release_tx.send(()).expect("release held health lock");
        blocker
            .recv_timeout(Duration::from_secs(3))
            .expect("mutating lock holder completes");

        assert_eq!(report.status, HealthStatus::Degraded);
    }
}