fix(gateway): coalesce concurrent process completions

gateway/run.py

@@ -6573,6 +6573,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         self._completion_deliveries_inflight: set[tuple[str, str, object]] = set()
         self._completion_deliveries_delivered: "OrderedDict[tuple[str, str, object], None]" = OrderedDict()
         self._completion_delivery_retention = 2048
+        # Agent-triggered terminal completions from one conversation often land
+        # in the same scheduler tick.  Hold them briefly so the agent receives
+        # one synthetic turn instead of one turn per process (#70300).
+        self._completion_notification_batches: dict[tuple[str, ...], list[tuple[str, dict, asyncio.Future]]] = {}
+        self._completion_notification_batch_tasks: dict[tuple[str, ...], asyncio.Task] = {}
+        self._completion_notification_batch_window = 0.1
 
         # Cache AIAgent instances per session to preserve prompt caching.
         # Without this, a new AIAgent is created per message, rebuilding the
@@ -23972,6 +23978,135 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 except Exception:
                     logger.debug("Could not release durable completion claim", exc_info=True)
 
+    @staticmethod
+    def _completion_notification_batch_key(evt: dict) -> tuple[str, ...]:
+        """Return a routing-complete key for short-window process fan-in."""
+        return tuple(str(evt.get(field) or "") for field in (
+            "session_key",
+            "platform",
+            "chat_type",
+            "chat_id",
+            "thread_id",
+            "user_id",
+        ))
+
+    @staticmethod
+    def _format_coalesced_process_completions(entries: list[tuple[str, dict, asyncio.Future]]) -> str:
+        """Build one bounded synthetic event from several redacted completions."""
+        lines = [
+            f"[IMPORTANT: {len(entries)} background processes completed for this session.",
+            "Treat these results as one completion batch and send at most one "
+            "consolidated user-facing response.",
+        ]
+        shown = entries[:10]
+        for _text, evt, _future in shown:
+            session_id = str(evt.get("session_id") or "unknown")
+            exit_code = evt.get("exit_code")
+            reason = str(evt.get("completion_reason") or "exited")
+            output = str(evt.get("output") or "").strip()
+            if len(output) > 800:
+                output = f"[… truncated …]\n{output[-800:]}"
+            lines.append(
+                f"\n- {session_id}: exit_code={exit_code}, reason={reason}"
+            )
+            if output:
+                lines.append(output)
+        omitted = len(entries) - len(shown)
+        if omitted:
+            lines.append(
+                f"\n- … and {omitted} more completion(s); inspect them with "
+                "the process tool if they affect the conclusion."
+            )
+        lines.append(
+            "If a result does not change the current conclusion, absorb it silently.]"
+        )
+        return "\n".join(lines)
+
+    def _record_coalesced_completion_siblings(self, events: list[dict]) -> None:
+        """Extend a successful primary delivery claim to its batched siblings."""
+        with self._completion_delivery_lock:
+            for evt in events:
+                identity = self._completion_delivery_identity(evt)
+                if identity is None:
+                    continue
+                self._completion_deliveries_inflight.discard(identity)
+                self._completion_deliveries_delivered[identity] = None
+            while (
+                len(self._completion_deliveries_delivered)
+                > self._completion_delivery_retention
+            ):
+                self._completion_deliveries_delivered.popitem(last=False)
+
+    async def _flush_process_completion_batch(self, key: tuple[str, ...]) -> None:
+        """Deliver one short-window completion batch and resolve its waiters."""
+        current_task = asyncio.current_task()
+        entries: list[tuple[str, dict, asyncio.Future]] = []
+        delivered: Optional[bool] = False
+        try:
+            await asyncio.sleep(self._completion_notification_batch_window)
+            entries = self._completion_notification_batches.pop(key, [])
+            # Detach before adapter delivery.  A completion that arrives while
+            # this batch is in flight must be able to schedule the next flush.
+            if self._completion_notification_batch_tasks.get(key) is current_task:
+                self._completion_notification_batch_tasks.pop(key, None)
+            if not entries:
+                return
+            try:
+                if len(entries) == 1:
+                    synth_text = entries[0][0]
+                else:
+                    synth_text = self._format_coalesced_process_completions(entries)
+
+                # A duplicate primary can legitimately return None from the
+                # lifecycle dedupe seam.  Try the next batch identity so a
+                # fresh sibling is never discarded with that duplicate.
+                delivered = None
+                for _text, candidate_evt, _future in entries:
+                    delivered = await self._deliver_completion_notification(
+                        synth_text, candidate_evt,
+                    )
+                    if delivered is not None:
+                        break
+                if delivered is True and len(entries) > 1:
+                    self._record_coalesced_completion_siblings(
+                        [evt for _text, evt, _future in entries]
+                    )
+            except Exception:
+                logger.exception("Coalesced process completion delivery failed")
+                delivered = False
+            finally:
+                # Never strand watcher futures if formatting or delivery fails.
+                # False follows the existing watcher retry path.
+                for _text, _evt, future in entries:
+                    if not future.done():
+                        future.set_result(delivered)
+        finally:
+            # Do not remove a newer flush task that reused the same route key.
+            if self._completion_notification_batch_tasks.get(key) is current_task:
+                self._completion_notification_batch_tasks.pop(key, None)
+
+    async def _enqueue_process_completion_notification(
+        self, synth_text: str, evt: dict,
+    ) -> Optional[bool]:
+        """Fan in concurrent process completions that share one conversation."""
+        # Some unit tests construct GatewayRunner with object.__new__.  Keep the
+        # batching seam lazy so those focused lifecycle tests remain valid.
+        if not hasattr(self, "_completion_notification_batches"):
+            self._completion_notification_batches = {}
+            self._completion_notification_batch_tasks = {}
+            self._completion_notification_batch_window = 0.1
+
+        key = self._completion_notification_batch_key(evt)
+        future = asyncio.get_running_loop().create_future()
+        self._completion_notification_batches.setdefault(key, []).append(
+            (synth_text, evt, future)
+        )
+        if key not in self._completion_notification_batch_tasks:
+            self._completion_notification_batch_tasks[key] = asyncio.create_task(
+                self._flush_process_completion_batch(key)
+            )
+        return await future
+
     def _enrich_async_delegation_routing(self, evt: dict) -> None:
         """Fill platform/chat_id/thread_id/chat_type on an async-delegation event.
 
@@ -24145,7 +24280,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     synth_text = format_process_notification(completion_evt)
                     if not synth_text:
                         break
-                    delivered = await self._deliver_completion_notification(
+                    delivered = await self._enqueue_process_completion_notification(
                         synth_text, completion_evt,
                     )
                     if delivered is False:

tests/gateway/test_completion_delivery.py

@@ -316,3 +316,198 @@ def test_autonomous_completion_redacts_real_command_and_output_secrets(monkeypat
     delivered = adapter.handle_message.await_args.args[0]
     assert secret not in delivered.text
     assert "HOME=/home/user" in delivered.text
+
+
+def test_concurrent_process_watchers_coalesce_one_session_completion_turn(monkeypatch):
+    """Concurrent terminal watchers for one session must re-enter the agent once."""
+    import tools.process_registry as pr_module
+
+    registry = ProcessRegistry()
+    watchers = []
+    for index in range(3):
+        session = ProcessSession(
+            id=f"proc_batch_{index}",
+            command=f"printf batch-{index}",
+            task_id=f"task-{index}",
+            started_at=1000.0 + index,
+            output_buffer=f"batch-{index}\n",
+            exited=True,
+            exit_code=0,
+            notify_on_complete=True,
+        )
+        registry._finished[session.id] = session
+        watchers.append({
+            "session_id": session.id,
+            "check_interval": 0,
+            "session_key": "agent:main:telegram:dm:123",
+            "platform": "telegram",
+            "chat_type": "dm",
+            "chat_id": "123",
+            "notify_on_complete": True,
+        })
+    monkeypatch.setattr(pr_module, "process_registry", registry)
+
+    adapter = SimpleNamespace(handle_message=AsyncMock())
+    runner = _runner(adapter)
+
+    async def _exercise():
+        await asyncio.gather(*(
+            runner._run_process_watcher(watcher)
+            for watcher in watchers
+        ))
+
+    asyncio.run(_exercise())
+
+    adapter.handle_message.assert_awaited_once()
+    delivered = adapter.handle_message.await_args.args[0]
+    assert "3 background processes completed" in delivered.text
+    for index in range(3):
+        assert f"proc_batch_{index}" in delivered.text
+
+
+def test_completion_arriving_during_batch_delivery_schedules_next_flush():
+    """A new event cannot be stranded behind an in-flight batch for its route."""
+    first_delivery_entered = asyncio.Event()
+    release_first_delivery = asyncio.Event()
+    delivery_count = 0
+
+    async def _deliver(_event):
+        nonlocal delivery_count
+        delivery_count += 1
+        if delivery_count == 1:
+            first_delivery_entered.set()
+            await release_first_delivery.wait()
+
+    adapter = SimpleNamespace(handle_message=AsyncMock(side_effect=_deliver))
+    runner = _runner(adapter)
+
+    async def _exercise():
+        first = asyncio.create_task(runner._enqueue_process_completion_notification(
+            "first completion",
+            _completion_event(started_at=1.0, session_id="proc_first"),
+        ))
+        await first_delivery_entered.wait()
+        second = asyncio.create_task(runner._enqueue_process_completion_notification(
+            "second completion",
+            _completion_event(started_at=2.0, session_id="proc_second"),
+        ))
+        release_first_delivery.set()
+        assert await first is True
+        assert await asyncio.wait_for(second, timeout=1.0) is True
+
+    asyncio.run(_exercise())
+
+    assert adapter.handle_message.await_count == 2
+
+
+def test_completion_batches_do_not_cross_conversation_routes():
+    adapter = SimpleNamespace(handle_message=AsyncMock())
+    runner = _runner(adapter)
+
+    first = _completion_event(started_at=1.0, session_id="proc_route_a")
+    second = _completion_event(started_at=2.0, session_id="proc_route_b")
+    second["session_key"] = "agent:main:telegram:dm:456"
+    second["chat_id"] = "456"
+
+    async def _exercise():
+        return await asyncio.gather(
+            runner._enqueue_process_completion_notification("first", first),
+            runner._enqueue_process_completion_notification("second", second),
+        )
+
+    assert asyncio.run(_exercise()) == [True, True]
+    assert adapter.handle_message.await_count == 2
+
+
+def test_failed_coalesced_delivery_retries_all_entries():
+    attempts = 0
+
+    async def _deliver(_event):
+        nonlocal attempts
+        attempts += 1
+        if attempts == 1:
+            raise RuntimeError("temporary adapter failure")
+
+    adapter = SimpleNamespace(handle_message=AsyncMock(side_effect=_deliver))
+    runner = _runner(adapter)
+    events = [
+        _completion_event(started_at=float(index), session_id=f"proc_retry_{index}")
+        for index in range(2)
+    ]
+
+    async def _enqueue_all():
+        return await asyncio.gather(*(
+            runner._enqueue_process_completion_notification(f"event-{index}", event)
+            for index, event in enumerate(events)
+        ))
+
+    async def _exercise():
+        assert await _enqueue_all() == [False, False]
+        assert await _enqueue_all() == [True, True]
+
+    asyncio.run(_exercise())
+    assert adapter.handle_message.await_count == 2
+
+
+def test_coalesced_success_records_every_completion_identity():
+    adapter = SimpleNamespace(handle_message=AsyncMock())
+    runner = _runner(adapter)
+    events = [
+        _completion_event(started_at=float(index), session_id=f"proc_ledger_{index}")
+        for index in range(3)
+    ]
+
+    async def _exercise():
+        return await asyncio.gather(*(
+            runner._enqueue_process_completion_notification(f"event-{index}", event)
+            for index, event in enumerate(events)
+        ))
+
+    assert asyncio.run(_exercise()) == [True, True, True]
+    for event in events:
+        identity = runner._completion_delivery_identity(event)
+        assert identity in runner._completion_deliveries_delivered
+
+
+def test_duplicate_primary_does_not_discard_fresh_batch_sibling():
+    adapter = SimpleNamespace(handle_message=AsyncMock())
+    runner = _runner(adapter)
+    duplicate = _completion_event(started_at=1.0, session_id="proc_duplicate")
+    fresh = _completion_event(started_at=2.0, session_id="proc_fresh")
+    duplicate_identity = runner._completion_delivery_identity(duplicate)
+    runner._completion_deliveries_delivered[duplicate_identity] = None
+
+    async def _exercise():
+        return await asyncio.gather(
+            runner._enqueue_process_completion_notification("duplicate", duplicate),
+            runner._enqueue_process_completion_notification("fresh", fresh),
+        )
+
+    assert asyncio.run(_exercise()) == [True, True]
+    adapter.handle_message.assert_awaited_once()
+    fresh_identity = runner._completion_delivery_identity(fresh)
+    assert fresh_identity in runner._completion_deliveries_delivered
+
+
+def test_batch_format_failure_resolves_waiters_for_retry(monkeypatch):
+    adapter = SimpleNamespace(handle_message=AsyncMock())
+    runner = _runner(adapter)
+    monkeypatch.setattr(
+        runner,
+        "_format_coalesced_process_completions",
+        MagicMock(side_effect=ValueError("bad batch")),
+    )
+    events = [
+        _completion_event(started_at=float(index), session_id=f"proc_format_{index}")
+        for index in range(2)
+    ]
+
+    async def _exercise():
+        pending = asyncio.gather(*(
+            runner._enqueue_process_completion_notification(f"event-{index}", event)
+            for index, event in enumerate(events)
+        ))
+        return await asyncio.wait_for(pending, timeout=1.0)
+
+    assert asyncio.run(_exercise()) == [False, False]
+    adapter.handle_message.assert_not_awaited()