refactor: merge watch_queue into completion_queue

Unified queue with 'type' field distinguishing 'completion', 'watch_match', and 'watch_disabled' events. Extracted _format_process_notification() in CLI and gateway to handle all event types in a single drain loop. Removes duplication across both CLI drain sites and the gateway.
2026-04-11 01:56:21 -07:00
parent 346b947fc5
commit bf2b74cef0
4 changed files with 117 additions and 121 deletions
@@ -1171,6 +1171,45 @@ def _resolve_attachment_path(raw_path: str) -> Path | None:
    return resolved


+def _format_process_notification(evt: dict) -> "str | None":
+    """Format a process notification event into a [SYSTEM: ...] message.
+
+    Handles both completion events (notify_on_complete) and watch pattern
+    match events from the unified completion_queue.
+    """
+    evt_type = evt.get("type", "completion")
+    _sid = evt.get("session_id", "unknown")
+    _cmd = evt.get("command", "unknown")
+
+    if evt_type == "watch_disabled":
+        return f"[SYSTEM: {evt.get('message', '')}]"
+
+    if evt_type == "watch_match":
+        _pat = evt.get("pattern", "?")
+        _out = evt.get("output", "")
+        _sup = evt.get("suppressed", 0)
+        text = (
+            f"[SYSTEM: Background process {_sid} matched "
+            f"watch pattern \"{_pat}\".\n"
+            f"Command: {_cmd}\n"
+            f"Matched output:\n{_out}"
+        )
+        if _sup:
+            text += f"\n({_sup} earlier matches were suppressed by rate limit)"
+        text += "]"
+        return text
+
+    # Default: completion event
+    _exit = evt.get("exit_code", "?")
+    _out = evt.get("output", "")
+    return (
+        f"[SYSTEM: Background process {_sid} completed "
+        f"(exit code {_exit}).\n"
+        f"Command: {_cmd}\n"
+        f"Output:\n{_out}]"
+    )
+
+
 def _detect_file_drop(user_input: str) -> "dict | None":
    """Detect if *user_input* starts with a real local file path.

@@ -8870,45 +8909,15 @@ class HermesCLI:
                        # Periodic config watcher — auto-reload MCP on mcp_servers change
                        if not self._agent_running:
                            self._check_config_mcp_changes()
-                            # Check for background process completion notifications
-                            # while the agent is idle (user hasn't typed anything yet).
+                            # Check for background process notifications (completions
+                            # and watch pattern matches) while agent is idle.
                            try:
                                from tools.process_registry import process_registry
                                if not process_registry.completion_queue.empty():
-                                    completion = process_registry.completion_queue.get_nowait()
-                                    _exit = completion.get("exit_code", "?")
-                                    _cmd = completion.get("command", "unknown")
-                                    _sid = completion.get("session_id", "unknown")
-                                    _out = completion.get("output", "")
-                                    _synth = (
-                                        f"[SYSTEM: Background process {_sid} completed "
-                                        f"(exit code {_exit}).\n"
-                                        f"Command: {_cmd}\n"
-                                        f"Output:\n{_out}]"
-                                    )
-                                    self._pending_input.put(_synth)
-                                # Check for watch pattern match notifications
-                                if not process_registry.watch_queue.empty():
-                                    watch_evt = process_registry.watch_queue.get_nowait()
-                                    _wsid = watch_evt.get("session_id", "unknown")
-                                    _wcmd = watch_evt.get("command", "unknown")
-                                    _wtype = watch_evt.get("type", "watch_match")
-                                    if _wtype == "watch_disabled":
-                                        _wsynth = f"[SYSTEM: {watch_evt.get('message', '')}]"
-                                    else:
-                                        _wpat = watch_evt.get("pattern", "?")
-                                        _wout = watch_evt.get("output", "")
-                                        _wsup = watch_evt.get("suppressed", 0)
-                                        _wsynth = (
-                                            f"[SYSTEM: Background process {_wsid} matched "
-                                            f"watch pattern \"{_wpat}\".\n"
-                                            f"Command: {_wcmd}\n"
-                                            f"Matched output:\n{_wout}"
-                                        )
-                                        if _wsup:
-                                            _wsynth += f"\n({_wsup} earlier matches were suppressed by rate limit)"
-                                        _wsynth += "]"
-                                    self._pending_input.put(_wsynth)
+                                    evt = process_registry.completion_queue.get_nowait()
+                                    _synth = _format_process_notification(evt)
+                                    if _synth:
+                                        self._pending_input.put(_synth)
                            except Exception:
                                pass
                        continue
@@ -9026,45 +9035,15 @@ class HermesCLI:
                                    _cprint(f"{_DIM}Voice auto-restart failed: {e}{_RST}")
                            threading.Thread(target=_restart_recording, daemon=True).start()

-                        # Drain process completion and watch notifications — any
-                        # background process events that arrived while the agent
-                        # was running get auto-injected as a new user message.
+                        # Drain process notifications (completions + watch matches)
+                        # that arrived while the agent was running.
                        try:
                            from tools.process_registry import process_registry
                            while not process_registry.completion_queue.empty():
-                                completion = process_registry.completion_queue.get_nowait()
-                                _exit = completion.get("exit_code", "?")
-                                _cmd = completion.get("command", "unknown")
-                                _sid = completion.get("session_id", "unknown")
-                                _out = completion.get("output", "")
-                                _synth = (
-                                    f"[SYSTEM: Background process {_sid} completed "
-                                    f"(exit code {_exit}).\n"
-                                    f"Command: {_cmd}\n"
-                                    f"Output:\n{_out}]"
-                                )
-                                self._pending_input.put(_synth)
-                            while not process_registry.watch_queue.empty():
-                                watch_evt = process_registry.watch_queue.get_nowait()
-                                _wsid = watch_evt.get("session_id", "unknown")
-                                _wcmd = watch_evt.get("command", "unknown")
-                                _wtype = watch_evt.get("type", "watch_match")
-                                if _wtype == "watch_disabled":
-                                    _wsynth = f"[SYSTEM: {watch_evt.get('message', '')}]"
-                                else:
-                                    _wpat = watch_evt.get("pattern", "?")
-                                    _wout = watch_evt.get("output", "")
-                                    _wsup = watch_evt.get("suppressed", 0)
-                                    _wsynth = (
-                                        f"[SYSTEM: Background process {_wsid} matched "
-                                        f"watch pattern \"{_wpat}\".\n"
-                                        f"Command: {_wcmd}\n"
-                                        f"Matched output:\n{_wout}"
-                                    )
-                                    if _wsup:
-                                        _wsynth += f"\n({_wsup} earlier matches were suppressed by rate limit)"
-                                    _wsynth += "]"
-                                self._pending_input.put(_wsynth)
+                                evt = process_registry.completion_queue.get_nowait()
+                                _synth = _format_process_notification(evt)
+                                if _synth:
+                                    self._pending_input.put(_synth)
                        except Exception:
                            pass  # Non-fatal — don't break the main loop

@@ -476,6 +476,33 @@ def _resolve_hermes_bin() -> Optional[list[str]]:
    return None


+def _format_gateway_process_notification(evt: dict) -> "str | None":
+    """Format a watch pattern event from completion_queue into a [SYSTEM:] message."""
+    evt_type = evt.get("type", "completion")
+    _sid = evt.get("session_id", "unknown")
+    _cmd = evt.get("command", "unknown")
+
+    if evt_type == "watch_disabled":
+        return f"[SYSTEM: {evt.get('message', '')}]"
+
+    if evt_type == "watch_match":
+        _pat = evt.get("pattern", "?")
+        _out = evt.get("output", "")
+        _sup = evt.get("suppressed", 0)
+        text = (
+            f"[SYSTEM: Background process {_sid} matched "
+            f"watch pattern \"{_pat}\".\n"
+            f"Command: {_cmd}\n"
+            f"Matched output:\n{_out}"
+        )
+        if _sup:
+            text += f"\n({_sup} earlier matches were suppressed by rate limit)"
+        text += "]"
+        return text
+
+    return None
+
+
 class GatewayRunner:
    """
    Main gateway controller.
@@ -3430,34 +3457,26 @@ class GatewayRunner:
            except Exception as e:
                logger.error("Process watcher setup error: %s", e)

-            # Drain watch pattern notifications that arrived during the agent run
+            # Drain watch pattern notifications that arrived during the agent run.
+            # Watch events and completions share the same queue; completions are
+            # already handled by the per-process watcher task above, so we only
+            # inject watch-type events here.
            try:
                from tools.process_registry import process_registry as _pr
-                while not _pr.watch_queue.empty():
-                    watch_evt = _pr.watch_queue.get_nowait()
-                    _wsid = watch_evt.get("session_id", "unknown")
-                    _wcmd = watch_evt.get("command", "unknown")
-                    _wtype = watch_evt.get("type", "watch_match")
-                    if _wtype == "watch_disabled":
-                        synth_text = f"[SYSTEM: {watch_evt.get('message', '')}]"
-                    else:
-                        _wpat = watch_evt.get("pattern", "?")
-                        _wout = watch_evt.get("output", "")
-                        _wsup = watch_evt.get("suppressed", 0)
-                        synth_text = (
-                            f"[SYSTEM: Background process {_wsid} matched "
-                            f"watch pattern \"{_wpat}\".\n"
-                            f"Command: {_wcmd}\n"
-                            f"Matched output:\n{_wout}"
-                        )
-                        if _wsup:
-                            synth_text += f"\n({_wsup} earlier matches were suppressed by rate limit)"
-                        synth_text += "]"
-                    # Inject as synthetic message to trigger a new agent turn
-                    try:
-                        await self._inject_watch_notification(synth_text, event)
-                    except Exception as e2:
-                        logger.error("Watch notification injection error: %s", e2)
+                _watch_events = []
+                while not _pr.completion_queue.empty():
+                    evt = _pr.completion_queue.get_nowait()
+                    evt_type = evt.get("type", "completion")
+                    if evt_type in ("watch_match", "watch_disabled"):
+                        _watch_events.append(evt)
+                    # else: completion events are handled by the watcher task
+                for evt in _watch_events:
+                    synth_text = _format_gateway_process_notification(evt)
+                    if synth_text:
+                        try:
+                            await self._inject_watch_notification(synth_text, event)
+                        except Exception as e2:
+                            logger.error("Watch notification injection error: %s", e2)
            except Exception as e:
                logger.debug("Watch queue drain error: %s", e)

@@ -73,20 +73,20 @@ class TestCheckWatchPatterns:
        """No watch_patterns → no notifications."""
        session = _make_session(watch_patterns=[])
        registry._check_watch_patterns(session, "ERROR: something broke\n")
-        assert registry.watch_queue.empty()
+        assert registry.completion_queue.empty()

    def test_no_match_no_notification(self, registry):
        """Output that doesn't match any pattern → no notification."""
        session = _make_session(watch_patterns=["ERROR", "FAIL"])
        registry._check_watch_patterns(session, "INFO: all good\nDEBUG: fine\n")
-        assert registry.watch_queue.empty()
+        assert registry.completion_queue.empty()

    def test_basic_match(self, registry):
        """Single matching line triggers a notification."""
        session = _make_session(watch_patterns=["ERROR"])
        registry._check_watch_patterns(session, "INFO: ok\nERROR: disk full\n")
-        assert not registry.watch_queue.empty()
-        evt = registry.watch_queue.get_nowait()
+        assert not registry.completion_queue.empty()
+        evt = registry.completion_queue.get_nowait()
        assert evt["type"] == "watch_match"
        assert evt["pattern"] == "ERROR"
        assert "disk full" in evt["output"]
@@ -96,7 +96,7 @@ class TestCheckWatchPatterns:
        """First matching pattern is reported."""
        session = _make_session(watch_patterns=["WARN", "ERROR"])
        registry._check_watch_patterns(session, "ERROR: bad\nWARN: hmm\n")
-        evt = registry.watch_queue.get_nowait()
+        evt = registry.completion_queue.get_nowait()
        # ERROR appears first in the output, and we check patterns in order
        # so "WARN" won't match "ERROR: bad" but "ERROR" will
        assert evt["pattern"] == "ERROR"
@@ -107,7 +107,7 @@ class TestCheckWatchPatterns:
        session = _make_session(watch_patterns=["ERROR"])
        session._watch_disabled = True
        registry._check_watch_patterns(session, "ERROR: boom\n")
-        assert registry.watch_queue.empty()
+        assert registry.completion_queue.empty()

    def test_hit_counter_increments(self, registry):
        """Each delivered notification increments _watch_hits."""
@@ -123,7 +123,7 @@ class TestCheckWatchPatterns:
        # Generate 30 matching lines (more than the 20-line cap)
        text = "\n".join(f"X line {i}" for i in range(30)) + "\n"
        registry._check_watch_patterns(session, text)
-        evt = registry.watch_queue.get_nowait()
+        evt = registry.completion_queue.get_nowait()
        # Should only have 20 lines max
        assert evt["output"].count("\n") <= 20

@@ -138,7 +138,7 @@ class TestRateLimiting:
        session = _make_session(watch_patterns=["E"])
        for i in range(WATCH_MAX_PER_WINDOW):
            registry._check_watch_patterns(session, f"E {i}\n")
-        assert registry.watch_queue.qsize() == WATCH_MAX_PER_WINDOW
+        assert registry.completion_queue.qsize() == WATCH_MAX_PER_WINDOW

    def test_exceeds_window_limit(self, registry):
        """Notifications beyond the rate limit are suppressed."""
@@ -146,7 +146,7 @@ class TestRateLimiting:
        for i in range(WATCH_MAX_PER_WINDOW + 5):
            registry._check_watch_patterns(session, f"E {i}\n")
        # Only WATCH_MAX_PER_WINDOW should be in the queue
-        assert registry.watch_queue.qsize() == WATCH_MAX_PER_WINDOW
+        assert registry.completion_queue.qsize() == WATCH_MAX_PER_WINDOW
        assert session._watch_suppressed == 5

    def test_window_resets(self, registry):
@@ -163,7 +163,7 @@ class TestRateLimiting:
        session._watch_window_start = time.time() - WATCH_WINDOW_SECONDS - 1
        registry._check_watch_patterns(session, "E after reset\n")
        # Should deliver now (window reset)
-        assert registry.watch_queue.qsize() == WATCH_MAX_PER_WINDOW + 1
+        assert registry.completion_queue.qsize() == WATCH_MAX_PER_WINDOW + 1

    def test_suppressed_count_in_next_delivery(self, registry):
        """Suppressed count is reported in the next successful delivery."""
@@ -180,8 +180,8 @@ class TestRateLimiting:
        registry._check_watch_patterns(session, "E back\n")
        # Drain to the last event
        last_evt = None
-        while not registry.watch_queue.empty():
-            last_evt = registry.watch_queue.get_nowait()
+        while not registry.completion_queue.empty():
+            last_evt = registry.completion_queue.get_nowait()
        assert last_evt["suppressed"] == 3
        assert session._watch_suppressed == 0  # reset after delivery

@@ -207,8 +207,8 @@ class TestOverloadKillSwitch:
        assert session._watch_disabled is True
        # Should have a watch_disabled event in the queue
        disabled_evts = []
-        while not registry.watch_queue.empty():
-            evt = registry.watch_queue.get_nowait()
+        while not registry.completion_queue.empty():
+            evt = registry.completion_queue.get_nowait()
            if evt.get("type") == "watch_disabled":
                disabled_evts.append(evt)
        assert len(disabled_evts) == 1
@@ -127,16 +127,13 @@ class ProcessRegistry:
        # Side-channel for check_interval watchers (gateway reads after agent run)
        self.pending_watchers: List[Dict[str, Any]] = []

-        # Completion notifications — processes with notify_on_complete push here
-        # on exit.  CLI process_loop and gateway drain this after each agent turn
-        # to auto-trigger a new agent turn with the process results.
+        # Notification queue — unified queue for all background process events.
+        # Completion notifications (notify_on_complete) and watch pattern matches
+        # both land here, distinguished by "type" field.  CLI process_loop and
+        # gateway drain this after each agent turn to auto-trigger new turns.
        import queue as _queue_mod
        self.completion_queue: _queue_mod.Queue = _queue_mod.Queue()

-        # Watch pattern notifications — processes with watch_patterns push here
-        # when output matches a pattern.  Same consumption model as completion_queue.
-        self.watch_queue: _queue_mod.Queue = _queue_mod.Queue()
-
    @staticmethod
    def _clean_shell_noise(text: str) -> str:
        """Strip shell startup warnings from the beginning of output."""
@@ -186,7 +183,7 @@ class ProcessRegistry:
                    session._watch_overload_since = now
                elif now - session._watch_overload_since > WATCH_OVERLOAD_KILL_SECONDS:
                    session._watch_disabled = True
-                    self.watch_queue.put({
+                    self.completion_queue.put({
                        "session_id": session.id,
                        "command": session.command,
                        "type": "watch_disabled",
@@ -214,7 +211,7 @@ class ProcessRegistry:
        if len(output) > 2000:
            output = output[:2000] + "\n...(truncated)"

-        self.watch_queue.put({
+        self.completion_queue.put({
            "session_id": session.id,
            "command": session.command,
            "type": "watch_match",
@@ -605,6 +602,7 @@ class ProcessRegistry:
            from tools.ansi_strip import strip_ansi
            output_tail = strip_ansi(session.output_buffer[-2000:]) if session.output_buffer else ""
            self.completion_queue.put({
+                "type": "completion",
                "session_id": session.id,
                "command": session.command,
                "exit_code": session.exit_code,