diff --git a/cli.py b/cli.py index 011045092e..0969a060ba 100644 --- a/cli.py +++ b/cli.py @@ -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 diff --git a/gateway/run.py b/gateway/run.py index e154ef3647..df69a498c1 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -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) diff --git a/tests/tools/test_watch_patterns.py b/tests/tools/test_watch_patterns.py index 510d5d55b1..e31844f9f6 100644 --- a/tests/tools/test_watch_patterns.py +++ b/tests/tools/test_watch_patterns.py @@ -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 diff --git a/tools/process_registry.py b/tools/process_registry.py index 6ba3340a35..1be9b89f66 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -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,