fix(nudge): wake gate requires a real NudgeQueue

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 7886d3b763)

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

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