From a6752cb64582bd6ea69cc92e0ebffffe8cd3e084 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Tue, 7 Jul 2026 16:02:34 -0700 Subject: [PATCH] fix(nudge): wake gate requires a real NudgeQueue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A session whose _nudge_queue answers has_pending truthily while its deliver_wake_nudge_from_queue consumes nothing turns the worker-exit backstop into an infinite respawn loop: the gate passes, the wake worker no-ops, the exit backstop re-runs the gate, forever. Mock-backed test sessions riding real Workstreams are exactly that shape, and one worker on such a pairing is enough to ignite a wake-thread storm that trips the leaked-thread guard in every subsequent test. The wake contract requires real drain semantics — the spawned worker must CONSUME what the gate saw — so the gate now refuses on type, not just presence. (cherry picked from commit 7886d3b7633d94d94d087af18a98510ba0d0e239) --- tests/test_idle_nudge_watcher.py | 17 +++++++++++++++++ turnstone/core/idle_nudge_watcher.py | 15 ++++++++++----- 2 files changed, 27 insertions(+), 5 deletions(-) diff --git a/tests/test_idle_nudge_watcher.py b/tests/test_idle_nudge_watcher.py index fcde8285..8744107a 100644 --- a/tests/test_idle_nudge_watcher.py +++ b/tests/test_idle_nudge_watcher.py @@ -231,6 +231,23 @@ class TestWakeWorkstreamIfPending: assert wake_workstream_if_pending(ws) is False assert mock_send.call_count == 0 + def test_refuses_non_nudgequeue_stub(self, fake_mgr_and_ws): + """The gate refuses on TYPE, not just presence: a mock session's + auto-created ``_nudge_queue`` answers ``has_pending`` truthily + while its ``deliver_wake_nudge_from_queue`` consumes nothing — + with the worker-exit backstop re-running this gate after every + exit, one worker on such a session would respawn wake workers + forever (the storm that took down the full-suite CI run). Only + a real :class:`NudgeQueue` carries the drain semantics the wake + contract needs.""" + from unittest.mock import MagicMock + + _mgr, ws = fake_mgr_and_ws + ws.session._nudge_queue = MagicMock() # truthy has_pending, no real drain + with patch("turnstone.core.session_worker.send") as mock_send: + assert wake_workstream_if_pending(ws) is False + assert mock_send.call_count == 0 + def test_dispatched_path_logs_trigger(self, fake_mgr_and_ws, caplog): """A fresh spawn — ``send`` returns True without touching the passed ``enqueue`` — emits ``nudge_wake.dispatched`` tagged with diff --git a/turnstone/core/idle_nudge_watcher.py b/turnstone/core/idle_nudge_watcher.py index 47a78a91..e92ac325 100644 --- a/turnstone/core/idle_nudge_watcher.py +++ b/turnstone/core/idle_nudge_watcher.py @@ -26,7 +26,7 @@ from typing import TYPE_CHECKING, Any from turnstone.core import session_worker from turnstone.core.log import get_logger -from turnstone.core.nudge_queue import USER_DRAIN +from turnstone.core.nudge_queue import USER_DRAIN, NudgeQueue from turnstone.core.workstream import WorkstreamState if TYPE_CHECKING: @@ -56,9 +56,14 @@ def wake_workstream_if_pending(ws: Workstream, *, trigger: str = "unspecified") Gates, in order: * ``ws.session is None`` — workstream tracked but session not - built — or a bare stub session without a NudgeQueue (watch-style - dispatchers drive sessions that aren't installed on the - workstream). + built — or a session whose ``_nudge_queue`` is not a real + :class:`NudgeQueue` (bare stubs; mock sessions). The wake + contract REQUIRES real drain semantics: the spawned worker's + ``deliver_wake_nudge_from_queue`` must actually CONSUME what + ``has_pending`` saw, or the worker-exit backstop respawns wake + workers forever — a mock queue's truthy ``has_pending`` plus a + no-op deliver is exactly that storm, so the gate refuses on + TYPE, not just presence. * ``ws._closed`` — ``close()`` already ran (or is racing us); its storage row says ``closed`` and a wake send would drive a torn-down session. Lockless FAST-PATH only: a stale ``False`` @@ -92,7 +97,7 @@ def wake_workstream_if_pending(ws: Workstream, *, trigger: str = "unspecified") if session is None or ws._closed or ws.state is not WorkstreamState.IDLE: return False nudge_queue = getattr(session, "_nudge_queue", None) - if nudge_queue is None or not nudge_queue.has_pending(USER_DRAIN): + if not isinstance(nudge_queue, NudgeQueue) or not nudge_queue.has_pending(USER_DRAIN): return False deferred = False