From a3b3ba8c09c1ce863499c480ceaef2e8d6c5047c Mon Sep 17 00:00:00 2001 From: De-pitcher Date: Mon, 7 Sep 2026 06:38:13 +0100 Subject: [PATCH 1/2] feat(agent): implement Phase 36 self-healing, plan drift detection and auto-rollback --- agentcli/agent/__init__.py | 20 ++ agentcli/agent/drift_detector.py | 258 ++++++++++++++++++ agentcli/agent/events.py | 34 +++ agentcli/agent/loop.py | 112 ++++++++ agentcli/agent/rollback.py | 294 +++++++++++++++++++++ agentcli/cli.py | 51 ++++ agentcli/session.py | 8 + agentcli/ui/prompt.py | 27 +- agentcli/ui/render.py | 48 ++++ agentcli/ui/tui_app.py | 60 +++++ tests/test_phase36_self_healing.py | 402 +++++++++++++++++++++++++++++ 11 files changed, 1313 insertions(+), 1 deletion(-) create mode 100644 agentcli/agent/drift_detector.py create mode 100644 agentcli/agent/rollback.py create mode 100644 tests/test_phase36_self_healing.py diff --git a/agentcli/agent/__init__.py b/agentcli/agent/__init__.py index 439d2e7..cb0d60e 100644 --- a/agentcli/agent/__init__.py +++ b/agentcli/agent/__init__.py @@ -19,7 +19,15 @@ LoopIterationLimitError — raised when max_iterations is exceeded """ +from .drift_detector import ( + ActionRecord, + DriftReport, + DriftSeverity, + PlanDriftDetector, +) from .events import ( + AutoHealingRollbackEvent, + DriftDetectedEvent, FinishEvent, LoopErrorEvent, LoopEvent, @@ -27,22 +35,34 @@ ReflectEvent, StepResultEvent, StepStartEvent, + StrategyRecoveryEvent, ) from .loop import AgentLoop, LoopIterationLimitError from .reflector import DefaultReflector, ReflectDecision from .registry import ToolRegistry +from .rollback import AutoHealingManager, HealingSnapshot __all__ = [ + "ActionRecord", "AgentLoop", + "AutoHealingManager", + "AutoHealingRollbackEvent", "DefaultReflector", + "DriftDetectedEvent", + "DriftReport", + "DriftSeverity", "FinishEvent", + "HealingSnapshot", "LoopErrorEvent", "LoopEvent", "LoopIterationLimitError", + "PlanDriftDetector", "PlanEvent", "ReflectDecision", "ReflectEvent", "StepResultEvent", "StepStartEvent", + "StrategyRecoveryEvent", "ToolRegistry", ] + diff --git a/agentcli/agent/drift_detector.py b/agentcli/agent/drift_detector.py new file mode 100644 index 0000000..d53138c --- /dev/null +++ b/agentcli/agent/drift_detector.py @@ -0,0 +1,258 @@ +"""Plan Drift & Action Cycle Detector (Phase 36). + +Tracks execution history, measures drift against planned goals, identifies +oscillating actions (loops/ping-pong), and calculates drift severity scores. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import time +from dataclasses import dataclass, field +from enum import Enum +from typing import Any + +logger = logging.getLogger(__name__) + + +class DriftSeverity(str, Enum): + """Drift severity levels.""" + + LOW = "low" + MODERATE = "moderate" + CRITICAL = "critical" + + +@dataclass +class ActionRecord: + """Record of a single executed action or subagent step.""" + + iteration: int + step_index: int + agent_type: str + payload: dict[str, Any] + success: bool + error: str | None = None + timestamp: float = field(default_factory=time.time) + action_signature: str = "" + + def __post_init__(self) -> None: + if not self.action_signature: + self.action_signature = self.compute_signature(self.agent_type, self.payload) + + @staticmethod + def compute_signature(agent_type: str, payload: dict[str, Any]) -> str: + """Derive a canonical normalized signature for an action.""" + # Extract salient keys + salient: dict[str, Any] = { + "type": agent_type, + "cmd": payload.get("command") or payload.get("cmd") or "", + "path": payload.get("path") or payload.get("file") or payload.get("target") or "", + "op": payload.get("operation") or payload.get("op") or "", + "query": payload.get("query") or payload.get("pattern") or "", + } + # Strip empty fields + filtered = {k: v for k, v in salient.items() if v} + serialized = json.dumps(filtered, sort_keys=True) + h = hashlib.sha256(serialized.encode("utf-8")).hexdigest()[:12] + readable = f"{agent_type}:{salient.get('op') or salient.get('cmd') or salient.get('path') or salient.get('query') or 'action'}" + return f"{readable}#{h}" + + +@dataclass +class DriftReport: + """Diagnostic report describing current plan drift and cycle status.""" + + drift_score: float # 0.0 (aligned) to 1.0 (severely drifted/stuck) + severity: DriftSeverity + is_loop_detected: bool + cycle_signature: str | None + reasons: list[str] = field(default_factory=list) + consecutive_failures: int = 0 + unique_actions_count: int = 0 + total_actions_count: int = 0 + + +class PlanDriftDetector: + """Detects loops, oscillating edits, repeated errors, and execution drift.""" + + def __init__( + self, + loop_threshold: int = 2, + max_drift_score: float = 0.7, + ) -> None: + self.loop_threshold = loop_threshold + self.max_drift_score = max_drift_score + self._history: list[ActionRecord] = [] + self._consecutive_failures: int = 0 + + def record_action( + self, + iteration: int, + step_index: int, + agent_type: str, + payload: dict[str, Any], + success: bool, + error: str | None = None, + ) -> ActionRecord: + """Record an executed step and update failure counters.""" + rec = ActionRecord( + iteration=iteration, + step_index=step_index, + agent_type=agent_type, + payload=payload, + success=success, + error=error, + ) + self._history.append(rec) + if success: + self._consecutive_failures = 0 + else: + self._consecutive_failures += 1 + return rec + + def detect_cycles(self) -> tuple[bool, str | None, list[str]]: + """Detect execution loops, repeated errors, or ping-pong oscillations. + + Returns: + (is_loop_detected, cycle_signature, reasons) + """ + reasons: list[str] = [] + if len(self._history) < 2: + return False, None, [] + + # 1. Consecutive identical failures on the same action signature + if self._consecutive_failures >= self.loop_threshold: + recent_failed = [r for r in self._history[-self._consecutive_failures :] if not r.success] + if len(recent_failed) >= self.loop_threshold: + sigs = [r.action_signature for r in recent_failed] + if len(set(sigs)) == 1: + sig = sigs[0] + reasons.append( + f"Action '{sig}' failed {len(recent_failed)} consecutive times without recovery." + ) + return True, sig, reasons + + # 2. Ping-pong oscillation: A -> B -> A -> B + if len(self._history) >= 4: + s1, s2, s3, s4 = [r.action_signature for r in self._history[-4:]] + if s1 == s3 and s2 == s4 and s1 != s2: + cycle_sig = f"{s1} <-> {s2}" + reasons.append(f"Oscillating ping-pong execution detected between '{s1}' and '{s2}'.") + return True, cycle_sig, reasons + + # 3. Repeated identical tool actions (>= 3 times in recent 5 actions) + recent = self._history[-5:] + if len(recent) >= 3: + sig_counts: dict[str, int] = {} + for r in recent: + sig_counts[r.action_signature] = sig_counts.get(r.action_signature, 0) + 1 + for sig, count in sig_counts.items(): + if count >= 3: + reasons.append( + f"Repeated execution of action '{sig}' ({count} times in last {len(recent)} steps)." + ) + return True, sig, reasons + + # 4. Same file modified repeatedly with errors + recent_file_errors: list[str] = [] + for r in reversed(self._history[-6:]): + if not r.success and r.payload.get("path"): + recent_file_errors.append(str(r.payload["path"])) + if len(recent_file_errors) >= 3: + most_frequent_path = max(set(recent_file_errors), key=recent_file_errors.count) + if recent_file_errors.count(most_frequent_path) >= 2: + reasons.append( + f"Repeated mutation failures on file '{most_frequent_path}'." + ) + return True, f"file:{most_frequent_path}", reasons + + return False, None, [] + + def evaluate_drift( + self, + current_plan: list[dict[str, Any]] | None = None, + planned_total_steps: int | None = None, + ) -> DriftReport: + """Calculate overall drift score and classify severity.""" + total = len(self._history) + if total == 0: + return DriftReport( + drift_score=0.0, + severity=DriftSeverity.LOW, + is_loop_detected=False, + cycle_signature=None, + reasons=[], + consecutive_failures=0, + unique_actions_count=0, + total_actions_count=0, + ) + + is_loop, cycle_sig, loop_reasons = self.detect_cycles() + reasons = list(loop_reasons) + + failures = sum(1 for r in self._history if not r.success) + failure_rate = failures / total + + unique_sigs = {r.action_signature for r in self._history} + repetition_factor = 1.0 - (len(unique_sigs) / total) + + # Baseline score from failure rate and repetition + score = (failure_rate * 0.5) + (repetition_factor * 0.3) + + # Consecutive failure penalty + if self._consecutive_failures > 0: + score += min(0.4, self._consecutive_failures * 0.15) + if self._consecutive_failures >= 2: + reasons.append(f"{self._consecutive_failures} consecutive step failures recorded.") + + # Cycle penalty + if is_loop: + score = max(score, 0.75) + + # Iteration inflation penalty (executing far more steps than planned) + expected_steps = planned_total_steps or (len(current_plan) if current_plan else 5) + if total > expected_steps * 2: + inflation = min(0.3, (total - expected_steps * 2) * 0.05) + score += inflation + reasons.append(f"Step count ({total}) significantly exceeded expected plan budget ({expected_steps}).") + + score = min(1.0, max(0.0, round(score, 3))) + + # Classify severity + if score >= self.max_drift_score or is_loop or self._consecutive_failures >= 3: + severity = DriftSeverity.CRITICAL + elif score >= 0.4 or self._consecutive_failures >= 2: + severity = DriftSeverity.MODERATE + else: + severity = DriftSeverity.LOW + + return DriftReport( + drift_score=score, + severity=severity, + is_loop_detected=is_loop, + cycle_signature=cycle_sig, + reasons=reasons, + consecutive_failures=self._consecutive_failures, + unique_actions_count=len(unique_sigs), + total_actions_count=total, + ) + + def get_action_history(self) -> list[ActionRecord]: + """Return shallow copy of recorded action history.""" + return list(self._history) + + def reset(self) -> None: + """Clear recorded history and counters.""" + self._history.clear() + self._consecutive_failures = 0 + + +__all__ = [ + "ActionRecord", + "DriftReport", + "DriftSeverity", + "PlanDriftDetector", +] diff --git a/agentcli/agent/events.py b/agentcli/agent/events.py index d85afa3..607dd1d 100644 --- a/agentcli/agent/events.py +++ b/agentcli/agent/events.py @@ -71,7 +71,39 @@ class LoopErrorEvent(LoopEvent): error: str = "" +@dataclass +class DriftDetectedEvent(LoopEvent): + """Emitted when plan drift, oscillation, or action cycles are detected.""" + + drift_score: float = 0.0 + severity: str = "low" # 'low' | 'moderate' | 'critical' + is_loop_detected: bool = False + cycle_signature: str | None = None + reasons: list[str] = field(default_factory=list) + + +@dataclass +class AutoHealingRollbackEvent(LoopEvent): + """Emitted when auto-healing reverts filesystem changes after failures/drift.""" + + snapshot_id: str = "" + trigger: str = "" # 'critical_drift' | 'consecutive_errors' | 'cycle_detected' + reverted_files: list[str] = field(default_factory=list) + error: str | None = None + + +@dataclass +class StrategyRecoveryEvent(LoopEvent): + """Emitted when a strategy recovery prompt is synthesized for plan correction.""" + + strategy_prompt: str = "" + alternative_actions: list[str] = field(default_factory=list) + diagnostics: str = "" + + __all__ = [ + "AutoHealingRollbackEvent", + "DriftDetectedEvent", "FinishEvent", "LoopErrorEvent", "LoopEvent", @@ -79,4 +111,6 @@ class LoopErrorEvent(LoopEvent): "ReflectEvent", "StepResultEvent", "StepStartEvent", + "StrategyRecoveryEvent", ] + diff --git a/agentcli/agent/loop.py b/agentcli/agent/loop.py index 52e5113..6f23fd5 100644 --- a/agentcli/agent/loop.py +++ b/agentcli/agent/loop.py @@ -32,7 +32,10 @@ from ..routing.router import Router from ..subagents.base import SubAgentResult, SubAgentTask, SubAgentType from ..subagents.planner import PlannerAgent +from .drift_detector import DriftSeverity, PlanDriftDetector from .events import ( + AutoHealingRollbackEvent, + DriftDetectedEvent, FinishEvent, LoopErrorEvent, LoopEvent, @@ -40,14 +43,17 @@ ReflectEvent, StepResultEvent, StepStartEvent, + StrategyRecoveryEvent, ) from .protocols import ExecutorProtocol, PlannerProtocol, ReflectorProtocol from .reflector import DefaultReflector, LLMReflector, ReflectDecision, ReflectOutcome from .registry import ToolRegistry +from .rollback import AutoHealingManager logger = logging.getLogger(__name__) + class LoopIterationLimitError(Exception): """Raised when the loop hits its max_iterations ceiling.""" @@ -82,6 +88,8 @@ def __init__( run_id: str | None = None, initial_context: str | None = None, max_cost_usd: float | None = None, + drift_detector: PlanDriftDetector | None = None, + auto_healing: AutoHealingManager | None = None, ) -> None: from ..files import expand_file_references @@ -114,6 +122,14 @@ def __init__( self.max_cost_usd = max_cost_usd self.cumulative_cost_usd: float = 0.0 + self.drift_detector = ( + drift_detector if drift_detector is not None else PlanDriftDetector() + ) + self.auto_healing = ( + auto_healing if auto_healing is not None else AutoHealingManager() + ) + self._recovery_guidance: str | None = None + # If using default PlannerAgent, pass config for LLM-based planning if self._config is not None and isinstance(self.planner, PlannerAgent): self.planner._set_config(self._config) @@ -122,6 +138,7 @@ def __init__( self._running_tasks: list[asyncio.Task[Any]] = [] self._start_time: float | None = None + def cancel(self) -> None: """Explicitly cancel in-flight tasks in this loop run.""" for task in list(self._running_tasks): @@ -190,6 +207,21 @@ async def _run_impl(self) -> AsyncIterator[LoopEvent]: agent_type = step.get("agent_type", SubAgentType.CODE_ANALYZER.value) payload = dict(step.get("payload", {})) + # Auto-healing snapshot capture before mutation + step_snapshot_id = "" + path_to_snapshot = ( + payload.get("path") + or payload.get("file") + or payload.get("target") + ) + if path_to_snapshot and self.auto_healing.is_enabled: + step_snapshot_id = self.auto_healing.take_snapshot( + description=f"Pre-step {step_index + 1} ({agent_type})", + iteration=iteration, + paths=[str(path_to_snapshot)], + metadata={"agent_type": agent_type, "payload": payload}, + ) + yield StepStartEvent( iteration=iteration, run_id=self.run_id, @@ -204,6 +236,27 @@ async def _run_impl(self) -> AsyncIterator[LoopEvent]: step_results.append(result) self._all_results.append(result) + # Record action in drift detector + self.drift_detector.record_action( + iteration=iteration, + step_index=step_index, + agent_type=agent_type, + payload=payload, + success=result.success, + error=result.error, + ) + + if not result.success: + self.auto_healing.record_failure( + result.error or "Step failed", + step_info={"agent_type": agent_type, "payload": payload}, + ) + if step_snapshot_id: + self.auto_healing.rollback_to_snapshot(step_snapshot_id) + else: + self.auto_healing.reset_consecutive_failures() + + # Accumulate estimated step cost if model specified model_used = payload.get("model", "") if model_used: @@ -230,6 +283,60 @@ async def _run_impl(self) -> AsyncIterator[LoopEvent]: duration_seconds=round(step_duration, 4), ) + # Evaluate drift and check for loops / consecutive failures + drift_report = self.drift_detector.evaluate_drift(current_plan=current_plan) + if ( + drift_report.severity in (DriftSeverity.MODERATE, DriftSeverity.CRITICAL) + or drift_report.is_loop_detected + ): + yield DriftDetectedEvent( + iteration=iteration, + run_id=self.run_id, + drift_score=drift_report.drift_score, + severity=drift_report.severity.value, + is_loop_detected=drift_report.is_loop_detected, + cycle_signature=drift_report.cycle_signature, + reasons=drift_report.reasons, + ) + + if ( + drift_report.severity == DriftSeverity.CRITICAL + or drift_report.is_loop_detected + or drift_report.consecutive_failures >= 2 + ): + trigger_name = ( + "cycle_detected" + if drift_report.is_loop_detected + else ( + "consecutive_errors" + if drift_report.consecutive_failures >= 2 + else "critical_drift" + ) + ) + rollback_res = self.auto_healing.rollback_last() + yield AutoHealingRollbackEvent( + iteration=iteration, + run_id=self.run_id, + snapshot_id=rollback_res.get("snapshot_id", ""), + trigger=trigger_name, + reverted_files=rollback_res.get("reverted_files", []), + error=rollback_res.get("error"), + ) + + recovery_prompt = self.auto_healing.synthesize_recovery_prompt( + drift_report=drift_report, + goal=self.goal, + ) + self._recovery_guidance = recovery_prompt + yield StrategyRecoveryEvent( + iteration=iteration, + run_id=self.run_id, + strategy_prompt=recovery_prompt, + alternative_actions=drift_report.reasons, + diagnostics=f"Score: {drift_report.drift_score}, Cycle: {drift_report.cycle_signature}", + ) + + if ( self.max_cost_usd is not None and self.cumulative_cost_usd >= self.max_cost_usd @@ -380,9 +487,14 @@ async def _plan(self, previous_plan: list[dict[str, Any]] | None) -> list[dict[s ) context_parts.append("\n".join(feedback_lines)) + if self._recovery_guidance: + context_parts.append(self._recovery_guidance) + self._recovery_guidance = None + if context_parts: payload["context"] = "\n\n".join(context_parts) + task = SubAgentTask( agent_type=SubAgentType.PLANNER, payload=payload, diff --git a/agentcli/agent/rollback.py b/agentcli/agent/rollback.py new file mode 100644 index 0000000..401c4af --- /dev/null +++ b/agentcli/agent/rollback.py @@ -0,0 +1,294 @@ +"""Agent Auto-Healing & File Rollback Manager (Phase 36). + +Provides transactional filesystem snapshots, automatic rollbacks when loops +or critical drift occur, and recovery prompt synthesis for resilient self-healing. +""" + +from __future__ import annotations + +import logging +import time +import uuid +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from .checkpoints import MAX_SNAPSHOT_FILE_BYTES, FileSnapshot +from .drift_detector import DriftReport + +logger = logging.getLogger(__name__) + +MAX_HEALING_SNAPSHOTS = 20 + + +@dataclass +class HealingSnapshot: + """Snapshot of files captured prior to a risky or mutative agent action.""" + + id: str + iteration: int + description: str + timestamp: float = field(default_factory=time.time) + files: dict[str, FileSnapshot] = field(default_factory=dict) + metadata: dict[str, Any] = field(default_factory=dict) + + +class AutoHealingManager: + """Orchestrates automatic snapshot capture, rollbacks, and recovery prompts.""" + + def __init__( + self, + root_dir: Path | str = ".", + max_snapshots: int = MAX_HEALING_SNAPSHOTS, + enabled: bool = True, + ) -> None: + self.root_dir = Path(root_dir).resolve() + self.max_snapshots = max_snapshots + self._enabled = enabled + self._snapshots: list[HealingSnapshot] = [] + self._recent_errors: list[str] = [] + self._consecutive_failures: int = 0 + + @property + def is_enabled(self) -> bool: + """Return True if auto-healing snapshots and rollbacks are enabled.""" + return self._enabled + + def enable(self) -> None: + """Enable auto-healing snapshots and rollbacks.""" + self._enabled = True + + def disable(self) -> None: + """Disable auto-healing snapshots and rollbacks.""" + self._enabled = False + + def take_snapshot( + self, + description: str = "Pre-step snapshot", + iteration: int = 0, + paths: list[str] | None = None, + metadata: dict[str, Any] | None = None, + ) -> str: + """Capture filesystem snapshot before executing a step.""" + if not self._enabled: + return "" + + snapshot_id = uuid.uuid4().hex[:8] + snapshot = HealingSnapshot( + id=snapshot_id, + iteration=iteration, + description=description, + metadata=metadata or {}, + ) + self._snapshots.append(snapshot) + + if paths: + for p in paths: + self.record_file_before_mutation(p, snapshot_id=snapshot_id) + + if len(self._snapshots) > self.max_snapshots: + self._snapshots.pop(0) + + return snapshot_id + + def record_file_before_mutation( + self, rel_path: str, snapshot_id: str | None = None + ) -> None: + """Record the state of a file in the active or latest snapshot.""" + if not self._enabled: + return + + target_snap: HealingSnapshot | None = None + if snapshot_id: + for s in self._snapshots: + if s.id == snapshot_id: + target_snap = s + break + else: + target_snap = self._snapshots[-1] if self._snapshots else None + + if target_snap is None: + return + + norm_path = Path(rel_path).as_posix() + if norm_path in target_snap.files: + return # Already recorded in this snapshot + + full_path = (self.root_dir / norm_path).resolve() + if not full_path.exists(): + target_snap.files[norm_path] = FileSnapshot(rel_path=norm_path, exists=False) + return + + try: + if full_path.stat().st_size > MAX_SNAPSHOT_FILE_BYTES: + target_snap.files[norm_path] = FileSnapshot( + rel_path=norm_path, exists=True, is_binary=True + ) + return + + text = full_path.read_text(encoding="utf-8", errors="replace") + target_snap.files[norm_path] = FileSnapshot( + rel_path=norm_path, exists=True, content=text + ) + except OSError: + target_snap.files[norm_path] = FileSnapshot( + rel_path=norm_path, exists=True, is_binary=True + ) + + def record_failure( + self, error: str, step_info: dict[str, Any] | None = None + ) -> int: + """Record an execution error and increment consecutive failure counter.""" + self._recent_errors.append(error) + if len(self._recent_errors) > 10: + self._recent_errors.pop(0) + self._consecutive_failures += 1 + return self._consecutive_failures + + def reset_consecutive_failures(self) -> None: + """Reset the failure counter following a successful step.""" + self._consecutive_failures = 0 + + def get_recent_errors(self) -> list[str]: + """Return list of recent error messages.""" + return list(self._recent_errors) + + def rollback_last(self) -> dict[str, Any]: + """Roll back the most recent snapshot.""" + return self.rollback_to_snapshot(None) + + def rollback_to_snapshot( + self, snapshot_id: str | None = None + ) -> dict[str, Any]: + """Revert all file mutations recorded in the specified snapshot.""" + if not self._snapshots: + return { + "success": False, + "error": "No auto-healing snapshots available to restore", + "reverted_files": [], + } + + target_snap: HealingSnapshot | None = None + if snapshot_id is None: + target_snap = self._snapshots[-1] + else: + for s in self._snapshots: + if s.id == snapshot_id: + target_snap = s + break + + if target_snap is None: + return { + "success": False, + "error": f"Snapshot '{snapshot_id}' not found", + "reverted_files": [], + } + + reverted: list[str] = [] + errors: list[str] = [] + + for rel_path, snap in target_snap.files.items(): + full_path = (self.root_dir / rel_path).resolve() + try: + if not snap.exists: + if full_path.exists(): + full_path.unlink() + reverted.append(f"deleted {rel_path}") + else: + full_path.parent.mkdir(parents=True, exist_ok=True) + if snap.content is not None: + full_path.write_text(snap.content, encoding="utf-8") + reverted.append(f"restored {rel_path}") + except OSError as exc: + errors.append(f"Failed to restore {rel_path}: {exc}") + + # Remove the restored snapshot from active list + if target_snap in self._snapshots: + self._snapshots.remove(target_snap) + + logger.info( + "AutoHealing rollback performed on snapshot '%s': %d files reverted (%s)", + target_snap.id, + len(reverted), + ", ".join(reverted) if reverted else "none", + ) + + return { + "success": len(errors) == 0, + "snapshot_id": target_snap.id, + "description": target_snap.description, + "reverted_files": reverted, + "errors": errors, + } + + def synthesize_recovery_prompt( + self, + drift_report: DriftReport | None = None, + recent_errors: list[str] | None = None, + goal: str = "", + ) -> str: + """Synthesize a structured recovery strategy prompt for the planner/reflector.""" + errors = recent_errors or self._recent_errors + err_snippet = "\n".join(f"- {e}" for e in errors[-3:]) if errors else "- Unknown error" + + reasons = drift_report.reasons if drift_report else [] + reasons_snippet = "\n".join(f"- {r}" for r in reasons) if reasons else "- Plan execution drifted from primary goal." + + severity = drift_report.severity.value if drift_report else "critical" + cycle_sig = drift_report.cycle_signature if drift_report else None + + guidelines = [ + "1. Do NOT repeat the exact same failed command, query, or file mutation.", + "2. If a specific tool keeps failing, switch to an alternative tool or approach (e.g. read before edit, run tests sequentially, inspect diagnostics).", + "3. Decompose complex multi-step operations into smaller, verifiable sub-steps.", + "4. Verify preconditions before modifying files or executing commands.", + ] + + if cycle_sig: + guidelines.insert( + 0, + f"🚨 BREAK CYCLE: An execution loop was detected on '{cycle_sig}'. You MUST avoid this action and choose an alternative path.", + ) + + prompt_lines = [ + f"=== AGENT SELF-HEALING RECOVERY GUIDANCE (Severity: {severity.upper()}) ===", + f"Goal: {goal}" if goal else "", + "", + "Recent Failures / Root Causes:", + err_snippet, + "", + "Drift & Cycle Diagnostics:", + reasons_snippet, + "", + "Mandatory Recovery Strategy:", + "\n".join(guidelines), + "===============================================================", + ] + + return "\n".join(line for line in prompt_lines if line is not None) + + def list_snapshots(self) -> list[dict[str, Any]]: + """List summary of recorded snapshots.""" + return [ + { + "id": s.id, + "iteration": s.iteration, + "description": s.description, + "timestamp": s.timestamp, + "file_count": len(s.files), + "files": list(s.files.keys()), + } + for s in reversed(self._snapshots) + ] + + def reset(self) -> None: + """Clear all snapshots and failure records.""" + self._snapshots.clear() + self._recent_errors.clear() + self._consecutive_failures = 0 + + +__all__ = [ + "AutoHealingManager", + "HealingSnapshot", +] diff --git a/agentcli/cli.py b/agentcli/cli.py index f7ae3a0..fec7351 100644 --- a/agentcli/cli.py +++ b/agentcli/cli.py @@ -1139,8 +1139,59 @@ async def run_chat(args: argparse.Namespace, config: Config) -> int: print("Usage: /branch [list | create | status | diff | merge | discard | prune]") continue + if user_input.startswith(("/healing", "/heal")): + h_parts = user_input.split(maxsplit=2) + h_subcmd = h_parts[1].lower() if len(h_parts) > 1 else "status" + h_target = h_parts[2].strip() if len(h_parts) > 2 else "" + + if h_subcmd in ("status", "info"): + enabled_str = "Enabled" if session.auto_healing.is_enabled else "Disabled" + snaps = session.auto_healing.list_snapshots() + report = session.drift_detector.evaluate_drift() + print(f"\n--- Agent Self-Healing Status [{enabled_str}] ---") + print(f" Drift Severity: {report.severity.value.upper()} (Score: {report.drift_score:.3f})") + print(f" Cycle/Loop Detected: {'Yes' if report.is_loop_detected else 'No'}") + if report.cycle_signature: + print(f" Cycle Signature: {report.cycle_signature}") + print(f" Consecutive Failures: {report.consecutive_failures}") + print(f" Total Actions Tracked: {report.total_actions_count} ({report.unique_actions_count} unique)") + print(f" Active Snapshots: {len(snaps)}") + if report.reasons: + print(" Active Drift Warnings:") + for r in report.reasons: + print(f" - {r}") + if snaps: + print(" Recent Snapshots:") + for s in snaps[:3]: + print(f" • [{s['id']}] {s['description']} ({s['file_count']} files)") + print("------------------------------------------------\n") + print("Actions: /healing rollback [id] | /healing reset | /healing [enable|disable]\n") + elif h_subcmd == "rollback": + h_target_id: str | None = h_target if h_target else None + h_res: dict[str, Any] = session.auto_healing.rollback_to_snapshot(h_target_id) + if h_res.get("success"): + reverted = ", ".join(h_res.get("reverted_files", [])) or "none" + print(f"\nSuccessfully rolled back snapshot '{h_res.get('snapshot_id')}': {reverted}\n") + else: + print(f"\nRollback failed: {h_res.get('error')}\n") + + elif h_subcmd == "reset": + session.auto_healing.reset() + session.drift_detector.reset() + print("Self-healing and drift detector history reset successfully.") + elif h_subcmd in ("enable", "on"): + session.auto_healing.enable() + print("Auto-healing snapshots and rollbacks enabled.") + elif h_subcmd in ("disable", "off"): + session.auto_healing.disable() + print("Auto-healing snapshots and rollbacks disabled.") + else: + print("Usage: /healing [status | rollback | reset | enable | disable]") + continue + if user_input in {"/clear", "/cls"}: + renderer.clear() continue diff --git a/agentcli/session.py b/agentcli/session.py index c507a8e..11c8a4e 100644 --- a/agentcli/session.py +++ b/agentcli/session.py @@ -8,9 +8,11 @@ from typing import Any from .agent.checkpoints import CheckpointManager +from .agent.drift_detector import PlanDriftDetector from .agent.events import LoopEvent from .agent.loop import AgentLoop, is_agentic_task from .agent.registry import ToolRegistry +from .agent.rollback import AutoHealingManager from .agent.tasks import TaskManager from .config import Config from .files import load_agents_md @@ -120,12 +122,15 @@ def __init__( self.router: Router | None = None self.mcp_manager: MCPClientManager = MCPClientManager(config=self.config) self.checkpoint_manager: CheckpointManager = CheckpointManager() + self.auto_healing: AutoHealingManager = AutoHealingManager() + self.drift_detector: PlanDriftDetector = PlanDriftDetector() self.task_manager: TaskManager = TaskManager() self.skill_engine: SkillEngine = SkillEngine() self.worktree_manager: WorktreeManager = WorktreeManager() self.active_worktree_path: Path | None = None + if config.routing.enabled: self.registry = ModelRegistry(config.routing) if not forced_model: @@ -451,8 +456,11 @@ async def run_loop(self, goal: str, run_id: str | None = None) -> AsyncIterator[ run_id=run_id, initial_context=initial_context, max_cost_usd=max_cost, + drift_detector=self.drift_detector, + auto_healing=self.auto_healing, ) + try: async for event in loop.run(): yield event diff --git a/agentcli/ui/prompt.py b/agentcli/ui/prompt.py index 0f7e31c..bed244e 100644 --- a/agentcli/ui/prompt.py +++ b/agentcli/ui/prompt.py @@ -57,6 +57,8 @@ def resolve_slash_command(text: str) -> str: "/worktree": "/branch", "/wt": "/branch", "/branches": "/branch", + "/heal": "/healing", + "/selfheal": "/healing", } if raw_cmd in aliases: @@ -74,6 +76,7 @@ def resolve_slash_command(text: str) -> str: "/tasks", "/skill", "/branch", + "/healing", "/tokens", "/cost", "/clear", @@ -92,7 +95,7 @@ def resolve_slash_command(text: str) -> str: class SlashAndFileCompleter(Completer): - """Completer for slash commands (/models, /model, /undo, /tasks, /skill, /branch, /budget, /history, /exit, etc.), model arguments, and @file references.""" + """Completer for slash commands (/models, /model, /undo, /tasks, /skill, /branch, /healing, /budget, /history, /exit, etc.), model arguments, and @file references.""" SLASH_COMMANDS: ClassVar[list[tuple[str, str]]] = [ ("/help", "Show help, slash commands, and shortcuts"), @@ -106,6 +109,7 @@ class SlashAndFileCompleter(Completer): ("/tasks", "List or manage background tasks (/tasks, /tasks kill )"), ("/skill", "Run or inspect custom skills and recipes (/skill list, /skill run )"), ("/branch", "Manage Git worktrees and sandboxes (/branch list, /branch create )"), + ("/healing", "Manage self-healing, drift detection, and auto-rollback (/healing status, /healing rollback)"), ("/tokens", "Show current session token usage breakdown"), ("/cost", "Show current session estimated cost"), ("/clear", "Clear terminal screen"), @@ -126,9 +130,12 @@ class SlashAndFileCompleter(Completer): "/worktree": "/branch", "/wt": "/branch", "/branches": "/branch", + "/heal": "/healing", + "/selfheal": "/healing", } + def __init__(self) -> None: self.path_completer = PathCompleter(expanduser=True) @@ -247,6 +254,24 @@ def get_completions(self, document: Document, complete_event: CompleteEvent) -> ) return + # Complete healing arguments after /healing or /heal + if text.startswith(("/healing ", "\\healing ", "/heal ", "\\heal ")): + arg = text.split(maxsplit=1)[1] if len(text.split(maxsplit=1)) > 1 else "" + arg_lower = arg.lower() + + healing_actions = [ + ("status", "[ACTION] Show self-healing, drift detector, and snapshot status"), + ("rollback", "[ACTION] Revert last or specific snapshot (/healing rollback [id])"), + ("reset", "[ACTION] Reset drift history, failure counters, and snapshots"), + ("enable", "[ACTION] Enable auto-healing snapshots and rollbacks"), + ("disable", "[ACTION] Disable auto-healing snapshots and rollbacks"), + ] + for act, desc in healing_actions: + if act.startswith(arg_lower): + yield Completion(act, start_position=-len(arg), display_meta=desc) + return + + # Complete slash commands at the start of input (support both / and \) if text.startswith(("/", "\\")): diff --git a/agentcli/ui/render.py b/agentcli/ui/render.py index ec4eaae..90ae989 100644 --- a/agentcli/ui/render.py +++ b/agentcli/ui/render.py @@ -160,6 +160,37 @@ def render_loop_event(self, event: Any, *, verbose: bool = False) -> None: self.console.print( f" [magenta]🔍 reflect[/magenta] [bold]{event.decision}[/bold] [dim]— {event.reason}[/dim]" ) + elif event_name == "DriftDetectedEvent": + sev_color = ( + "yellow" + if getattr(event, "severity", "") == "moderate" + else ("red" if getattr(event, "severity", "") == "critical" else "dim") + ) + loop_flag = ( + " [bold red][CYCLE DETECTED][/bold red]" + if getattr(event, "is_loop_detected", False) + else "" + ) + score_val = getattr(event, "drift_score", 0.0) + sev_val = getattr(event, "severity", "low") + self.console.print( + f" [{sev_color}]⚠️ plan drift ({sev_val})[/{sev_color}]{loop_flag} [dim]score: {score_val:.2f}[/dim]" + ) + if verbose and getattr(event, "reasons", None): + for r in event.reasons: + self.console.print(f" [dim]• {r}[/dim]") + elif event_name == "AutoHealingRollbackEvent": + reverted = getattr(event, "reverted_files", []) + reverted_str = f" ({len(reverted)} file(s) reverted)" if reverted else "" + trigger = getattr(event, "trigger", "drift") + self.console.print( + f" [bold yellow]🔄 auto-healing rollback[/bold yellow] [dim]trigger: {trigger}[/dim]{reverted_str}" + ) + elif event_name == "StrategyRecoveryEvent": + diag = getattr(event, "diagnostics", "") + self.console.print( + f" [bold cyan]💡 strategy recovery synthesized[/bold cyan] [dim]— {diag}[/dim]" + ) elif event_name == "FinishEvent": self.console.print(f"\n[bold green]✨ Done:[/bold green] {event.summary}") out = getattr(event, "output", None) @@ -202,6 +233,22 @@ def render_loop_event(self, event: Any, *, verbose: bool = False) -> None: safe_print(f" [step {event.step_index + 1}] [{status}]{err}{timing}") elif event_name == "ReflectEvent": safe_print(f" [reflect] {event.decision} - {event.reason}") + elif event_name == "DriftDetectedEvent": + loop_flag = ( + " [CYCLE DETECTED]" if getattr(event, "is_loop_detected", False) else "" + ) + score_val = getattr(event, "drift_score", 0.0) + sev_val = getattr(event, "severity", "low") + safe_print(f" [drift] severity={sev_val} score={score_val:.2f}{loop_flag}") + elif event_name == "AutoHealingRollbackEvent": + reverted = getattr(event, "reverted_files", []) + trigger = getattr(event, "trigger", "drift") + safe_print( + f" [auto-healing] rollback trigger={trigger} files={len(reverted)}" + ) + elif event_name == "StrategyRecoveryEvent": + diag = getattr(event, "diagnostics", "") + safe_print(f" [strategy-recovery] {diag}") elif event_name == "FinishEvent": safe_print(f"\n[done] {event.summary}") out = getattr(event, "output", None) @@ -210,6 +257,7 @@ def render_loop_event(self, event: Any, *, verbose: bool = False) -> None: elif event_name == "LoopErrorEvent": safe_print(f"\n[loop-error] {event.error}") + def render_sessions_table( self, sessions: list[Any], diff --git a/agentcli/ui/tui_app.py b/agentcli/ui/tui_app.py index b0c9285..659e78a 100644 --- a/agentcli/ui/tui_app.py +++ b/agentcli/ui/tui_app.py @@ -692,6 +692,66 @@ async def _handle_slash_command(self, text: str, timestamp: str, event: KeyPress self.add_message("system", "Worktree manager not available in active session.", timestamp) return + if cmd in ("/healing", "/heal"): + parts = text.split(maxsplit=2) + subcmd = parts[1].lower() if len(parts) > 1 else "status" + target = parts[2].strip() if len(parts) > 2 else "" + + if self.session and hasattr(self.session, "auto_healing"): + healing = self.session.auto_healing + detector = self.session.drift_detector + if subcmd in ("status", "info"): + status_str = "Enabled" if healing.is_enabled else "Disabled" + report = detector.evaluate_drift() + snaps = healing.list_snapshots() + lines = [ + f"Agent Self-Healing [{status_str}]:", + f" • Drift Severity: {report.severity.value.upper()} (Score: {report.drift_score:.2f})", + f" • Cycle Detected: {'Yes' if report.is_loop_detected else 'No'}", + f" • Consecutive Failures: {report.consecutive_failures}", + f" • Snapshots Available: {len(snaps)}", + ] + if report.cycle_signature: + lines.append(f" • Cycle Signature: {report.cycle_signature}") + if report.reasons: + lines.append(" • Diagnostics:") + for r in report.reasons: + lines.append(f" - {r}") + self.add_message("system", "\n".join(lines), timestamp) + elif subcmd == "rollback": + res = healing.rollback_to_snapshot(target or None) + if res.get("success"): + reverted = ", ".join(res.get("reverted_files", [])) or "none" + self.add_message( + "system", + f"Rolled back snapshot '{res.get('snapshot_id')}': {reverted}", + timestamp, + ) + else: + self.add_message( + "system", f"Rollback failed: {res.get('error')}", timestamp + ) + elif subcmd == "reset": + healing.reset() + detector.reset() + self.add_message("system", "Self-healing and drift history reset.", timestamp) + elif subcmd in ("enable", "on"): + healing.enable() + self.add_message("system", "Auto-healing enabled.", timestamp) + elif subcmd in ("disable", "off"): + healing.disable() + self.add_message("system", "Auto-healing disabled.", timestamp) + else: + self.add_message( + "system", + "Usage: /healing [status | rollback | reset | enable | disable]", + timestamp, + ) + else: + self.add_message("system", "Auto-healing not available in session.", timestamp) + return + + if cmd in ("/budget", "/cost", "/tokens"): if self.session and hasattr(self.session, "governor"): parts = text.split(maxsplit=2) diff --git a/tests/test_phase36_self_healing.py b/tests/test_phase36_self_healing.py new file mode 100644 index 0000000..ba8f5b7 --- /dev/null +++ b/tests/test_phase36_self_healing.py @@ -0,0 +1,402 @@ +"""Tests for Phase 36: Agent Self-Healing, Plan Drift Detection & Auto-Rollback.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest +from prompt_toolkit.document import Document + +from agentcli.agent.drift_detector import ( + DriftReport, + DriftSeverity, + PlanDriftDetector, +) +from agentcli.agent.events import ( + LoopEvent, +) +from agentcli.agent.loop import AgentLoop +from agentcli.agent.protocols import ExecutorProtocol, PlannerProtocol, ReflectorProtocol +from agentcli.agent.reflector import ReflectDecision, ReflectOutcome +from agentcli.agent.rollback import AutoHealingManager +from agentcli.config import Config +from agentcli.session import AgentSession +from agentcli.subagents.base import SubAgentResult, SubAgentTask, SubAgentType +from agentcli.ui.prompt import SlashAndFileCompleter + +# --------------------------------------------------------------------------- +# Test PlanDriftDetector +# --------------------------------------------------------------------------- + + +class TestPlanDriftDetector: + def test_drift_detector_initial_state(self) -> None: + detector = PlanDriftDetector() + report = detector.evaluate_drift() + assert report.drift_score == 0.0 + assert report.severity == DriftSeverity.LOW + assert not report.is_loop_detected + assert report.cycle_signature is None + assert report.consecutive_failures == 0 + assert report.total_actions_count == 0 + assert len(detector.get_action_history()) == 0 + + def test_drift_detector_record_actions_success(self) -> None: + detector = PlanDriftDetector() + rec1 = detector.record_action( + iteration=1, + step_index=0, + agent_type="code_analyzer", + payload={"path": "main.py"}, + success=True, + ) + assert rec1.success is True + assert "main.py" in rec1.action_signature + assert detector.evaluate_drift().severity == DriftSeverity.LOW + assert detector.evaluate_drift().consecutive_failures == 0 + + def test_drift_detector_consecutive_identical_failures_cycle(self) -> None: + detector = PlanDriftDetector(loop_threshold=2) + detector.record_action( + iteration=1, + step_index=0, + agent_type="shell_execution", + payload={"command": "pytest tests/test_core.py"}, + success=False, + error="Process failed with exit code 1", + ) + rep1 = detector.evaluate_drift() + assert rep1.consecutive_failures == 1 + assert rep1.severity in (DriftSeverity.LOW, DriftSeverity.MODERATE) + + # Second failure with same action signature + detector.record_action( + iteration=1, + step_index=1, + agent_type="shell_execution", + payload={"command": "pytest tests/test_core.py"}, + success=False, + error="Process failed with exit code 1", + ) + rep2 = detector.evaluate_drift() + assert rep2.is_loop_detected is True + assert rep2.cycle_signature is not None + assert rep2.severity == DriftSeverity.CRITICAL + assert any("pytest tests/test_core.py" in r for r in rep2.reasons) + + def test_drift_detector_ping_pong_oscillation_cycle(self) -> None: + detector = PlanDriftDetector() + # Ping-pong sequence: A -> B -> A -> B + detector.record_action( + iteration=1, step_index=0, agent_type="file_ops", payload={"path": "a.py"}, success=True + ) + detector.record_action( + iteration=1, step_index=1, agent_type="file_ops", payload={"path": "b.py"}, success=True + ) + detector.record_action( + iteration=2, step_index=0, agent_type="file_ops", payload={"path": "a.py"}, success=True + ) + detector.record_action( + iteration=2, step_index=1, agent_type="file_ops", payload={"path": "b.py"}, success=True + ) + + rep = detector.evaluate_drift() + assert rep.is_loop_detected is True + assert rep.cycle_signature is not None + assert "<->" in rep.cycle_signature + assert rep.severity == DriftSeverity.CRITICAL + + def test_drift_detector_repeated_tool_actions_cycle(self) -> None: + detector = PlanDriftDetector() + for i in range(3): + detector.record_action( + iteration=1, + step_index=i, + agent_type="workspace", + payload={"query": "find_symbol"}, + success=True, + ) + rep = detector.evaluate_drift() + assert rep.is_loop_detected is True + assert rep.severity == DriftSeverity.CRITICAL + assert any("Repeated execution" in r for r in rep.reasons) + + def test_drift_detector_file_mutation_cycle(self) -> None: + detector = PlanDriftDetector() + detector.record_action( + iteration=1, step_index=0, agent_type="file_ops", payload={"path": "src/app.py", "op": "write"}, success=False, error="Syntax error" + ) + detector.record_action( + iteration=1, step_index=1, agent_type="code_analyzer", payload={"path": "src/other.py"}, success=True + ) + detector.record_action( + iteration=2, step_index=0, agent_type="file_ops", payload={"path": "src/app.py", "op": "edit"}, success=False, error="Type error" + ) + detector.record_action( + iteration=2, step_index=1, agent_type="file_ops", payload={"path": "src/app.py", "op": "patch"}, success=False, error="Test failure" + ) + rep = detector.evaluate_drift() + assert rep.is_loop_detected is True + assert "src/app.py" in str(rep.cycle_signature) + + def test_drift_detector_reset(self) -> None: + detector = PlanDriftDetector() + detector.record_action( + iteration=1, step_index=0, agent_type="tool", payload={"cmd": "ls"}, success=False + ) + assert len(detector.get_action_history()) == 1 + detector.reset() + assert len(detector.get_action_history()) == 0 + assert detector.evaluate_drift().consecutive_failures == 0 + + +# --------------------------------------------------------------------------- +# Test AutoHealingManager +# --------------------------------------------------------------------------- + + +class TestAutoHealingManager: + def test_auto_healing_enable_disable(self) -> None: + mgr = AutoHealingManager(enabled=True) + assert mgr.is_enabled is True + mgr.disable() + assert mgr.is_enabled is False + mgr.enable() + assert mgr.is_enabled is True + + def test_auto_healing_snapshot_and_rollback(self, tmp_path: Path) -> None: + mgr = AutoHealingManager(root_dir=tmp_path, enabled=True) + + test_file = tmp_path / "hello.txt" + test_file.write_text("original content", encoding="utf-8") + + new_file = tmp_path / "created.txt" + + # Capture snapshot + snap_id = mgr.take_snapshot( + description="Before file edits", + paths=["hello.txt", "created.txt"], + ) + assert snap_id != "" + + # Mutate existing file and create new file + test_file.write_text("corrupted content", encoding="utf-8") + new_file.write_text("transient file", encoding="utf-8") + assert test_file.read_text(encoding="utf-8") == "corrupted content" + assert new_file.exists() + + # Perform rollback + res = mgr.rollback_to_snapshot(snap_id) + assert res["success"] is True + assert "restored hello.txt" in res["reverted_files"] + assert "deleted created.txt" in res["reverted_files"] + + # Verify restoration + assert test_file.read_text(encoding="utf-8") == "original content" + assert not new_file.exists() + + def test_auto_healing_rollback_last(self, tmp_path: Path) -> None: + mgr = AutoHealingManager(root_dir=tmp_path, enabled=True) + f = tmp_path / "data.json" + f.write_text('{"v": 1}', encoding="utf-8") + + mgr.take_snapshot(description="Snap 1", paths=["data.json"]) + f.write_text('{"v": 2}', encoding="utf-8") + + res = mgr.rollback_last() + assert res["success"] is True + assert f.read_text(encoding="utf-8") == '{"v": 1}' + + def test_auto_healing_empty_rollback_error(self, tmp_path: Path) -> None: + mgr = AutoHealingManager(root_dir=tmp_path, enabled=True) + res = mgr.rollback_last() + assert res["success"] is False + assert "No auto-healing snapshots available" in res["error"] + + def test_auto_healing_failure_tracking(self) -> None: + mgr = AutoHealingManager() + assert mgr.record_failure("Err 1") == 1 + assert mgr.record_failure("Err 2") == 2 + assert len(mgr.get_recent_errors()) == 2 + mgr.reset_consecutive_failures() + assert mgr.record_failure("Err 3") == 1 + + def test_auto_healing_synthesize_recovery_prompt(self) -> None: + mgr = AutoHealingManager() + mgr.record_failure("ImportError: module missing") + report = DriftReport( + drift_score=0.85, + severity=DriftSeverity.CRITICAL, + is_loop_detected=True, + cycle_signature="shell_execution:pytest#123", + reasons=["Repeated execution of failing test suite"], + ) + prompt = mgr.synthesize_recovery_prompt(drift_report=report, goal="Fix tests") + assert "SELF-HEALING RECOVERY GUIDANCE" in prompt + assert "CRITICAL" in prompt + assert "BREAK CYCLE" in prompt + assert "ImportError: module missing" in prompt + + def test_auto_healing_list_and_reset(self, tmp_path: Path) -> None: + mgr = AutoHealingManager(root_dir=tmp_path) + mgr.take_snapshot(description="Test snap") + assert len(mgr.list_snapshots()) == 1 + mgr.reset() + assert len(mgr.list_snapshots()) == 0 + + +# --------------------------------------------------------------------------- +# Test Events and Loop Integration +# --------------------------------------------------------------------------- + + +class MockFailingExecutor(ExecutorProtocol): + def __init__(self, tmp_path: Path) -> None: + self.tmp_path = tmp_path + self.call_count = 0 + + def list_tools(self) -> list[str]: + return ["file_ops", "code_analyzer", "shell_execution"] + + async def execute(self, agent_type: str, payload: dict[str, Any]) -> SubAgentResult: + self.call_count += 1 + + # Mutate file on attempt + target_path = payload.get("path") + if target_path: + p = (self.tmp_path / target_path).resolve() + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text("bad code that fails build", encoding="utf-8") + + return SubAgentResult( + task_id="step-fail", + agent_type=SubAgentType.FILE_OPS, + success=False, + error="SyntaxError: invalid syntax", + ) + + +class MockRecoveryPlanner(PlannerProtocol): + def __init__(self) -> None: + self.call_count = 0 + self.received_contexts: list[str] = [] + + async def run(self, task: SubAgentTask) -> SubAgentResult: + self.call_count += 1 + ctx = task.payload.get("context", "") + self.received_contexts.append(ctx) + + if self.call_count == 1: + plan = [ + {"agent_type": "file_ops", "payload": {"path": "module.py", "op": "write"}}, + {"agent_type": "file_ops", "payload": {"path": "module.py", "op": "write"}}, + ] + else: + plan = [ + {"agent_type": "code_analyzer", "payload": {"path": "module.py"}}, + ] + + return SubAgentResult( + task_id="plan-1", + agent_type=SubAgentType.PLANNER, + success=True, + output={"plan": plan}, + ) + + +class MockSimpleReflector(ReflectorProtocol): + def __init__(self) -> None: + self.iteration = 0 + + def reflect( + self, + goal: str, + plan: list[dict[str, Any]], + results: list[SubAgentResult], + ) -> ReflectOutcome: + self.iteration += 1 + if self.iteration == 1: + return ReflectOutcome(decision=ReflectDecision.REPLAN, reason="Steps failed, replan") + return ReflectOutcome(decision=ReflectDecision.FINISH, reason="All complete") + + +class TestLoopAutoHealingIntegration: + @pytest.mark.asyncio + async def test_loop_auto_healing_triggers_rollback_and_recovery(self, tmp_path: Path) -> None: + mod_file = tmp_path / "module.py" + mod_file.write_text("initial clean code", encoding="utf-8") + + executor = MockFailingExecutor(tmp_path) + planner = MockRecoveryPlanner() + reflector = MockSimpleReflector() + + drift_detector = PlanDriftDetector(loop_threshold=2) + auto_healing = AutoHealingManager(root_dir=tmp_path, enabled=True) + + loop = AgentLoop( + goal="Refactor module without breaking tests", + registry=executor, + planner=planner, + reflector=reflector, + max_iterations=3, + drift_detector=drift_detector, + auto_healing=auto_healing, + ) + + events: list[LoopEvent] = [] + async for ev in loop.run(): + events.append(ev) + + # Check emitted events + event_types = [type(ev).__name__ for ev in events] + assert "PlanEvent" in event_types + assert "StepStartEvent" in event_types + assert "StepResultEvent" in event_types + assert "DriftDetectedEvent" in event_types + assert "AutoHealingRollbackEvent" in event_types + assert "StrategyRecoveryEvent" in event_types + assert "FinishEvent" in event_types + + # Verify auto-rollback restored original file + assert mod_file.read_text(encoding="utf-8") == "initial clean code" + + # Verify recovery guidance was received by planner on re-planning + assert len(planner.received_contexts) >= 2 + assert "SELF-HEALING RECOVERY GUIDANCE" in planner.received_contexts[1] + + +# --------------------------------------------------------------------------- +# Test Session and CLI Autocomplete Integration +# --------------------------------------------------------------------------- + + +class TestSessionAndPromptIntegration: + def test_session_has_healing_and_drift_detector(self) -> None: + cfg = Config() + session = AgentSession(cfg) + try: + assert isinstance(session.auto_healing, AutoHealingManager) + assert isinstance(session.drift_detector, PlanDriftDetector) + assert session.auto_healing.is_enabled is True + finally: + session.close() + + def test_slash_completer_healing(self) -> None: + completer = SlashAndFileCompleter() + + # Check /healing in root slash commands + doc = Document("/heal") + completions = list(completer.get_completions(doc, None)) # type: ignore[arg-type] + matches = [c.text for c in completions] + assert "/healing" in matches + + # Check /healing subcommands + doc_sub = Document("/healing ") + sub_completions = list(completer.get_completions(doc_sub, None)) # type: ignore[arg-type] + sub_matches = [c.text for c in sub_completions] + assert "status" in sub_matches + assert "rollback" in sub_matches + assert "reset" in sub_matches + assert "enable" in sub_matches + assert "disable" in sub_matches From 59b180d0d65f13d55c9b09813a4bd0b3cea6aaeb Mon Sep 17 00:00:00 2001 From: De-pitcher Date: Mon, 7 Sep 2026 06:41:08 +0100 Subject: [PATCH 2/2] test: add mock env var for session initialization in Phase 36 tests --- tests/test_phase36_self_healing.py | 46 +++++++++++++++++++++++++++++- 1 file changed, 45 insertions(+), 1 deletion(-) diff --git a/tests/test_phase36_self_healing.py b/tests/test_phase36_self_healing.py index ba8f5b7..631e721 100644 --- a/tests/test_phase36_self_healing.py +++ b/tests/test_phase36_self_healing.py @@ -372,7 +372,10 @@ async def test_loop_auto_healing_triggers_rollback_and_recovery(self, tmp_path: class TestSessionAndPromptIntegration: - def test_session_has_healing_and_drift_detector(self) -> None: + def test_session_has_healing_and_drift_detector( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("OPENROUTER_API_KEY", "sk-test-key-1234") cfg = Config() session = AgentSession(cfg) try: @@ -400,3 +403,44 @@ def test_slash_completer_healing(self) -> None: assert "reset" in sub_matches assert "enable" in sub_matches assert "disable" in sub_matches + + def test_renderer_renders_healing_events(self) -> None: + from agentcli.agent.events import ( + AutoHealingRollbackEvent, + DriftDetectedEvent, + StrategyRecoveryEvent, + ) + from agentcli.ui.render import ConsoleRenderer + + ev1 = DriftDetectedEvent( + iteration=1, + drift_score=0.85, + severity="critical", + is_loop_detected=True, + reasons=["Loop detected on file edit"], + ) + ev2 = AutoHealingRollbackEvent( + iteration=1, + snapshot_id="snap1", + trigger="cycle_detected", + reverted_files=["app.py"], + ) + ev3 = StrategyRecoveryEvent( + iteration=1, + diagnostics="Score: 0.85, Cycle: action#1", + strategy_prompt="recovery guidance", + ) + + renderer_plain = ConsoleRenderer() + renderer_plain._rich_available = False + renderer_plain.render_loop_event(ev1, verbose=True) + renderer_plain.render_loop_event(ev2, verbose=True) + renderer_plain.render_loop_event(ev3, verbose=True) + + renderer_rich = ConsoleRenderer() + renderer_rich._rich_available = True + renderer_rich.render_loop_event(ev1, verbose=True) + renderer_rich.render_loop_event(ev2, verbose=True) + renderer_rich.render_loop_event(ev3, verbose=True) + +