10.4Heartbeat Monitoring

For long-running agent sessions (nightly exploratory tests), implement a heartbeat:

import threading
import time
import sys

class HeartbeatMonitor:
    """Kills the process if the agent stops making progress."""

    def __init__(self, max_idle_seconds: int = 120):
        self.max_idle = max_idle_seconds
        self.last_heartbeat = time.time()
        self._running = True
        self._thread = threading.Thread(target=self._monitor, daemon=True)
        self._thread.start()

    def beat(self):
        """Call this every time the agent takes an action."""
        self.last_heartbeat = time.time()

    def _monitor(self):
        while self._running:
            idle = time.time() - self.last_heartbeat
            if idle > self.max_idle:
                print(f"DEAD MAN'S SWITCH: No heartbeat for {idle:.0f}s. "
                      f"Killing process.", file=sys.stderr)
                os._exit(1)  # Hard exit — bypass cleanup
            time.sleep(10)  # Check every 10 seconds

    def stop(self):
        self._running = False

# Usage in the harness
monitor = HeartbeatMonitor(max_idle_seconds=120)

class HarnessWithHeartbeat(ConstrainedTestHarness):
    def execute(self, objective):
        for step in range(self.config.max_steps):
            monitor.beat()  # Signal that we are still making progress
            action = self.agent.next_action()
            result = self.agent.execute_action(action)
            if result.is_terminal:
                monitor.stop()
                return result