fix(gateway): complete /loop ticks after streamed already_sent turns
Streamed replies return None so the adapter does not send twice. The /loop hook then saw empty text and never ran, so awaiting_response stayed true and later ticks never fired. Stash the delivered text on the event and use it for the post-turn hooks. /goal uses the same path. Tests: tests/gateway/test_loop_command.py
This commit is contained in:
+29
-5
@@ -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,
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user