"""Crash recovery and reconciliation for evolution commits.""" from __future__ import annotations import hashlib import json import shutil import time from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any from openspace.skill_engine.patch import collect_skill_snapshot from openspace.skill_engine.registry import write_skill_id from openspace.skill_engine.skill_utils import validate_skill_dir from openspace.utils.logging import Logger from .audit import EvolutionActionRecord logger = Logger.get_logger(__name__) DEFAULT_STALE_JOB_TIMEOUT_S = 30 * 60.0 DEFAULT_STAGING_RETENTION_S = 7 * 24 * 60 * 60.0 _TERMINAL_ACTION_STATUSES = { "committed", "committed_reconciled", "failed", "failed_needs_review", } @dataclass(frozen=True, slots=True) class EvolutionRecoveryResult: stale_jobs_recovered: int = 0 actions_reconciled: int = 0 actions_failed: int = 0 actions_needing_review: int = 0 staging_dirs_removed: int = 0 errors: list[str] = field(default_factory=list) def to_dict(self) -> dict[str, Any]: return asdict(self) class EvolutionRecovery: """Best-effort startup recovery across TriggerJobs, EvidenceStore, and skills.""" def __init__( self, *, evidence_store: Any, skill_store: Any | None = None, registry: Any | None = None, trigger_engine: Any | None = None, trigger_store: Any | None = None, stale_job_timeout_s: float = DEFAULT_STALE_JOB_TIMEOUT_S, staging_retention_s: float = DEFAULT_STAGING_RETENTION_S, ) -> None: self.evidence_store = evidence_store self.skill_store = skill_store self.registry = registry self.trigger_engine = trigger_engine self.trigger_store = trigger_store or getattr(trigger_engine, "store", None) self.stale_job_timeout_s = max(0.0, float(stale_job_timeout_s)) self.staging_retention_s = max(0.0, float(staging_retention_s)) def run_startup_recovery(self) -> EvolutionRecoveryResult: """Run all startup recovery steps without raising to runtime init.""" stale_jobs = 0 reconciled = 0 failed = 0 needs_review = 0 staging_removed = 0 errors: list[str] = [] try: stale_jobs = self._recover_stale_jobs() except Exception as exc: logger.debug("Evolution stale job recovery failed", exc_info=True) errors.append(f"stale_jobs: {exc}") try: action_result = self._reconcile_committing_actions() reconciled += action_result.actions_reconciled failed += action_result.actions_failed needs_review += action_result.actions_needing_review errors.extend(action_result.errors) except Exception as exc: logger.debug("Evolution action reconciliation failed", exc_info=True) errors.append(f"actions: {exc}") try: staging_removed = self._cleanup_stale_staging_dirs() except Exception as exc: logger.debug("Evolution staging cleanup failed", exc_info=True) errors.append(f"staging_cleanup: {exc}") return EvolutionRecoveryResult( stale_jobs_recovered=stale_jobs, actions_reconciled=reconciled, actions_failed=failed, actions_needing_review=needs_review, staging_dirs_removed=staging_removed, errors=errors, ) def _recover_stale_jobs(self) -> int: recover = getattr(self.trigger_engine, "recover_stale_jobs", None) if callable(recover): return int(recover(timeout_s=self.stale_job_timeout_s) or 0) store_recover = getattr(self.trigger_store, "recover_stale_jobs", None) if callable(store_recover): return int(store_recover(timeout_s=self.stale_job_timeout_s) or 0) return 0 def _reconcile_committing_actions(self) -> EvolutionRecoveryResult: if self.skill_store is None: return EvolutionRecoveryResult() list_actions = getattr(self.evidence_store, "list_actions", None) if not callable(list_actions): return EvolutionRecoveryResult() actions = list_actions(status="committing", limit=100) reconciled = 0 failed = 0 needs_review = 0 errors: list[str] = [] for action in actions: try: status = self._reconcile_action(action) if status == "committed_reconciled": reconciled += 1 elif status == "failed_needs_review": needs_review += 1 elif status == "failed": failed += 1 except Exception as exc: logger.debug( "Evolution action reconciliation failed for %s", getattr(action, "action_id", ""), exc_info=True, ) errors.append(f"{getattr(action, 'action_id', '')}: {exc}") self._record_failure(action, "recovery", "failed_retryable", str(exc)) return EvolutionRecoveryResult( actions_reconciled=reconciled, actions_failed=failed, actions_needing_review=needs_review, errors=errors, ) def _reconcile_action(self, action: EvolutionActionRecord) -> str: target_dir = _path_or_none(action.active_target_dir) proposed_dir = _resolve_proposed_dir(action) record = self._load_skill_record(action) if record is not None: mismatch_reason = self._record_reconcile_mismatch( action, record, target_dir, proposed_dir, ) if mismatch_reason: self._finalize_failed(action, "failed_needs_review", mismatch_reason) return "failed_needs_review" self._finalize_committed(action, record) return "committed_reconciled" backup_dir = _path_or_none(action.backup_dir) target_unchanged = _dirs_equivalent(target_dir, backup_dir) missing_proposed = proposed_dir is None or not proposed_dir.is_dir() if target_unchanged or missing_proposed: reason = ( "recovery: active files unchanged" if target_unchanged else "recovery: proposed staging content missing" ) self._finalize_failed(action, "failed", reason) return "failed" if target_dir is not None and _dirs_equivalent(target_dir, proposed_dir): restored = self._restore_backup(action, target_dir, backup_dir) if restored: self._finalize_failed( action, "failed", "recovery: active files matched unrecorded proposed staging; backup restored", ) return "failed" self._finalize_failed( action, "failed_needs_review", "recovery: active files matched unrecorded proposed staging but backup restore failed", ) return "failed_needs_review" self._finalize_failed( action, "failed_needs_review", "recovery: active target state is unknown and SkillStore record is missing", ) return "failed_needs_review" def _load_skill_record(self, action: EvolutionActionRecord) -> Any | None: if self.skill_store is None: return None if action.skill_id: load_record = getattr(self.skill_store, "load_record", None) if callable(load_record): record = load_record(action.skill_id) if record is not None: return record return self._find_skill_record_for_action(action) def _find_skill_record_for_action(self, action: EvolutionActionRecord) -> Any | None: records = self._load_all_skill_records() if not records: return None action_id = str(action.action_id or "") target_dir = _path_or_none(action.active_target_dir) staging_id = Path(action.staging_dir).name if action.staging_dir else "" evidence_refs = {str(ref) for ref in action.evidence_refs if ref} path_matched_records = [ record for record in records if _record_path_matches_target(record, target_dir) ] if action_id: for record in records: if _record_lineage_value(record, "evolution_action_id") == action_id: return record if staging_id: for record in path_matched_records: if staging_id in _record_search_blob(record): return record if evidence_refs: for record in path_matched_records: provenance_refs = { str(ref) for ref in (_record_lineage_value(record, "provenance_refs") or []) if ref } if evidence_refs.intersection(provenance_refs): return record if action_id: for record in path_matched_records: if action_id in _record_search_blob(record): return record proposed_dir = _resolve_proposed_dir(action) if proposed_dir is not None and proposed_dir.is_dir(): proposed_hash = _snapshot_hash(proposed_dir) for record in path_matched_records: if _record_snapshot_hash(record) == proposed_hash: return record return None def _load_all_skill_records(self) -> list[Any]: if self.skill_store is None: return [] load_all = getattr(self.skill_store, "load_all", None) if not callable(load_all): return [] records = load_all(active_only=False) if isinstance(records, dict): return list(records.values()) return list(records or []) def _record_reconcile_mismatch( self, action: EvolutionActionRecord, record: Any, target_dir: Path | None, proposed_dir: Path | None, ) -> str | None: record_dir = _record_skill_dir(record) if (target_dir is None or not target_dir.is_dir()) and record_dir is not None: target_dir = record_dir if target_dir is None or not target_dir.is_dir(): return "recovery: active target missing; cannot verify SkillStore record" if proposed_dir is None or not proposed_dir.is_dir(): return "recovery: proposed staging content missing; cannot verify SkillStore record" if not _record_path_matches_target(record, target_dir): return "recovery: SkillStore record path does not match active target" proposed_hash = _snapshot_hash(proposed_dir) if _snapshot_hash(target_dir) != proposed_hash: return "recovery: active target does not match proposed staging content" record_hash = _record_snapshot_hash(record) if record_hash and record_hash != proposed_hash: return "recovery: SkillStore record snapshot does not match proposed staging content" return None def _finalize_committed(self, action: EvolutionActionRecord, record: Any) -> None: target_dir = _path_or_none(action.active_target_dir) record_dir = _record_skill_dir(record) if (target_dir is None or not target_dir.is_dir()) and record_dir is not None: target_dir = record_dir if target_dir is None: raise RuntimeError("active target missing") validation_error = validate_skill_dir(target_dir) if validation_error: raise RuntimeError(f"active target invalid: {validation_error}") write_skill_id(target_dir, record.skill_id, raise_on_error=True) self._refresh_registry(action, target_dir, record) self.evidence_store.finalize_action( action.action_id, status="committed_reconciled", skill_id=record.skill_id, changed_files=action.changed_files, backup_dir=action.backup_dir, raw_backrefs=action.evidence_refs, ) def _refresh_registry( self, action: EvolutionActionRecord, target_dir: Path, record: Any, ) -> None: if self.registry is None: return load_skill_from_dir = getattr(self.registry, "load_skill_from_dir", None) if not callable(load_skill_from_dir): return meta = load_skill_from_dir(target_dir) if meta is None: raise RuntimeError(f"registry could not load skill from {target_dir}") if str(action.action_type).upper() == "FIX": old_skill_id = action.parent_skill_ids[0] if action.parent_skill_ids else record.skill_id update_skill = getattr(self.registry, "update_skill", None) if callable(update_skill): update_skill(old_skill_id, meta) return add_skill = getattr(self.registry, "add_skill", None) if callable(add_skill): add_skill(meta) def _restore_backup( self, action: EvolutionActionRecord, target_dir: Path | None, backup_dir: Path | None, ) -> bool: if str(action.action_type).upper() != "FIX": return False if target_dir is None or backup_dir is None or not backup_dir.is_dir(): return False try: if target_dir.exists(): shutil.rmtree(target_dir) shutil.copytree(backup_dir, target_dir) return True except Exception as exc: self._record_failure(action, "backup_restore", "failed_needs_review", str(exc)) return False def _finalize_failed( self, action: EvolutionActionRecord, status: str, reason: str, ) -> None: self._record_failure(action, "recovery", status, reason) self.evidence_store.finalize_action( action.action_id, status=status, skill_id=action.skill_id, changed_files=action.changed_files, backup_dir=action.backup_dir, failure_reason=reason, raw_backrefs=action.evidence_refs, ) def _record_failure( self, action: EvolutionActionRecord, phase: str, status: str, reason: str, ) -> None: recorder = getattr(self.evidence_store, "record_action_failure", None) if not callable(recorder): return try: recorder(action.action_id, phase=phase, status=status, error=reason) except Exception: logger.debug("Failed to record evolution recovery failure", exc_info=True) def _cleanup_stale_staging_dirs(self) -> int: list_actions = getattr(self.evidence_store, "list_actions", None) if not callable(list_actions): return 0 actions = list_actions(limit=1000) active_staging = { str(Path(action.staging_dir).expanduser().resolve()) for action in actions if action.staging_dir and action.commit_status not in _TERMINAL_ACTION_STATUSES } now = time.time() removed = 0 for action in actions: if action.commit_status not in _TERMINAL_ACTION_STATUSES: continue if not action.staging_dir: continue staging = Path(action.staging_dir).expanduser().resolve() if str(staging) in active_staging or not staging.is_dir(): continue try: if now - staging.stat().st_mtime < self.staging_retention_s: continue shutil.rmtree(staging) removed += 1 except Exception: logger.debug("Failed to clean evolution staging dir %s", staging, exc_info=True) return removed def _resolve_proposed_dir(action: EvolutionActionRecord) -> Path | None: staging = _path_or_none(action.staging_dir) target = _path_or_none(action.active_target_dir) if staging is None: return None proposed_root = staging / "proposed" if not proposed_root.is_dir(): return None if target is not None: preferred = proposed_root / target.name if preferred.is_dir(): return preferred dirs = [child for child in proposed_root.iterdir() if child.is_dir()] if len(dirs) == 1: return dirs[0] return None def _dirs_equivalent(left: Path | None, right: Path | None) -> bool: if left is None or right is None: return False if not left.exists() and not right.exists(): return True if not left.is_dir() or not right.is_dir(): return False return _snapshot_hash(left) == _snapshot_hash(right) def _snapshot_hash(path: Path) -> str: snapshot = collect_skill_snapshot(path) return _snapshot_mapping_hash(snapshot) def _snapshot_mapping_hash(snapshot: dict[str, str]) -> str: encoded = json.dumps(snapshot, ensure_ascii=False, sort_keys=True).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def _record_snapshot_hash(record: Any) -> str | None: snapshot = _record_lineage_value(record, "content_snapshot") if not isinstance(snapshot, dict) or not snapshot: return None normalized = {str(key): str(value) for key, value in snapshot.items()} return _snapshot_mapping_hash(normalized) def _record_path_matches_target(record: Any, target_dir: Path | None) -> bool: if target_dir is None: return False skill_dir = _record_skill_dir(record) if skill_dir is None: return False return skill_dir == target_dir def _record_skill_dir(record: Any) -> Path | None: raw_path = str(getattr(record, "path", "") or "") if not raw_path: return None try: path = Path(raw_path).expanduser().resolve() except Exception: return None return path.parent if path.name == "SKILL.md" else path def _record_lineage_value(record: Any, name: str) -> Any: lineage = getattr(record, "lineage", None) if lineage is None: return None if isinstance(lineage, dict): return lineage.get(name) return getattr(lineage, name, None) def _record_search_blob(record: Any) -> str: parts = [ getattr(record, "skill_id", ""), getattr(record, "name", ""), getattr(record, "path", ""), _record_lineage_value(record, "source_task_id"), _record_lineage_value(record, "change_summary"), _record_lineage_value(record, "evolution_action_id"), ] provenance = _record_lineage_value(record, "provenance_refs") or [] parts.extend(provenance) return "\n".join(str(part or "") for part in parts) def _path_or_none(value: str | None) -> Path | None: if not value: return None try: return Path(value).expanduser().resolve() except Exception: return None __all__ = [ "DEFAULT_STAGING_RETENTION_S", "DEFAULT_STALE_JOB_TIMEOUT_S", "EvolutionRecovery", "EvolutionRecoveryResult", ]