From 2f6bbfbcbc87a755b0dae9c1ccb0887b37fe8b73 Mon Sep 17 00:00:00 2001 From: fangliquanflq Date: Sat, 15 Aug 2026 20:58:59 +0800 Subject: [PATCH] fix(gateway): release loop ticks after empty responses --- gateway/run.py | 79 ++++++++++++++++-------------- tests/gateway/test_loop_command.py | 28 ++++++++++- 2 files changed, 69 insertions(+), 38 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index e59af63550..dd16dce8ff 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -17450,45 +17450,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "protect the transcript, this message was not processed. " "Wait for the active turn to finish, then resend it." ) - # Goal continuation: after the agent returns a final response - # for this turn, check any standing /goal — the judge will - # either mark it done, pause it (budget), or enqueue a - # continuation prompt back through the adapter FIFO so the - # next turn makes more progress. Wrapped in try/except so a - # broken judge never breaks normal message handling. try: - _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 - # Skip for empty responses (interrupted / errored) — the - # judge would almost always say "continue" and we'd loop - # on error. Let the user drive the next turn. - if _final_text.strip(): - try: - session_entry = await self.async_session_store.get_or_create_session( - source, - touch_activity=not is_internal, - ) - except Exception: - session_entry = None - if session_entry is not None: - await self._post_turn_goal_continuation( - session_entry=session_entry, - source=source, - final_response=_final_text, - ) - # /loop tick completion: if this turn was a loop - # wakeup, evaluate it (LOOP_COMPLETE marker, --until - # judge, caps) and schedule the next tick. - await self._post_turn_loop_completion( - session_entry=session_entry, - source=source, - final_response=_final_text, - ) + await self._run_post_turn_hooks( + agent_result=_agent_result, + source=source, + is_internal=is_internal, + ) except Exception as _goal_exc: - logger.debug("goal continuation hook failed: %s", _goal_exc) + logger.debug("post-turn hook failed: %s", _goal_exc) return _agent_result finally: # MoA one-shot restore must run on EVERY exit path, not just @@ -21060,6 +21029,42 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew except Exception as exc: logger.debug("goal continuation: enqueue failed: %s", exc) + async def _run_post_turn_hooks( + self, + *, + agent_result: Any, + source: Any, + is_internal: bool, + ) -> 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 + + try: + session_entry = await self.async_session_store.get_or_create_session( + source, + touch_activity=not is_internal, + ) + except Exception: + return + + # Empty interrupted/errored responses must not drive /goal, but an + # in-flight /loop tick still needs to be released and rescheduled. + if final_text.strip(): + await self._post_turn_goal_continuation( + session_entry=session_entry, + source=source, + final_response=final_text, + ) + await self._post_turn_loop_completion( + session_entry=session_entry, + source=source, + final_response=final_text, + ) + async def _post_turn_loop_completion( diff --git a/tests/gateway/test_loop_command.py b/tests/gateway/test_loop_command.py index c11876a995..4a8cfdbf9d 100644 --- a/tests/gateway/test_loop_command.py +++ b/tests/gateway/test_loop_command.py @@ -1,6 +1,7 @@ """Gateway /loop command tests — dispatch, routing capture, mid-run guard.""" import time +from unittest.mock import AsyncMock import pytest @@ -19,7 +20,7 @@ class _FakeSessionStore: def __init__(self): self.entry = _FakeSessionEntry() - def get_or_create_session(self, source): + def get_or_create_session(self, source, *, touch_activity=True): return self.entry def _generate_session_key(self, source): @@ -137,3 +138,28 @@ async def test_post_turn_loop_completion_noop_without_inflight_tick(loop_env): reloaded = loops.load_loop("sid-gateway-loop") assert reloaded.status == "active" assert reloaded.ticks_fired == 0 + + +@pytest.mark.asyncio +async def test_empty_agent_result_releases_inflight_loop_tick(loop_env): + 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 + + runner._post_turn_goal_continuation = AsyncMock() + await GatewayRunner._run_post_turn_hooks( + runner, + agent_result={"final_response": ""}, + source=_make_event("wakeup").source, + is_internal=True, + ) + + runner._post_turn_goal_continuation.assert_not_awaited() + reloaded = loops.load_loop("sid-gateway-loop") + assert reloaded.awaiting_response is False + assert reloaded.status == "active" + assert reloaded.next_due_at > time.time()