diff --git a/gateway/run.py b/gateway/run.py index 6c187fe9cb..a4fcde2d49 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -17510,6 +17510,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew agent_result=_agent_result, source=source, is_internal=is_internal, + event=event, ) except Exception as _goal_exc: logger.debug("post-turn hook failed: %s", _goal_exc) @@ -20278,6 +20279,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew ) except Exception as _e: logger.debug("trailing footer send failed: %s", _e) + # This branch returns None so the adapter does not send the + # body twice. /loop and /goal hooks in _handle_message read + # the return value, so stash the delivered text on the event + # or those hooks never run and a /loop tick stays awaiting. + try: + event._streamed_final_response = str(response or "") + except Exception: + pass return None return response @@ -21098,13 +21107,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew agent_result: Any, source: Any, is_internal: bool, + event: Any = None, ) -> None: """Run goal and loop bookkeeping after an agent turn returns.""" - final_text = "" - if isinstance(agent_result, dict): - final_text = str(agent_result.get("final_response") or "") - elif isinstance(agent_result, str): - final_text = agent_result + final_text = self._final_text_for_post_turn_hooks(agent_result, event) try: session_entry = await self.async_session_store.get_or_create_session( @@ -21135,7 +21141,25 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew except Exception as exc: logger.debug("loop completion hook failed: %s", exc) + @staticmethod + def _final_text_for_post_turn_hooks(agent_result, event=None) -> str: + """Text for /goal and /loop after a gateway turn. + Streamed turns return None from _handle_message_with_agent + (already_sent). The delivered reply is stashed on the event so + those hooks still see it. + """ + text = "" + if isinstance(agent_result, dict): + text = str(agent_result.get("final_response") or "") + elif isinstance(agent_result, str): + text = agent_result + if text.strip(): + return text + streamed = getattr(event, "_streamed_final_response", None) + if isinstance(streamed, str) and streamed.strip(): + return streamed + return text async def _post_turn_loop_completion( self, diff --git a/tests/gateway/test_loop_command.py b/tests/gateway/test_loop_command.py index 1720086f35..73d63317f7 100644 --- a/tests/gateway/test_loop_command.py +++ b/tests/gateway/test_loop_command.py @@ -141,6 +141,51 @@ async def test_post_turn_loop_completion_noop_without_inflight_tick(loop_env): assert reloaded.ticks_fired == 0 +def test_streamed_already_sent_none_recovers_text_for_hooks(): + """Streamed turns return None. Hooks must still see the delivered reply.""" + event = _make_event("wakeup") + event._streamed_final_response = "CI is green.\nLOOP_COMPLETE" + assert GatewayRunner._final_text_for_post_turn_hooks(None, event) == ( + "CI is green.\nLOOP_COMPLETE" + ) + assert GatewayRunner._final_text_for_post_turn_hooks(None, _make_event("x")) == "" + assert ( + GatewayRunner._final_text_for_post_turn_hooks( + {"final_response": "from dict"}, event + ) + == "from dict" + ) + + +@pytest.mark.asyncio +async def test_streamed_already_sent_completes_loop_tick(loop_env): + """A streamed wakeup must not leave awaiting_response stuck.""" + runner = _make_runner() + await GatewayRunner._handle_loop_command(runner, _make_event("/loop 5m poll CI")) + + mgr = loops.LoopManager(session_id="sid-gateway-loop") + mgr.state.next_due_at = time.time() - 1 + assert mgr.fire_tick() is not None + assert mgr.state.awaiting_response is True + assert mgr.is_due() is False + + event = _make_event("wakeup") + event._streamed_final_response = "CI is done.\nLOOP_COMPLETE" + # Same inputs the already_sent branch leaves for _handle_message. + final_text = GatewayRunner._final_text_for_post_turn_hooks(None, event) + assert final_text.strip() + + await GatewayRunner._post_turn_loop_completion( + runner, + session_entry=_FakeSessionEntry(), + source=None, + final_response=final_text, + ) + reloaded = loops.load_loop("sid-gateway-loop") + assert reloaded.awaiting_response is False + assert reloaded.status == "done" + + @pytest.mark.asyncio async def test_empty_agent_result_releases_inflight_loop_tick(loop_env): runner = _make_runner()