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;
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,
];
#[derive(Debug, Clone, Default)]
pub struct SelfhealSourceProbe {
explicit_checkout: Option<PathBuf>,
candidates: Vec<PathBuf>,
setup_refusal: Option<String>,
}
impl SelfhealSourceProbe {
pub fn empty() -> Self {
Self::default()
}
pub fn explicit(path: PathBuf) -> Self {
Self {
explicit_checkout: Some(path),
..Self::default()
}
}
pub fn candidates(paths: impl IntoIterator<Item = PathBuf>) -> Self {
Self {
candidates: paths.into_iter().collect(),
..Self::default()
}
}
pub fn with_candidates(mut self, paths: impl IntoIterator<Item = PathBuf>) -> Self {
self.candidates.extend(paths);
dedup_paths(&mut self.candidates);
self
}
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())
}
#[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("; ")
))
}
}
}
#[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),
}
}
}
#[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,
}
}
}
#[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>,
}
#[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>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub eligible: Option<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub reconstructed_call: Option<ReconstructedToolCall>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub reconstructed_call_path: Option<PathBuf>,
#[serde(default, skip_serializing_if = "is_zero_u32")]
pub auto_fix_attempts: u32,
#[serde(default, skip_serializing_if = "is_false")]
pub auto_fix_exhausted: bool,
#[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>,
#[serde(default, skip_serializing_if = "is_false")]
pub auto_fix_awaiting_review: bool,
#[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>,
}
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,
})
}
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))
.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,
("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(¶ms_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: ¶ms_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}"))?;
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)
.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();
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,
};
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,
));
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)
}
}
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 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(§ion.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(§ion.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),
})
}
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(¶ms_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");
}
}