"""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, }