fix(nudge): log refused wakes; correct the already-dispatched hold-clear comment

The wake gate documented exactly one info line per call past its
gates, but a send() refusal (the authoritative under-lock _closed
re-check catching a teardown the gate's lockless peek missed) emitted
nothing — a dropped wake should stay traceable to its trigger, so the
refusal now logs nudge_wake.refused.

The already-dispatched branch's comment claimed a held reminder can
coexist with the terminal mark via a redelivery whose commit raised —
impossible with the current control flow (_redeliver_pending clears
the hold before committing).  Reworded to what the clear actually is:
the last line of defense against any coexisting hold leaking forever
once this branch deactivates the row, since inactive rows never
re-list.  Test comment updated to match.

(cherry picked from commit cb59afe443)

tests/test_idle_nudge_watcher.py

@@ -296,3 +296,23 @@ class TestWakeWorkstreamIfPending:
         assert deferred[0].levelno == logging.INFO
         assert "trigger=" in deferred[0].getMessage()
         assert not any("nudge_wake.dispatched" in r.getMessage() for r in caplog.records)
+
+    def test_refused_path_logs_refusal(self, fake_mgr_and_ws, caplog):
+        """``send`` refusing outright — its authoritative under-lock
+        ``_closed`` re-check caught a teardown the gate's lockless peek
+        missed — emits ``nudge_wake.refused``: a dropped wake must stay
+        traceable to its trigger, not vanish silently."""
+        _mgr, ws = fake_mgr_and_ws
+        ws.session._nudge_queue.enqueue("watch_triggered", "output", "any")
+        with (
+            patch("turnstone.core.session_worker.send", return_value=False) as mock_send,
+            caplog.at_level(logging.INFO, logger="turnstone.core.idle_nudge_watcher"),
+        ):
+            assert wake_workstream_if_pending(ws, trigger="watch-fire") is False
+            assert mock_send.call_count == 1
+        refused = [r for r in caplog.records if "nudge_wake.refused" in r.getMessage()]
+        assert len(refused) == 1
+        assert refused[0].levelno == logging.INFO
+        assert "trigger=" in refused[0].getMessage()
+        assert not any("nudge_wake.dispatched" in r.getMessage() for r in caplog.records)
+        assert not any("nudge_wake.deferred_worker_busy" in r.getMessage() for r in caplog.records)

tests/test_watch.py

@@ -775,10 +775,12 @@ class TestWatchRunnerDeliveryRetry:
             assert "ws-1" not in runner._restoring
 
     def test_pending_cleared_on_already_dispatched_retry(self):
-        # A re-delivery that succeeded but whose row-commit raised leaves
-        # the id in BOTH _terminal_dispatched and _pending_delivery.  The
-        # next tick's already-dispatched branch must clear the hold too, or
-        # it leaks once the row deactivates.
+        # Constructs the id-in-both-sets state DIRECTLY: no current path
+        # produces it (_redeliver_pending clears the hold before its
+        # commit), but the already-dispatched branch deactivates the row —
+        # after which it never re-lists — so it is the last line of
+        # defense against any such hold leaking forever.  Pin that it
+        # clears the hold alongside the retry-deactivate.
         storage = MagicMock()
         storage.update_watch.return_value = True
         runner = self._make_runner(storage)

turnstone/core/idle_nudge_watcher.py

@@ -87,6 +87,11 @@ def wake_workstream_if_pending(ws: Workstream, *, trigger: str = "unspecified")
       queued; the owning worker's exit backstop (or its next drain
       seam) delivers it.
     * ``nudge_wake.dispatched`` — a fresh wake daemon was spawned.
+    * ``nudge_wake.refused`` — ``session_worker.send`` declined the
+      spawn: its authoritative under-lock ``_closed`` re-check caught a
+      teardown this gate's lockless peek missed.  The entry dies with
+      the workstream; logged here so a dropped wake is traceable to its
+      trigger during production troubleshooting.
 
     Returns ``True`` iff the wake was handed to
     ``session_worker.send`` — which may still downgrade it to a no-op
@@ -116,6 +121,8 @@ def wake_workstream_if_pending(ws: Workstream, *, trigger: str = "unspecified")
         log.info("nudge_wake.deferred_worker_busy ws=%s trigger=%s", ws.id[:8], trigger)
     elif ok:
         log.info("nudge_wake.dispatched ws=%s trigger=%s", ws.id[:8], trigger)
+    else:
+        log.info("nudge_wake.refused ws=%s trigger=%s", ws.id[:8], trigger)
     return ok
 
 

turnstone/core/watch.py

@@ -653,11 +653,14 @@ class WatchRunner:
                 self._storage.update_watch(watch_id, active=False, next_poll="")
                 with self._terminal_dispatched_lock:
                     self._terminal_dispatched.discard(watch_id)
-                # A held reminder for an already-dispatched watch means a
-                # re-delivery SUCCEEDED but its row-commit raised (leaving
-                # the id in both sets).  The reminder is delivered, so drop
-                # the hold here — otherwise, once we deactivate the row it
-                # never re-lists and the entry would leak forever.
+                # Belt-and-braces: no current path leaves an id in both
+                # ``_terminal_dispatched`` AND ``_pending_delivery``
+                # (``_redeliver_pending`` clears the hold BEFORE its
+                # commit), but this branch deactivates the row — after
+                # which it never re-lists — so any hold that ever DID
+                # coexist with the terminal mark would leak forever
+                # without this clear.  The mark means the reminder was
+                # delivered; a coexisting hold is by definition stale.
                 self._clear_pending_delivery(watch_id)
             except Exception:
                 log.exception("watch_runner.retry_deactivate_failed", extra={"watch_id": watch_id})