fix(gateway): don't re-deliver consumed background completions as raw watcher messages

process(wait) marks a completion consumed and returns the exit code +
output inline. The gateway process watcher's agent-notify branch honored
that (skipping the synthetic agent turn), but its skip FELL THROUGH to
the plain text-notification branch, which re-sent the same completion to
the chat as a raw '[Background process ... finished with exit code ...]'
message — a duplicate delivery of output the agent had already read and
was summarizing (observed on Slack with
display.background_process_notifications: all, but platform-agnostic).

Guard the raw-notification branch on is_completion_consumed(), same as
the agent-notify branch. poll() stays read-only and never marks consumed
(#10156), so status checks still can't suppress autonomous delivery.

Fixes #65379. Reported-by: hergert

gateway/run.py

@@ -17828,6 +17828,23 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     break
 
                 # --- Normal text-only notification ---
+                # Skip when the agent already consumed this completion via
+                # wait/log (#65379): process(wait) returned the exit code and
+                # output inline, so the raw "[Background process ... finished
+                # with exit code ...]" message would be a duplicate delivery
+                # of the same completion. The agent_notify branch above
+                # already honors _completion_consumed; without this check its
+                # skip FALLS THROUGH to this block and re-delivers the output
+                # the agent is actively summarizing. poll() is read-only and
+                # intentionally does not mark consumed (#10156), so a status
+                # check never suppresses this message.
+                if _pr_check.is_completion_consumed(session_id):
+                    logger.debug(
+                        "Process watcher: completion for %s already consumed "
+                        "via wait/log — skipping raw notification (#65379)",
+                        session_id,
+                    )
+                    break
                 # Decide whether to notify based on mode
                 should_notify = (
                     notify_mode in {"all", "result"}

tests/gateway/test_background_process_notifications.py

@@ -24,8 +24,9 @@ from gateway.run import GatewayRunner, _parse_session_key
 class _FakeRegistry:
     """Return pre-canned sessions, then None once exhausted."""
 
-    def __init__(self, sessions):
+    def __init__(self, sessions, consumed=False):
         self._sessions = list(sessions)
+        self._consumed = consumed
 
     def get(self, session_id):
         if self._sessions:
@@ -33,7 +34,7 @@ class _FakeRegistry:
         return None
 
     def is_completion_consumed(self, session_id):
-        return False
+        return self._consumed
 
 
 def _build_runner(monkeypatch, tmp_path, mode: str) -> GatewayRunner:
@@ -248,6 +249,70 @@ async def test_no_thread_id_sends_no_metadata(monkeypatch, tmp_path):
     assert kwargs["metadata"] is None
 
 
+@pytest.mark.asyncio
+async def test_consumed_completion_skips_raw_notification(monkeypatch, tmp_path):
+    """#65379: after process(wait) already returned the completion inline,
+    the gateway watcher must NOT also push the raw
+    "[Background process ... finished with exit code ...]" message.
+
+    The agent-notify branch already honored _completion_consumed, but its
+    skip fell through to the text-notification branch, double-delivering the
+    same output to the chat (observed on Slack with
+    background_process_notifications: all)."""
+    import tools.process_registry as pr_module
+
+    sessions = [SimpleNamespace(
+        output_buffer="done\n", exited=True, exit_code=0, command="sleep 1; echo done",
+    )]
+    monkeypatch.setattr(
+        pr_module, "process_registry", _FakeRegistry(sessions, consumed=True)
+    )
+
+    async def _instant_sleep(*_a, **_kw):
+        pass
+    monkeypatch.setattr(asyncio, "sleep", _instant_sleep)
+
+    runner = _build_runner(monkeypatch, tmp_path, "all")
+    adapter = runner.adapters[Platform.TELEGRAM]
+
+    # notify_on_complete=True mirrors the reported scenario: the watcher's
+    # agent-notify skip must not fall through to a raw adapter.send().
+    watcher = _watcher_dict()
+    watcher["notify_on_complete"] = True
+    await runner._run_process_watcher(watcher)
+
+    adapter.send.assert_not_awaited()
+    adapter.handle_message.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_consumed_completion_skips_raw_notification_without_agent_notify(
+    monkeypatch, tmp_path
+):
+    """#65379 variant: same double-delivery guard for plain watchers
+    (notify_on_complete=False) — wait/log consumption suppresses the raw
+    completion message in every mode."""
+    import tools.process_registry as pr_module
+
+    sessions = [SimpleNamespace(
+        output_buffer="done\n", exited=True, exit_code=0, command="echo done",
+    )]
+    monkeypatch.setattr(
+        pr_module, "process_registry", _FakeRegistry(sessions, consumed=True)
+    )
+
+    async def _instant_sleep(*_a, **_kw):
+        pass
+    monkeypatch.setattr(asyncio, "sleep", _instant_sleep)
+
+    runner = _build_runner(monkeypatch, tmp_path, "all")
+    adapter = runner.adapters[Platform.TELEGRAM]
+
+    await runner._run_process_watcher(_watcher_dict())
+
+    adapter.send.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_inject_watch_notification_routes_from_session_store_origin(monkeypatch, tmp_path):
     from gateway.session import SessionSource