car_server_core/

selfheal.rs

//! Daemon wiring for deterministic self-healing detection and bounded PR fixes.
//!
//! The subsystem reads evidence CAR already owns, runs the pure
//! `car-selfheal` detectors, and appends detections/dismissals/ticks/coder
//! attempts to one private JSONL ledger. A bounded source probe renders local
//! handoff documents; only a redactor-clean recurring builtin failure reaches
//! the registry-owned PR template. Auto-fix never edits the maintainer's working
//! tree or merges: it fetches `origin/main`, creates an isolated worktree, and
//! delegates one contract-gated round to `car code-task`.

use car_eventlog::{Alert, AlertThresholds, EventKind};
use car_ir::ActionProposal;
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";
const DEFAULT_AUTO_FIX: bool = true;
const DEFAULT_MAX_CONCURRENT: u32 = 1;
const DEFAULT_MAX_PER_DAY: u32 = 3;
const DEFAULT_MAX_ROUNDS_PER_KEY: u32 = 3;
// The JSON-RPC dispatcher abandons a handler after 1800 seconds. Bound the
// child well inside that ceiling so the daemon can persist its terminal result.
const SELFHEAL_CODER_WALL_SECS: u64 = 15 * 60;

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),
        }
    }
}

/// How a tick proves that its validated checkout already contains the replay
/// verb. Production builds compile that checkout's CLI and invoke its help;
/// the fixed variants are deterministic seams for integration tests.
#[derive(Debug, Clone, Default)]
pub enum SelfhealReplayVerbProbe {
    #[default]
    CheckoutBuild,
    #[doc(hidden)]
    Available,
    #[doc(hidden)]
    Unavailable(String),
}

impl SelfhealReplayVerbProbe {
    #[doc(hidden)]
    pub fn available_for_tests() -> Self {
        Self::Available
    }

    #[doc(hidden)]
    pub fn unavailable_for_tests(reason: impl Into<String>) -> Self {
        Self::Unavailable(reason.into())
    }

    async fn check(&self, checkout: &Path, target_dir: &Path) -> Result<(), String> {
        match self {
            Self::Available => Ok(()),
            Self::Unavailable(reason) => Err(reason.clone()),
            Self::CheckoutBuild => check_checkout_replay_verb(checkout, target_dir).await,
        }
    }
}

/// 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 auto_fix_enabled: bool,
    pub max_concurrent: u32,
    pub max_per_day: u32,
    pub max_rounds_per_key: u32,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub auto_fix_refusal_reason: Option<String>,
    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>,
}

/// A secret-safe tool invocation reconstructed from the accepted proposal
/// preimage in the same journal as its `ActionFailed` event.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReconstructedToolCall {
    pub tool: String,
    pub params: BTreeMap<String, serde_json::Value>,
}

#[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>,
    /// Present only for `recurring_tool_failure`. True requires a recovered
    /// call, builtin provenance, and a byte-unchanged redactor pass.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub eligible: Option<bool>,
    /// Omitted when parameters were unavailable or the redactor changed them;
    /// unsafe raw parameters are never copied out of `ProposalReceived`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub reconstructed_call: Option<ReconstructedToolCall>,
    /// Owner-private sidecar beside `local_issue_path`, when routed locally.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub reconstructed_call_path: Option<PathBuf>,
    /// Number of durable coder rounds started for this stable key.
    #[serde(default, skip_serializing_if = "is_zero_u32")]
    pub auto_fix_attempts: u32,
    /// True once the configured/hard three-round ceiling is reached.
    #[serde(default, skip_serializing_if = "is_false")]
    pub auto_fix_exhausted: bool,
    /// True when a durable start marker has no terminal attempt record yet.
    #[serde(default, skip_serializing_if = "is_false")]
    pub auto_fix_in_progress: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub last_auto_fix_attempt: Option<FixAttemptResult>,
    /// An already-open stable-key PR exists remotely; the review loop owns it.
    #[serde(default, skip_serializing_if = "is_false")]
    pub auto_fix_awaiting_review: bool,
    /// The stable-key PR was closed without merge and must never be reopened.
    #[serde(default, skip_serializing_if = "is_false")]
    pub auto_fix_parked: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub remote_pr_number: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub remote_pr_url: Option<String>,
}

fn is_zero_u32(value: &u32) -> bool {
    *value == 0
}

fn is_false(value: &bool) -> bool {
    !*value
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FixTrigger {
    Cadence,
    Manual,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RemotePrDisposition {
    AwaitingReview,
    Parked,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FixAttemptResult {
    pub dedup_key: String,
    pub round: u32,
    pub trigger: FixTrigger,
    pub started_at: DateTime<Utc>,
    pub completed_at: DateTime<Utc>,
    pub target_branch: String,
    pub workspace_dir: PathBuf,
    pub spawned: bool,
    pub exit_code: Option<i32>,
    pub failure_class: String,
}

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>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        eligible: Option<bool>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        reconstructed_call: Option<ReconstructedToolCall>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        reconstructed_call_path: Option<PathBuf>,
    },
    Dismissal {
        dismissed_at: DateTime<Utc>,
        dedup_key: String,
    },
    Tick {
        summary: TickSummary,
    },
    FixStarted {
        started_at: DateTime<Utc>,
        dedup_key: String,
        round: u32,
        trigger: FixTrigger,
        target_branch: String,
        workspace_dir: PathBuf,
    },
    FixAttempt {
        result: FixAttemptResult,
    },
    RemotePr {
        observed_at: DateTime<Utc>,
        dedup_key: String,
        disposition: RemotePrDisposition,
        pr_number: u64,
        pr_url: String,
    },
}

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

/// Daemon-wide detector, private ledger, and bounded PR-fix 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,
    replay_verb_probe: SelfhealReplayVerbProbe,
    auto_fix: AutoFixConfig,
    auto_fix_refusal_reason: Option<String>,
}

impl SelfhealService {
    pub fn open(
        ledger_path: PathBuf,
        interval_secs: u64,
        evidence_override: Option<SelfhealEvidence>,
        source_probe: SelfhealSourceProbe,
        replay_verb_probe: SelfhealReplayVerbProbe,
    ) -> 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())?;
        let (auto_fix, auto_fix_refusal_reason) =
            match read_auto_fix_config(&state_root.join("config.toml")) {
                Ok(config) => (config, None),
                Err(reason) => (AutoFixConfig::disabled(), Some(reason)),
            };
        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,
            replay_verb_probe,
            auto_fix,
            auto_fix_refusal_reason,
        })
    }

    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, self.auto_fix.max_rounds_per_key);
        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 { .. }
            | LedgerRecord::FixStarted { .. }
            | LedgerRecord::FixAttempt { .. }
            | LedgerRecord::RemotePr { .. } => 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,
            auto_fix_enabled: self.auto_fix.auto_fix && self.auto_fix_refusal_reason.is_none(),
            max_concurrent: self.auto_fix.max_concurrent,
            max_per_day: self.auto_fix.max_per_day,
            max_rounds_per_key: self.auto_fix.max_rounds_per_key,
            auto_fix_refusal_reason: self.auto_fix_refusal_reason.clone(),
            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: if self.auto_fix.auto_fix
                && self.auto_fix_refusal_reason.is_none()
                && route.route == SelfhealRoute::Local
            {
                "pr-only"
            } else {
                "watch-only"
            },
        }
    }

    pub async fn detections(&self, query: DetectionQuery) -> DetectionPage {
        let state = self.state.lock().await;
        let (active, _) = fold_detections(&state.records, self.auto_fix.max_rounds_per_key);
        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, self.auto_fix.max_rounds_per_key);
        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,
        })
    }

    /// Start one explicit coder round for an eligible recurring-tool key.
    /// The off switch controls unattended cadence only; an operator invoking
    /// `selfheal.fix` may still run a bounded round deliberately.
    pub async fn fix(&self, dedup_key: &str) -> Result<FixAttemptResult, String> {
        validate_dedup_key(dedup_key)?;
        let _operation = self
            .operation
            .try_lock()
            .map_err(|_| "another self-heal operation is already running".to_string())?;
        match self.reconcile_remote_pr_unlocked(dedup_key).await? {
            RemoteDedupResult::Proceed => {
                self.run_fix_unlocked(dedup_key, FixTrigger::Manual).await
            }
            RemoteDedupResult::AwaitingReview => Err(format!(
                "self-heal detection '{dedup_key}' is awaiting review on its open remote PR"
            )),
            RemoteDedupResult::Parked => Err(format!(
                "self-heal detection '{dedup_key}' is parked after its PR closed unmerged"
            )),
        }
    }

    async fn run_auto_fix_unlocked(&self) -> Result<Option<FixAttemptResult>, String> {
        if !self.auto_fix.auto_fix || self.auto_fix_refusal_reason.is_some() {
            return Ok(None);
        }
        let candidate = {
            let state = self.state.lock().await;
            if starts_today(&state.records, Utc::now()) >= self.auto_fix.max_per_day {
                return Ok(None);
            }
            let (active, _) = fold_detections(&state.records, self.auto_fix.max_rounds_per_key);
            active
                .into_values()
                .filter(|detection| detection.eligible == Some(true))
                // Cadence owns only the initial round. Review-driven/manual
                // rounds reuse the key and stable PR branch explicitly.
                .filter(|detection| detection.auto_fix_attempts == 0)
                .filter(|detection| {
                    !detection.auto_fix_awaiting_review && !detection.auto_fix_parked
                })
                .min_by(|left, right| {
                    left.detection
                        .last_observed_at()
                        .cmp(&right.detection.last_observed_at())
                        .then_with(|| left.dedup_key().cmp(right.dedup_key()))
                })
                .map(|detection| detection.dedup_key().to_string())
        };
        match candidate {
            Some(key) => match self.reconcile_remote_pr_unlocked(&key).await? {
                RemoteDedupResult::Proceed => self
                    .run_fix_unlocked(&key, FixTrigger::Cadence)
                    .await
                    .map(Some),
                RemoteDedupResult::AwaitingReview | RemoteDedupResult::Parked => Ok(None),
            },
            None => Ok(None),
        }
    }

    async fn reconcile_remote_pr_unlocked(
        &self,
        dedup_key: &str,
    ) -> Result<RemoteDedupResult, String> {
        let route = self.source_probe.resolve();
        let Some(checkout) = route.source_checkout else {
            return Ok(RemoteDedupResult::Proceed);
        };
        let attempts = {
            let state = self.state.lock().await;
            let (active, _) = fold_detections(&state.records, self.auto_fix.max_rounds_per_key);
            active
                .get(dedup_key)
                .map_or(0, |detection| detection.auto_fix_attempts)
        };
        if attempts > 0 {
            return Ok(RemoteDedupResult::Proceed);
        }

        let Some(remote) = query_remote_selfheal_pr(&checkout, dedup_key).await? else {
            return Ok(RemoteDedupResult::Proceed);
        };
        let disposition = match (remote.state.as_str(), remote.merged_at.as_deref()) {
            ("OPEN", _) => RemotePrDisposition::AwaitingReview,
            ("CLOSED", None) => RemotePrDisposition::Parked,
            // A merged stable-key PR also makes a fresh initial round invalid;
            // never recreate the branch after delivery completed.
            ("CLOSED", Some(_)) => RemotePrDisposition::Parked,
            (state, _) => {
                return Err(format!("remote self-heal PR has unknown state '{state}'"));
            }
        };
        let record = LedgerRecord::RemotePr {
            observed_at: Utc::now(),
            dedup_key: dedup_key.to_string(),
            disposition,
            pr_number: remote.number,
            pr_url: remote.url,
        };
        let mut state = self.state.lock().await;
        append_records(&self.ledger_path, std::slice::from_ref(&record))?;
        state.records.push(record);
        Ok(match disposition {
            RemotePrDisposition::AwaitingReview => RemoteDedupResult::AwaitingReview,
            RemotePrDisposition::Parked => RemoteDedupResult::Parked,
        })
    }

    async fn run_fix_unlocked(
        &self,
        dedup_key: &str,
        trigger: FixTrigger,
    ) -> Result<FixAttemptResult, String> {
        if let Some(reason) = &self.auto_fix_refusal_reason {
            return Err(format!("self-heal auto-fix config refused: {reason}"));
        }
        let route = self.source_probe.resolve();
        let checkout = route.source_checkout.ok_or_else(|| {
            format!(
                "self-heal key '{dedup_key}' cannot auto-fix from route {}: {}",
                route_name(route.route),
                route
                    .refusal_reason
                    .unwrap_or_else(|| "no validated source checkout".to_string())
            )
        })?;
        let (detection, round) = {
            let state = self.state.lock().await;
            let (active, _) = fold_detections(&state.records, self.auto_fix.max_rounds_per_key);
            let detection = active
                .get(dedup_key)
                .cloned()
                .ok_or_else(|| format!("unknown or dismissed self-heal detection '{dedup_key}'"))?;
            if detection.eligible != Some(true) {
                return Err(format!(
                    "self-heal detection '{dedup_key}' is not eligible for auto-fix"
                ));
            }
            if detection.auto_fix_in_progress {
                return Err(format!(
                    "self-heal detection '{dedup_key}' already has a coder round in progress"
                ));
            }
            if detection.auto_fix_attempts >= self.auto_fix.max_rounds_per_key {
                return Err(format!(
                    "self-heal detection '{dedup_key}' is auto-fix exhausted"
                ));
            }
            if starts_today(&state.records, Utc::now()) >= self.auto_fix.max_per_day {
                return Err(format!(
                    "self-heal daily auto-fix limit ({}) is exhausted",
                    self.auto_fix.max_per_day
                ));
            }
            let round = detection.auto_fix_attempts + 1;
            (detection, round)
        };
        let call = detection
            .reconstructed_call
            .as_ref()
            .ok_or_else(|| format!("eligible self-heal key '{dedup_key}' has no safe call"))?;

        let attempt_dir = self
            .state_root
            .join("selfheal")
            .join("attempts")
            .join(dedup_key);
        car_secrets::ensure_private_dir(&attempt_dir)
            .map_err(|error| format!("create self-heal attempt directory: {error}"))?;
        let params_path = attempt_dir.join("params.json");
        persist_private_json(&params_path, &call.params)?;
        let rendered =
            crate::selfheal_templates::render(&crate::selfheal_templates::TemplateContext {
                dedup_key,
                detector_id: detection.detection.detector_id(),
                locator: detection.detection.locator(),
                tool: &call.tool,
                params_path: &params_path,
            })?;
        let intent_path = attempt_dir.join("intent.md");
        persist_private_bytes(&intent_path, rendered.intent.as_bytes())?;
        let contract_path = attempt_dir.join("contract.json");
        persist_private_json(&contract_path, &rendered.contract)?;

        let target_branch = format!("car/selfheal/{dedup_key}");
        let workspace_dir = checkout
            .join(".worktrees")
            .join(format!("selfheal-{dedup_key}"));
        let target_dir = self.state_root.join("selfheal").join("target");
        car_secrets::ensure_private_dir(&target_dir)
            .map_err(|error| format!("create self-heal cargo target directory: {error}"))?;
        // Resolve the orchestrator before prepending the isolated target to its
        // child PATH. Otherwise round two could launch round one's freshly
        // built (and not yet re-verified) `car` binary as the orchestrator.
        let car_executable = resolve_car_cli_executable();
        let mut path_entries = vec![target_dir.join("debug")];
        if let Some(path) = std::env::var_os("PATH") {
            path_entries.extend(std::env::split_paths(&path));
        }
        let code_task_path = std::env::join_paths(path_entries)
            .map_err(|error| format!("build self-heal code-task PATH: {error}"))?;
        let started_at = Utc::now();
        let started = LedgerRecord::FixStarted {
            started_at,
            dedup_key: dedup_key.to_string(),
            round,
            trigger: trigger.clone(),
            target_branch: target_branch.clone(),
            workspace_dir: workspace_dir.clone(),
        };
        {
            let mut state = self.state.lock().await;
            append_records(&self.ledger_path, std::slice::from_ref(&started))?;
            state.records.push(started);
        }

        let fetch = tokio::process::Command::new("git")
            .current_dir(&checkout)
            .args(["fetch", "origin", "main"])
            .kill_on_drop(true)
            .output()
            .await;
        match fetch {
            Err(_) => {
                return self
                    .finish_attempt(
                        dedup_key,
                        round,
                        trigger,
                        started_at,
                        target_branch,
                        workspace_dir,
                        false,
                        None,
                        "fetch_spawn_failed".to_string(),
                    )
                    .await;
            }
            Ok(output) if !output.status.success() => {
                return self
                    .finish_attempt(
                        dedup_key,
                        round,
                        trigger,
                        started_at,
                        target_branch,
                        workspace_dir,
                        false,
                        output.status.code(),
                        "fetch_failed".to_string(),
                    )
                    .await;
            }
            Ok(_) => {}
        }

        let Some(car_executable) = car_executable else {
            return self
                .finish_attempt(
                    dedup_key,
                    round,
                    trigger,
                    started_at,
                    target_branch,
                    workspace_dir,
                    false,
                    None,
                    "spawn_failed".to_string(),
                )
                .await;
        };
        let marker = format!("<!-- car-selfheal:key={dedup_key} -->");
        let transcript_path = attempt_dir.join(format!("code-task-round-{round}.jsonl"));
        persist_private_bytes(&transcript_path, b"")?;
        let argv = vec![
            "code-task".to_string(),
            "--repo".to_string(),
            checkout.to_string_lossy().into_owned(),
            "--intent-file".to_string(),
            intent_path.to_string_lossy().into_owned(),
            "--contract-file".to_string(),
            contract_path.to_string_lossy().into_owned(),
            "--deliver".to_string(),
            "pr".to_string(),
            "--target-branch".to_string(),
            target_branch.clone(),
            "--pr-base".to_string(),
            "main".to_string(),
            "--workspace-dir".to_string(),
            workspace_dir.to_string_lossy().into_owned(),
            "--body-prefix".to_string(),
            marker,
            "--max-session-wall-secs".to_string(),
            SELFHEAL_CODER_WALL_SECS.to_string(),
            "--transcript".to_string(),
            transcript_path.to_string_lossy().into_owned(),
            "--json".to_string(),
        ];
        let output = tokio::process::Command::new(car_executable)
            .args(&argv)
            .env("CARGO_TARGET_DIR", &target_dir)
            // The first contract line builds this worktree's CLI into the
            // isolated target. The second line's literal `car tools call`
            // must therefore resolve that fresh binary, not the installed
            // daemon binary whose source predates the repair.
            .env("PATH", code_task_path)
            .kill_on_drop(true)
            .output()
            .await;
        match output {
            Err(_) => {
                self.finish_attempt(
                    dedup_key,
                    round,
                    trigger,
                    started_at,
                    target_branch,
                    workspace_dir,
                    false,
                    None,
                    "spawn_failed".to_string(),
                )
                .await
            }
            Ok(output) => {
                let failure_class = classify_code_task_output(&output.stdout);
                self.finish_attempt(
                    dedup_key,
                    round,
                    trigger,
                    started_at,
                    target_branch,
                    workspace_dir,
                    true,
                    output.status.code(),
                    failure_class,
                )
                .await
            }
        }
    }

    #[allow(clippy::too_many_arguments)]
    async fn finish_attempt(
        &self,
        dedup_key: &str,
        round: u32,
        trigger: FixTrigger,
        started_at: DateTime<Utc>,
        target_branch: String,
        workspace_dir: PathBuf,
        spawned: bool,
        exit_code: Option<i32>,
        failure_class: String,
    ) -> Result<FixAttemptResult, String> {
        let result = FixAttemptResult {
            dedup_key: dedup_key.to_string(),
            round,
            trigger,
            started_at,
            completed_at: Utc::now(),
            target_branch,
            workspace_dir,
            spawned,
            exit_code,
            failure_class,
        };
        let record = LedgerRecord::FixAttempt {
            result: result.clone(),
        };
        let mut state = self.state.lock().await;
        append_records(&self.ledger_path, std::slice::from_ref(&record))?;
        state.records.push(record);
        Ok(result)
    }

    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 { .. }
                | LedgerRecord::FixStarted { .. }
                | LedgerRecord::FixAttempt { .. }
                | LedgerRecord::RemotePr { .. } => 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();
        let summary = self.run_with_evidence(started_at, evidence, route).await?;
        if let Err(error) = self.run_auto_fix_unlocked().await {
            tracing::warn!(target: "car::selfheal", %error, "self-heal auto-fix round failed");
        }
        Ok(summary)
    }

    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 snapshots = {
            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();
                }
            }
            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 replay_verb_refusal = if route.route == SelfhealRoute::Local
            && unique.values().any(|detection| {
                detection.detector_id() == car_selfheal::detectors::tools::DETECTOR_ID
            }) {
            let checkout = route
                .source_checkout
                .as_ref()
                .expect("local route always carries a checkout");
            self.replay_verb_probe
                .check(
                    checkout,
                    &self.state_root.join("selfheal/replay-probe-target"),
                )
                .await
                .err()
                .map(|reason| format!("checkout predates the replay verb: {reason}"))
        } else {
            None
        };
        let mut state = self.state.lock().await;
        let (previous, dismissed) =
            fold_detections(&state.records, self.auto_fix.max_rounds_per_key);
        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 assessment = assess_reconstructed_call(&detection, &evidence.events, &redactor);
            let mut eligible = assessment.as_ref().map(|assessment| assessment.eligible);
            let reconstructed_call =
                assessment.and_then(|assessment| assessment.reconstructed_call);
            let ineligible_reason = if eligible == Some(true) {
                replay_verb_refusal.as_deref()
            } else {
                None
            };
            if ineligible_reason.is_some() {
                eligible = Some(false);
            }
            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, eligible, ineligible_reason)?;
                Some(path)
            } else {
                None
            };
            let reconstructed_call_path =
                if route.route == SelfhealRoute::Local && eligible.is_some() {
                    let path = self
                        .state_root
                        .join("selfheal")
                        .join("issues")
                        .join(format!("{key}.call.json"));
                    persist_reconstructed_call(
                        &path,
                        eligible.unwrap_or(false),
                        reconstructed_call.as_ref(),
                    )?;
                    Some(path)
                } else {
                    None
                };
            let mut routed = RoutedDetection {
                detection,
                route: route.route,
                local_issue_path,
                eligible,
                reconstructed_call,
                reconstructed_call_path,
                auto_fix_attempts: 0,
                auto_fix_exhausted: false,
                auto_fix_in_progress: false,
                last_auto_fix_attempt: None,
                auto_fix_awaiting_review: false,
                auto_fix_parked: false,
                remote_pr_number: None,
                remote_pr_url: None,
            };
            apply_attempt_state(
                &mut routed,
                &state.records,
                self.auto_fix.max_rounds_per_key,
            );
            apply_remote_pr_state(&mut routed, &state.records);
            let should_append = match previous.get(&key) {
                None => {
                    appended += 1;
                    true
                }
                Some(old) if old != &routed => {
                    appended += 1;
                    changed += 1;
                    true
                }
                Some(_) => false,
            };
            if should_append {
                records.push(LedgerRecord::Detection {
                    recorded_at: observed_at,
                    detection: routed.detection,
                    route: routed.route,
                    local_issue_path: routed.local_issue_path,
                    eligible: routed.eligible,
                    reconstructed_call: routed.reconstructed_call,
                    reconstructed_call_path: routed.reconstructed_call_path,
                });
            }
        }
        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: if self.auto_fix.auto_fix
                && self.auto_fix_refusal_reason.is_none()
                && route.route == SelfhealRoute::Local
            {
                "pr-only"
            } else {
                "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 detection and eligible auto-fix 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<_>>();
        // A tool may start before the prior tick's watermark and fail after
        // it. Include only older ProposalReceived rows as reconstruction
        // context; detectors ignore this kind, so prior failures are never
        // made active again.
        let proposal_context = since.map_or_else(Vec::new, |cutoff| {
            log.events()
                .iter()
                .filter(|event| {
                    event.kind == EventKind::ProposalReceived
                        && 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(),
        }));
        events.extend(proposal_context.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>,
    #[serde(default = "default_auto_fix")]
    auto_fix: bool,
    #[serde(default = "default_max_concurrent")]
    max_concurrent: u32,
    #[serde(default = "default_max_per_day")]
    max_per_day: u32,
    #[serde(default = "default_max_rounds_per_key")]
    max_rounds_per_key: u32,
}

impl Default for SelfhealConfigSection {
    fn default() -> Self {
        Self {
            source_checkout: None,
            auto_fix: DEFAULT_AUTO_FIX,
            max_concurrent: DEFAULT_MAX_CONCURRENT,
            max_per_day: DEFAULT_MAX_PER_DAY,
            max_rounds_per_key: DEFAULT_MAX_ROUNDS_PER_KEY,
        }
    }
}

#[derive(Debug, Clone)]
struct AutoFixConfig {
    auto_fix: bool,
    max_concurrent: u32,
    max_per_day: u32,
    max_rounds_per_key: u32,
}

impl Default for AutoFixConfig {
    fn default() -> Self {
        Self {
            auto_fix: DEFAULT_AUTO_FIX,
            max_concurrent: DEFAULT_MAX_CONCURRENT,
            max_per_day: DEFAULT_MAX_PER_DAY,
            max_rounds_per_key: DEFAULT_MAX_ROUNDS_PER_KEY,
        }
    }
}

impl AutoFixConfig {
    fn disabled() -> Self {
        Self {
            auto_fix: false,
            ..Self::default()
        }
    }
}

const fn default_auto_fix() -> bool {
    DEFAULT_AUTO_FIX
}

const fn default_max_concurrent() -> u32 {
    DEFAULT_MAX_CONCURRENT
}

const fn default_max_per_day() -> u32 {
    DEFAULT_MAX_PER_DAY
}

const fn default_max_rounds_per_key() -> u32 {
    DEFAULT_MAX_ROUNDS_PER_KEY
}

fn read_auto_fix_config(path: &Path) -> Result<AutoFixConfig, String> {
    let text = match std::fs::read_to_string(path) {
        Ok(text) => text,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
            return Ok(AutoFixConfig::default())
        }
        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()
        )
    })?;
    let section = config.selfheal.unwrap_or_default();
    if section.max_concurrent != DEFAULT_MAX_CONCURRENT {
        return Err(format!(
            "[selfheal] max_concurrent must be 1, got {}",
            section.max_concurrent
        ));
    }
    if !(1..=DEFAULT_MAX_PER_DAY).contains(&section.max_per_day) {
        return Err(format!(
            "[selfheal] max_per_day must be between 1 and {DEFAULT_MAX_PER_DAY}, got {}",
            section.max_per_day
        ));
    }
    if !(1..=DEFAULT_MAX_ROUNDS_PER_KEY).contains(&section.max_rounds_per_key) {
        return Err(format!(
            "[selfheal] max_rounds_per_key must be between 1 and {DEFAULT_MAX_ROUNDS_PER_KEY}, got {}",
            section.max_rounds_per_key
        ));
    }
    Ok(AutoFixConfig {
        auto_fix: section.auto_fix,
        max_concurrent: section.max_concurrent,
        max_per_day: section.max_per_day,
        max_rounds_per_key: section.max_rounds_per_key,
    })
}

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)
}

#[derive(Debug)]
struct CallAssessment {
    eligible: bool,
    reconstructed_call: Option<ReconstructedToolCall>,
}

fn assess_reconstructed_call(
    detection: &Detection,
    events: &[EventEvidence],
    redactor: &Redactor,
) -> Option<CallAssessment> {
    if detection.kind() != DetectionKind::RecurringToolFailure {
        return None;
    }
    let Some((builtin, reconstructed_call)) = reconstruct_failed_call(detection, events) else {
        return Some(CallAssessment {
            eligible: false,
            reconstructed_call: None,
        });
    };
    if !params_pass_redactor(&reconstructed_call.params, redactor) {
        return Some(CallAssessment {
            eligible: false,
            reconstructed_call: None,
        });
    }
    Some(CallAssessment {
        eligible: auto_fix_tool_allowed(&reconstructed_call.tool, builtin),
        reconstructed_call: Some(reconstructed_call),
    })
}

/// Auto-fix may replay data-oriented commodity builtins, never the nested GUI
/// agent that can click and type in an ambient desktop session. Provenance
/// alone identifies who owns a tool; it does not make every builtin safe to
/// replay unattended.
fn auto_fix_tool_allowed(tool: &str, builtin: bool) -> bool {
    builtin && tool != "delegate_gui"
}

fn params_pass_redactor(params: &BTreeMap<String, serde_json::Value>, redactor: &Redactor) -> bool {
    let Ok(params_json) = serde_json::to_string(params) else {
        return false;
    };
    redactor.redact(&params_json) == params_json
        && params.iter().all(|(key, value)| {
            redactor.redact(key) == *key && json_value_passes_redactor(value, redactor)
        })
}

fn json_value_passes_redactor(value: &serde_json::Value, redactor: &Redactor) -> bool {
    match value {
        serde_json::Value::String(value) => redactor.redact(value) == *value,
        serde_json::Value::Array(values) => values
            .iter()
            .all(|value| json_value_passes_redactor(value, redactor)),
        serde_json::Value::Object(values) => values.iter().all(|(key, value)| {
            redactor.redact(key) == *key && json_value_passes_redactor(value, redactor)
        }),
        serde_json::Value::Null | serde_json::Value::Bool(_) | serde_json::Value::Number(_) => true,
    }
}

fn reconstruct_failed_call(
    detection: &Detection,
    events: &[EventEvidence],
) -> Option<(bool, ReconstructedToolCall)> {
    let failure = events
        .iter()
        .filter(|record| record.event.kind == EventKind::ActionFailed)
        .filter(|record| failure_belongs_to_detection(detection, record))
        .max_by(|left, right| {
            left.event
                .timestamp
                .cmp(&right.event.timestamp)
                .then_with(|| left.event.proposal_id.cmp(&right.event.proposal_id))
                .then_with(|| left.event.action_id.cmp(&right.event.action_id))
        })?;
    let journal = failure.path.as_deref()?;
    let proposal_id = failure.event.proposal_id.as_deref()?;
    let action_id = failure.event.action_id.as_deref()?;
    let failed_tool = failure.event.data.get("tool")?.as_str()?;
    let builtin = failure
        .event
        .data
        .get("tool_source")
        .and_then(serde_json::Value::as_str)
        == Some("builtin");

    let proposal_event = events
        .iter()
        .filter(|record| {
            record.event.kind == EventKind::ProposalReceived
                && record.path.as_deref() == Some(journal)
                && record.event.proposal_id.as_deref() == Some(proposal_id)
                && record.event.timestamp <= failure.event.timestamp
        })
        .max_by_key(|record| record.event.timestamp)?;
    let proposal: ActionProposal =
        serde_json::from_value(proposal_event.event.data.get("proposal")?.clone()).ok()?;
    if proposal.id != proposal_id {
        return None;
    }
    let action = proposal
        .actions
        .into_iter()
        .find(|action| action.id == action_id)?;
    let tool = action.tool?;
    if tool != failed_tool {
        return None;
    }
    Some((
        builtin,
        ReconstructedToolCall {
            tool,
            params: action.parameters.into_iter().collect(),
        },
    ))
}

fn failure_belongs_to_detection(detection: &Detection, failure: &EventEvidence) -> bool {
    let Some(journal) = failure.path.as_deref() else {
        return false;
    };
    detection.provenance().iter().any(|provenance| {
        provenance.path() == Some(journal)
            && provenance.run_id() == failure.event.run_id.as_deref()
            && provenance.event_id().is_some_and(|event_id| {
                failure.event.hash.as_deref() == Some(event_id)
                    || failure.event.action_id.as_deref() == Some(event_id)
            })
    })
}

#[derive(Serialize)]
struct PersistedReconstructedCall<'a> {
    eligible: bool,
    reconstructed_call: Option<&'a ReconstructedToolCall>,
}

fn persist_reconstructed_call(
    path: &Path,
    eligible: bool,
    reconstructed_call: Option<&ReconstructedToolCall>,
) -> Result<(), String> {
    let parent = path
        .parent()
        .ok_or_else(|| "reconstructed self-heal call path has no parent".to_string())?;
    car_secrets::ensure_private_dir(parent)
        .map_err(|error| format!("create reconstructed self-heal call directory: {error}"))?;
    let mut bytes = serde_json::to_vec_pretty(&PersistedReconstructedCall {
        eligible,
        reconstructed_call,
    })
    .map_err(|error| format!("serialize reconstructed self-heal call: {error}"))?;
    bytes.push(b'\n');
    let mut file = if path.exists() {
        car_secrets::open_private_truncate(path)
    } else {
        car_secrets::create_private_file(path)
    }
    .map_err(|error| {
        format!(
            "open reconstructed self-heal call {}: {error}",
            path.display()
        )
    })?;
    file.write_all(&bytes)
        .map_err(|error| format!("write reconstructed self-heal call: {error}"))?;
    file.flush()
        .map_err(|error| format!("flush reconstructed self-heal call: {error}"))?;
    file.sync_all()
        .map_err(|error| format!("sync reconstructed self-heal call: {error}"))?;
    car_secrets::revalidate_private_path(path, &file)
        .map_err(|error| format!("revalidate reconstructed self-heal call: {error}"))
}

fn persist_private_json<T: Serialize>(path: &Path, value: &T) -> Result<(), String> {
    let mut bytes = serde_json::to_vec_pretty(value)
        .map_err(|error| format!("serialize private self-heal artifact: {error}"))?;
    bytes.push(b'\n');
    persist_private_bytes(path, &bytes)
}

fn persist_private_bytes(path: &Path, bytes: &[u8]) -> Result<(), String> {
    let parent = path
        .parent()
        .ok_or_else(|| "private self-heal artifact path has no parent".to_string())?;
    car_secrets::ensure_private_dir(parent)
        .map_err(|error| format!("create private self-heal artifact directory: {error}"))?;
    let mut file = if path.exists() {
        car_secrets::open_private_truncate(path)
    } else {
        car_secrets::create_private_file(path)
    }
    .map_err(|error| {
        format!(
            "open private self-heal artifact {}: {error}",
            path.display()
        )
    })?;
    file.write_all(bytes)
        .map_err(|error| format!("write private self-heal artifact: {error}"))?;
    file.flush()
        .map_err(|error| format!("flush private self-heal artifact: {error}"))?;
    file.sync_all()
        .map_err(|error| format!("sync private self-heal artifact: {error}"))?;
    car_secrets::revalidate_private_path(path, &file)
        .map_err(|error| format!("revalidate private self-heal artifact: {error}"))
}

fn resolve_car_cli_executable() -> Option<PathBuf> {
    let executable_name = if cfg!(windows) { "car.exe" } else { "car" };
    if let Ok(current) = std::env::current_exe() {
        if let Some(sibling) = current.parent().map(|parent| parent.join(executable_name)) {
            if sibling.is_file() {
                return Some(sibling);
            }
        }
    }
    std::env::var_os("PATH").and_then(|path| {
        std::env::split_paths(&path)
            .map(|entry| entry.join(executable_name))
            .find(|candidate| candidate.is_file())
    })
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RemoteDedupResult {
    Proceed,
    AwaitingReview,
    Parked,
}

#[derive(Debug, Deserialize)]
struct RemoteSelfhealPr {
    number: u64,
    url: String,
    state: String,
    #[serde(rename = "mergedAt")]
    merged_at: Option<String>,
    body: String,
}

async fn check_checkout_replay_verb(checkout: &Path, target_dir: &Path) -> Result<(), String> {
    car_secrets::ensure_private_dir(target_dir)
        .map_err(|error| format!("create replay probe target: {error}"))?;
    let build = tokio::process::Command::new("cargo")
        .current_dir(checkout.join("car-rs"))
        .args(["build", "-p", "car-cli"])
        .env("CARGO_TARGET_DIR", target_dir)
        .kill_on_drop(true)
        .output()
        .await
        .map_err(|error| format!("could not build checkout car CLI: {error}"))?;
    if !build.status.success() {
        return Err(format!(
            "checkout car CLI build failed with exit {}: {}",
            build.status.code().unwrap_or(-1),
            output_tail(&build.stderr, 8)
        ));
    }

    #[cfg(windows)]
    let car = target_dir.join("debug/car.exe");
    #[cfg(not(windows))]
    let car = target_dir.join("debug/car");
    let help = tokio::process::Command::new(&car)
        .args(["tools", "call", "--help"])
        .kill_on_drop(true)
        .output()
        .await
        .map_err(|error| format!("checkout-built car has no runnable replay verb: {error}"))?;
    if help.status.success() {
        Ok(())
    } else {
        Err(format!(
            "checkout-built `car tools call --help` exited {}: {}",
            help.status.code().unwrap_or(-1),
            output_tail(&help.stderr, 8)
        ))
    }
}

async fn query_remote_selfheal_pr(
    checkout: &Path,
    dedup_key: &str,
) -> Result<Option<RemoteSelfhealPr>, String> {
    let head = format!("car/selfheal/{dedup_key}");
    let output = tokio::process::Command::new("gh")
        .current_dir(checkout)
        .args([
            "pr",
            "list",
            "--repo",
            "Parslee-ai/car",
            "--head",
            &head,
            "--state",
            "all",
            "--json",
            "number,url,state,mergedAt,body",
        ])
        .kill_on_drop(true)
        .output()
        .await
        .map_err(|error| format!("remote self-heal PR lookup could not start: {error}"))?;
    if !output.status.success() {
        return Err(format!(
            "remote self-heal PR lookup failed with exit {}: {}",
            output.status.code().unwrap_or(-1),
            output_tail(&output.stderr, 8)
        ));
    }
    let mut prs: Vec<RemoteSelfhealPr> = serde_json::from_slice(&output.stdout)
        .map_err(|error| format!("parse remote self-heal PR lookup: {error}"))?;
    let marker = format!("<!-- car-selfheal:key={dedup_key} -->");
    if prs.iter().any(|pr| !pr.body.contains(&marker)) {
        return Err(format!(
            "remote branch '{head}' has a PR without the trusted self-heal marker"
        ));
    }
    prs.sort_by_key(|pr| if pr.state == "OPEN" { 0 } else { 1 });
    Ok(prs.into_iter().next())
}

fn output_tail(bytes: &[u8], lines: usize) -> String {
    let text = String::from_utf8_lossy(bytes);
    text.lines()
        .rev()
        .take(lines)
        .collect::<Vec<_>>()
        .into_iter()
        .rev()
        .collect::<Vec<_>>()
        .join("\n")
}

fn classify_code_task_output(stdout: &[u8]) -> String {
    code_task_failure_class(stdout).unwrap_or_else(|| "missing_run_end".to_string())
}

fn code_task_failure_class(stdout: &[u8]) -> Option<String> {
    String::from_utf8_lossy(stdout)
        .lines()
        .rev()
        .filter_map(|line| serde_json::from_str::<serde_json::Value>(line).ok())
        .find(|event| event.get("type").and_then(serde_json::Value::as_str) == Some("run_end"))
        .and_then(|event| {
            event
                .get("failure_class")
                .and_then(serde_json::Value::as_str)
                .map(str::to_string)
        })
}

fn route_name(route: SelfhealRoute) -> &'static str {
    match route {
        SelfhealRoute::Local => "local",
        SelfhealRoute::Feedback => "feedback",
        SelfhealRoute::LedgerOnly => "ledger-only",
    }
}

fn render_local_issue(
    path: &Path,
    detection: &Detection,
    eligible: Option<bool>,
    ineligible_reason: Option<&str>,
) -> 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\
Auto-Fix-Eligible: {}\n\
Auto-Fix-Ineligible-Reason: {}\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(),
        eligible.map_or("not-applicable", |value| if value {
            "true"
        } else {
            "false"
        }),
        ineligible_reason.unwrap_or("(none)"),
        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],
    max_rounds_per_key: u32,
) -> (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,
                eligible,
                reconstructed_call,
                reconstructed_call_path,
                ..
            } => {
                detections.insert(
                    detection.dedup_key().to_string(),
                    RoutedDetection {
                        detection: detection.clone(),
                        route: *route,
                        local_issue_path: local_issue_path.clone(),
                        eligible: *eligible,
                        reconstructed_call: reconstructed_call.clone(),
                        reconstructed_call_path: reconstructed_call_path.clone(),
                        auto_fix_attempts: 0,
                        auto_fix_exhausted: false,
                        auto_fix_in_progress: false,
                        last_auto_fix_attempt: None,
                        auto_fix_awaiting_review: false,
                        auto_fix_parked: false,
                        remote_pr_number: None,
                        remote_pr_url: None,
                    },
                );
            }
            LedgerRecord::Dismissal { dedup_key, .. } => {
                dismissed.insert(dedup_key.clone());
            }
            LedgerRecord::Tick { .. }
            | LedgerRecord::FixStarted { .. }
            | LedgerRecord::FixAttempt { .. }
            | LedgerRecord::RemotePr { .. } => {}
        }
    }
    for (key, detection) in &mut detections {
        apply_attempt_state(detection, records, max_rounds_per_key);
        apply_remote_pr_state(detection, records);
        if dismissed.contains(key) {
            detection.auto_fix_in_progress = false;
        }
    }
    for key in &dismissed {
        detections.remove(key);
    }
    (detections, dismissed)
}

fn apply_attempt_state(
    detection: &mut RoutedDetection,
    records: &[LedgerRecord],
    max_rounds_per_key: u32,
) {
    let key = detection.dedup_key().to_string();
    let started = records
        .iter()
        .filter_map(|record| match record {
            LedgerRecord::FixStarted {
                dedup_key, round, ..
            } if dedup_key == &key => Some(*round),
            _ => None,
        })
        .collect::<BTreeSet<_>>();
    let finished = records
        .iter()
        .filter_map(|record| match record {
            LedgerRecord::FixAttempt { result } if result.dedup_key == key => Some(result.round),
            _ => None,
        })
        .collect::<BTreeSet<_>>();
    detection.auto_fix_attempts = u32::try_from(started.len()).unwrap_or(u32::MAX);
    detection.auto_fix_exhausted = detection.auto_fix_attempts >= max_rounds_per_key;
    detection.auto_fix_in_progress = started.iter().any(|round| !finished.contains(round));
    detection.last_auto_fix_attempt = records.iter().rev().find_map(|record| match record {
        LedgerRecord::FixAttempt { result } if result.dedup_key == key => Some(result.clone()),
        _ => None,
    });
}

fn apply_remote_pr_state(detection: &mut RoutedDetection, records: &[LedgerRecord]) {
    let key = detection.dedup_key().to_string();
    if let Some((disposition, number, url)) = records.iter().rev().find_map(|record| match record {
        LedgerRecord::RemotePr {
            dedup_key,
            disposition,
            pr_number,
            pr_url,
            ..
        } if dedup_key == &key => Some((*disposition, *pr_number, pr_url.clone())),
        _ => None,
    }) {
        detection.auto_fix_awaiting_review = disposition == RemotePrDisposition::AwaitingReview;
        detection.auto_fix_parked = disposition == RemotePrDisposition::Parked;
        detection.remote_pr_number = Some(number);
        detection.remote_pr_url = Some(url);
    }
}

fn starts_today(records: &[LedgerRecord], now: DateTime<Utc>) -> u32 {
    u32::try_from(
        records
            .iter()
            .filter(|record| {
                matches!(
                    record,
                    LedgerRecord::FixStarted { started_at, .. }
                        if started_at.date_naive() == now.date_naive()
                )
            })
            .count(),
    )
    .unwrap_or(u32::MAX)
}

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(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn auto_fix_config_defaults_on_and_refuses_raised_hard_limits() {
        let temp = tempfile::TempDir::new().unwrap();
        let path = temp.path().join("config.toml");
        let default = read_auto_fix_config(&path).unwrap();
        assert!(default.auto_fix);
        assert_eq!(default.max_concurrent, 1);
        assert_eq!(default.max_per_day, 3);
        assert_eq!(default.max_rounds_per_key, 3);

        std::fs::write(
            &path,
            "[selfheal]\nauto_fix=false\nmax_per_day=2\nmax_rounds_per_key=1\n",
        )
        .unwrap();
        let lowered = read_auto_fix_config(&path).unwrap();
        assert!(!lowered.auto_fix);
        assert_eq!(lowered.max_per_day, 2);
        assert_eq!(lowered.max_rounds_per_key, 1);

        for (config, refusal) in [
            ("[selfheal]\nmax_concurrent=2\n", "max_concurrent must be 1"),
            (
                "[selfheal]\nmax_per_day=4\n",
                "max_per_day must be between 1 and 3",
            ),
            (
                "[selfheal]\nmax_rounds_per_key=4\n",
                "max_rounds_per_key must be between 1 and 3",
            ),
        ] {
            std::fs::write(&path, config).unwrap();
            assert!(read_auto_fix_config(&path).unwrap_err().contains(refusal));
        }
    }

    #[test]
    fn auto_fix_excludes_side_effecting_gui_delegation_even_when_builtin() {
        assert!(auto_fix_tool_allowed("read_file", true));
        assert!(!auto_fix_tool_allowed("delegate_gui", true));
        assert!(!auto_fix_tool_allowed("read_file", false));
    }

    #[test]
    fn code_task_terminal_failure_class_comes_from_run_end_record() {
        assert_eq!(
            code_task_failure_class(
                b"{\"type\":\"progress\"}\n{\"type\":\"run_end\",\"failure_class\":\"contract_not_green\"}\n"
            )
            .as_deref(),
            Some("contract_not_green")
        );
        assert!(code_task_failure_class(b"not json\n").is_none());
        assert_eq!(classify_code_task_output(b"not json\n"), "missing_run_end");
    }
}