fix(tools,gateway): format watch_overflow events instead of dropping them

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

@@ -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)

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

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

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