fix(gateway): release loop ticks after empty responses

This commit is contained in:
fangliquanflq
2026-08-15 20:58:59 +08:00
committed by Teknium
parent a15de34545
commit 2f6bbfbcbc
2 changed files with 69 additions and 38 deletions
+42 -37
View File
@@ -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(
+27 -1
View File
@@ -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()