diff --git a/hermes_cli/config.py b/hermes_cli/config.py index a981b1bbbf14b..a835289c8407e 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -483,6 +483,7 @@ def ensure_hermes_home(): "api_key": "", # API key for delegation.base_url (falls back to OPENAI_API_KEY) "max_iterations": 50, # per-subagent iteration cap (each subagent gets its own budget, # independent of the parent's max_iterations) + "auto_route_coding_to_codex_yolo": True, }, # Ephemeral prefill messages file — JSON list of {role, content} dicts diff --git a/tests/conftest.py b/tests/conftest.py index 313a3cecfd65b..ce1ddb0d156f4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -77,32 +77,24 @@ def _ensure_current_event_loop(request): """Provide a default event loop for sync tests that call get_event_loop(). Python 3.11+ no longer guarantees a current loop for plain synchronous tests. - A number of gateway tests still use asyncio.get_event_loop().run_until_complete(...). - Ensure they always have a usable loop without interfering with pytest-asyncio's - own loop management for @pytest.mark.asyncio tests. + A number of sync tests still use asyncio.get_event_loop().run_until_complete(...). + Create a fresh loop for those tests without touching pytest-asyncio's own + loop management for @pytest.mark.asyncio tests. """ if request.node.get_closest_marker("asyncio") is not None: yield return - try: - loop = asyncio.get_event_loop_policy().get_event_loop() - except RuntimeError: - loop = None - - created = loop is None or loop.is_closed() - if created: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) try: yield finally: - if created and loop is not None: - try: - loop.close() - finally: - asyncio.set_event_loop(None) + try: + loop.close() + finally: + asyncio.set_event_loop(None) @pytest.fixture(autouse=True) diff --git a/tests/tools/test_notify_on_complete.py b/tests/tools/test_notify_on_complete.py index 8cf17bfbf6d42..a44e792bceaf8 100644 --- a/tests/tools/test_notify_on_complete.py +++ b/tests/tools/test_notify_on_complete.py @@ -11,6 +11,9 @@ import json import os import queue +import shlex +import sys +import threading import time import pytest from pathlib import Path @@ -50,6 +53,24 @@ def _make_session( return s +def _wait_until(predicate, timeout=5, interval=0.1): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(interval) + return False + + +def _drain_completion_queue(registry): + items = [] + while True: + try: + items.append(registry.completion_queue.get_nowait()) + except queue.Empty: + return items + + # ========================================================================= # ProcessSession field # ========================================================================= @@ -162,6 +183,142 @@ def test_multiple_completions_queued(self, registry): # Checkpoint persistence # ========================================================================= +class TestEndToEndNotifyFlow: + def test_long_running_poll_log_then_wait_still_notifies(self, registry, tmp_path): + session = ProcessSession( + id="proc_long_running", + command="fake long task", + task_id="t1", + started_at=time.time(), + cwd=str(tmp_path), + notify_on_complete=True, + ) + registry._running[session.id] = session + + def finish_in_background(): + with session._lock: + session.output_buffer += "start\n" + time.sleep(0.4) + with session._lock: + session.output_buffer += "middle\n" + time.sleep(0.8) + with session._lock: + session.output_buffer += "done\n" + session.exited = True + session.exit_code = 0 + registry._move_to_finished(session) + + worker = threading.Thread(target=finish_in_background, daemon=True) + worker.start() + + assert registry.completion_queue.empty() + assert _wait_until( + lambda: registry.poll(session.id)["status"] == "running" + and "start" in registry.poll(session.id)["output_preview"], + timeout=2, + ) + + poll_result = registry.poll(session.id) + assert poll_result["status"] == "running" + assert "start" in poll_result["output_preview"] + assert registry.completion_queue.empty() + + log_result = registry.read_log(session.id) + assert log_result["status"] == "running" + assert "start" in log_result["output"] + assert log_result["total_lines"] >= 1 + assert registry.completion_queue.empty() + + wait_result = registry.wait(session.id, timeout=5) + worker.join(timeout=1) + assert wait_result["status"] == "exited" + assert wait_result["exit_code"] == 0 + assert "done" in wait_result["output"] + + assert _wait_until(lambda: not registry.completion_queue.empty(), timeout=2) + completion = registry.completion_queue.get_nowait() + assert completion["session_id"] == session.id + assert completion["exit_code"] == 0 + assert "start" in completion["output"] + assert "middle" in completion["output"] + assert "done" in completion["output"] + + +class TestTerminalAndProcessIntegration: + def test_terminal_background_poll_log_wait_still_enqueues_completion(self, monkeypatch, tmp_path): + from tools.process_registry import _handle_process, process_registry + from tools.terminal_tool import cleanup_vm, terminal_tool + + monkeypatch.setenv("TERMINAL_ENV", "local") + task_id = f"notify_flow_{time.time_ns()}" + _drain_completion_queue(process_registry) + + python_code = ( + "import time; " + "print('start', flush=True); " + "time.sleep(0.4); " + "print('middle', flush=True); " + "time.sleep(0.8); " + "print('done', flush=True)" + ) + command = f"{shlex.quote(sys.executable)} -c {shlex.quote(python_code)}" + + try: + start_result = json.loads( + terminal_tool( + command=command, + background=True, + notify_on_complete=True, + task_id=task_id, + workdir=str(tmp_path), + ) + ) + session_id = start_result["session_id"] + assert start_result["notify_on_complete"] is True + assert start_result["exit_code"] == 0 + assert process_registry.completion_queue.empty() + + assert _wait_until( + lambda: ( + poll := json.loads(_handle_process({"action": "poll", "session_id": session_id})) + )["status"] == "running" + and "start" in poll["output_preview"], + timeout=3, + ) + + poll_result = json.loads(_handle_process({"action": "poll", "session_id": session_id})) + assert poll_result["status"] == "running" + assert "start" in poll_result["output_preview"] + assert process_registry.completion_queue.empty() + + log_result = json.loads(_handle_process({"action": "log", "session_id": session_id})) + assert log_result["status"] == "running" + assert "start" in log_result["output"] + assert log_result["total_lines"] >= 1 + assert process_registry.completion_queue.empty() + + wait_result = json.loads( + _handle_process({"action": "wait", "session_id": session_id, "timeout": 5}) + ) + assert wait_result["status"] == "exited" + assert wait_result["exit_code"] == 0 + assert "done" in wait_result["output"] + + assert _wait_until(lambda: not process_registry.completion_queue.empty(), timeout=2) + completion = process_registry.completion_queue.get_nowait() + assert completion["session_id"] == session_id + assert completion["exit_code"] == 0 + assert "start" in completion["output"] + assert "middle" in completion["output"] + assert "done" in completion["output"] + finally: + session = process_registry.get(start_result["session_id"]) if 'start_result' in locals() else None + if session and not session.exited: + process_registry.kill_process(start_result["session_id"]) + cleanup_vm(task_id) + _drain_completion_queue(process_registry) + + class TestCheckpointNotify: def test_checkpoint_includes_notify(self, registry, tmp_path): with patch("tools.process_registry.CHECKPOINT_PATH", tmp_path / "procs.json"): diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index a148a31f059eb..f17ef6c613bbc 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -38,6 +38,8 @@ MAX_DEPTH = 2 # parent (0) -> child (1) -> grandchild rejected (2) DEFAULT_MAX_ITERATIONS = 50 DEFAULT_TOOLSETS = ["terminal", "file", "web"] +DEFAULT_CODEX_YOLO_COMMAND = "/home/ges/.local/bin/codex-yolo" +DEFAULT_CODEX_YOLO_ARGS = ["--acp", "--stdio"] def check_delegate_requirements() -> bool: @@ -113,6 +115,59 @@ def _strip_blocked_tools(toolsets: List[str]) -> List[str]: return [t for t in toolsets if t not in blocked_toolset_names] +def _looks_like_coding_task( + goal: Optional[str], + context: Optional[str] = None, + toolsets: Optional[List[str]] = None, +) -> bool: + """Best-effort detection for implementation-oriented coding tasks.""" + if toolsets and not any(t in set(toolsets) for t in ("terminal", "file")): + return False + + haystack = "\n".join([ + str(goal or ""), + str(context or ""), + " ".join(toolsets or []), + ]).lower() + + positive_markers = [ + "implement", "implementation", "build", "create", "write code", + "coding", "fix bug", "bugfix", "debug", "refactor", "patch", + "edit files", "modify files", "run tests", "failing test", + "test failure", "repair", + ] + negative_markers = [ + "research", "analyze", "analysis", "review", "code review", + "plan", "planning", "design doc", "spec", "investigate only", + "read-only", "summarize", + ] + + if any(marker in haystack for marker in negative_markers): + return False + return any(marker in haystack for marker in positive_markers) + + +def _resolve_default_acp_override( + goal: Optional[str], + context: Optional[str], + toolsets: Optional[List[str]], + acp_command: Optional[str], + acp_args: Optional[List[str]], + cfg: Optional[Dict[str, Any]] = None, +) -> tuple[Optional[str], Optional[List[str]]]: + """Prefer codex-yolo for clear coding tasks when no ACP override is provided.""" + if acp_command or acp_args: + return acp_command, acp_args + auto_route_enabled = True if cfg is None else bool(cfg.get("auto_route_coding_to_codex_yolo", True)) + if not auto_route_enabled: + return acp_command, acp_args + if not _looks_like_coding_task(goal, context, toolsets): + return acp_command, acp_args + if not os.path.isfile(DEFAULT_CODEX_YOLO_COMMAND) or not os.access(DEFAULT_CODEX_YOLO_COMMAND, os.X_OK): + return acp_command, acp_args + return DEFAULT_CODEX_YOLO_COMMAND, list(DEFAULT_CODEX_YOLO_ARGS) + + def _build_child_progress_callback(task_index: int, parent_agent, task_count: int = 1) -> Optional[callable]: """Build a callback that relays child agent tool calls to the parent display. @@ -589,15 +644,24 @@ def delegate_task( children = [] try: for i, t in enumerate(task_list): + task_toolsets = t.get("toolsets") or toolsets + default_acp_command, default_acp_args = _resolve_default_acp_override( + goal=t["goal"], + context=t.get("context"), + toolsets=task_toolsets, + acp_command=t.get("acp_command") or acp_command, + acp_args=t.get("acp_args") or acp_args, + cfg=cfg, + ) child = _build_child_agent( task_index=i, goal=t["goal"], context=t.get("context"), - toolsets=t.get("toolsets") or toolsets, model=creds["model"], + toolsets=task_toolsets, model=creds["model"], max_iterations=effective_max_iter, parent_agent=parent_agent, override_provider=creds["provider"], override_base_url=creds["base_url"], override_api_key=creds["api_key"], override_api_mode=creds["api_mode"], - override_acp_command=t.get("acp_command") or acp_command, - override_acp_args=t.get("acp_args") or acp_args, + override_acp_command=default_acp_command, + override_acp_args=default_acp_args, ) # Override with correct parent tool names (before child construction mutated global) child._delegate_saved_tool_names = _parent_tool_names diff --git a/tools/process_registry.py b/tools/process_registry.py index b935f49c337f6..0ce35c05a8853 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -33,6 +33,7 @@ import logging import os import platform +import select import shlex import signal import subprocess @@ -361,18 +362,48 @@ def spawn_via_env( def _reader_loop(self, session: ProcessSession): """Background thread: read stdout from a local Popen process.""" first_chunk = True + + def _append_chunk(chunk: str): + nonlocal first_chunk + if not chunk: + return + if first_chunk: + chunk = self._clean_shell_noise(chunk) + first_chunk = False + with session._lock: + session.output_buffer += chunk + if len(session.output_buffer) > session.max_output_chars: + session.output_buffer = session.output_buffer[-session.max_output_chars:] + try: - while True: - chunk = session.process.stdout.read(4096) - if not chunk: - break - if first_chunk: - chunk = self._clean_shell_noise(chunk) - first_chunk = False - with session._lock: - session.output_buffer += chunk - if len(session.output_buffer) > session.max_output_chars: - session.output_buffer = session.output_buffer[-session.max_output_chars:] + if _IS_WINDOWS: + while True: + chunk = session.process.stdout.readline() + if chunk: + _append_chunk(chunk) + continue + if session.process.poll() is not None: + break + time.sleep(0.05) + else: + fd = session.process.stdout.fileno() + while True: + ready, _, _ = select.select([fd], [], [], 0.1) + if ready: + chunk = os.read(fd, 4096) + if chunk: + _append_chunk(chunk.decode("utf-8", errors="replace")) + continue + if session.process.poll() is not None: + while True: + ready, _, _ = select.select([fd], [], [], 0) + if not ready: + break + chunk = os.read(fd, 4096) + if not chunk: + break + _append_chunk(chunk.decode("utf-8", errors="replace")) + break except Exception as e: logger.debug("Process stdout reader ended: %s", e)