"""Regression tests for subagent timeout diagnostic dump (issue #14726). When delegate_task's child subagent times out without having made any API call, a structured diagnostic file is written under ``~/.hermes/logs/subagent-timeout--.log``. This gives users a concrete artifact to inspect (worker thread stack, system prompt size, tool schema bytes, credential pool state, etc.) instead of the previous opaque "subagent timed out" error. These tests pin: - the diagnostic writer's output format and content - the timeout branch in _run_single_child only dumps when api_calls == 0 - the error message surfaces the diagnostic path - api_calls > 0 timeouts do NOT write a dump (the old "stuck on slow API call" explanation still applies) """ from __future__ import annotations import threading from pathlib import Path from unittest.mock import MagicMock import pytest @pytest.fixture def hermes_home(tmp_path, monkeypatch): home = tmp_path / ".hermes" home.mkdir() monkeypatch.setenv("HERMES_HOME", str(home)) return home class _StubChild: """Minimal stand-in for an AIAgent subagent.""" def __init__( self, *, api_call_count: int = 0, hang_seconds: float = 5.0, subagent_id: str = "sa-0-stubabc", tool_schema=None, ): self._subagent_id = subagent_id self._delegate_depth = 1 self._delegate_role = "leaf" self.model = "test/model" self.provider = "testprov" self.api_mode = "chat_completions" self.base_url = "https://example.test/v1" self.max_iterations = 30 self.quiet_mode = True self.skip_memory = True self.skip_context_files = True self.platform = "cli" self.ephemeral_system_prompt = "sys prompt" self.enabled_toolsets = ["web", "terminal"] self.valid_tool_names = {"web_search", "terminal"} self.tools = tool_schema if tool_schema is not None else [ {"name": "web_search", "description": "search"}, {"name": "terminal", "description": "shell"}, ] self._api_call_count = api_call_count self._hang = threading.Event() self._hang_seconds = hang_seconds def get_activity_summary(self): return { "api_call_count": self._api_call_count, "max_iterations": self.max_iterations, "current_tool": None, "seconds_since_activity": 60, } def run_conversation(self, user_message, task_id=None, stream_callback=None): self._hang.wait(self._hang_seconds) return {"final_response": "", "completed": False, "api_calls": self._api_call_count} def interrupt(self): self._hang.set() # ── _dump_subagent_timeout_diagnostic ────────────────────────────────── class TestDumpSubagentTimeoutDiagnostic: def test_returns_none_on_unwritable_logs_dir(self, tmp_path, monkeypatch): # Point HERMES_HOME at an unwritable path so logs/ can't be created # (simulates permission-denied). Helper must not raise. from tools.delegate_tool import _dump_subagent_timeout_diagnostic bogus = tmp_path / "does-not-exist" / ".hermes" monkeypatch.setenv("HERMES_HOME", str(bogus)) child = _StubChild() # Make the logs dir itself unwritable by creating it as a FILE # so mkdir(exist_ok=True) → NotADirectoryError and we fall through. bogus.parent.mkdir(parents=True, exist_ok=True) bogus.mkdir() (bogus / "logs").write_text("not a dir") result = _dump_subagent_timeout_diagnostic( child=child, task_index=0, timeout_seconds=300.0, duration_seconds=300.0, worker_thread=None, goal="x", ) child.interrupt() # Either None (mkdir failed) or a real path; must never raise. # We assert no exception propagates — the return value is advisory. assert result is None or Path(result).exists() def test_timeout_diagnostic_marks_long_goal_as_non_original(self, hermes_home): # #121572: an elided goal must carry the non-imitable counted marker, # never the bare "...[truncated]" the model could copy. from agent.compression_marker import _COMPRESSION_MARKER_RE from tools.delegate_tool import _dump_subagent_timeout_diagnostic child = _StubChild() path = _dump_subagent_timeout_diagnostic( child=child, task_index=0, timeout_seconds=300.0, duration_seconds=300.0, worker_thread=None, goal="g" * 1200, ) text = Path(path).read_text(encoding="utf-8") assert _COMPRESSION_MARKER_RE.search(text) assert "g" * 100 in text assert "...[truncated]" not in text # ── _run_single_child timeout branch wiring ─────────────────────────── class TestRunSingleChildTimeoutDump: """The timeout branch in _run_single_child must emit the diagnostic dump when api_calls == 0, and must NOT emit it when api_calls > 0.""" def _invoke_with_short_timeout(self, child, monkeypatch): """Run _run_single_child with a tiny timeout to force the timeout branch.""" from tools import delegate_tool # Force a 0.3s timeout so the test is fast monkeypatch.setattr(delegate_tool, "_get_child_timeout", lambda: 0.3) parent = MagicMock() parent._touch_activity = MagicMock() parent._current_task_id = None return delegate_tool._run_single_child( task_index=0, goal="test goal", child=child, parent_agent=parent, ) def test_zero_api_calls_writes_dump_and_surfaces_path(self, hermes_home, monkeypatch): child = _StubChild(api_call_count=0, hang_seconds=10.0) result = self._invoke_with_short_timeout(child, monkeypatch) assert result["status"] == "timeout" assert result["api_calls"] == 0 assert result["diagnostic_path"] is not None dump_path = Path(result["diagnostic_path"]) assert dump_path.is_file() assert dump_path.parent == hermes_home / "logs" # Error message surfaces the path and the "no API call" phrasing assert "without making any API call" in result["error"] assert "Diagnostic:" in result["error"] assert str(dump_path) in result["error"] # ── explicit timeout metadata (#51690, salvaged from PR #60378) ──── def test_non_timeout_error_has_null_timeout_metadata(self, hermes_home, monkeypatch): """The metadata fields are timeout-specific — a child that raises must report them as None so consumers can key on presence.""" from tools import delegate_tool monkeypatch.setattr(delegate_tool, "_get_child_timeout", lambda: 30.0) child = _StubChild(api_call_count=1, hang_seconds=0.0) def _boom(*a, **kw): raise RuntimeError("child crashed") child.run_conversation = _boom parent = MagicMock() parent._touch_activity = MagicMock() parent._current_task_id = None result = delegate_tool._run_single_child( task_index=0, goal="test goal", child=child, parent_agent=parent, ) assert result["status"] == "error" assert result["timeout_seconds"] is None assert result["timed_out_after_seconds"] is None assert result["timeout_phase"] is None