fix(gateway): honor notification-off watch reinjection

Signed-off-by: Lidang-Jiang <lidangjiang@gmail.com>

gateway/run.py

@@ -19299,14 +19299,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             # single consumer — so we leave them on the queue here.
             try:
                 from tools.process_registry import process_registry as _pr
-                _watch_events = _drain_gateway_watch_events(_pr.completion_queue)
-                for evt in _watch_events:
-                    synth_text = _format_gateway_process_notification(evt)
-                    if synth_text:
-                        try:
-                            await self._inject_watch_notification(synth_text, evt)
-                        except Exception as e2:
-                            logger.error("Watch notification injection error: %s", e2)
+                await self._drain_watch_notifications(_pr.completion_queue)
             except Exception as e:
                 logger.debug("Watch queue drain error: %s", e)
 
@@ -23558,6 +23551,26 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             scope_id=scope_id,
         )
 
+    async def _drain_watch_notifications(self, completion_queue) -> None:
+        """Consume queued watch events and inject them when notifications are enabled.
+
+        The queue is ALWAYS drained (so watch events don't rot or requeue-spin)
+        but injection is skipped entirely when
+        ``display.background_process_notifications`` is ``off`` (#9290).
+        """
+        watch_events = _drain_gateway_watch_events(completion_queue)
+        if self._load_background_notifications_mode() == "off":
+            return
+
+        for evt in watch_events:
+            synth_text = _format_gateway_process_notification(evt)
+            if not synth_text:
+                continue
+            try:
+                await self._inject_watch_notification(synth_text, evt)
+            except Exception as exc:
+                logger.error("Watch notification injection error: %s", exc)
+
     async def _inject_watch_notification(
         self, synth_text: str, evt: dict,
     ) -> Optional[bool]:

tests/gateway/test_background_process_notifications.py

@@ -8,6 +8,7 @@ Contributed by @PeterFile (PR #593), reimplemented on current main.
 """
 
 import asyncio
+import queue
 from types import SimpleNamespace
 from unittest.mock import AsyncMock
 
@@ -66,6 +67,17 @@ def _watcher_dict(session_id="proc_test", thread_id=""):
     return d
 
 
+def _watch_event(session_id="proc_watch", thread_id="42"):
+    return {
+        "type": "watch_match",
+        "session_id": session_id,
+        "session_key": f"agent:main:telegram:dm:123:{thread_id}",
+        "pattern": "READY",
+        "command": "build",
+        "output": "READY\n",
+    }
+
+
 # ---------------------------------------------------------------------------
 # _load_background_notifications_mode unit tests
 # ---------------------------------------------------------------------------
@@ -201,6 +213,54 @@ async def test_inject_watch_notification_routes_from_session_store_origin(monkey
     assert synth_event.source.user_name == "Emiliyan"
 
 
+@pytest.mark.asyncio
+async def test_post_turn_watch_drain_off_consumes_without_injecting(monkeypatch, tmp_path):
+    runner = _build_runner(monkeypatch, tmp_path, "off")
+    adapter = runner.adapters[Platform.TELEGRAM]
+    completion_queue = queue.Queue()
+    completion_queue.put(_watch_event("proc_one"))
+    completion_queue.put(_watch_event("proc_two"))
+    async_event = {"type": "async_delegation", "session_id": "delegate_one"}
+    completion_queue.put(async_event)
+
+    await runner._drain_watch_notifications(completion_queue)
+
+    adapter.handle_message.assert_not_awaited()
+    assert completion_queue.qsize() == 1
+    assert completion_queue.get_nowait() is async_event
+
+
+@pytest.mark.asyncio
+async def test_post_turn_watch_drain_all_injects_from_queued_event_origin(monkeypatch, tmp_path):
+    from gateway.session import SessionSource
+
+    runner = _build_runner(monkeypatch, tmp_path, "all")
+    adapter = runner.adapters[Platform.TELEGRAM]
+    runner.session_store._entries["agent:main:telegram:dm:123:42"] = SimpleNamespace(
+        origin=SessionSource(
+            platform=Platform.TELEGRAM,
+            chat_id="123",
+            chat_type="dm",
+            thread_id="42",
+            user_id="proc_owner",
+            user_name="alice",
+        )
+    )
+    completion_queue = queue.Queue()
+    completion_queue.put(_watch_event())
+    async_event = {"type": "async_delegation", "session_id": "delegate_one"}
+    completion_queue.put(async_event)
+
+    await runner._drain_watch_notifications(completion_queue)
+
+    adapter.handle_message.assert_awaited_once()
+    synth_event = adapter.handle_message.await_args.args[0]
+    assert synth_event.source.thread_id == "42"
+    assert synth_event.source.user_id == "proc_owner"
+    assert completion_queue.qsize() == 1
+    assert completion_queue.get_nowait() is async_event
+
+
 @pytest.mark.asyncio
 async def test_inject_watch_notification_carries_message_id_reply_anchor(monkeypatch, tmp_path):
     from gateway.session import SessionSource