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