fix: coalesce multiple background process completion notifications by session_key

gateway/run.py

@@ -2899,6 +2899,11 @@ def _format_gateway_process_notification(evt: dict) -> "str | None":
         from tools.process_registry import format_process_notification
         return format_process_notification(evt)
 
+    if evt_type == "completion" or evt_type is None:
+        # Standard process completions — also use the rich formatter.
+        from tools.process_registry import format_process_notification
+        return format_process_notification(evt)
+
     return None
 
 
@@ -13839,13 +13844,7 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
             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._coalesce_and_inject_watch_events(_watch_events)
             except Exception as e:
                 logger.debug("Watch queue drain error: %s", e)
 
@@ -17510,6 +17509,82 @@ def _build_process_event_source(self, evt: dict):
             user_name=str(evt.get("user_name") or "").strip() or None,
         )
 
+    async def _coalesce_and_inject_watch_events(
+        self, watch_events: list[dict],
+    ) -> None:
+        """Inject watch events, coalescing completion notifications by session_key.
+
+        When multiple background processes finish in the same gateway tick,
+        injecting each one individually floods the agent session with redundant
+        notifications.  Instead, group standard completions by session_key and
+        deliver a single batched message.
+        """
+        if not watch_events:
+            return
+
+        # Separate standard completions from watch/wake events.
+        completions: list[dict] = []
+        others: list[dict] = []
+        for evt in watch_events:
+            if evt.get("type") in (None, "completion"):
+                completions.append(evt)
+            else:
+                others.append(evt)
+
+        # Coalesce completions by session_key.
+        by_key: dict[str, list[dict]] = {}
+        for evt in completions:
+            sk = str(evt.get("session_key") or evt.get("session_id") or "")
+            by_key.setdefault(sk, []).append(evt)
+
+        for sk, group in by_key.items():
+            if len(group) == 1:
+                evt = group[0]
+                synth_text = _format_gateway_process_notification(evt)
+                if synth_text:
+                    try:
+                        await self._inject_watch_notification(synth_text, evt)
+                    except Exception as e:
+                        logger.error(
+                            "Watch notification injection error: %s", e,
+                        )
+            else:
+                # Batched completion notification.
+                count = len(group)
+                first = group[0]
+                ids = [str(evt.get("session_id", "?")) for evt in group]
+                if len(ids) <= 5:
+                    id_list = ", ".join(ids)
+                else:
+                    id_list = ", ".join(ids[:5]) + f", ...and {len(ids)-5} more"
+                logger.debug(
+                    "Coalesced %d completion events into 1 notification for session_key=%s",
+                    count, sk,
+                )
+                synth_text = (
+                    f"[IMPORTANT: {count} background processes finished "
+                    f"({id_list}) — results batched to avoid session flood. "
+                    f"Use process(id=N, action='log') to inspect individual outputs.]"
+                )
+                coalesced_evt = dict(first)
+                coalesced_evt["coalesced"] = True
+                coalesced_evt["coalesced_count"] = count
+                try:
+                    await self._inject_watch_notification(synth_text, coalesced_evt)
+                except Exception as e:
+                    logger.error(
+                        "Coalesced watch notification injection error: %s", e,
+                    )
+
+        # Non-completion events (watch_disabled, watch_match, async_delegation).
+        for evt in others:
+            synth_text = _format_gateway_process_notification(evt)
+            if synth_text:
+                try:
+                    await self._inject_watch_notification(synth_text, evt)
+                except Exception as e:
+                    logger.error("Watch notification injection error: %s", e)
+
     async def _inject_watch_notification(
         self, synth_text: str, evt: dict,
     ) -> Optional[bool]:

tests/gateway/test_background_process_notifications.py

@@ -689,4 +689,238 @@ async def fake_self_post(adapter, *, text, session_id):
     }
     result = await runner._inject_watch_notification("[SYSTEM: done]", evt)
     assert result is True
-    assert posts == ["raw-origin-sid"]
+
+
+# ---------------------------------------------------------------------------
+# _coalesce_and_inject_watch_events tests
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_single_completion_event_passes_through_unchanged(monkeypatch, tmp_path):
+    """A lone completion event should be injected as-is, no coalescing needed."""
+    runner = _build_runner(monkeypatch, tmp_path, "all")
+    adapter = runner.adapters[Platform.TELEGRAM]
+    adapter.handle_message = AsyncMock()
+
+    evt = {
+        "type": "completion",
+        "session_id": "proc_abc",
+        "session_key": "agent:main:telegram:dm:123:42",
+        "platform": "telegram",
+        "chat_id": "123",
+        "thread_id": "42",
+    }
+
+    await runner._coalesce_and_inject_watch_events([evt])
+
+    assert adapter.handle_message.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_multiple_completions_same_session_key_are_coalesced(monkeypatch, tmp_path):
+    """Multiple completions for the same session_key produce exactly one injection."""
+    runner = _build_runner(monkeypatch, tmp_path, "all")
+    adapter = runner.adapters[Platform.TELEGRAM]
+    adapter.handle_message = AsyncMock()
+
+    session_key = "agent:main:telegram:dm:123:42"
+    events = [
+        {
+            "type": "completion",
+            "session_id": f"proc_{i}",
+            "session_key": session_key,
+            "platform": "telegram",
+            "chat_id": "123",
+            "thread_id": "42",
+        }
+        for i in range(6)
+    ]
+
+    await runner._coalesce_and_inject_watch_events(events)
+
+    # 6 events → 1 coalesced injection
+    assert adapter.handle_message.await_count == 1
+    injected_text = adapter.handle_message.await_args.args[0].text
+    assert "6 background processes finished" in injected_text
+    assert "batched" in injected_text.lower()
+    # Verify session_ids are listed in the message
+    for i in range(5):
+        assert f"proc_{i}" in injected_text
+    assert "...and 1 more" in injected_text
+
+
+@pytest.mark.asyncio
+async def test_coalesced_message_truncates_ids_after_5(monkeypatch, tmp_path):
+    """When >5 processes, only first 5 IDs are listed, rest shown as count."""
+    runner = _build_runner(monkeypatch, tmp_path, "all")
+    adapter = runner.adapters[Platform.TELEGRAM]
+    adapter.handle_message = AsyncMock()
+
+    session_key = "agent:main:telegram:dm:123:42"
+    events = [
+        {
+            "type": "completion",
+            "session_id": f"proc_{i}",
+            "session_key": session_key,
+            "platform": "telegram",
+            "chat_id": "123",
+            "thread_id": "42",
+        }
+        for i in range(8)
+    ]
+
+    await runner._coalesce_and_inject_watch_events(events)
+
+    assert adapter.handle_message.await_count == 1
+    injected_text = adapter.handle_message.await_args.args[0].text
+    assert "8 background processes finished" in injected_text
+    # First 5 should be listed
+    for i in range(5):
+        assert f"proc_{i}" in injected_text
+    # Beyond 5 should be truncated
+    assert "...and 3 more" in injected_text
+    assert "proc_5" not in injected_text
+
+
+@pytest.mark.asyncio
+async def test_type_none_treated_as_completion(monkeypatch, tmp_path):
+    """Events with type=None should be treated as completions and coalesced."""
+    runner = _build_runner(monkeypatch, tmp_path, "all")
+    adapter = runner.adapters[Platform.TELEGRAM]
+    adapter.handle_message = AsyncMock()
+
+    session_key = "agent:main:telegram:dm:123:42"
+    events = [
+        {
+            "type": None,
+            "session_id": "proc_a",
+            "session_key": session_key,
+            "platform": "telegram",
+            "chat_id": "123",
+            "thread_id": "42",
+        },
+        {
+            "type": "completion",
+            "session_id": "proc_b",
+            "session_key": session_key,
+            "platform": "telegram",
+            "chat_id": "123",
+            "thread_id": "42",
+        },
+    ]
+
+    await runner._coalesce_and_inject_watch_events(events)
+
+    # Both should be coalesced into 1
+    assert adapter.handle_message.await_count == 1
+    injected_text = adapter.handle_message.await_args.args[0].text
+    assert "2 background processes finished" in injected_text
+    assert "proc_a" in injected_text
+    assert "proc_b" in injected_text
+
+
+@pytest.mark.asyncio
+async def test_completions_different_session_keys_not_coalesced(monkeypatch, tmp_path):
+    """Completions for DIFFERENT session_keys remain separate injections."""
+    runner = _build_runner(monkeypatch, tmp_path, "all")
+    adapter = runner.adapters[Platform.TELEGRAM]
+    adapter.handle_message = AsyncMock()
+
+    events = [
+        {
+            "type": "completion",
+            "session_id": f"proc_a{i}",
+            "session_key": f"agent:main:telegram:dm:100:{i}",
+            "platform": "telegram",
+            "chat_id": "100",
+            "thread_id": str(i),
+        }
+        for i in range(3)
+    ]
+
+    await runner._coalesce_and_inject_watch_events(events)
+
+    # Different session_keys → separate injections
+    assert adapter.handle_message.await_count == 3
+
+
+@pytest.mark.asyncio
+async def test_watch_match_events_are_not_coalesced(monkeypatch, tmp_path):
+    """watch_match events pass through individually, not coalesced."""
+    runner = _build_runner(monkeypatch, tmp_path, "all")
+    adapter = runner.adapters[Platform.TELEGRAM]
+    adapter.handle_message = AsyncMock()
+
+    session_key = "agent:main:telegram:dm:123:42"
+    events = [
+        {
+            "type": "watch_match",
+            "session_id": "proc_1",
+            "session_key": session_key,
+            "platform": "telegram",
+            "chat_id": "123",
+            "thread_id": "42",
+            "pattern": "DONE",
+            "output": "Build DONE",
+            "command": "make build",
+        },
+        {
+            "type": "watch_match",
+            "session_id": "proc_2",
+            "session_key": session_key,
+            "platform": "telegram",
+            "chat_id": "123",
+            "thread_id": "42",
+            "pattern": "ERROR",
+            "output": "Build ERROR",
+            "command": "make test",
+        },
+    ]
+
+    await runner._coalesce_and_inject_watch_events(events)
+
+    # watch_match events should NOT be coalesced — each is individually important
+    assert adapter.handle_message.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_mixed_completions_and_watch_events(monkeypatch, tmp_path):
+    """Completions are coalesced; watch events pass through individually."""
+    runner = _build_runner(monkeypatch, tmp_path, "all")
+    adapter = runner.adapters[Platform.TELEGRAM]
+    adapter.handle_message = AsyncMock()
+
+    session_key = "agent:main:telegram:dm:123:42"
+    events = [
+        {"type": "completion", "session_id": "proc_a", "session_key": session_key,
+         "platform": "telegram", "chat_id": "123", "thread_id": "42"},
+        {"type": "completion", "session_id": "proc_b", "session_key": session_key,
+         "platform": "telegram", "chat_id": "123", "thread_id": "42"},
+        {"type": "completion", "session_id": "proc_c", "session_key": session_key,
+         "platform": "telegram", "chat_id": "123", "thread_id": "42"},
+        {"type": "watch_match", "session_id": "proc_w", "session_key": session_key,
+         "platform": "telegram", "chat_id": "123", "thread_id": "42",
+         "pattern": "DONE", "output": "ok", "command": "cmd"},
+    ]
+
+    await runner._coalesce_and_inject_watch_events(events)
+
+    # 3 completions coalesced into 1 + 1 watch_match = 2 injections
+    assert adapter.handle_message.await_count == 2
+    # Verify coalesced text mentions the count
+    texts = [c.args[0].text for c in adapter.handle_message.await_args_list]
+    coalesced = [t for t in texts if "background processes finished" in t]
+    assert len(coalesced) == 1
+    assert "3 background processes finished" in coalesced[0]
+
+
+@pytest.mark.asyncio
+async def test_empty_events_list_is_noop(monkeypatch, tmp_path):
+    """Empty event list should not cause any injection."""
+    runner = _build_runner(monkeypatch, tmp_path, "all")
+    adapter = runner.adapters[Platform.TELEGRAM]
+    adapter.handle_message = AsyncMock()
+
+    await runner._coalesce_and_inject_watch_events([])
+
+    adapter.handle_message.assert_not_awaited()