File size: 10,935 Bytes
b5b9c2e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 | """Tests for cron job inactivity-based timeout.
Tests cover:
- Active agent runs indefinitely (no inactivity timeout)
- Idle agent triggers inactivity timeout with diagnostic info
- Unlimited timeout (HERMES_CRON_TIMEOUT=0)
- Backward compat: HERMES_CRON_TIMEOUT env var still works
- Error message includes activity summary
"""
import concurrent.futures
import os
import sys
import time
import threading
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
# Ensure project root is importable
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
class FakeAgent:
"""Mock agent with controllable activity summary for timeout tests."""
def __init__(self, idle_seconds=0.0, activity_desc="tool_call",
current_tool=None, api_call_count=5, max_iterations=90):
self._idle_seconds = idle_seconds
self._activity_desc = activity_desc
self._current_tool = current_tool
self._api_call_count = api_call_count
self._max_iterations = max_iterations
self._interrupted = False
self._interrupt_msg = None
def get_activity_summary(self):
return {
"last_activity_ts": time.time() - self._idle_seconds,
"last_activity_desc": self._activity_desc,
"seconds_since_activity": self._idle_seconds,
"current_tool": self._current_tool,
"api_call_count": self._api_call_count,
"max_iterations": self._max_iterations,
}
def interrupt(self, msg):
self._interrupted = True
self._interrupt_msg = msg
def run_conversation(self, prompt):
"""Simulate a quick agent run that finishes immediately."""
return {"final_response": "Done", "messages": []}
class SlowFakeAgent(FakeAgent):
"""Agent that runs for a while, simulating active work then going idle."""
def __init__(self, run_duration=0.5, idle_after=None, **kwargs):
super().__init__(**kwargs)
self._run_duration = run_duration
self._idle_after = idle_after # seconds before becoming idle
self._start_time = None
def get_activity_summary(self):
summary = super().get_activity_summary()
if self._idle_after is not None and self._start_time:
elapsed = time.time() - self._start_time
if elapsed > self._idle_after:
# Agent has gone idle
idle_time = elapsed - self._idle_after
summary["seconds_since_activity"] = idle_time
summary["last_activity_desc"] = "api_call_streaming"
else:
summary["seconds_since_activity"] = 0.0
return summary
def run_conversation(self, prompt):
self._start_time = time.time()
time.sleep(self._run_duration)
return {"final_response": "Completed after work", "messages": []}
class TestInactivityTimeout:
"""Test the inactivity-based timeout polling loop in cron scheduler."""
def test_active_agent_completes_normally(self):
"""An agent that finishes quickly should return its result."""
agent = FakeAgent(idle_seconds=0.0)
_cron_inactivity_limit = 10.0
_POLL_INTERVAL = 0.1
pool = concurrent.futures.ThreadPoolExecutor(max_workers=1)
future = pool.submit(agent.run_conversation, "test prompt")
_inactivity_timeout = False
result = None
while True:
done, _ = concurrent.futures.wait({future}, timeout=_POLL_INTERVAL)
if done:
result = future.result()
break
_idle_secs = 0.0
if hasattr(agent, "get_activity_summary"):
_act = agent.get_activity_summary()
_idle_secs = _act.get("seconds_since_activity", 0.0)
if _idle_secs >= _cron_inactivity_limit:
_inactivity_timeout = True
break
pool.shutdown(wait=False)
assert result is not None
assert result["final_response"] == "Done"
assert not _inactivity_timeout
assert not agent._interrupted
def test_idle_agent_triggers_timeout(self):
"""An agent that goes idle should be detected and interrupted."""
# Agent will run for 0.3s, then become idle after 0.1s of that
agent = SlowFakeAgent(
run_duration=5.0, # would run forever without timeout
idle_after=0.1, # goes idle almost immediately
activity_desc="api_call_streaming",
current_tool="web_search",
api_call_count=3,
max_iterations=50,
)
_cron_inactivity_limit = 0.5 # 0.5s inactivity triggers timeout
_POLL_INTERVAL = 0.1
pool = concurrent.futures.ThreadPoolExecutor(max_workers=1)
future = pool.submit(agent.run_conversation, "test prompt")
_inactivity_timeout = False
result = None
while True:
done, _ = concurrent.futures.wait({future}, timeout=_POLL_INTERVAL)
if done:
result = future.result()
break
_idle_secs = 0.0
if hasattr(agent, "get_activity_summary"):
try:
_act = agent.get_activity_summary()
_idle_secs = _act.get("seconds_since_activity", 0.0)
except Exception:
pass
if _idle_secs >= _cron_inactivity_limit:
_inactivity_timeout = True
break
pool.shutdown(wait=False, cancel_futures=True)
assert _inactivity_timeout is True
assert result is None # Never got a result — interrupted
def test_unlimited_timeout(self):
"""HERMES_CRON_TIMEOUT=0 means no timeout at all."""
agent = FakeAgent(idle_seconds=0.0)
_cron_inactivity_limit = None # unlimited
pool = concurrent.futures.ThreadPoolExecutor(max_workers=1)
future = pool.submit(agent.run_conversation, "test prompt")
# With unlimited, we just await the result directly.
result = future.result()
pool.shutdown(wait=False)
assert result["final_response"] == "Done"
def test_timeout_env_var_parsing(self, monkeypatch):
"""HERMES_CRON_TIMEOUT env var is respected."""
monkeypatch.setenv("HERMES_CRON_TIMEOUT", "1200")
_cron_timeout = float(os.getenv("HERMES_CRON_TIMEOUT", 600))
assert _cron_timeout == 1200.0
_cron_inactivity_limit = _cron_timeout if _cron_timeout > 0 else None
assert _cron_inactivity_limit == 1200.0
def test_timeout_zero_means_unlimited(self, monkeypatch):
"""HERMES_CRON_TIMEOUT=0 yields None (unlimited)."""
monkeypatch.setenv("HERMES_CRON_TIMEOUT", "0")
_cron_timeout = float(os.getenv("HERMES_CRON_TIMEOUT", 600))
_cron_inactivity_limit = _cron_timeout if _cron_timeout > 0 else None
assert _cron_inactivity_limit is None
def test_timeout_error_includes_diagnostics(self):
"""The TimeoutError message should include last activity info."""
agent = SlowFakeAgent(
run_duration=5.0,
idle_after=0.05,
activity_desc="api_call_streaming",
current_tool="delegate_task",
api_call_count=7,
max_iterations=90,
)
_cron_inactivity_limit = 0.3
_POLL_INTERVAL = 0.1
pool = concurrent.futures.ThreadPoolExecutor(max_workers=1)
future = pool.submit(agent.run_conversation, "test")
_inactivity_timeout = False
while True:
done, _ = concurrent.futures.wait({future}, timeout=_POLL_INTERVAL)
if done:
break
_idle_secs = 0.0
if hasattr(agent, "get_activity_summary"):
try:
_act = agent.get_activity_summary()
_idle_secs = _act.get("seconds_since_activity", 0.0)
except Exception:
pass
if _idle_secs >= _cron_inactivity_limit:
_inactivity_timeout = True
break
pool.shutdown(wait=False, cancel_futures=True)
assert _inactivity_timeout
# Build the diagnostic message like the scheduler does
_activity = agent.get_activity_summary()
_last_desc = _activity.get("last_activity_desc", "unknown")
_secs_ago = _activity.get("seconds_since_activity", 0)
err_msg = (
f"Cron job 'test-job' idle for "
f"{int(_secs_ago)}s (limit {int(_cron_inactivity_limit)}s) "
f"— last activity: {_last_desc}"
)
assert "idle for" in err_msg
assert "api_call_streaming" in err_msg
def test_agent_without_activity_summary_uses_wallclock_fallback(self):
"""If agent lacks get_activity_summary, idle_secs stays 0 (never times out).
This ensures backward compat if somehow an old agent is used.
The polling loop will eventually complete when the task finishes.
"""
class BareAgent:
def run_conversation(self, prompt):
return {"final_response": "no activity tracker", "messages": []}
agent = BareAgent()
_cron_inactivity_limit = 0.1 # tiny limit
_POLL_INTERVAL = 0.1
pool = concurrent.futures.ThreadPoolExecutor(max_workers=1)
future = pool.submit(agent.run_conversation, "test")
_inactivity_timeout = False
while True:
done, _ = concurrent.futures.wait({future}, timeout=_POLL_INTERVAL)
if done:
result = future.result()
break
_idle_secs = 0.0
if hasattr(agent, "get_activity_summary"):
try:
_act = agent.get_activity_summary()
_idle_secs = _act.get("seconds_since_activity", 0.0)
except Exception:
pass
if _idle_secs >= _cron_inactivity_limit:
_inactivity_timeout = True
break
pool.shutdown(wait=False)
# Should NOT have timed out — bare agent has no get_activity_summary
assert not _inactivity_timeout
assert result["final_response"] == "no activity tracker"
class TestSysPathOrdering:
"""Test that sys.path is set before repo-level imports."""
def test_hermes_time_importable(self):
"""hermes_time should be importable when cron.scheduler loads."""
# This import would fail if sys.path.insert comes after the import
from cron.scheduler import _hermes_now
assert callable(_hermes_now)
def test_hermes_constants_importable(self):
"""hermes_constants should be importable from cron context."""
from hermes_constants import get_hermes_home
assert callable(get_hermes_home)
|