fix(gateway): guard empty task_id reaps and prefer the finished worker's result
Two follow-ups from review of the salvaged fix: - _reap_gateway_turn_processes now returns 0 for a blank task_id. ProcessSession.task_id defaults to empty for sessionless callers, so a blank turn id would match (and kill) every unrelated empty-task process instead of the turn's own. - The asyncio poll loop checks executor completion BEFORE the watchdog's timeout flag. When both race in the same window, the completed run has already persisted its real reply to session history; surfacing the 'agent inactive' diagnostic would contradict the stored transcript. This matches _abandon_timed_out_gateway_turn's own worker-done-wins tiebreak.
This commit is contained in:
+15
-3
@@ -2709,6 +2709,11 @@ def _reap_gateway_turn_processes(
|
||||
process. The newer turn snapshots its own baseline independently, so
|
||||
skipping here does not leave anything permanently unreaped.
|
||||
"""
|
||||
if not task_id:
|
||||
# ProcessSession.task_id defaults to "" for sessionless callers, so a
|
||||
# blank id would match (and kill) every unrelated empty-task process
|
||||
# instead of this turn's own. Nothing session-scoped to reap.
|
||||
return 0
|
||||
if is_still_current is not None:
|
||||
try:
|
||||
if not is_still_current():
|
||||
@@ -24402,12 +24407,19 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
done, _ = await asyncio.wait(
|
||||
{_executor_task}, timeout=_POLL_INTERVAL
|
||||
)
|
||||
if done:
|
||||
# Prefer the real result when the worker finished,
|
||||
# even if the watchdog fired in the same window: the
|
||||
# completed run already persisted its reply to session
|
||||
# history, so surfacing the "agent inactive" diagnostic
|
||||
# here would contradict the stored transcript. This
|
||||
# mirrors _abandon_timed_out_gateway_turn's own
|
||||
# worker_done-wins tiebreak (under cleanup_lock).
|
||||
response = _executor_task.result()
|
||||
break
|
||||
if _turn_timeout_fired.is_set():
|
||||
_inactivity_timeout = True
|
||||
break
|
||||
if done:
|
||||
response = _executor_task.result()
|
||||
break
|
||||
# Agent still running — check inactivity.
|
||||
_agent_ref = agent_holder[0]
|
||||
_idle_secs = 0.0
|
||||
|
||||
@@ -188,6 +188,26 @@ def test_reap_fails_open_when_is_still_current_raises(monkeypatch):
|
||||
assert calls == [True]
|
||||
|
||||
|
||||
def test_reap_skips_empty_task_id(monkeypatch):
|
||||
"""ProcessSession.task_id defaults to "" — a blank turn id must never
|
||||
fan out into killing unrelated sessionless processes (#76188 review)."""
|
||||
calls = []
|
||||
monkeypatch.setattr(
|
||||
process_registry,
|
||||
"kill_started_since",
|
||||
lambda *_a, **_k: calls.append(True) or 1,
|
||||
)
|
||||
|
||||
killed = _reap_gateway_turn_processes(
|
||||
"",
|
||||
frozenset(),
|
||||
source="gateway_turn_timeout",
|
||||
)
|
||||
|
||||
assert killed == 0
|
||||
assert calls == []
|
||||
|
||||
|
||||
def test_timeout_abandon_propagates_is_still_current_to_the_reap(monkeypatch):
|
||||
agent = _IdleAgent()
|
||||
worker_done, timeout_fired, cleanup_lock = _state()
|
||||
|
||||
Reference in New Issue
Block a user