Upload aco/doom_detector.py with huggingface_hub

aco/doom_detector.py

@@ -1,287 +1,91 @@
-"""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
+"""Early Termination / Doom Detector: Stop runs unlikely to succeed."""
+from typing import Dict, List, Optional
 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 DoomSignal:
+    signal_type: str
+    severity: float  # 0-1
+    evidence: str
 
 @dataclass
 class DoomAssessment:
-    action: DoomAction
-    confidence: float
+    doomed: bool
+    severity: float  # 0-1
+    signals: List[DoomSignal]
+    recommended_action: str  # "continue","stop","mark_blocked","ask_question","switch_strategy"
     reasoning: str
-    signals_triggered: List[str]
-    recommended_action: Optional[str] = None
-    question_to_ask: Optional[str] = None
 
+DOOM_THRESHOLDS = {
+    "max_failed_tool_calls": 3,
+    "max_repeated_planning": 2,
+    "max_cost_without_progress": 2.0,
+    "max_context_confusion": 0.7,
+    "max_escalation_loops": 2,
+}
 
 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, doom_threshold: float = 0.7):
+        self.doom_threshold = doom_threshold
+        self.assessments: List[DoomAssessment] = []
 
-    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."""
-        
+    def assess(self, steps: List[Dict], current_cost: float, max_cost: float,
+               model_tier: int, verifier_disagreements: int = 0) -> DoomAssessment:
         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,
-        }
+        severity = 0.0
+        # Check signals
+        failed_tools = sum(1 for s in steps for tc in s.get("tool_calls",[])
+                          if not tc.get("success", True))
+        if failed_tools >= DOOM_THRESHOLDS["max_failed_tool_calls"]:
+            signals.append(DoomSignal("failed_tool_calls", 0.6,
+                          f"{failed_tools} failed tool calls"))
+            severity += 0.3
+
+        no_progress = all(not s.get("artifacts_created") for s in steps)
+        if no_progress and len(steps) > 3:
+            signals.append(DoomSignal("no_artifact_progress", 0.5,
+                          "no artifacts after 3+ steps"))
+            severity += 0.25
+
+        if current_cost > DOOM_THRESHOLDS["max_cost_without_progress"] and no_progress:
+            signals.append(DoomSignal("growing_cost_no_progress", 0.7,
+                          f"cost={current_cost:.2f} with no progress"))
+            severity += 0.35
+
+        planning_steps = sum(1 for s in steps if s.get("retry_num",0) > 0)
+        if planning_steps >= DOOM_THRESHOLDS["max_repeated_planning"]:
+            signals.append(DoomSignal("repeated_planning", 0.4,
+                          f"{planning_steps} retry steps"))
+            severity += 0.2
+
+        if verifier_disagreements >= 2:
+            signals.append(DoomSignal("verifier_disagreement", 0.6,
+                          f"{verifier_disagreements} verifier disagreements"))
+            severity += 0.3
+
+        if current_cost >= max_cost * 0.9:
+            signals.append(DoomSignal("approaching_cost_limit", 0.5,
+                          f"cost={current_cost:.2f} / {max_cost:.2f}"))
+            severity += 0.4
+
+        severity = min(severity, 1.0)
+        doomed = severity >= self.doom_threshold
+        # Determine action
+        if not doomed:
+            action = "continue"
+            reasoning = f"severity={severity:.2f} < threshold={self.doom_threshold}"
+        elif failed_tools >= 3 and no_progress:
+            action = "mark_blocked"
+            reasoning = "too many failures with no progress"
+        elif current_cost >= max_cost * 0.9:
+            action = "stop"
+            reasoning = "approaching cost limit"
+        elif verifier_disagreements >= 2:
+            action = "switch_strategy"
+            reasoning = "verifier disagreement suggests wrong approach"
+        else:
+            action = "ask_question"
+            reasoning = "run may be recoverable with user input"
+        assessment = DoomAssessment(doomed, severity, signals, action, reasoning)
+        self.assessments.append(assessment)
+        return assessment