Watchdog: When Your AI Agent Needs a Babysitter

Here's the problem. Your agent has a self_modify tool. It can change its own config, rewrite its prompts, update its system instructions. Sounds great until it enters a death spiral.

I've seen it happen. Agent decides its temperature is too low. Bumps it to 0.9. Output gets weirder. Agent decides it needs MORE creativity. Bumps to 1.2. Now it's hallucinating database schemas. Agent tries to "fix" itself by rewriting its system prompt. Now it thinks it's a pirate. You get billed $47 before someone notices.

The fix? A watchdog. Not the Kubernetes kind. Something that watches the agent itself.

What Actually Breaks

Three failure modes. You've seen all of them.

Infinite loops. Agent calls a tool. Tool returns data. Agent calls the same tool with the same args. Again. Again. Token count hits 500K. You're paying for nothing.

Unresponsive LLM. API timeout. Rate limit. Network blip. Agent doesn't handle it gracefully — just sits there waiting. No error. No retry. Just silence.

Self-modify degradation. Agent changes its prompt from "helpful assistant" to "maximize engagement." Suddenly it's arguing with users. Or it drops the temperature too low and produces the same response to every question.

The Watchdog Pattern

class AgentWatchdog:
    def __init__(self, agent, config: WatchdogConfig):
        self.agent = agent
        self.config = config
        self.metrics = MetricsCollector()
        self.anomaly_detector = AnomalyDetector()

    async def monitor(self):
        while self.agent.is_running:
            snapshot = await self.collect_snapshot()

            if self.anomaly_detector.detect(snapshot):
                await self.handle_anomaly(snapshot)

            await asyncio.sleep(self.config.poll_interval)

Three things to watch. Always.

1. Loop Detection

class LoopDetector:
    def __init__(self, max_repeats: int = 5, window_seconds: int = 60):
        self.max_repeats = max_repeats
        self.call_history = deque(maxlen=100)

    def check(self, tool_name: str, args: dict) -> bool:
        self.call_history.append((tool_name, args, time.time()))

        # Count identical calls in window
        recent = [c for c in self.call_history 
                  if c[2] > time.time() - self.window_seconds]

        repeats = sum(1 for c in recent 
                     if c[0] == tool_name and c[1] == args)

        return repeats >= self.max_repeats

Simple. Effective. Catches the "call the same search API with the same query 47 times" pattern.

2. Health Check

class HealthChecker:
    def __init__(self, timeout_seconds: int = 30):
        self.timeout = timeout_seconds

    async def check(self, agent) -> HealthStatus:
        try:
            start = time.time()
            response = await asyncio.wait_for(
                agent.ping(), 
                timeout=self.timeout
            )
            latency = time.time() - start

            return HealthStatus(
                alive=True,
                latency_ms=latency * 1000,
                last_response=response
            )
        except asyncio.TimeoutError:
            return HealthStatus(alive=False, error="timeout")
        except Exception as e:
            return HealthStatus(alive=False, error=str(e))

Ping the LLM. If it doesn't respond in 30 seconds, something's wrong. Don't wait for the user to notice.

3. Anomaly Detection

class AnomalyDetector:
    def __init__(self, baseline_window: int = 100):
        self.baseline = deque(maxlen=baseline_window)
        self.threshold_multiplier = 3.0  # Standard deviation multiplier

    def update_baseline(self, metrics: dict):
        self.baseline.append(metrics)

    def is_anomalous(self, current: dict) -> bool:
        if len(self.baseline) < 10:
            return False  # Not enough data yet

        for key in current:
            values = [m[key] for m in self.baseline if key in m]
            if not values:
                continue

            mean = sum(values) / len(values)
            variance = sum((v - mean) ** 2 for v in values) / len(values)
            std_dev = variance ** 0.5

            if abs(current[key] - mean) > self.threshold_multiplier * std_dev:
                return True

        return False

Token count suddenly 10x normal? Latency spiking? Cost per step went from $0.02 to $0.50? Flag it.

The Automatic Response

Detection is useless without action. Here's what you do:

class WatchdogAction:
    def __init__(self, git_repo: str):
        self.repo = git_repo

    async def handle_anomaly(self, snapshot: AgentSnapshot):
        # 1. Log everything
        await self.log_incident(snapshot)

        # 2. Kill the current execution
        await self.agent.stop()

        # 3. Revert to last known good state
        last_good = await self.find_last_good_commit()
        await self.git_revert(last_good)

        # 4. Notify
        await self.send_alert(snapshot)

    async def git_revert(self, commit_hash: str):
        subprocess.run(["git", "revert", "--no-commit", commit_hash])
        subprocess.run(["git", "commit", "-m", 
                       f"auto-revert: watchdog detected anomaly"])
        subprocess.run(["git", "push"])

Three steps. Stop the bleeding. Revert the damage. Tell someone.

The Hard Parts

This sounds simple. It's not. Two things will bite you.

False positives. Your anomaly detector flags a legit spike in usage. Now you've killed a running agent and reverted configs for nothing. Solution: require multiple consecutive anomalies before acting. Or use a confirmation window.

What's "good"? The agent modifies itself constantly. Which commit is the "last good" one? You need a baseline — a snapshot taken right after deployment, before any self-modification. Mark it as golden.

class GoldenBaseline:
    def __init__(self):
        self.golden_commit = None

    def mark_golden(self):
        result = subprocess.run(["git", "rev-parse", "HEAD"], 
                               capture_output=True, text=True)
        self.golden_commit = result.stdout.strip()
        subprocess.run(["git", "tag", "golden", self.golden_commit])

What This Looks Like in Practice

An agent runs for 6 hours. It's modified its system prompt 12 times. Token count per step is stable at ~2K. Then it hits a weird edge case. The self_modify tool runs and sets max_tokens to 999999. Next LLM call tries to generate a million tokens. Cost spikes from $0.03 to $15 per step.

The watchdog catches it. Token count is 500x baseline. Latency is 30x normal. It kills the agent, reverts to the golden commit, and sends a Slack message. Total time from anomaly to recovery: 12 seconds.

Without the watchdog? Someone notices the billing alert 4 hours later.

The Missing Piece

You need one more thing. The watchdog itself needs monitoring. If it crashes, you're blind.

python
class WatchdogSupervisor:
    def __init__(self):
        self.watchdog = AgentWatchdog()
        self.heartbeat_interval = 5  # seconds

    async def run(self):
        while True:
            try:
                await asyncio.wait_for(
                    self.watchdog.monitor(),
                    timeout=self.heartbeat_interval * 3
                )
            except as

---

**Debugging AI agents shouldn't feel like reading The Matrix.** 
Join other engineers who are building reliable autonomous workflows in our community: [TracePilot Discord](https://discord.gg/KzXRAXFM8)