From d2d77667503c255e4f2f36efb02f83ab927bb0fe Mon Sep 17 00:00:00 2001 From: spfcraze Date: Fri, 31 Jul 2026 10:43:50 -0400 Subject: [PATCH] fix(tools,gateway): format watch_overflow events instead of dropping them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit format_process_notification had no case for watch_overflow_tripped / watch_overflow_released, so a watch-pattern notification flood surfaced as '[IMPORTANT: Background process exited (exit code ?)]' — a phantom exit notification for a process that never existed — while the actual 'watch flood, N notifications suppressed' summary in the event's message field was silently dropped. The gateway delivery path was worse: _drain_gateway_watch_events retained only watch_match and watch_disabled, discarding overflow events entirely before formatting. Route both event types through the message field in the shared formatter and the gateway formatter, and retain them in the gateway drain. --- gateway/run.py | 13 ++++++- .../test_background_process_notifications.py | 35 +++++++++++++++++++ tests/tools/test_watch_patterns.py | 29 +++++++++++++++ tools/process_registry.py | 6 ++++ 4 files changed, 82 insertions(+), 1 deletion(-) diff --git a/gateway/run.py b/gateway/run.py index 0327a4ca136d0..47d476d891c1d 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -3540,6 +3540,12 @@ def _format_gateway_process_notification(evt: dict) -> "str | None": if evt_type == "watch_disabled": return f"[IMPORTANT: {evt.get('message', '')}]" + # Overflow events carry their human-readable summary in `message`, + # like watch_disabled — see the shared formatter in + # tools/process_registry.py. + if evt_type in ("watch_overflow_tripped", "watch_overflow_released"): + return f"[IMPORTANT: {evt.get('message', '')}]" + if evt_type == "watch_match": _pat = evt.get("pattern", "?") _out = evt.get("output", "") @@ -3581,7 +3587,12 @@ def _drain_gateway_watch_events(completion_queue) -> "list[dict]": except Exception: break evt_type = evt.get("type", "completion") - if evt_type in {"watch_match", "watch_disabled"}: + if evt_type in { + "watch_match", + "watch_disabled", + "watch_overflow_tripped", + "watch_overflow_released", + }: watch_events.append(evt) elif evt_type == "async_delegation": requeue.append(evt) diff --git a/tests/gateway/test_background_process_notifications.py b/tests/gateway/test_background_process_notifications.py index 4d32e63d3c40f..3dfff7a20603a 100644 --- a/tests/gateway/test_background_process_notifications.py +++ b/tests/gateway/test_background_process_notifications.py @@ -537,3 +537,38 @@ async def test_inject_watch_notification_origin_session_id_wins(monkeypatch, tmp result = await runner._inject_watch_notification("[SYSTEM: done]", evt) assert result is True assert posts == ["raw-origin-sid"] + + +def test_gateway_drain_retains_and_formats_overflow_events(): + """watch_overflow_* events must survive the gateway drain and render + their summary — previously they were discarded at the drain (only + watch_match/watch_disabled were retained) and had no formatter branch.""" + import asyncio + from gateway.run import ( + _drain_gateway_watch_events, + _format_gateway_process_notification, + ) + + queue = asyncio.Queue() + tripped = { + "type": "watch_overflow_tripped", + "message": "watch flood detected: 47 notifications suppressed for pattern 'ERROR'", + "session_id": "proc_a1b2", + } + released = { + "type": "watch_overflow_released", + "message": "watch flood released: notifications resumed for pattern 'ERROR'", + "session_id": "proc_a1b2", + } + queue.put_nowait(tripped) + queue.put_nowait(released) + + retained = _drain_gateway_watch_events(queue) + assert retained == [tripped, released] + + out_tripped = _format_gateway_process_notification(tripped) + assert "47 notifications suppressed" in out_tripped + assert "exit code" not in out_tripped + out_released = _format_gateway_process_notification(released) + assert "notifications resumed" in out_released + assert "exit code" not in out_released diff --git a/tests/tools/test_watch_patterns.py b/tests/tools/test_watch_patterns.py index 5fe2467cb1709..760e3ede6aa97 100644 --- a/tests/tools/test_watch_patterns.py +++ b/tests/tools/test_watch_patterns.py @@ -336,3 +336,32 @@ class TestGlobalCircuitBreaker: admitted = True assert released assert admitted + + +class TestOverflowNotificationFormatting: + """watch_overflow_* events must surface their summary, not fall through + to the completion formatter as a phantom 'process exited (exit code ?)'.""" + + def test_overflow_tripped_formats_message(self): + from tools.process_registry import format_process_notification + + evt = { + "type": "watch_overflow_tripped", + "message": "watch flood detected: 47 notifications suppressed for pattern 'ERROR'", + "session_id": "proc_a1b2", + } + out = format_process_notification(evt) + assert "47 notifications suppressed" in out + assert "exit code" not in out + + def test_overflow_released_formats_message(self): + from tools.process_registry import format_process_notification + + evt = { + "type": "watch_overflow_released", + "message": "watch flood released: notifications resumed for pattern 'ERROR'", + "session_id": "proc_a1b2", + } + out = format_process_notification(evt) + assert "notifications resumed" in out + assert "exit code" not in out diff --git a/tools/process_registry.py b/tools/process_registry.py index 6d61b3ab690d3..0883fa1c225ea 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -2812,6 +2812,12 @@ def format_process_notification(evt: dict) -> "str | None": if evt_type == "watch_disabled": return f"[IMPORTANT: {evt.get('message', '')}]" + # Overflow events carry their human-readable summary in `message` — + # without this case they fall through to the completion formatter and + # surface as a phantom "process exited (exit code ?)" notification. + if evt_type in ("watch_overflow_tripped", "watch_overflow_released"): + return f"[IMPORTANT: {evt.get('message', '')}]" + if evt_type == "watch_match": _pat = evt.get("pattern", "?") _out = evt.get("output", "")