Upload aco/doom_detector.py

aco/doom_detector.py

@@ -0,0 +1,287 @@
+"""Early Termination / Doom Detector - Module 10.
+
+Detects runs that are unlikely to succeed without more information or intervention.
+
+Signals:
+- repeated failed tool calls
+- no artifact progress
+- growing cost without new evidence
+- repeated planning
+- verifier disagreement
+- context confusion
+- escalating retries
+- model loop behavior
+
+Actions:
+- stop
+- mark BLOCKED
+- ask one targeted question
+- switch strategy
+- escalate model
+- escalate human
+"""
+
+from typing import Dict, List, Optional, Any
+from dataclasses import dataclass
+from enum import Enum
+
+from .trace_schema import AgentTrace, TraceStep, Outcome, FailureTag
+from .config import ACOConfig
+
+
+class DoomAction(Enum):
+    STOP = "stop"
+    MARK_BLOCKED = "mark_blocked"
+    ASK_TARGETED_QUESTION = "ask_targeted_question"
+    SWITCH_STRATEGY = "switch_strategy"
+    ESCALATE_MODEL = "escalate_model"
+    ESCALATE_HUMAN = "escalate_human"
+    CONTINUE = "continue"
+
+
+@dataclass
+class DoomAssessment:
+    action: DoomAction
+    confidence: float
+    reasoning: str
+    signals_triggered: List[str]
+    recommended_action: Optional[str] = None
+    question_to_ask: Optional[str] = None
+
+
+class DoomDetector:
+    """Detects doomed agent runs and recommends intervention."""
+
+    # Signal weights
+    SIGNAL_WEIGHTS = {
+        "repeated_tool_failures": 0.3,
+        "no_artifact_progress": 0.25,
+        "cost_explosion": 0.3,
+        "repeated_planning": 0.2,
+        "verifier_disagreement": 0.25,
+        "context_confusion": 0.2,
+        "escalating_retries": 0.35,
+        "model_loop": 0.4,
+        "stagnant_context": 0.15,
+    }
+
+    # Doom threshold
+    DOOM_THRESHOLD = 0.6
+    BLOCKED_THRESHOLD = 0.8
+
+    def __init__(self, config: Optional[ACOConfig] = None):
+        self.config = config or ACOConfig()
+        self.assessment_history: List[DoomAssessment] = []
+
+    def assess(
+        self,
+        trace: AgentTrace,
+        current_step: TraceStep,
+        predicted_cost: float,
+        predicted_steps: int,
+    ) -> DoomAssessment:
+        """Assess whether a run is doomed and what action to take."""
+        
+        signals = []
+        score = 0.0
+        
+        # Signal 1: Repeated failed tool calls
+        tool_failures = self._count_recent_tool_failures(trace)
+        if tool_failures >= 3:
+            signals.append("repeated_tool_failures")
+            score += self.SIGNAL_WEIGHTS["repeated_tool_failures"] * min(tool_failures / 5, 1.0)
+        
+        # Signal 2: No artifact progress
+        if len(trace.final_artifacts) == 0 and len(trace.steps) > 5:
+            signals.append("no_artifact_progress")
+            score += self.SIGNAL_WEIGHTS["no_artifact_progress"]
+        
+        # Signal 3: Cost explosion
+        total_cost = trace.total_cost_computed
+        cost_ratio = total_cost / max(predicted_cost, 0.0001)
+        if cost_ratio > self.config.doom_max_cost_ratio:
+            signals.append("cost_explosion")
+            score += self.SIGNAL_WEIGHTS["cost_explosion"] * min(cost_ratio / 5, 1.0)
+        
+        # Signal 4: Repeated planning (re-planning without execution)
+        replan_count = self._count_replanning(trace)
+        if replan_count >= 2:
+            signals.append("repeated_planning")
+            score += self.SIGNAL_WEIGHTS["repeated_planning"] * min(replan_count / 4, 1.0)
+        
+        # Signal 5: Verifier disagreement
+        verifier_disagree = self._count_verifier_disagreement(trace)
+        if verifier_disagree >= self.config.doom_verifier_disagreement_threshold:
+            signals.append("verifier_disagreement")
+            score += self.SIGNAL_WEIGHTS["verifier_disagreement"] * min(verifier_disagree / 4, 1.0)
+        
+        # Signal 6: Context confusion (rapid context size oscillation)
+        if len(trace.steps) >= 3:
+            ctx_sizes = [s.context_size_tokens for s in trace.steps[-3:]]
+            if max(ctx_sizes) - min(ctx_sizes) > 5000:
+                signals.append("context_confusion")
+                score += self.SIGNAL_WEIGHTS["context_confusion"]
+        
+        # Signal 7: Escalating retries
+        total_retries = trace.total_retries
+        if total_retries >= self.config.doom_max_retries * len(trace.steps) * 0.5:
+            signals.append("escalating_retries")
+            score += self.SIGNAL_WEIGHTS["escalating_retries"] * min(total_retries / 10, 1.0)
+        
+        # Signal 8: Model loop (same model producing same outputs)
+        loop_detected = self._detect_model_loop(trace)
+        if loop_detected:
+            signals.append("model_loop")
+            score += self.SIGNAL_WEIGHTS["model_loop"]
+        
+        # Signal 9: Stagnant context (no new information in last N steps)
+        if self._is_context_stagnant(trace):
+            signals.append("stagnant_context")
+            score += self.SIGNAL_WEIGHTS["stagnant_context"]
+        
+        # Cap score
+        score = min(score, 1.0)
+        
+        if score < 0.3:
+            return DoomAssessment(
+                action=DoomAction.CONTINUE,
+                confidence=1.0 - score,
+                reasoning="Run appears healthy.",
+                signals_triggered=signals,
+            )
+        
+        if score < self.DOOM_THRESHOLD:
+            return DoomAssessment(
+                action=DoomAction.ASK_TARGETED_QUESTION,
+                confidence=score,
+                reasoning=f"Early warning: {', '.join(signals)}. Asking targeted question.",
+                signals_triggered=signals,
+                question_to_ask=self._generate_targeted_question(trace, signals),
+            )
+        
+        if score < self.BLOCKED_THRESHOLD:
+            # Decide between strategy switch, model escalation, or stop
+            if current_step.model_call and current_step.model_call.model_id:
+                current_tier = self._infer_tier(current_step.model_call.model_id)
+                if current_tier < 4:
+                    return DoomAssessment(
+                        action=DoomAction.ESCALATE_MODEL,
+                        confidence=score,
+                        reasoning=f"Run struggling ({score:.2f} doom score). Escalating model.",
+                        signals_triggered=signals,
+                    )
+            
+            return DoomAssessment(
+                action=DoomAction.SWITCH_STRATEGY,
+                confidence=score,
+                reasoning=f"Run struggling ({score:.2f} doom score). Switching strategy.",
+                signals_triggered=signals,
+                recommended_action="Change approach: retrieve more context, simplify task decomposition",
+            )
+        
+        # High doom score — mark blocked or escalate human
+        if score > 0.95:
+            return DoomAssessment(
+                action=DoomAction.ESCALATE_HUMAN,
+                confidence=score,
+                reasoning=f"Critical failure pattern ({score:.2f} doom score). Human escalation required.",
+                signals_triggered=signals,
+            )
+        
+        return DoomAssessment(
+            action=DoomAction.MARK_BLOCKED,
+            confidence=score,
+            reasoning=f"Doom threshold exceeded ({score:.2f}). Marking BLOCKED.",
+            signals_triggered=signals,
+        )
+
+    def _count_recent_tool_failures(self, trace: AgentTrace, window: int = 5) -> int:
+        recent_steps = trace.steps[-window:] if len(trace.steps) > window else trace.steps
+        return sum(
+            1 for step in recent_steps
+            for tc in step.tool_calls
+            if tc.failed
+        )
+
+    def _count_replanning(self, trace: AgentTrace) -> int:
+        replan_count = 0
+        for step in trace.steps:
+            # Heuristic: step mentions plan but no tool calls or artifacts
+            if step.planned_next and not step.tool_calls and not step.artifacts_created:
+                replan_count += 1
+        return replan_count
+
+    def _count_verifier_disagreement(self, trace: AgentTrace) -> int:
+        disagreements = 0
+        for step in trace.steps:
+            verifiers = step.verifier_calls
+            if len(verifiers) >= 2:
+                results = [v.passed for v in verifiers]
+                if any(results) and not all(results):
+                    disagreements += 1
+            elif len(verifiers) == 1:
+                # Verifier rejected but step proceeded anyway
+                if not verifiers[0].passed:
+                    disagreements += 1
+        return disagreements
+
+    def _detect_model_loop(self, trace: AgentTrace) -> bool:
+        if len(trace.steps) < 4:
+            return False
+        # Check if last 4 steps have identical or very similar tool call patterns
+        last4 = trace.steps[-4:]
+        patterns = [
+            tuple(tc.tool_name for tc in s.tool_calls)
+            for s in last4
+        ]
+        return len(set(patterns)) <= 2 and len(patterns) == 4
+
+    def _is_context_stagnant(self, trace: AgentTrace, window: int = 3) -> bool:
+        if len(trace.steps) < window:
+            return False
+        recent = trace.steps[-window:]
+        sources = [set(s.context_sources) for s in recent]
+        # If no new sources introduced
+        if len(sources) >= 2:
+            for i in range(1, len(sources)):
+                if sources[i] - sources[i-1]:
+                    return False
+            return True
+        return False
+
+    def _generate_targeted_question(self, trace: AgentTrace, signals: List[str]) -> str:
+        if "repeated_tool_failures" in signals:
+            return "The requested tools are failing repeatedly. Can you provide the correct parameters or clarify the task scope?"
+        if "no_artifact_progress" in signals:
+            return "No progress has been made on the expected deliverable. Is there a specific format or file you need?"
+        if "cost_explosion" in signals:
+            return "This task is taking more resources than expected. Can you narrow the scope or clarify priorities?"
+        if "repeated_planning" in signals:
+            return "The agent keeps re-planning. What is the single most important next step?"
+        return "Can you clarify or narrow the task requirements to help the agent proceed more efficiently?"
+
+    def _infer_tier(self, model_id: str) -> int:
+        # Simplified tier inference
+        if "frontier" in model_id.lower() or "gpt-4" in model_id.lower():
+            return 4
+        if "medium" in model_id.lower():
+            return 3
+        if "small" in model_id.lower() or "mini" in model_id.lower():
+            return 2
+        return 1
+
+    def get_stats(self) -> Dict[str, Any]:
+        """Return doom detection statistics."""
+        total = len(self.assessment_history)
+        if total == 0:
+            return {"total_assessments": 0}
+        
+        action_counts = {}
+        for a in self.assessment_history:
+            action_counts[a.action.value] = action_counts.get(a.action.value, 0) + 1
+        
+        return {
+            "total_assessments": total,
+            "action_distribution": action_counts,
+            "avg_confidence": sum(a.confidence for a in self.assessment_history) / total,
+        }