car-server-core 0.52.1

//! Watch-only daemon wiring for deterministic self-healing detection.
//!
//! The subsystem reads evidence CAR already owns, runs the pure
//! `car-selfheal` detectors, and appends detections/operator dismissals/tick
//! summaries to one private JSONL ledger. A bounded, file-read-only source
//! probe may additionally render private local handoff documents under
//! CAR_HOME. It has no checkout write, off-machine filing, remediation, or
//! network path.

use car_eventlog::{Alert, AlertThresholds};
use car_selfheal::{
    agent_gave_up, agent_log_errors, capability_miss, metrics_alerts, recurring_tool_failure,
    AgentDetectorConfig, AgentLogDetectorConfig, Detection, DetectionKind, EventEvidence,
    EvidenceSource, Redactor, Severity, SupervisorAgentState, SupervisorSnapshot,
    ToolFailureConfig,
};
use chrono::{DateTime, Duration, Utc};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::io::{BufRead, BufReader, Read, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::sync::Mutex;

use crate::session::ServerState;

pub const DEFAULT_SELFHEAL_INTERVAL_SECS: u64 = 15 * 60;
pub const SELFHEAL_INTERVAL_ENV: &str = "CAR_SELFHEAL_INTERVAL_SECS";
const MAX_PAGE_SIZE: usize = 500;
const DEFAULT_PAGE_SIZE: usize = 100;
const ACTIVITY_TAIL_LINES: usize = 40;
const STDERR_TAIL_LINES: usize = 50;
const EXPECTED_ORIGIN: &str = "Parslee-ai/car";
const TRUST_TIER: &str = "trusted";

pub const DETECTOR_IDS: [&str; 5] = [
    car_selfheal::detectors::metrics::DETECTOR_ID,
    car_selfheal::detectors::agent::DETECTOR_ID,
    car_selfheal::detectors::agent_logs::DETECTOR_ID,
    car_selfheal::detectors::tools::DETECTOR_ID,
    car_selfheal::detectors::capability::DETECTOR_ID,
];

/// Bounded filesystem inputs for source-checkout detection. The standalone
/// daemon builds this once from its explicit CAR_HOME config, running-binary
/// ancestors, and `.car/` project discovery. Each tick validates these paths
/// using file reads only: never a subprocess, network request, or broad scan.
#[derive(Debug, Clone, Default)]
pub struct SelfhealSourceProbe {
    explicit_checkout: Option<PathBuf>,
    candidates: Vec<PathBuf>,
    setup_refusal: Option<String>,
}

impl SelfhealSourceProbe {
    /// Fail-closed probe with no source candidates. Suitable for embedders that
    /// have not explicitly established a CAR source checkout.
    pub fn empty() -> Self {
        Self::default()
    }

    /// An explicit checkout is authoritative: if it fails validation, auto
    /// candidates are not consulted.
    pub fn explicit(path: PathBuf) -> Self {
        Self {
            explicit_checkout: Some(path),
            ..Self::default()
        }
    }

    /// Deterministic candidate order for embedders and tests.
    pub fn candidates(paths: impl IntoIterator<Item = PathBuf>) -> Self {
        Self {
            candidates: paths.into_iter().collect(),
            ..Self::default()
        }
    }

    /// Add bounded fallback candidates. They are ignored when an explicit
    /// checkout was configured, including when that explicit path is refused.
    pub fn with_candidates(mut self, paths: impl IntoIterator<Item = PathBuf>) -> Self {
        self.candidates.extend(paths);
        dedup_paths(&mut self.candidates);
        self
    }

    /// Build the production probe without launching `git` or touching the
    /// network. Explicit config is `[selfheal] source_checkout = "..."` in
    /// `<CAR_HOME>/config.toml`; relative values resolve under CAR_HOME.
    /// Auto-candidates are bounded to running-binary ancestors and the root of
    /// the `.car/` project discovered from `$CAR_PROJECT_DIR` or cwd.
    pub fn from_process() -> Self {
        let car_home = car_home::root_or_relative();
        let executable = std::env::current_exe().ok();
        let anchor = std::env::var_os("CAR_PROJECT_DIR")
            .map(PathBuf::from)
            .or_else(|| std::env::current_dir().ok());
        Self::from_local_paths(&car_home, executable.as_deref(), anchor.as_deref())
    }

    /// Deterministic seam for verifying the exact production discovery path.
    /// `executable` is the binary file, and `project_anchor` is where `.car/`
    /// walk-up starts.
    #[doc(hidden)]
    pub fn from_local_paths(
        car_home: &Path,
        executable: Option<&Path>,
        project_anchor: Option<&Path>,
    ) -> Self {
        let config_path = car_home.join("config.toml");
        let mut probe = Self::default();
        match read_source_checkout_config(&config_path) {
            Ok(Some(path)) => {
                probe.explicit_checkout = Some(if path.is_absolute() {
                    path
                } else {
                    car_home.join(path)
                });
            }
            Ok(None) => {}
            Err(error) => probe.setup_refusal = Some(error),
        }

        if let Some(parent) = executable.and_then(Path::parent) {
            probe
                .candidates
                .extend(parent.ancestors().map(Path::to_path_buf));
        }
        if let Some(car_dir) = project_anchor.and_then(car_memgine::project::discover_project) {
            if let Some(project_root) = car_dir.parent() {
                probe.candidates.push(project_root.to_path_buf());
            }
        }
        dedup_paths(&mut probe.candidates);
        probe
    }

    fn resolve(&self) -> SourceRouteDecision {
        if let Some(reason) = &self.setup_refusal {
            return SourceRouteDecision::ledger_only(reason.clone());
        }
        if let Some(candidate) = &self.explicit_checkout {
            return match validate_source_checkout(candidate) {
                Ok(path) => SourceRouteDecision::local(path),
                Err(reason) => SourceRouteDecision::ledger_only(format!(
                    "explicit selfheal.source_checkout {} refused: {reason}",
                    candidate.display()
                )),
            };
        }

        let mut refusals = Vec::new();
        for candidate in &self.candidates {
            match validate_source_checkout(candidate) {
                Ok(path) => return SourceRouteDecision::local(path),
                Err(reason) => refusals.push(format!("{}: {reason}", candidate.display())),
            }
        }
        if refusals.is_empty() {
            SourceRouteDecision::ledger_only(
                "no bounded source-checkout candidates were discovered".to_string(),
            )
        } else {
            SourceRouteDecision::ledger_only(format!(
                "no candidate validated as {EXPECTED_ORIGIN}; refusals: {}",
                refusals.join("; ")
            ))
        }
    }
}

/// Where a self-heal detection was routed. `feedback` is reserved in the wire
/// shape for the separately gated feedback sink; this implementation never
/// selects it and fails closed to `ledger-only`.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum SelfhealRoute {
    Local,
    Feedback,
    #[default]
    LedgerOnly,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SourceRouteDecision {
    pub route: SelfhealRoute,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub source_checkout: Option<PathBuf>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub refusal_reason: Option<String>,
}

impl SourceRouteDecision {
    fn local(path: PathBuf) -> Self {
        Self {
            route: SelfhealRoute::Local,
            source_checkout: Some(path),
            refusal_reason: None,
        }
    }

    fn ledger_only(reason: String) -> Self {
        Self {
            route: SelfhealRoute::LedgerOnly,
            source_checkout: None,
            refusal_reason: Some(reason),
        }
    }
}

/// Evidence assembled by the daemon for one tick. Public only so embedders and
/// integration tests can supply deterministic evidence without touching the
/// real process supervisor or user event journals.
#[derive(Debug, Clone, Default)]
pub struct SelfhealEvidence {
    pub events: Vec<EventEvidence>,
    pub metric_alerts: Vec<Alert>,
    pub metric_provenance: Vec<EvidenceSource>,
    pub supervisor_snapshot: Option<SupervisorSnapshot>,
    pub registry_agents: usize,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TickSummary {
    pub started_at: DateTime<Utc>,
    pub completed_at: DateTime<Utc>,
    #[serde(default)]
    pub route: SelfhealRoute,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub source_checkout: Option<PathBuf>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub refusal_reason: Option<String>,
    pub events_scanned: usize,
    pub supervisor_agents: usize,
    pub registry_agents: usize,
    pub detections_found: usize,
    pub appended: usize,
    pub changed: usize,
    pub suppressed_dismissed: usize,
    pub filing_mode: String,
}

#[derive(Debug, Clone, Serialize)]
pub struct SelfhealStatus {
    pub cadence_secs: u64,
    pub last_tick_at: Option<DateTime<Utc>>,
    pub route: SelfhealRoute,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub source_checkout: Option<PathBuf>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub refusal_reason: Option<String>,
    pub detectors: Vec<&'static str>,
    pub detection_count: usize,
    pub warning_count: usize,
    pub critical_count: usize,
    pub dismissed_count: usize,
    pub filing_mode: &'static str,
}

#[derive(Debug, Clone, Deserialize, Default)]
pub struct DetectionQuery {
    #[serde(default)]
    pub kind: Option<DetectionKind>,
    #[serde(default)]
    pub severity: Option<Severity>,
    #[serde(default)]
    pub since: Option<DateTime<Utc>>,
    #[serde(default)]
    pub offset: usize,
    #[serde(default)]
    pub limit: Option<usize>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RoutedDetection {
    #[serde(flatten)]
    pub detection: Detection,
    pub route: SelfhealRoute,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub local_issue_path: Option<PathBuf>,
}

impl RoutedDetection {
    pub fn dedup_key(&self) -> &str {
        self.detection.dedup_key()
    }
}

#[derive(Debug, Clone, Serialize)]
pub struct DetectionPage {
    pub detections: Vec<RoutedDetection>,
    pub total: usize,
    pub offset: usize,
    pub limit: usize,
    pub next_offset: Option<usize>,
}

#[derive(Debug, Clone, Serialize)]
pub struct DismissResult {
    pub dedup_key: String,
    pub dismissed: bool,
    pub already_dismissed: bool,
    pub dismissed_at: DateTime<Utc>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "record_type", rename_all = "snake_case")]
enum LedgerRecord {
    Detection {
        recorded_at: DateTime<Utc>,
        detection: Detection,
        #[serde(default)]
        route: SelfhealRoute,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        local_issue_path: Option<PathBuf>,
    },
    Dismissal {
        dismissed_at: DateTime<Utc>,
        dedup_key: String,
    },
    Tick {
        summary: TickSummary,
    },
}

#[derive(Default)]
struct LedgerState {
    records: Vec<LedgerRecord>,
    supervisor_snapshots: VecDeque<SupervisorSnapshot>,
}

/// Daemon-wide watch-only detector, ledger, and private local-handoff owner.
pub struct SelfhealService {
    ledger_path: PathBuf,
    state_root: PathBuf,
    interval_secs: u64,
    state: Mutex<LedgerState>,
    operation: Mutex<()>,
    evidence_override: Option<SelfhealEvidence>,
    source_probe: SelfhealSourceProbe,
}

impl SelfhealService {
    pub fn open(
        ledger_path: PathBuf,
        interval_secs: u64,
        evidence_override: Option<SelfhealEvidence>,
        source_probe: SelfhealSourceProbe,
    ) -> Result<Self, String> {
        let records = load_records(&ledger_path)?;
        let state_root = ledger_path
            .parent()
            .and_then(Path::parent)
            .map(Path::to_path_buf)
            .ok_or_else(|| "self-heal ledger must live under <CAR_HOME>/selfheal".to_string())?;
        Ok(Self {
            ledger_path,
            state_root,
            interval_secs: interval_secs.max(1),
            state: Mutex::new(LedgerState {
                records,
                supervisor_snapshots: VecDeque::new(),
            }),
            operation: Mutex::new(()),
            evidence_override,
            source_probe,
        })
    }

    pub fn interval_secs(&self) -> u64 {
        self.interval_secs
    }

    pub fn ledger_path(&self) -> &Path {
        &self.ledger_path
    }

    pub async fn status(&self) -> SelfhealStatus {
        let state = self.state.lock().await;
        let (active, dismissed) = fold_detections(&state.records);
        let mut warning_count = 0;
        let mut critical_count = 0;
        for detection in active.values() {
            match detection.detection.severity() {
                Severity::Warning => warning_count += 1,
                Severity::Critical => critical_count += 1,
            }
        }
        let latest_tick = state.records.iter().rev().find_map(|record| match record {
            LedgerRecord::Tick { summary } => Some(summary),
            LedgerRecord::Detection { .. } | LedgerRecord::Dismissal { .. } => None,
        });
        let route = latest_tick.map_or_else(
            || SourceRouteDecision::ledger_only("source route not evaluated yet".to_string()),
            |summary| SourceRouteDecision {
                route: summary.route,
                source_checkout: summary.source_checkout.clone(),
                refusal_reason: summary.refusal_reason.clone(),
            },
        );
        SelfhealStatus {
            cadence_secs: self.interval_secs,
            last_tick_at: latest_tick.map(|summary| summary.completed_at),
            route: route.route,
            source_checkout: route.source_checkout,
            refusal_reason: route.refusal_reason,
            detectors: DETECTOR_IDS.to_vec(),
            detection_count: active.len(),
            warning_count,
            critical_count,
            dismissed_count: dismissed.len(),
            filing_mode: "watch-only",
        }
    }

    pub async fn detections(&self, query: DetectionQuery) -> DetectionPage {
        let state = self.state.lock().await;
        let (active, _) = fold_detections(&state.records);
        let mut detections: Vec<_> = active
            .into_values()
            .filter(|detection| {
                query
                    .kind
                    .is_none_or(|kind| detection.detection.kind() == kind)
                    && query
                        .severity
                        .is_none_or(|severity| detection.detection.severity() == severity)
                    && query
                        .since
                        .is_none_or(|since| detection.detection.last_observed_at() >= since)
            })
            .collect();
        detections.sort_by(|a, b| {
            b.detection
                .last_observed_at()
                .cmp(&a.detection.last_observed_at())
                .then_with(|| a.dedup_key().cmp(b.dedup_key()))
        });
        let total = detections.len();
        let limit = query
            .limit
            .unwrap_or(DEFAULT_PAGE_SIZE)
            .clamp(1, MAX_PAGE_SIZE);
        let page = detections
            .into_iter()
            .skip(query.offset)
            .take(limit)
            .collect::<Vec<_>>();
        let consumed = query.offset.saturating_add(page.len());
        DetectionPage {
            detections: page,
            total,
            offset: query.offset,
            limit,
            next_offset: (consumed < total).then_some(consumed),
        }
    }

    pub async fn dismiss(&self, dedup_key: &str) -> Result<DismissResult, String> {
        validate_dedup_key(dedup_key)?;
        let _operation = self.operation.lock().await;
        let dismissed_at = Utc::now();
        let mut state = self.state.lock().await;
        let known = state.records.iter().any(|record| {
            matches!(
                record,
                LedgerRecord::Detection { detection, .. } if detection.dedup_key() == dedup_key
            )
        });
        if !known {
            return Err(format!(
                "unknown self-heal detection dedup_key '{dedup_key}'"
            ));
        }
        let (_, dismissed) = fold_detections(&state.records);
        let already_dismissed = dismissed.contains(dedup_key);
        let record = LedgerRecord::Dismissal {
            dismissed_at,
            dedup_key: dedup_key.to_string(),
        };
        append_records(&self.ledger_path, std::slice::from_ref(&record))?;
        state.records.push(record);
        Ok(DismissResult {
            dedup_key: dedup_key.to_string(),
            dismissed: true,
            already_dismissed,
            dismissed_at,
        })
    }

    pub async fn run_tick(&self, server: &Arc<ServerState>) -> Result<TickSummary, String> {
        let _operation = self
            .operation
            .try_lock()
            .map_err(|_| "self-heal tick already running".to_string())?;
        let started_at = Utc::now();
        // Use tick start as the event-log watermark, not completion. Events
        // appended while a slow scan is in flight are therefore deferred to
        // (not skipped by) the next slice.
        let since = {
            let state = self.state.lock().await;
            state.records.iter().rev().find_map(|record| match record {
                LedgerRecord::Tick { summary } => Some(summary.started_at),
                LedgerRecord::Detection { .. } | LedgerRecord::Dismissal { .. } => None,
            })
        };
        let evidence = match self.evidence_override.clone() {
            Some(evidence) => evidence,
            None => gather_live_evidence(server, &self.state_root, since, started_at).await,
        };
        // Validate once for this tick, then reuse the immutable result for its
        // summary and every routed detection.
        let route = self.source_probe.resolve();
        self.run_with_evidence(started_at, evidence, route).await
    }

    async fn run_with_evidence(
        &self,
        started_at: DateTime<Utc>,
        evidence: SelfhealEvidence,
        route: SourceRouteDecision,
    ) -> Result<TickSummary, String> {
        let redactor = Redactor::from_env(std::env::vars());
        let car_version = env!("CARGO_PKG_VERSION");
        let observed_at = Utc::now();
        let mut found = Vec::new();
        found.extend(metrics_alerts(
            &evidence.metric_alerts,
            car_version,
            observed_at,
            &evidence.metric_provenance,
            &redactor,
        ));
        found.extend(recurring_tool_failure(
            &evidence.events,
            ToolFailureConfig::default(),
            car_version,
            &redactor,
        ));
        found.extend(capability_miss(&evidence.events, car_version, &redactor));

        let supervisor_agents = evidence
            .supervisor_snapshot
            .as_ref()
            .map_or(0, |snapshot| snapshot.agents.len());
        if let Some(snapshot) = evidence.supervisor_snapshot.as_ref() {
            found.extend(agent_log_errors(
                snapshot,
                AgentLogDetectorConfig::default(),
                car_version,
                &redactor,
            ));
        }
        let mut state = self.state.lock().await;
        if let Some(snapshot) = evidence.supervisor_snapshot {
            state.supervisor_snapshots.push_back(snapshot);
            let cutoff =
                observed_at - Duration::seconds(AgentDetectorConfig::default().window_secs as i64);
            while state
                .supervisor_snapshots
                .front()
                .is_some_and(|snapshot| snapshot.captured_at < cutoff)
            {
                state.supervisor_snapshots.pop_front();
            }
        }
        let snapshots = state
            .supervisor_snapshots
            .iter()
            .cloned()
            .collect::<Vec<_>>();
        found.extend(agent_gave_up(
            &snapshots,
            AgentDetectorConfig::default(),
            car_version,
            &redactor,
        ));

        // One latest value per stable key. This also collapses the same
        // machine-wide metrics signal observed in multiple session logs.
        let mut unique = BTreeMap::new();
        for detection in found {
            unique.insert(detection.dedup_key().to_string(), detection);
        }
        let detections_found = unique.len();
        let (previous, dismissed) = fold_detections(&state.records);
        let mut appended = 0;
        let mut changed = 0;
        let mut suppressed_dismissed = 0;
        let mut records = Vec::new();
        for (key, detection) in unique {
            if dismissed.contains(&key) {
                suppressed_dismissed += 1;
                continue;
            }
            let local_issue_path = if route.route == SelfhealRoute::Local {
                let path = self
                    .state_root
                    .join("selfheal")
                    .join("issues")
                    .join(format!("{key}.md"));
                render_local_issue(&path, &detection)?;
                Some(path)
            } else {
                None
            };
            let routed = RoutedDetection {
                detection,
                route: route.route,
                local_issue_path,
            };
            match previous.get(&key) {
                None => {
                    appended += 1;
                    records.push(LedgerRecord::Detection {
                        recorded_at: observed_at,
                        detection: routed.detection,
                        route: routed.route,
                        local_issue_path: routed.local_issue_path,
                    });
                }
                Some(old) if old != &routed => {
                    appended += 1;
                    changed += 1;
                    records.push(LedgerRecord::Detection {
                        recorded_at: observed_at,
                        detection: routed.detection,
                        route: routed.route,
                        local_issue_path: routed.local_issue_path,
                    });
                }
                Some(_) => {}
            }
        }
        let summary = TickSummary {
            started_at,
            completed_at: Utc::now(),
            route: route.route,
            source_checkout: route.source_checkout,
            refusal_reason: route.refusal_reason,
            events_scanned: evidence.events.len(),
            supervisor_agents,
            registry_agents: evidence.registry_agents,
            detections_found,
            appended,
            changed,
            suppressed_dismissed,
            filing_mode: "watch-only".to_string(),
        };
        records.push(LedgerRecord::Tick {
            summary: summary.clone(),
        });
        append_records(&self.ledger_path, &records)?;
        state.records.extend(records);
        Ok(summary)
    }
}

/// Spawn the default-on watch-only cadence. The first tick runs immediately so
/// an already-Errored supervised agent is visible after daemon boot. A slow
/// tick is never overlapped: `run_tick` refuses while its operation guard is
/// held. The task dies with the daemon runtime.
pub fn spawn_selfheal_cadence(
    state: Arc<ServerState>,
    interval_secs: u64,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        let mut ticker =
            tokio::time::interval(std::time::Duration::from_secs(interval_secs.max(1)));
        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        loop {
            ticker.tick().await;
            if let Err(error) = state.selfheal.run_tick(&state).await {
                tracing::warn!(target: "car::selfheal", %error, "self-heal detection tick failed");
            }
        }
    })
}

async fn log_file_observation(path: &str) -> (Option<DateTime<Utc>>, Option<u64>) {
    let Ok(metadata) = tokio::fs::metadata(path).await else {
        return (None, None);
    };
    let modified_at = metadata.modified().ok().map(DateTime::<Utc>::from);
    (modified_at, Some(metadata.len()))
}

async fn gather_live_evidence(
    server: &Arc<ServerState>,
    state_root: &Path,
    since: Option<DateTime<Utc>>,
    until: DateTime<Utc>,
) -> SelfhealEvidence {
    let sessions = server
        .sessions
        .lock()
        .await
        .values()
        .cloned()
        .collect::<Vec<_>>();
    let mut events = Vec::new();
    let mut metric_alerts_found = Vec::new();
    let mut metric_provenance = Vec::new();
    let thresholds = AlertThresholds {
        max_cost_usd: None,
        max_error_rate: Some(0.5),
        max_avg_latency_ms: None,
        max_goals_ungrounded: Some(0),
        min_actions: Some(5),
    };
    for session in sessions {
        let handle = session.runtime.event_log_handle();
        let log = handle.lock().await;
        let slice = log
            .events()
            .iter()
            .filter(|event| {
                since.is_none_or(|cutoff| event.timestamp > cutoff) && event.timestamp <= until
            })
            .cloned()
            .collect::<Vec<_>>();
        let summary = car_eventlog::summarize(&slice);
        metric_alerts_found.extend(car_eventlog::evaluate_alerts(&summary, &thresholds));
        let path = Some(format!("journals/{}.jsonl", session.client_id));
        metric_provenance.extend(slice.iter().map(|event| EvidenceSource {
            event_id: None,
            run_id: event.run_id.clone(),
            path: path.clone(),
        }));
        events.extend(slice.into_iter().map(|event| EventEvidence {
            event,
            path: path.clone(),
        }));
    }

    let supervisor_snapshot = if let Some(supervisor) = server.supervisor_if_installed() {
        let managed = supervisor.list().await;
        let mut agents = Vec::with_capacity(managed.len());
        for agent in &managed {
            let activity = supervisor
                .read_log(
                    &agent.spec.id,
                    car_registry::supervisor::LogStream::Stdout,
                    ACTIVITY_TAIL_LINES,
                    0,
                )
                .await
                .ok();
            let stderr = supervisor
                .read_log(
                    &agent.spec.id,
                    car_registry::supervisor::LogStream::Stderr,
                    STDERR_TAIL_LINES,
                    0,
                )
                .await
                .ok();
            let (activity_modified_at, activity_bytes) = match activity.as_ref() {
                Some(tail) => log_file_observation(&tail.stdout_path).await,
                None => (None, None),
            };
            let stderr_bytes = match stderr.as_ref() {
                Some(tail) => log_file_observation(&tail.stderr_path).await.1,
                None => None,
            };
            let mut state = SupervisorAgentState::from_managed(
                agent,
                stderr
                    .as_ref()
                    .map(|tail| tail.stderr.join("\n"))
                    .unwrap_or_default(),
                Some(format!("logs/{}.stderr.log", agent.spec.id)),
            );
            state.activity_tail = activity
                .as_ref()
                .map(|tail| tail.stdout.join("\n"))
                .unwrap_or_default();
            state.activity_path = Some(format!("logs/{}.stdout.log", agent.spec.id));
            state.activity_modified_at = activity_modified_at;
            state.activity_bytes = activity_bytes;
            state.stderr_bytes = stderr_bytes;
            agents.push(state);
        }
        Some(SupervisorSnapshot {
            captured_at: Utc::now(),
            agents,
        })
    } else if let Some(manifest) = server.observer_manifest_path() {
        car_registry::supervisor::Supervisor::list_from_manifest(manifest)
            .ok()
            .map(|managed| SupervisorSnapshot {
                captured_at: Utc::now(),
                agents: managed
                    .iter()
                    .map(|agent| SupervisorAgentState::from_managed(agent, "", None))
                    .collect(),
            })
    } else {
        None
    };

    SelfhealEvidence {
        events,
        metric_alerts: metric_alerts_found,
        metric_provenance,
        supervisor_snapshot,
        registry_agents: count_registry_agents(&state_root.join("registry")),
    }
}

#[derive(Deserialize)]
struct SelfhealConfigFile {
    #[serde(default)]
    selfheal: Option<SelfhealConfigSection>,
}

#[derive(Deserialize)]
struct SelfhealConfigSection {
    #[serde(default)]
    source_checkout: Option<PathBuf>,
}

fn read_source_checkout_config(path: &Path) -> Result<Option<PathBuf>, String> {
    let text = match std::fs::read_to_string(path) {
        Ok(text) => text,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(error) => {
            return Err(format!(
                "could not read self-heal config {}: {error}",
                path.display()
            ))
        }
    };
    let config: SelfhealConfigFile = toml::from_str(&text).map_err(|error| {
        format!(
            "could not parse self-heal config {}: {error}",
            path.display()
        )
    })?;
    Ok(config.selfheal.and_then(|section| section.source_checkout))
}

fn dedup_paths(paths: &mut Vec<PathBuf>) {
    let mut seen = BTreeSet::new();
    paths.retain(|path| seen.insert(path.clone()));
}

fn validate_source_checkout(candidate: &Path) -> Result<PathBuf, String> {
    if !candidate.is_dir() {
        return Err("directory does not exist".to_string());
    }
    let checkout = std::fs::canonicalize(candidate)
        .map_err(|error| format!("canonicalize candidate: {error}"))?;
    let dot_git = checkout.join(".git");
    if !dot_git.is_dir() && !dot_git.is_file() {
        return Err("missing .git directory or worktree file".to_string());
    }
    let config_path = git_config_path(&checkout, &dot_git)?;
    let config = std::fs::read_to_string(&config_path)
        .map_err(|error| format!("read git config {}: {error}", config_path.display()))?;
    let remote = origin_remote(&config)
        .ok_or_else(|| format!("git config {} has no origin remote", config_path.display()))?;
    if !is_expected_origin(&remote) {
        return Err(format!(
            "origin remote is '{}', expected {EXPECTED_ORIGIN}",
            remote_for_display(&remote)
        ));
    }
    Ok(checkout)
}

fn git_config_path(checkout: &Path, dot_git: &Path) -> Result<PathBuf, String> {
    if dot_git.is_dir() {
        return Ok(dot_git.join("config"));
    }
    let marker = std::fs::read_to_string(dot_git)
        .map_err(|error| format!("read worktree marker {}: {error}", dot_git.display()))?;
    let raw_git_dir = marker
        .lines()
        .find_map(|line| line.trim().strip_prefix("gitdir:"))
        .map(str::trim)
        .filter(|path| !path.is_empty())
        .ok_or_else(|| format!("invalid worktree marker {}", dot_git.display()))?;
    let git_dir = resolve_relative(checkout, Path::new(raw_git_dir));
    let common_dir_path = git_dir.join("commondir");
    if let Ok(raw_common_dir) = std::fs::read_to_string(&common_dir_path) {
        let common_dir = resolve_relative(&git_dir, Path::new(raw_common_dir.trim()));
        Ok(common_dir.join("config"))
    } else {
        Ok(git_dir.join("config"))
    }
}

fn resolve_relative(base: &Path, path: &Path) -> PathBuf {
    if path.is_absolute() {
        path.to_path_buf()
    } else {
        base.join(path)
    }
}

fn origin_remote(config: &str) -> Option<String> {
    let mut in_origin = false;
    for raw_line in config.lines() {
        let line = raw_line.trim();
        if line.starts_with('[') && line.ends_with(']') {
            in_origin = line.eq_ignore_ascii_case(r#"[remote "origin"]"#);
            continue;
        }
        if !in_origin || line.starts_with('#') || line.starts_with(';') {
            continue;
        }
        let Some((key, value)) = line.split_once('=') else {
            continue;
        };
        if key.trim().eq_ignore_ascii_case("url") {
            return Some(value.trim().trim_matches('"').to_string());
        }
    }
    None
}

fn remote_for_display(remote: &str) -> String {
    let Some((scheme, rest)) = remote.split_once("://") else {
        return remote.to_string();
    };
    let Some((authority, tail)) = rest.split_once('/') else {
        return remote.to_string();
    };
    if authority.contains('@') {
        format!(
            "{scheme}://[REDACTED]@{}/{tail}",
            authority.rsplit('@').next().unwrap_or(authority)
        )
    } else {
        remote.to_string()
    }
}

fn is_expected_origin(remote: &str) -> bool {
    let remote = remote.trim().trim_end_matches('/').trim_end_matches(".git");
    let (host, path) = if remote.contains("://") {
        let without_scheme = remote.split_once("://").map_or(remote, |(_, rest)| rest);
        let Some((host, path)) = without_scheme.split_once('/') else {
            return false;
        };
        (host.rsplit('@').next().unwrap_or(host), path)
    } else if let Some((left, path)) = remote.split_once(':') {
        (left.rsplit('@').next().unwrap_or(left), path)
    } else {
        let Some((host, path)) = remote.split_once('/') else {
            return false;
        };
        (host.rsplit('@').next().unwrap_or(host), path)
    };
    host.eq_ignore_ascii_case("github.com") && path.eq_ignore_ascii_case(EXPECTED_ORIGIN)
}

fn render_local_issue(path: &Path, detection: &Detection) -> Result<(), String> {
    let occurrence_count = local_issue_occurrence_count(path)?.saturating_add(1);
    let parent = path
        .parent()
        .ok_or_else(|| "local self-heal issue path has no parent".to_string())?;
    car_secrets::ensure_private_dir(parent)
        .map_err(|error| format!("create local self-heal issue directory: {error}"))?;

    let event_ids = detection
        .provenance()
        .iter()
        .filter_map(|source| source.event_id())
        .collect::<Vec<_>>();
    let run_ids = detection
        .provenance()
        .iter()
        .filter_map(|source| source.run_id())
        .collect::<Vec<_>>();
    let paths = detection
        .provenance()
        .iter()
        .filter_map(|source| source.path())
        .collect::<Vec<_>>();
    let evidence = if detection.evidence().is_empty() {
        "- (none)".to_string()
    } else {
        detection
            .evidence()
            .iter()
            .map(|excerpt| format!("- {excerpt}"))
            .collect::<Vec<_>>()
            .join("\n")
    };
    let issue = format!(
        "# CAR self-heal detection: {}\n\n\
Trust-Tier: {TRUST_TIER}\n\
Dedup-Key: {}\n\
Detector-ID: {}\n\
Severity: {:?}\n\
Route: local\n\
Occurrence-Count: {occurrence_count}\n\
First-Observed: {}\n\
Last-Observed: {}\n\n\
## Provenance\n\n\
- CAR-Version: {}\n\
- Platform: {}/{}\n\
- Event-IDs: {}\n\
- Run-IDs: {}\n\
- Evidence-Paths: {}\n\n\
## REDACTED Evidence Excerpt\n\n\
{evidence}\n\n\
## Repro Hints\n\n\
- Re-run `selfheal.run` and correlate the detector identity and provenance above.\n\
- Inspect the named local CAR evidence source; do not send it off-machine.\n",
        detection.locator(),
        detection.dedup_key(),
        detection.detector_id(),
        detection.severity(),
        detection.first_observed_at().to_rfc3339(),
        detection.last_observed_at().to_rfc3339(),
        detection.car_version(),
        std::env::consts::OS,
        std::env::consts::ARCH,
        display_list(&event_ids),
        display_list(&run_ids),
        display_list(&paths),
    );

    let mut file = if path.exists() {
        car_secrets::open_private_truncate(path)
    } else {
        car_secrets::create_private_file(path)
    }
    .map_err(|error| format!("open local self-heal issue {}: {error}", path.display()))?;
    file.write_all(issue.as_bytes())
        .map_err(|error| format!("write local self-heal issue: {error}"))?;
    file.flush()
        .map_err(|error| format!("flush local self-heal issue: {error}"))?;
    file.sync_all()
        .map_err(|error| format!("sync local self-heal issue: {error}"))?;
    car_secrets::revalidate_private_path(path, &file)
        .map_err(|error| format!("revalidate local self-heal issue: {error}"))
}

fn local_issue_occurrence_count(path: &Path) -> Result<u64, String> {
    let mut file = match car_secrets::open_private_read(path) {
        Ok(file) => file,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(0),
        Err(error) => {
            return Err(format!(
                "open existing local self-heal issue {}: {error}",
                path.display()
            ))
        }
    };
    let mut text = String::new();
    file.read_to_string(&mut text)
        .map_err(|error| format!("read existing local self-heal issue: {error}"))?;
    text.lines()
        .find_map(|line| line.strip_prefix("Occurrence-Count: "))
        .ok_or_else(|| {
            format!(
                "existing local self-heal issue {} has no occurrence count",
                path.display()
            )
        })?
        .parse::<u64>()
        .map_err(|error| format!("parse local self-heal issue occurrence count: {error}"))
}

fn display_list(values: &[&str]) -> String {
    if values.is_empty() {
        "(none)".to_string()
    } else {
        values.join(", ")
    }
}

fn count_registry_agents(registry_dir: &Path) -> usize {
    let Ok(entries) = std::fs::read_dir(registry_dir) else {
        return 0;
    };
    entries
        .filter_map(Result::ok)
        .filter(|entry| entry.path().extension().and_then(|ext| ext.to_str()) == Some("json"))
        .filter(|entry| {
            std::fs::read(entry.path())
                .ok()
                .and_then(|bytes| serde_json::from_slice::<car_registry::AgentEntry>(&bytes).ok())
                .is_some()
        })
        .count()
}

fn fold_detections(
    records: &[LedgerRecord],
) -> (BTreeMap<String, RoutedDetection>, BTreeSet<String>) {
    let mut detections = BTreeMap::new();
    let mut dismissed = BTreeSet::new();
    for record in records {
        match record {
            LedgerRecord::Detection {
                detection,
                route,
                local_issue_path,
                ..
            } => {
                detections.insert(
                    detection.dedup_key().to_string(),
                    RoutedDetection {
                        detection: detection.clone(),
                        route: *route,
                        local_issue_path: local_issue_path.clone(),
                    },
                );
            }
            LedgerRecord::Dismissal { dedup_key, .. } => {
                dismissed.insert(dedup_key.clone());
            }
            LedgerRecord::Tick { .. } => {}
        }
    }
    for key in &dismissed {
        detections.remove(key);
    }
    (detections, dismissed)
}

fn validate_dedup_key(key: &str) -> Result<(), String> {
    if key.len() == 64 && key.bytes().all(|byte| byte.is_ascii_hexdigit()) {
        Ok(())
    } else {
        Err("dedup_key must be a 64-character SHA-256 hex string".to_string())
    }
}

fn load_records(path: &Path) -> Result<Vec<LedgerRecord>, String> {
    let file = match car_secrets::open_private_read(path) {
        Ok(file) => file,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
        Err(error) => return Err(format!("open self-heal ledger: {error}")),
    };
    car_secrets::revalidate_private_path(path, &file)
        .map_err(|error| format!("validate self-heal ledger: {error}"))?;
    let mut records = Vec::new();
    for (index, line) in BufReader::new(file).lines().enumerate() {
        let line =
            line.map_err(|error| format!("read self-heal ledger line {}: {error}", index + 1))?;
        if line.trim().is_empty() {
            continue;
        }
        records.push(
            serde_json::from_str(&line)
                .map_err(|error| format!("parse self-heal ledger line {}: {error}", index + 1))?,
        );
    }
    Ok(records)
}

fn append_records(path: &Path, records: &[LedgerRecord]) -> Result<(), String> {
    if records.is_empty() {
        return Ok(());
    }
    let parent = path
        .parent()
        .ok_or_else(|| "self-heal ledger path has no parent".to_string())?;
    car_secrets::ensure_private_dir(parent)
        .map_err(|error| format!("create self-heal ledger directory: {error}"))?;
    let mut file = car_secrets::open_private_append(path)
        .map_err(|error| format!("open self-heal ledger for append: {error}"))?;
    let original_len = file
        .metadata()
        .map_err(|error| format!("stat self-heal ledger: {error}"))?
        .len();
    let write_result = (|| -> Result<(), String> {
        for record in records {
            serde_json::to_writer(&mut file, record)
                .map_err(|error| format!("serialize self-heal ledger record: {error}"))?;
            file.write_all(b"\n")
                .map_err(|error| format!("append self-heal ledger newline: {error}"))?;
        }
        file.flush()
            .map_err(|error| format!("flush self-heal ledger: {error}"))?;
        file.sync_all()
            .map_err(|error| format!("sync self-heal ledger: {error}"))?;
        car_secrets::revalidate_private_path(path, &file)
            .map_err(|error| format!("revalidate self-heal ledger: {error}"))?;
        Ok(())
    })();
    if let Err(error) = write_result {
        let _ = file.set_len(original_len);
        return Err(error);
    }
    Ok(())
}