import time
import pytest
from amicus.monitor import HeartbeatMonitor
from amicus import core, server
def test_zombie_detection(temp_context_dir):
"""Test that zombie agents are detected and locks released."""
# Setup state
core.set_tracking_enabled(True)
state_file = core.get_state_file()
# Create a zombie state (old heartbeat)
zombie_state = {
"summary": "Working...",
"next_steps": [{"task": "Step 1"}],
"active_files": ["locked.py"],
"ask_user": False,
"last_heartbeat": time.time() - 100, # 100s ago
"timestamp": time.time() - 100,
"cluster_nodes": {
"Zombie-1": {
"role": "worker",
"last_heartbeat": time.time() - 100
}
}
}
core.write_with_lock(state_file, zombie_state)
# Run monitor check manually
monitor = HeartbeatMonitor(interval=0.1)
monitor._check_health()
# Verify cleanup
new_state = core.read_with_lock(state_file)
assert new_state["active_files"] == []
assert new_state["ask_user"] is True
summary = str(new_state.get("summary", ""))
assert "crashed" in summary and "Locks released" in summary
def test_active_agent_ignored(temp_context_dir):
"""Test that active agents are NOT touched."""
# Setup state
core.set_tracking_enabled(True)
state_file = core.get_state_file()
# Create active state
active_state = {
"summary": "Working...",
"next_steps": [{"task": "Step 1"}],
"active_files": ["locked.py"],
"ask_user": False,
"last_heartbeat": time.time(), # Now
"timestamp": time.time(),
"cluster_nodes": {
"BM-1": {
"role": "bootstrap_manager",
"last_heartbeat": time.time()
}
}
}
core.write_with_lock(state_file, active_state)
# Run monitor check
monitor = HeartbeatMonitor(interval=0.1)
monitor._check_health()
# Verify NO change
new_state = core.read_with_lock(state_file)
assert new_state["active_files"] == ["locked.py"]
assert new_state["ask_user"] is False
