diff --git a/gateway/run.py b/gateway/run.py index e8334fdcbceae..3f1718e077022 100644 --- a/gateway/run.py +++ b/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: diff --git a/tests/gateway/test_completion_delivery.py b/tests/gateway/test_completion_delivery.py index 3c3e09967f415..4aba541b8611e 100644 --- a/tests/gateway/test_completion_delivery.py +++ b/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()