fix(process): preserve completion notification after poll/log/wait on long background tasks

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

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)

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"):

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

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)