fix(watch): harden nudge/wake delivery across eviction, cancel, and identity rebinds

Wake path:
- Denial metacog nudge moves to the tool channel so it drains with the
  denied tool batch instead of the next user-message seam.
- wake_workstream_if_pending: shared wake gate for watch fires on
  already-idle workstreams (no IDLE transition for the watcher to
  observe), wired as wake_fn at every set_watch_runner site via the
  shared _watch_fire_wake_fn helper (closes over the Workstream OBJECT
  — after eviction+restore an id-keyed manager lookup would miss).
- session_worker exit backstop re-runs the wake gate the moment worker
  ownership clears: IDLE fans out on the worker thread, so
  transition-time wakes always landed on the reuse path and no-op'd
  (the coordinator idle_children strand).
- deliver_wake_nudge_from_queue contains GenerationCancelled — it is
  the wake worker's run() closure and only Exception is caught
  downstream.

Watch delivery:
- Terminal fires that cannot reach their workstream are HELD and
  redelivered on min(interval, 60s) without re-running the command,
  bounded by MAX_DELIVERY_ATTEMPTS per cycle and the watch's own
  max_polls across cycles; the poll charge commits durably at hold
  time so restarts stay budget-bounded.
- Restore admission control: per-ws dedup + MAX_CONCURRENT_RESTORES
  cap, presence-only re-check under the lock, detection-only stall
  alerts (reclaiming a wedged admission would trade capped degradation
  for total poll-pool collapse).
- Permanent-vs-transient restore taxonomy: corrupt persona stamp and
  genuinely-missing history (confirmed by a raising storage probe —
  the resume loader swallows read blips into []) deactivate the watch
  immediately; everything else holds and retries.
- Cancel-race defense: delivery paths re-check is_watch_active before
  stashing/dispatching, cancel paths write the row BEFORE
  forget_terminal_dispatched, the HTTP cancel endpoint clears runner
  state, and a per-tick sweep bounds the residual stash-after-clear
  interleaving to one check_interval.
- Abandon/exhaustion commits are write-then-clear so storage that can
  read but not write retries the row write instead of re-running the
  command every cycle; the fresh-fire unrestorable path stashes before
  its deactivation write for the same reason.

Registry follows identity:
- The dispatch registry is keyed by _ws_id at registration time; every
  rebind now moves it: non-fork resume() and /new go through
  _follow_watch_registration (new key live before the old is removed,
  never stealing a registration another live session holds), removals
  are owner-checked so tearing down a watch-restore shell or a
  resumed-away session cannot unregister a live pane, the restore
  shell yields to a registration that appears mid-restore, CLI
  --resume registers after the successful resume, and both the open
  path and the detail-GET lazy rehydrate wire the registration.

Teardown gating and backpressure honesty:
- cleanup_session_ui marks ws._closed FIRST under ws._lock — every
  teardown path (close, close_idle, evict, delete, discard) funnels
  through it — and session_worker.send re-checks under the same lock,
  so a wake can never spawn a worker on a torn-down workstream.
- Create responses carry initial_message_status when the initial
  message could not be delivered (queue_full / refused_closed) instead
  of reading as success; staged attachments survive for the retry;
  /send surfaces a closed workstream as 404 rather than queue_full.

Docs/spec: OpenAPI artifacts regenerated; api-reference documents the
new create-response field; TS SDK type extended.

Tests: ~30 new pins (cancel races, budget durability across restarts,
owner-checked registry moves, teardown gating, stall alerts,
backpressure surfaces, wait_until final re-check); wide subsystem
sweep green (2353 passed).

tests/_helpers.py

@@ -3,9 +3,37 @@ not fixtures, and several test files want to import them directly."""
 
 from __future__ import annotations
 
-from typing import Any
+import time
+from typing import TYPE_CHECKING, Any
 from unittest.mock import MagicMock
 
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+
+def wait_until(cond: Callable[[], bool], timeout: float = 5.0) -> None:
+    """Poll ``cond`` to True within ``timeout`` or fail the test.
+
+    The worker/wake tests can't join threads by identity:
+    ``session_worker.send`` assigns ``ws.worker_thread`` under the lock
+    BEFORE ``t.start()``, so the instant a dispatching call returns, a
+    fast worker may already have run its exit backstop and installed the
+    (not-yet-started) wake thread — joining whatever ``ws.worker_thread``
+    points at races ``RuntimeError: cannot join thread before it is
+    started``.  Poll outcomes instead.
+    """
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if cond():
+            return
+        time.sleep(0.005)
+    if cond():
+        # Final re-check: the condition can become true during the last
+        # sleep (or a CI descheduling stall past the deadline) — failing
+        # without re-looking makes the helper itself a flake source.
+        return
+    raise AssertionError("condition not met within timeout")
+
 
 def make_chat_session(**overrides: Any) -> Any:
     """Build a minimal ``ChatSession`` with sane test defaults.

tests/test_coordinator_adapter.py

@@ -245,6 +245,24 @@ def test_cleanup_ui_tolerates_missing_session_and_ui() -> None:
     ws.session = None
     ws.ui = None
     adapter.cleanup_ui(ws)  # no crash
+    assert ws._closed is True  # still marked dead
+
+
+def test_cleanup_ui_marks_workstream_closed() -> None:
+    """Every teardown path — close, close_idle, EVICTION, delete,
+    discard — funnels through cleanup_ui, which marks the object dead
+    under ``ws._lock`` BEFORE the teardown body runs.  The wake paths
+    that hold OBJECT references (the watch ``wake_fn``,
+    ``session_worker``'s exit backstop) gate on ``_closed``, and
+    ``session_worker.send`` re-checks it under the same lock — without
+    this write here, a wake racing an eviction or delete (which never
+    set the flag) would spawn a full unattended turn on the torn-down
+    session."""
+    adapter, _ = _make_adapter()
+    ws = _make_ws()
+    assert ws._closed is False
+    adapter.cleanup_ui(ws)
+    assert ws._closed is True
 
 
 # ---------------------------------------------------------------------------

tests/test_idle_nudge_wake_integration.py

@@ -28,8 +28,10 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 
+from tests._helpers import wait_until as _wait_until
 from tests.test_session_manager import FakeStorage
-from turnstone.core.idle_nudge_watcher import IdleNudgeWatcher
+from turnstone.core import session_worker
+from turnstone.core.idle_nudge_watcher import IdleNudgeWatcher, wake_workstream_if_pending
 from turnstone.core.session import ChatSession
 from turnstone.core.session_manager import SessionManager
 from turnstone.core.trajectory import dicts_from_turns, turn_from_dict
@@ -299,6 +301,72 @@ def test_idle_event_with_empty_queue_does_not_dispatch_wake(real_mgr, tmp_db):
         watcher.shutdown()
 
 
+def test_watch_fire_on_already_idle_session_drives_wake_send(real_mgr, tmp_db):
+    """A watch firing on an ALREADY-idle workstream sees no IDLE
+    transition, so :class:`IdleNudgeWatcher` never re-checks the queue —
+    the dispatch closure's ``wake_fn`` must drive the wake itself.
+
+    Boundary path under test (only the LLM stream is patched):
+      dispatch closure (real, built by ``set_watch_runner``)
+        → NudgeQueue.enqueue (real)
+        → wake_fn → wake_workstream_if_pending (real)
+        → session_worker.send (real) → daemon thread
+        → ChatSession.deliver_wake_nudge_from_queue (real)
+        → ChatSession.send("") → watch_triggered system turn in history
+    """
+    mgr, _adapter = real_mgr
+    ws = mgr.create(user_id="u1", name="watch-wake-int", skill=None)
+    assert ws.session is not None
+
+    captured: dict[str, Any] = {}
+
+    class _StubRunner:
+        def set_dispatch_fn(self, ws_id: str, fn: Any) -> None:
+            captured["fn"] = fn
+
+    # Production wiring shape (server.py): wake_fn closes over the
+    # Workstream OBJECT — not its id — so eviction+restore id drift
+    # can't strand the wake.
+    ws.session.set_watch_runner(
+        _StubRunner(), wake_fn=lambda: wake_workstream_if_pending(ws, trigger="watch-fire")
+    )
+
+    with (
+        patch.object(ws.session, "_create_stream_with_retry", return_value=iter([])),
+        patch.object(
+            ws.session,
+            "_stream_response",
+            return_value={"role": "assistant", "content": "ok"},
+        ),
+        patch.object(ws.session, "_update_token_table"),
+        patch.object(ws.session, "_print_status_line"),
+        patch.object(ws.session, "_visible_memory_count", return_value=0),
+        patch("turnstone.core.session.save_message"),
+    ):
+        ws.session._title_generated = True
+        # Idle all along — no worker, and no state transition coming.
+        assert ws.state is WorkstreamState.IDLE
+
+        # Simulate the WatchRunner poll thread delivering a fire.
+        captured["fn"]({"type": "watch_triggered", "text": "deploy finished: OK"}, "watch-1")
+
+        _wait_for_worker_done(ws)
+
+    # Queue drained by the wake — not parked until the next user message.
+    assert len(ws.session._nudge_queue) == 0
+
+    msgs = dicts_from_turns(ws.session.messages)
+    user_msgs = [m for m in msgs if m.get("role") == "user"]
+    assert user_msgs, "expected a synthesized user message from the wake"
+    assert user_msgs[-1]["content"] == ""
+    assert user_msgs[-1].get("_source") == "system_nudge"
+    sys_turns = [m for m in msgs if m.get("role") == "system"]
+    assert any(
+        m.get("_source") == "watch_triggered" and "deploy finished: OK" in m.get("content", "")
+        for m in sys_turns
+    ), f"expected a watch_triggered system turn, got {sys_turns!r}"
+
+
 @pytest.fixture
 def coord_mgr() -> tuple[SessionManager, _BuildRealSessionAdapter, FakeStorage]:
     """Real coord-side SessionManager with the adapter's kind set to
@@ -411,3 +479,120 @@ def test_coord_idle_with_active_children_emits_envelope_via_real_managers(coord_
     finally:
         watcher.shutdown()
         observer.shutdown()
+
+
+def test_coord_idle_emitted_from_worker_thread_still_wakes(coord_mgr, tmp_db):
+    """The production-shaped race the test above does NOT exercise: in
+    production, IDLE is emitted from INSIDE the worker (``set_state``
+    subscribers fire on the calling thread — the coord's send emits IDLE
+    before its worker exits).  The watcher's wake dispatch therefore
+    lands on ``session_worker.send``'s reuse path while the
+    transitioning worker still owns the flag, and no-ops.  Without the
+    ownership-clear backstop the ``idle_children`` nudge strands until
+    the next user message — a coord that forgot ``wait_for_workstream``
+    never revives.
+
+    Boundary path under test:
+      worker thread: mgr.set_state(IDLE)
+        → observer enqueues (real) → watcher wake no-ops (worker owns flag)
+        → run() returns → session_worker._runner finally clears the flag
+        → _retry_pending_wake → wake_workstream_if_pending (real)
+        → wake daemon → deliver_wake_nudge_from_queue → send("")
+        → idle_children system turn in history
+    """
+    from turnstone.console.coordinator_idle_observer import CoordinatorIdleObserver
+    from turnstone.core.workstream import WorkstreamKind as _Kind
+
+    mgr, adapter, storage = coord_mgr
+    observer = CoordinatorIdleObserver(mgr, storage)
+    observer.start()
+    watcher = IdleNudgeWatcher(mgr)
+    watcher.start()
+
+    try:
+        coord = mgr.create(user_id="u1", name="parent-coord-2", skill=None)
+        assert coord.session is not None
+
+        storage.register_workstream(
+            "child-x",
+            user_id="u1",
+            name="crawl-docs",
+            kind=_Kind.INTERACTIVE,
+            parent_ws_id=coord.id,
+            state="running",
+        )
+
+        coord.session.messages.append(turn_from_dict({"role": "user", "content": "spawn 1"}))
+        coord.session.messages.append(turn_from_dict({"role": "assistant", "content": "ok"}))
+
+        with (
+            patch.object(coord.session, "_create_stream_with_retry", return_value=iter([])),
+            patch.object(
+                coord.session,
+                "_stream_response",
+                return_value={"role": "assistant", "content": "ack"},
+            ),
+            patch.object(coord.session, "_full_messages", return_value=[]),
+            patch.object(coord.session, "_update_token_table"),
+            patch.object(coord.session, "_print_status_line"),
+            patch.object(coord.session, "_visible_memory_count", return_value=0),
+            patch("turnstone.core.session.save_message"),
+        ):
+            coord.session._title_generated = True
+
+            # Drive the IDLE transition from INSIDE a session_worker
+            # worker, as production does.
+            ok = session_worker.send(
+                coord,
+                enqueue=lambda: None,
+                run=lambda: mgr.set_state(coord.id, WorkstreamState.IDLE),
+                thread_name="coord-send-sim",
+            )
+            assert ok is True
+            # Without the backstop the queue never drains (the watcher's
+            # transition-time wake no-opped against the sim worker) and
+            # this poll times out.  Queue-empty implies the wake worker's
+            # drain ran, so the follow-up flag poll waits for ITS exit.
+            _wait_until(lambda: len(coord.session._nudge_queue) == 0)
+            _wait_for_worker_done(coord)
+
+        # Queue drained by the wake, not waiting on the next user message.
+        assert len(coord.session._nudge_queue) == 0
+        msgs = dicts_from_turns(coord.session.messages)
+        user_msgs = [m for m in msgs if m.get("role") == "user"]
+        wake_msg = user_msgs[-1]
+        assert wake_msg["content"] == ""
+        assert wake_msg.get("_source") == "system_nudge"
+        idle_turns = [
+            m for m in msgs if m.get("role") == "system" and m["_source"] == "idle_children"
+        ]
+        assert len(idle_turns) == 1
+        assert "crawl-docs" in idle_turns[0]["content"]
+        assert "wait_for_workstream" in idle_turns[0]["content"]
+    finally:
+        watcher.shutdown()
+        observer.shutdown()
+
+
+def test_wake_delivery_contains_generation_cancelled(tmp_db):
+    """A close/force-cancel racing the wake turn raises
+    ``GenerationCancelled`` (a BaseException) out of ``send("")`` — the
+    wake method must contain it: it IS the wake worker's ``run()``
+    closure, and ``session_worker._runner`` catches only ``Exception``,
+    so an escape would land in ``threading.excepthook`` as stderr noise
+    on every close-vs-wake race."""
+    from tests._helpers import make_chat_session
+    from turnstone.core.session import GenerationCancelled
+
+    session = make_chat_session()
+    session._nudge_queue.enqueue("idle_children", "kids waiting", "any")
+
+    def _cancelled_send(*_a: Any, **_k: Any) -> None:
+        raise GenerationCancelled
+
+    session.send = _cancelled_send  # type: ignore[method-assign]
+
+    session.deliver_wake_nudge_from_queue()  # must not raise
+
+    assert session._wake_source_tag == ""
+    assert session._wake_drained_reminders is None

tests/test_idle_nudge_watcher.py

@@ -9,13 +9,14 @@ module-level function to capture calls without spawning real threads.
 from __future__ import annotations
 
 import contextlib
+import logging
 import threading
 from typing import Any
 from unittest.mock import patch
 
 import pytest
 
-from turnstone.core.idle_nudge_watcher import IdleNudgeWatcher
+from turnstone.core.idle_nudge_watcher import IdleNudgeWatcher, wake_workstream_if_pending
 from turnstone.core.nudge_queue import NudgeQueue
 from turnstone.core.workstream import WorkstreamState
 
@@ -32,6 +33,7 @@ class _FakeSession:
 class _FakeWorkstream:
     def __init__(self, ws_id: str = "ws-test") -> None:
         self.id = ws_id
+        self.state = WorkstreamState.IDLE
         self.session: _FakeSession | None = _FakeSession()
         self._lock = threading.Lock()
         self._worker_running = False
@@ -163,3 +165,117 @@ class TestIdleNudgeWatcher:
         watcher.start()
         watcher.shutdown()
         watcher.shutdown()  # no error
+
+
+class TestWakeWorkstreamIfPending:
+    """Direct tests for the shared wake gate.
+
+    The IDLE-transition path (via the watcher) is covered above; these
+    pin the gates the watch dispatch closure relies on when it calls
+    the helper directly, with no state event involved.
+    """
+
+    def test_wakes_idle_ws_with_pending_entry(self, fake_mgr_and_ws):
+        _mgr, ws = fake_mgr_and_ws
+        ws.session._nudge_queue.enqueue("watch_triggered", "output", "any")
+        with patch("turnstone.core.session_worker.send", return_value=True) as mock_send:
+            assert wake_workstream_if_pending(ws) is True
+            assert mock_send.call_count == 1
+            kwargs = mock_send.call_args.kwargs
+            assert kwargs["enqueue"]() is None
+            kwargs["run"]()
+            assert ws.session.deliver_wake_nudge_from_queue_called == 1
+            assert kwargs["thread_name"].startswith("wake-nudge-")
+
+    def test_skips_session_none(self, fake_mgr_and_ws):
+        _mgr, ws = fake_mgr_and_ws
+        ws.session = None
+        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_skips_closed_ws(self, fake_mgr_and_ws):
+        """A workstream mid-``close()`` must not get a wake spawned on
+        its torn-down session, even while its ``state`` field still
+        reads IDLE (there is no CLOSED member — close uses the
+        ``_closed`` tombstone)."""
+        _mgr, ws = fake_mgr_and_ws
+        ws.session._nudge_queue.enqueue("watch_triggered", "output", "any")
+        ws._closed = True
+        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_skips_non_idle_states(self, fake_mgr_and_ws):
+        """Busy states imply a live worker that drains at its own seams;
+        ERROR stays parked for the operator — neither gets a wake."""
+        _mgr, ws = fake_mgr_and_ws
+        ws.session._nudge_queue.enqueue("watch_triggered", "output", "any")
+        with patch("turnstone.core.session_worker.send") as mock_send:
+            for state in (
+                WorkstreamState.RUNNING,
+                WorkstreamState.THINKING,
+                WorkstreamState.ATTENTION,
+                WorkstreamState.ERROR,
+            ):
+                ws.state = state
+                assert wake_workstream_if_pending(ws) is False
+            assert mock_send.call_count == 0
+
+    def test_skips_tool_only_entries(self, fake_mgr_and_ws):
+        """Tool-channel entries belong to the next tool-result seam — a
+        synthetic empty user turn can't drain them, so no wake."""
+        _mgr, ws = fake_mgr_and_ws
+        ws.session._nudge_queue.enqueue("tool_error", "check memories", "tool")
+        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
+        the trigger label (structlog renders the event name + ``%s``
+        placeholders into ``msg``; substring-match like the sibling
+        nudge_queue tests)."""
+        _mgr, ws = fake_mgr_and_ws
+        ws.session._nudge_queue.enqueue("watch_triggered", "output", "any")
+        with (
+            patch("turnstone.core.session_worker.send", return_value=True) as mock_send,
+            caplog.at_level(logging.INFO, logger="turnstone.core.idle_nudge_watcher"),
+        ):
+            assert wake_workstream_if_pending(ws, trigger="idle-transition") is True
+            assert mock_send.call_count == 1
+        dispatched = [r for r in caplog.records if "nudge_wake.dispatched" in r.getMessage()]
+        assert len(dispatched) == 1
+        assert dispatched[0].levelno == logging.INFO
+        assert "trigger=" in dispatched[0].getMessage()
+        # The reuse-path drop line must not appear on a fresh spawn.
+        assert not any("nudge_wake.deferred_worker_busy" in r.getMessage() for r in caplog.records)
+
+    def test_deferred_path_logs_worker_busy(self, fake_mgr_and_ws, caplog):
+        """The reuse path — ``send`` invokes the passed ``enqueue`` and
+        returns True — emits ``nudge_wake.deferred_worker_busy`` instead
+        of ``dispatched``.  The entry stays owed to the owning worker's
+        exit backstop; the return value is still True."""
+        _mgr, ws = fake_mgr_and_ws
+        ws.session._nudge_queue.enqueue("watch_triggered", "output", "any")
+
+        def _reuse_send(_ws: Any, *, enqueue: Any, run: Any, thread_name: Any) -> bool:
+            # Mimic a live worker owning the workstream: send routes the
+            # wake to the no-op enqueue rather than spawning a daemon.
+            enqueue()
+            return True
+
+        with (
+            patch("turnstone.core.session_worker.send", side_effect=_reuse_send) as mock_send,
+            caplog.at_level(logging.INFO, logger="turnstone.core.idle_nudge_watcher"),
+        ):
+            assert wake_workstream_if_pending(ws, trigger="idle-transition") is True
+            assert mock_send.call_count == 1
+        deferred = [
+            r for r in caplog.records if "nudge_wake.deferred_worker_busy" in r.getMessage()
+        ]
+        assert len(deferred) == 1
+        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)

tests/test_server_attachments_endpoints.py

@@ -459,6 +459,11 @@ class TestSendMessageAttachments:
         session = MagicMock()
         session._cancel_event = threading.Event()
         session.queue_message = MagicMock()
+        # A bare Mock's auto-created ``_nudge_queue`` (truthy, has_pending
+        # truthy, no-op deliver) turns the worker-exit wake backstop into an
+        # endless respawn loop; declare this a stub session WITHOUT a queue
+        # so the wake gate's stub-guard bails.
+        session._nudge_queue = None
         captured: dict = {}
 
         def fake_send(message, attachments=None, send_id=None):
@@ -480,6 +485,7 @@ class TestSendMessageAttachments:
         ws.session = session
         ws.worker_thread = None
         ws._worker_running = False
+        ws._closed = False  # a bare Mock attr is truthy → send() would refuse
         ws._lock = threading.RLock()
         mgr.get.return_value = ws
         return captured, session
@@ -658,6 +664,9 @@ class TestQueuedSendWithAttachments:
         session = MagicMock()
         session._cancel_event = threading.Event()
         session.queue_message = fake_queue_message
+        # Stub session without a NudgeQueue — see _wire_ws for why a bare
+        # Mock queue would feed the exit backstop an endless wake loop.
+        session._nudge_queue = None
 
         ui = MagicMock()
         ui._ws_lock = threading.Lock()
@@ -675,6 +684,7 @@ class TestQueuedSendWithAttachments:
         ws.session = session
         ws.worker_thread = worker
         ws._worker_running = True
+        ws._closed = False  # a bare Mock attr is truthy → send() would refuse
         ws._lock = threading.RLock()
         mgr.get.return_value = ws
         return captured
@@ -738,6 +748,7 @@ class TestBusyWorkerAttachments:
         ws.ui = ui
         ws.session = session
         ws.worker_thread = worker
+        ws._closed = False  # a bare Mock attr is truthy → send() would refuse
         ws._lock = threading.RLock()
         mgr.get.return_value = ws
         return ws, session

tests/test_server_attachments_on_create.py

@@ -434,6 +434,91 @@ class TestCreateMultipart:
         # pane's rehydrate can't observe it as still-staged.
         assert get_attachment_buffer().get(aid, ws_id=ws_id, user_id="userA") is None
 
+    def test_create_raced_by_live_worker_keeps_attachments_staged(self, app_client, monkeypatch):
+        """The enqueue branch (caller-supplied ws_id raced by a concurrent
+        /send claiming the worker first) can't deliver attachments through
+        the interjection seam — they must REMAIN STAGED so the composer
+        still shows them and the user's next send delivers them, while the
+        message text itself rides the queue."""
+        from turnstone.core import session_worker
+        from turnstone.core.attachment_buffer import get_attachment_buffer
+
+        client, _sessions, _gq = app_client
+        queued: list[str] = []
+
+        def _record_queue(self, text, *a, **k):
+            queued.append(text)
+            return ("", "normal", "msg-x")
+
+        monkeypatch.setattr(_FakeSession, "queue_message", _record_queue)
+
+        def _live_worker_send(ws, *, enqueue, run, thread_name=None):
+            enqueue()  # a worker already owns the ws — reuse path
+            return True
+
+        monkeypatch.setattr(session_worker, "send", _live_worker_send)
+
+        meta = {"name": "raced", "initial_message": "look at this file"}
+        resp = client.post(
+            "/v1/api/workstreams/new",
+            data={"meta": json.dumps(meta)},
+            files=[("file", ("notes.md", b"# hello\n", "text/markdown"))],
+            headers=_auth("userA"),
+        )
+        assert resp.status_code == 200, resp.text
+        ws_id = resp.json()["ws_id"]
+        aid = resp.json()["attachment_ids"][0]
+
+        assert queued == ["look at this file"]  # text preserved via the queue
+        # NOT drained: the upload stays staged, recoverable on the next send.
+        assert get_attachment_buffer().get(aid, ws_id=ws_id, user_id="userA") is not None
+        # Delivered path → no dropped-message marker on the response.
+        assert "initial_message_status" not in resp.json()
+
+    def test_create_raced_queue_full_reports_dropped_message(self, app_client, monkeypatch):
+        """``queue.Full`` on the raced enqueue path must not read as
+        success: it propagates out of ``_enqueue_init`` into
+        ``session_worker.send``'s backpressure branch (→ ``False``), and
+        the create response carries ``initial_message_status:
+        "queue_full"`` instead of a bare 200 implying the first message
+        was delivered.  Attachments stay staged for the retry."""
+        import queue as _queue
+
+        from turnstone.core import session_worker
+        from turnstone.core.attachment_buffer import get_attachment_buffer
+
+        client, _sessions, _gq = app_client
+
+        def _full_queue(self, *a, **k):
+            raise _queue.Full
+
+        monkeypatch.setattr(_FakeSession, "queue_message", _full_queue)
+
+        def _live_worker_send(ws, *, enqueue, run, thread_name=None):
+            # Mirror the real send()'s reuse-path backpressure contract:
+            # queue.Full → False, never a raise to the caller.
+            try:
+                enqueue()
+            except _queue.Full:
+                return False
+            return True
+
+        monkeypatch.setattr(session_worker, "send", _live_worker_send)
+
+        meta = {"name": "raced-full", "initial_message": "look at this file"}
+        resp = client.post(
+            "/v1/api/workstreams/new",
+            data={"meta": json.dumps(meta)},
+            files=[("file", ("notes.md", b"# hello\n", "text/markdown"))],
+            headers=_auth("userA"),
+        )
+        assert resp.status_code == 200, resp.text
+        body = resp.json()
+        assert body["initial_message_status"] == "queue_full"
+        # Attachments untouched — the composer chips survive for the retry.
+        aid = body["attachment_ids"][0]
+        assert get_attachment_buffer().get(aid, ws_id=body["ws_id"], user_id="userA") is not None
+
     def test_create_with_attachments_no_initial_message_keeps_staged(self, app_client):
         import hashlib
 
@@ -527,3 +612,28 @@ class TestCreateJsonStillWorks:
         assert data["ws_id"]
         # New optional field, but always emitted (empty list when absent)
         assert data["attachment_ids"] == []
+
+    def test_initial_message_routes_through_session_worker_send(self, app_client):
+        """The initial-message worker is dispatched via
+        ``session_worker.send`` (not an inlined ``threading.Thread``) so it
+        inherits the ownership-clear wake backstop.  Patching the module
+        attribute captures the wiring without spawning a thread — server.py
+        calls ``session_worker.send`` as a module attribute even from its
+        local import."""
+        from unittest.mock import patch
+
+        client, _sessions, _gq = app_client
+        with patch("turnstone.core.session_worker.send", return_value=True) as mock_send:
+            resp = client.post(
+                "/v1/api/workstreams/new",
+                json={"name": "init-dispatch", "initial_message": "go"},
+                headers=_auth("userA"),
+            )
+            assert resp.status_code == 200, resp.text
+            assert mock_send.call_count == 1
+            kwargs = mock_send.call_args.kwargs
+            assert kwargs["thread_name"].startswith("ws-init-")
+            # ``run`` is the init closure the shared dispatcher spawns; the
+            # dead-by-construction ``enqueue`` branch is still wired (loudly).
+            assert callable(kwargs["run"])
+            assert callable(kwargs["enqueue"])

tests/test_session.py

@@ -5086,6 +5086,55 @@ class TestMetacognitiveBuffers:
         assert sys_turn["_source"] == "tool_error"
         assert sys_turn["content"] == "you hit an error; check memory"
 
+    def test_denial_nudge_queues_on_tool_channel(self, tmp_db):
+        """A denial responds to the tool batch the user just rejected — the
+        producer must queue it on the TOOL channel so it drains through
+        ``_collect_advisories`` alongside the denied results (the same seam
+        tool_error / repeat use), not sit on the user channel until the next
+        user-message seam — by which point the model has already reacted to
+        the denial without the nudge.
+
+        Drives the REAL ``_execute_tools`` two-phase gate with real
+        ``_nudges_enabled`` / ``should_nudge`` gating; only the prepare
+        step and the UI approval are stubbed."""
+        from turnstone.core.metacognition import format_nudge
+
+        session = _make_session()
+        # ``should_nudge`` skips the very first message — give the session
+        # the natural pre-batch shape (user turn + assistant tool-call turn).
+        session.messages.append(turn_from_dict({"role": "user", "content": "do the thing"}))
+        session.messages.append(turn_from_dict({"role": "assistant", "content": "calling"}))
+
+        item = {
+            "call_id": "call_1",
+            "func_name": "notify",
+            "needs_approval": True,
+            # Must NOT run — a denied tool never executes.
+            "execute": lambda p: (p["call_id"], "EXECUTED — must not happen"),
+        }
+        with (
+            patch.object(session, "_safe_prepare_tool", return_value=item),
+            patch.object(session.ui, "approve_tools", return_value=(False, "use /tmp instead")),
+            patch.object(session, "_visible_memory_count", return_value=0),
+        ):
+            tool_calls = [
+                {
+                    "id": "call_1",
+                    "type": "function",
+                    "function": {"name": "notify", "arguments": "{}"},
+                }
+            ]
+            results, feedback = session._execute_tools(tool_calls)
+
+        # The denied item surfaced the operator's feedback as its result…
+        assert results == [("call_1", "Denied by user: use /tmp instead")]
+        assert feedback is None
+        # …and the denial nudge is queued on the TOOL channel, so the same
+        # batch's ``_collect_advisories`` drain delivers it; nothing defers
+        # to the next user turn.
+        assert session._nudge_queue.pending(channel="tool") == [("denial", format_nudge("denial"))]
+        assert session._nudge_queue.pending(channel="user") == []
+
     def test_queued_message_appends_system_turn_after_tool_batch(self, tmp_db):
         """A queued message arriving during a tool batch becomes a
         first-class ``{"role": "system", "_source": "user_interjection"}``

tests/test_session_worker.py

@@ -2,7 +2,7 @@
 
 The shared worker dispatch is load-bearing for both the interactive
 ``/v1/api/workstreams/{ws_id}/send`` HTTP handler and the coordinator
-``CoordinatorAdapter.send`` path. Tests cover the four invariants the
+``CoordinatorAdapter.send`` path. Tests cover the five invariants the
 module must hold:
 
 * live worker → enqueue, no thread spawn
@@ -10,10 +10,14 @@ module must hold:
 * concurrent ``send`` calls produce exactly one worker thread
   (Stage 1 bug-1 — the racy ``Thread.is_alive()`` gate stays caught)
 * ``_worker_running`` cleared in ``finally`` even on uncaught exception
+* ownership-clear wake backstop: a worker exiting with USER_DRAIN
+  nudges queued on an IDLE workstream spawns the wake send that the
+  IDLE fan-out (which ran on this worker's own thread) had to drop
 
-Callers pass no-arg closures, so this module never touches
-``ws.session`` — keeps the contract narrow and lets watch-style
-dispatchers drive a session that isn't installed on ``ws``.
+Callers pass no-arg closures, so dispatch never touches ``ws.session``;
+the exit backstop only PEEKS it defensively (``getattr`` for
+``_nudge_queue``, bail on stubs) — watch-style dispatchers can still
+drive a session that isn't installed on ``ws``.
 """
 
 from __future__ import annotations
@@ -22,8 +26,10 @@ import queue
 import threading
 from typing import Any
 
+from tests._helpers import wait_until as _wait_until
 from turnstone.core import session_worker
-from turnstone.core.workstream import Workstream
+from turnstone.core.nudge_queue import USER_DRAIN, NudgeQueue
+from turnstone.core.workstream import Workstream, WorkstreamState
 
 
 class _SendSession:
@@ -140,6 +146,41 @@ def test_enqueue_unexpected_exception_returns_false_logged() -> None:
     assert ws._worker_running is True
 
 
+def test_closed_workstream_refused_no_spawn() -> None:
+    """Authoritative closed-check: ``close()`` sets ``_closed`` under
+    ``ws._lock``, so a wake (or send) racing it must be refused HERE —
+    the wake gate's lockless peek can go stale, and a spawn past this
+    point would run a full unattended turn (inference, tool calls,
+    storage writes) on a workstream whose ``ws_closed`` already fired.
+    """
+    session = _SendSession()
+    ws = _make_ws(session)
+    ws._closed = True
+
+    ok = _send_message(ws, session, "hello")
+
+    assert ok is False
+    assert session.send_calls == []
+    assert session.queue_calls == []
+    assert ws.worker_thread is None
+    assert ws._worker_running is False
+
+
+def test_closed_workstream_refused_on_reuse_path_too() -> None:
+    """The refusal precedes the enqueue branch: no interjection is queued
+    onto a session whose workstream is already closed."""
+    session = _SendSession()
+    ws = _make_ws(session)
+    ws._worker_running = True
+    ws._closed = True
+
+    ok = _send_message(ws, session, "hello")
+
+    assert ok is False
+    assert session.queue_calls == []
+    assert ws._worker_running is True  # untouched — not ours to clear
+
+
 # ---------------------------------------------------------------------------
 # _worker_running lifecycle
 # ---------------------------------------------------------------------------
@@ -301,6 +342,141 @@ def test_thread_name_explicit_override() -> None:
     ws.worker_thread.join(timeout=2.0)
 
 
+class _WakeCapableSession(_SendSession):
+    """Adds the ChatSession surface the exit backstop peeks at."""
+
+    def __init__(self, **kwargs: Any) -> None:
+        super().__init__(**kwargs)
+        self._nudge_queue = NudgeQueue()
+        self.deliver_calls = 0
+        self.deliver_thread_names: list[str] = []
+        self.delivered = threading.Event()
+
+    def deliver_wake_nudge_from_queue(self) -> None:
+        # Mirror the real contract: the wake drains its own queue, so
+        # the wake worker's OWN exit backstop sees nothing pending and
+        # the chain converges instead of spawning wakes forever.
+        self.deliver_calls += 1
+        self.deliver_thread_names.append(threading.current_thread().name)
+        self._nudge_queue.drain(USER_DRAIN)
+        self.delivered.set()
+
+
+class TestWorkerExitWakeBackstop:
+    """A worker exiting while its (idle) workstream has USER_DRAIN
+    nudges queued spawns the wake send the IDLE fan-out had to drop.
+
+    Production shape being modelled: ``set_state(IDLE)`` fires its
+    subscribers on the worker thread from inside ``run()`` —
+    ``CoordinatorIdleObserver`` enqueues ``idle_children``, then
+    ``IdleNudgeWatcher``'s wake dispatch lands on the reuse path
+    (this very worker still owns the flag) and no-ops.  The enqueue
+    inside ``run`` below stands in for that observer enqueue.
+    """
+
+    def test_worker_exit_delivers_pending_wake(self) -> None:
+        session = _WakeCapableSession()
+        ws = _make_ws(session)
+        assert ws.state is WorkstreamState.IDLE  # dataclass default
+
+        def run() -> None:
+            # What the IDLE fan-out's observer does, on this thread.
+            session._nudge_queue.enqueue("idle_children", "kids waiting", "any")
+
+        ok = session_worker.send(ws, enqueue=lambda: None, run=run)
+        assert ok is True
+
+        # The wake is delivered on a fresh wake-named worker thread…
+        assert session.delivered.wait(timeout=2.0), (
+            "exit backstop did not deliver the pending nudge"
+        )
+        assert session.deliver_thread_names[0].startswith("wake-nudge-")
+        # …after which the wake worker's own exit backstop sees an empty
+        # queue and the chain converges: flag at rest, exactly one deliver.
+        _wait_until(lambda: ws._worker_running is False)
+        assert session.deliver_calls == 1
+        assert len(session._nudge_queue) == 0
+
+    def test_worker_exit_no_wake_when_queue_empty(self) -> None:
+        session = _WakeCapableSession()
+        ws = _make_ws(session)
+
+        ok = session_worker.send(ws, enqueue=lambda: None, run=lambda: None)
+        assert ok is True
+        original = ws.worker_thread
+        assert original is not None
+        original.join(timeout=2.0)
+
+        assert ws.worker_thread is original  # no wake spawned
+        assert session.deliver_calls == 0
+        assert ws._worker_running is False
+
+    def test_worker_exit_no_wake_for_stub_session_without_queue(self) -> None:
+        """The narrow-contract escape hatch: a session without a
+        ``_nudge_queue`` (watch-style stubs) is skipped by the shared
+        wake gate's own defensive peek — no AttributeError, no wake."""
+        session = _SendSession()
+        ws = _make_ws(session)
+
+        ok = _send_message(ws, session, "hello")
+        assert ok is True
+        original = ws.worker_thread
+        assert original is not None
+        original.join(timeout=2.0)
+
+        assert ws.worker_thread is original
+        assert ws._worker_running is False
+
+    def test_worker_exit_no_wake_when_state_not_idle(self) -> None:
+        """An ERROR exit stays parked for the operator — pending nudges
+        wait for the next real interaction rather than burning
+        unattended inference on a failed session."""
+        session = _WakeCapableSession()
+        ws = _make_ws(session)
+
+        def run() -> None:
+            session._nudge_queue.enqueue("idle_children", "kids waiting", "any")
+            ws.state = WorkstreamState.ERROR
+
+        ok = session_worker.send(ws, enqueue=lambda: None, run=run)
+        assert ok is True
+        original = ws.worker_thread
+        assert original is not None
+        original.join(timeout=2.0)
+
+        assert ws.worker_thread is original
+        assert session.deliver_calls == 0
+        assert len(session._nudge_queue) == 1  # still queued for later seams
+
+    def test_abandoned_worker_does_not_run_wake_backstop(self) -> None:
+        """Only the owner retries: an abandoned worker (successor claimed
+        the flag) finishing late must not spawn a wake — the successor's
+        own exit runs the backstop."""
+        send_gate = threading.Event()
+        session = _WakeCapableSession(send_gate=send_gate)
+        ws = _make_ws(session)
+
+        ok = _send_message(ws, session, "hello")
+        assert ok is True
+        abandoned = ws.worker_thread
+        assert abandoned is not None
+
+        session._nudge_queue.enqueue("idle_children", "kids waiting", "any")
+        sentinel = threading.Thread(target=lambda: None, name="successor")
+        with ws._lock:
+            ws.worker_thread = sentinel
+            ws._worker_running = True
+
+        send_gate.set()
+        abandoned.join(timeout=3.0)
+        assert not abandoned.is_alive()
+
+        # No wake spawned by the abandoned thread; ownership intact.
+        assert ws.worker_thread is sentinel
+        assert session.deliver_calls == 0
+        assert ws._worker_running is True
+
+
 def test_does_not_deadlock_when_run_briefly_grabs_ws_lock() -> None:
     """Sanity check: ``run`` is invoked OUTSIDE ``ws._lock``. A worker
     body that briefly takes the lock (e.g. to update worker state)

tests/test_watch.py

@@ -2,11 +2,15 @@
 
 from __future__ import annotations
 
+import threading
+import time
 from datetime import UTC, datetime
+from typing import Any
 from unittest.mock import MagicMock
 
 import pytest
 
+from tests._helpers import wait_until
 from turnstone.core.watch import (
     WatchRunner,
     build_watch_reminder,
@@ -545,8 +549,12 @@ class TestWatchRunner:
         assert runner.get_dispatch_fn("ws-1") is fn
         # Unknown ws → None.
         assert runner.get_dispatch_fn("ws-missing") is None
-        # After removal → None.
-        runner.remove_dispatch_fn("ws-1")
+        # Owner-checked removal: a non-owner's teardown must not remove a
+        # still-live registration (restore shell vs reopened pane).
+        runner.remove_dispatch_fn("ws-1", owner=MagicMock())
+        assert runner.get_dispatch_fn("ws-1") is fn
+        # The owner (or a blind removal) does remove it.
+        runner.remove_dispatch_fn("ws-1", owner=fn)
         assert runner.get_dispatch_fn("ws-1") is None
 
     def test_run_command_success(self):
@@ -566,3 +574,716 @@ class TestWatchRunner:
         output, code = runner._run_command("sleep 30")
         assert "timed out" in output.lower()
         assert code == -1
+
+
+def _watch_row(**over: Any) -> dict[str, Any]:
+    """A firing watch row (condition matches ``echo hello``); override
+    fields per test."""
+    row: dict[str, Any] = {
+        "watch_id": "abc123",
+        "ws_id": "ws-1",
+        "name": "test-watch",
+        "command": "echo hello",
+        "stop_on": '"hello" in output',
+        "max_polls": 100,
+        "poll_count": 0,
+        "last_output": None,
+        "interval_secs": 60,
+        "created": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S"),
+    }
+    row.update(over)
+    return row
+
+
+def _slow_restore(runner: WatchRunner, ws_id: str, calls: list[str], lock: threading.Lock) -> Any:
+    """Restore_fn stand-in: record the call, register a dispatch fn (as the
+    real restore does via ``set_watch_runner``), and sleep briefly so a
+    concurrent second caller is guaranteed to be waiting on ``_restore_lock``
+    when we return.
+    """
+    with lock:
+        calls.append(ws_id)
+    time.sleep(0.05)
+    fn = MagicMock()
+    runner.set_dispatch_fn(ws_id, fn)
+    return fn
+
+
+class TestWatchRunnerDeliveryRetry:
+    """Delivery failure HOLDS the built reminder and re-delivers it on a
+    later tick — never re-running the command, so a transient stop_on match
+    isn't lost — bounded by ``MAX_DELIVERY_ATTEMPTS``.  Until delivery lands
+    the row commits only ``next_poll`` plus the fire's durable poll charge:
+    no baseline advance and no deactivation of a fire the model never saw,
+    while a restart mid-hold (which re-runs the command) stays bounded by
+    ``max_polls``."""
+
+    def _make_runner(self, storage: Any, **kwargs: Any) -> WatchRunner:
+        return WatchRunner(
+            storage=storage,
+            node_id="test-node",
+            check_interval=0.1,
+            tool_timeout=5,
+            **kwargs,
+        )
+
+    def test_delivery_failure_holds_reminder_and_defers_row(self):
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        # No dispatch fn registered, no restore_fn → delivery fails.
+        runner = self._make_runner(storage)
+
+        runner._poll_watch(_watch_row())
+
+        # Row commit is the retry cadence + this fire's durable poll charge
+        # — the fire stays fully retryable, and a restart mid-hold (which
+        # re-runs the command) stays bounded by max_polls.
+        storage.update_watch.assert_called_once()
+        args, kwargs = storage.update_watch.call_args
+        assert args[0] == "abc123"
+        assert set(kwargs) == {"next_poll", "poll_count"}
+        assert kwargs["poll_count"] == 1  # charged durably at hold time
+        assert kwargs["next_poll"]  # advanced, not cleared
+        # The reminder is HELD for re-delivery; the row is NOT marked
+        # terminal-dispatched (the model never saw it).
+        with runner._pending_delivery_lock:
+            assert "abc123" in runner._pending_delivery
+            assert runner._pending_delivery["abc123"]["attempts"] == 1
+        with runner._terminal_dispatched_lock:
+            assert "abc123" not in runner._terminal_dispatched
+
+    def test_redelivery_uses_held_reminder_without_rerunning_command(self):
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        runner = self._make_runner(storage)
+
+        # Poll 1: fails → holds.  Capture the exact held reminder object.
+        runner._poll_watch(_watch_row())
+        with runner._pending_delivery_lock:
+            held = runner._pending_delivery["abc123"]["reminder"]
+
+        # ws restored: register a fn, and make _run_command explode so the
+        # test proves re-delivery does NOT re-run the command.
+        dispatch_fn = MagicMock()
+        runner.set_dispatch_fn("ws-1", dispatch_fn)
+        runner._run_command = MagicMock(  # type: ignore[method-assign]
+            side_effect=AssertionError("command must not re-run on re-delivery")
+        )
+
+        runner._poll_watch(_watch_row())
+
+        # Delivered the SAME held reminder; command untouched; committed
+        # terminal with the ORIGINAL fire's poll_count; hold cleared.
+        dispatch_fn.assert_called_once()
+        assert dispatch_fn.call_args[0][0] is held
+        runner._run_command.assert_not_called()
+        _a, kwargs = storage.update_watch.call_args
+        assert kwargs["active"] is False
+        assert kwargs["poll_count"] == 1  # retries consumed no ADDITIONAL budget
+        with runner._pending_delivery_lock:
+            assert "abc123" not in runner._pending_delivery
+
+    def test_transient_exhaustion_keeps_watch_active(self):
+        # A purely transient cause (no fn, no restore → returns False) that
+        # outlasts the attempt budget must NOT silently deactivate the watch
+        # — it drops the held reminder, charges ONE poll to the max_polls
+        # budget, and leaves the watch active to re-fire on its next
+        # interval.  The baseline (last_output) stays uncommitted so a
+        # delta-style stop_on re-fires on the change the model never saw.
+        from turnstone.core.watch import MAX_DELIVERY_ATTEMPTS
+
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        runner = self._make_runner(storage)  # always fails, transiently
+
+        # Poll 1 fires + holds (attempts=1); polls 2..MAX bump attempts, the
+        # MAX-th hitting the exhaustion ceiling.
+        for _ in range(MAX_DELIVERY_ATTEMPTS):
+            runner._poll_watch(_watch_row())
+
+        # Hold dropped, but the watch was NEVER deactivated — no active=False
+        # commit anywhere; the final commit charges the poll and re-schedules
+        # (no last_output → the fire re-detects next cycle).
+        with runner._pending_delivery_lock:
+            assert "abc123" not in runner._pending_delivery
+        deactivations = [
+            c for c in storage.update_watch.call_args_list if c.kwargs.get("active") is False
+        ]
+        assert deactivations == []
+        _a, kwargs = storage.update_watch.call_args  # last commit
+        assert set(kwargs) == {"poll_count", "next_poll"}
+        assert kwargs["poll_count"] == 1  # one poll charged to the budget
+
+    def test_transient_exhaustion_with_budget_spent_deactivates(self):
+        # The keep-alive-on-transient behavior is bounded by the watch's own
+        # max_polls budget: once poll_count reaches it, exhaustion commits
+        # the held (deactivating) update instead of re-running the command
+        # every interval forever against an unreachable workstream.
+        from turnstone.core.watch import MAX_DELIVERY_ATTEMPTS
+
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        runner = self._make_runner(storage)  # always fails, transiently
+
+        for _ in range(MAX_DELIVERY_ATTEMPTS):
+            runner._poll_watch(_watch_row(max_polls=1))  # budget spent on fire 1
+
+        with runner._pending_delivery_lock:
+            assert "abc123" not in runner._pending_delivery
+        _a, kwargs = storage.update_watch.call_args  # last commit
+        assert kwargs["active"] is False  # deactivated: budget spent
+        assert kwargs["poll_count"] == 1
+
+    def test_held_delivery_retries_on_capped_cadence_not_interval(self):
+        # Re-delivery is a cheap in-memory dispatch — a daily watch whose
+        # fire hit a busy restore slot must retry within
+        # DELIVERY_RETRY_CAP_SECS, not sit on the reminder for 24 h.
+        from turnstone.core.watch import DELIVERY_RETRY_CAP_SECS
+
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        runner = self._make_runner(storage)
+
+        runner._poll_watch(_watch_row(interval_secs=86_400))
+
+        _a, kwargs = storage.update_watch.call_args
+        assert set(kwargs) == {"next_poll", "poll_count"}
+        retry_at = datetime.strptime(kwargs["next_poll"], "%Y-%m-%dT%H:%M:%S").replace(tzinfo=UTC)
+        delta = (retry_at - datetime.now(UTC)).total_seconds()
+        assert 0 < delta <= DELIVERY_RETRY_CAP_SECS + 5  # capped, not 86400
+
+    def test_permanent_unrestorable_deactivates_immediately(self):
+        # A permanent failure (restore raises WatchWorkstreamUnrestorable,
+        # e.g. corrupt persona stamp) deactivates the watch on the FIRST
+        # fire — no held reminder, no waiting out the attempt budget.
+        from turnstone.core.watch import WatchWorkstreamUnrestorable
+
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        restore_fn = MagicMock(side_effect=WatchWorkstreamUnrestorable("ws-1"))
+        runner = self._make_runner(storage, restore_fn=restore_fn)
+
+        runner._poll_watch(_watch_row())
+
+        restore_fn.assert_called_once_with("ws-1")  # not retried 5×
+        _a, kwargs = storage.update_watch.call_args
+        assert kwargs["active"] is False  # deactivated now
+        with runner._pending_delivery_lock:
+            assert "abc123" not in runner._pending_delivery  # nothing held
+        # The admission slot is released even on the raising path.
+        with runner._restore_lock:
+            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.
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        runner = self._make_runner(storage)
+        with runner._terminal_dispatched_lock:
+            runner._terminal_dispatched.add("abc123")
+        runner._stash_pending_delivery("abc123", {"text": "x"}, {"active": False}, attempts=1)
+
+        runner._poll_watch(_watch_row())  # hits the already_dispatched branch
+
+        with runner._pending_delivery_lock:
+            assert "abc123" not in runner._pending_delivery
+        with runner._terminal_dispatched_lock:
+            assert "abc123" not in runner._terminal_dispatched
+
+    def test_restore_capacity_full_defers_without_restoring(self):
+        # When MAX_CONCURRENT_RESTORES restores are already in flight, a
+        # new evicted-ws poll must DEFER (return False, hold) rather than
+        # block a poll slot — and must not start a restore.
+        from turnstone.core.watch import MAX_CONCURRENT_RESTORES
+
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        restore_fn = MagicMock(return_value=MagicMock())
+        runner = self._make_runner(storage, restore_fn=restore_fn)
+        # Saturate the restore admission with other in-flight ws_ids.
+        with runner._restore_lock:
+            for i in range(MAX_CONCURRENT_RESTORES):
+                runner._restoring[f"other-{i}"] = time.monotonic()
+
+        result = runner._dispatch_result("ws-evicted", {"text": "x"}, "w1")
+
+        assert result is False  # deferred
+        restore_fn.assert_not_called()  # no restore admitted
+
+    def test_race_won_admission_delivers_outside_restore_lock(self):
+        # A dispatch fn registered between the fast-path miss and the
+        # admission check must be delivered WITHOUT holding _restore_lock:
+        # the closure can block (ws._lock, wake-thread spawn), and running
+        # it under the lock serialises every restore admission on the node
+        # behind one delivery.
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        restore_fn = MagicMock(return_value=None)
+        runner = self._make_runner(storage, restore_fn=restore_fn)
+
+        lock_free_during_dispatch: list[bool] = []
+
+        def probe(reminder: dict[str, Any], watch_id: str) -> None:
+            ok = runner._restore_lock.acquire(blocking=False)
+            lock_free_during_dispatch.append(ok)
+            if ok:
+                runner._restore_lock.release()
+
+        real_try = runner._try_dispatch_fn
+        calls = {"n": 0}
+
+        def fake_try(ws_id: str, reminder: dict[str, Any], watch_id: str) -> bool | None:
+            calls["n"] += 1
+            if calls["n"] == 1:
+                # Simulate a restore completing between the fast path and
+                # the admission check: the fn appears "while we waited".
+                runner.set_dispatch_fn("ws-1", probe)
+                return None
+            return real_try(ws_id, reminder, watch_id)
+
+        runner._try_dispatch_fn = fake_try  # type: ignore[method-assign]
+
+        result = runner._dispatch_result("ws-1", {"text": "x"}, "w1")
+
+        assert result is True  # race-won fn delivered
+        assert lock_free_during_dispatch == [True]  # ...outside the lock
+        restore_fn.assert_not_called()  # no restore admitted for a live fn
+
+    def test_restore_returning_none_holds(self):
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        restore_fn = MagicMock(return_value=None)  # e.g. all slots active
+        runner = self._make_runner(storage, restore_fn=restore_fn)
+
+        runner._poll_watch(_watch_row())
+
+        restore_fn.assert_called_once_with("ws-1")
+        _a, kwargs = storage.update_watch.call_args
+        assert set(kwargs) == {"next_poll", "poll_count"}
+        with runner._pending_delivery_lock:
+            assert "abc123" in runner._pending_delivery
+
+    def test_live_fn_raise_holds_without_restoring(self):
+        # A registered fn that RAISES means the ws is live; we must NOT fall
+        # through to restore (that would spawn a duplicate session on a live
+        # conversation).  The reminder is held for re-delivery instead.
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        restore_fn = MagicMock()
+        runner = self._make_runner(storage, restore_fn=restore_fn)
+        runner.set_dispatch_fn("ws-1", MagicMock(side_effect=RuntimeError("stale closure")))
+
+        runner._poll_watch(_watch_row())
+
+        restore_fn.assert_not_called()
+        with runner._pending_delivery_lock:
+            assert "abc123" in runner._pending_delivery
+        _a, kwargs = storage.update_watch.call_args
+        assert set(kwargs) == {"next_poll", "poll_count"}
+
+    def test_forget_terminal_dispatched_clears_held_reminder(self):
+        # User-cancel takes the row out of the due view; its held reminder
+        # must be dropped too or it would leak (never re-polled).
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        runner = self._make_runner(storage)
+        runner._poll_watch(_watch_row())
+        with runner._pending_delivery_lock:
+            assert "abc123" in runner._pending_delivery
+
+        runner.forget_terminal_dispatched("abc123")
+
+        with runner._pending_delivery_lock:
+            assert "abc123" not in runner._pending_delivery
+
+    def test_abandon_write_failure_keeps_hold_for_write_retry(self):
+        # Reads-succeed/writes-fail storage (e.g. disk-full SQLite): a
+        # failed abandon commit must keep the hold so the next tick retries
+        # the WRITE via the redeliver path — the clear-first order let the
+        # row re-list into a fresh COMMAND RUN every attempt-budget cycle,
+        # forever, with the poll budget never advancing.
+        from turnstone.core.watch import WatchWorkstreamUnrestorable
+
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        runner = self._make_runner(storage)
+        runner._poll_watch(_watch_row())  # fire → hold (write still OK here)
+        with runner._pending_delivery_lock:
+            pending = dict(runner._pending_delivery["abc123"])
+
+        runner._restore_fn = MagicMock(  # type: ignore[assignment]
+            side_effect=WatchWorkstreamUnrestorable("ws-1")
+        )
+        storage.update_watch.side_effect = RuntimeError("disk full")
+
+        with pytest.raises(RuntimeError):
+            runner._redeliver_pending(_watch_row(), pending)
+
+        with runner._pending_delivery_lock:
+            assert "abc123" in runner._pending_delivery  # hold survived
+
+    def test_unrestorable_abandon_write_failure_keeps_hold(self):
+        # Fresh-fire permanent failure whose deactivation write fails must
+        # not strand the still-active row into a fresh command run every
+        # tick: the stash routes the next tick into the redeliver path,
+        # which retries the WRITE — never the command.
+        from turnstone.core.watch import WatchWorkstreamUnrestorable
+
+        storage = MagicMock()
+        storage.update_watch.side_effect = RuntimeError("disk full")
+        restore_fn = MagicMock(side_effect=WatchWorkstreamUnrestorable("ws-1"))
+        runner = self._make_runner(storage, restore_fn=restore_fn)
+
+        with pytest.raises(RuntimeError):
+            runner._poll_watch(_watch_row())
+
+        with runner._pending_delivery_lock:
+            assert "abc123" in runner._pending_delivery  # hold survived
+
+        # Next tick: the write retries and lands; the command never re-runs.
+        runner._run_command = MagicMock(  # type: ignore[method-assign]
+            side_effect=AssertionError("command must not re-run")
+        )
+        storage.update_watch.side_effect = None
+        storage.update_watch.return_value = True
+
+        runner._poll_watch(_watch_row())
+
+        runner._run_command.assert_not_called()
+        _a, kwargs = storage.update_watch.call_args
+        assert kwargs["active"] is False  # deactivation landed on the retry
+        with runner._pending_delivery_lock:
+            assert "abc123" not in runner._pending_delivery
+
+    def test_exhaustion_write_failure_keeps_hold_for_write_retry(self):
+        # Same pathology on the transient-exhaustion branch: the charge
+        # commit failing must keep the hold (write retried next tick), not
+        # drop it into a fresh command cycle with the budget never durable.
+        from turnstone.core.watch import MAX_DELIVERY_ATTEMPTS
+
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        runner = self._make_runner(storage)  # no fn, no restore → transient
+        runner._poll_watch(_watch_row())  # fire → hold
+        with runner._pending_delivery_lock:
+            runner._pending_delivery["abc123"]["attempts"] = MAX_DELIVERY_ATTEMPTS - 1
+            pending = dict(runner._pending_delivery["abc123"])
+
+        storage.update_watch.side_effect = RuntimeError("disk full")
+
+        with pytest.raises(RuntimeError):
+            runner._redeliver_pending(_watch_row(), pending)
+
+        with runner._pending_delivery_lock:
+            assert "abc123" in runner._pending_delivery  # hold survived
+
+
+class TestWatchRunnerCancelRace:
+    """User-cancel racing the poll pool: the delivery paths re-check the
+    row's active state so a cancelled watch can neither deliver nor leak a
+    held reminder, and the tick sweep mops up the one interleaving the
+    point checks can't reach (a stash landing after the cancel path's
+    ``forget_terminal_dispatched`` already cleared)."""
+
+    def _make_runner(self, storage: Any, **kwargs: Any) -> WatchRunner:
+        return WatchRunner(
+            storage=storage,
+            node_id="test-node",
+            check_interval=0.1,
+            tool_timeout=5,
+            **kwargs,
+        )
+
+    def test_hold_dropped_when_watch_cancelled_mid_fire(self):
+        # Cancel lands while the fire's command is running: the hold path
+        # re-checks the row and DROPS instead of stashing — an inactive row
+        # never re-lists, so a stash here would leak for the process
+        # lifetime with nothing ever retrying or clearing it.
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        storage.is_watch_active.return_value = False
+        runner = self._make_runner(storage)  # no fn, no restore → would hold
+
+        runner._poll_watch(_watch_row())
+
+        with runner._pending_delivery_lock:
+            assert "abc123" not in runner._pending_delivery
+        # No retry-cadence commit either — the row already left the view.
+        storage.update_watch.assert_not_called()
+
+    def test_redelivery_dropped_when_watch_cancelled(self):
+        # Cancel lands between the due listing and the redelivery dispatch:
+        # deliver nothing (the model must not act on — nor a restore be
+        # spawned for — a watch the user just cancelled) and drop the hold.
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        runner = self._make_runner(storage)
+        runner._poll_watch(_watch_row())  # fails → holds (row still active)
+        with runner._pending_delivery_lock:
+            assert "abc123" in runner._pending_delivery
+
+        storage.is_watch_active.return_value = False  # user cancels
+        dispatch_fn = MagicMock()
+        runner.set_dispatch_fn("ws-1", dispatch_fn)  # ws even came back live
+
+        runner._poll_watch(_watch_row())
+
+        dispatch_fn.assert_not_called()
+        with runner._pending_delivery_lock:
+            assert "abc123" not in runner._pending_delivery
+        # The only row write remains the initial hold's cadence commit —
+        # no terminal commit lands over the cancel's row state.
+        assert storage.update_watch.call_count == 1
+
+    def test_tick_sweeps_cancelled_holds(self):
+        # The residual interleaving: a stash that landed AFTER the cancel's
+        # forget_terminal_dispatched cleared (its active re-check passed
+        # just before the cancel's row write).  The sweep drops it within
+        # one tick.
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        storage.list_due_watches.return_value = []
+        runner = self._make_runner(storage)
+        runner._stash_pending_delivery("abc123", {"text": "x"}, {"active": False}, attempts=1)
+        storage.is_watch_active.return_value = False  # row already cancelled
+
+        runner._tick()
+
+        with runner._pending_delivery_lock:
+            assert "abc123" not in runner._pending_delivery
+
+    def test_tick_sweep_keeps_active_holds(self):
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        storage.list_due_watches.return_value = []
+        runner = self._make_runner(storage)
+        runner._stash_pending_delivery("abc123", {"text": "x"}, {"active": False}, attempts=1)
+        storage.is_watch_active.return_value = True
+
+        runner._tick()
+
+        with runner._pending_delivery_lock:
+            assert "abc123" in runner._pending_delivery
+
+    def test_active_checks_bias_toward_delivery_on_storage_error(self):
+        # is_watch_active RAISING must not drop a fire: the sweep keeps the
+        # hold and the delivery paths proceed (bounded by their own attempt
+        # and poll budgets) — a storage blip is not a cancellation.
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        storage.list_due_watches.return_value = []
+        storage.is_watch_active.side_effect = RuntimeError("storage down")
+        runner = self._make_runner(storage)
+        runner._stash_pending_delivery("abc123", {"text": "x"}, {"active": False}, attempts=1)
+
+        runner._tick()  # sweep: biased active → kept
+
+        with runner._pending_delivery_lock:
+            assert "abc123" in runner._pending_delivery
+
+
+class TestWatchRunnerRestoreSerialization:
+    """Two watches on ONE evicted workstream, polled concurrently, must
+    trigger the restore path at most once — otherwise each spawns a live
+    auto-approved session racing writes into one conversation history."""
+
+    def test_concurrent_same_ws_restores_once(self):
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        restore_calls: list[str] = []
+        calls_lock = threading.Lock()
+
+        runner = WatchRunner(
+            storage=storage,
+            node_id="n",
+            check_interval=0.1,
+            tool_timeout=5,
+            restore_fn=lambda ws_id: _slow_restore(runner, ws_id, restore_calls, calls_lock),
+        )
+
+        reminder = {"type": "watch_triggered", "text": "x"}
+        barrier = threading.Barrier(2)
+        results: list[bool] = []
+        results_lock = threading.Lock()
+
+        def call(wid: str) -> None:
+            barrier.wait(timeout=2.0)
+            ok = runner._dispatch_result("ws-shared", reminder, wid)
+            with results_lock:
+                results.append(ok)
+
+        threads = [threading.Thread(target=call, args=(f"w{i}",)) for i in range(2)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join(timeout=3.0)
+
+        # Exactly one restore ran; the winner delivered (True) and the other
+        # DEFERRED (False, holds + re-delivers next tick) rather than blocking
+        # its poll slot on the in-flight restore or restoring a second time.
+        assert restore_calls == ["ws-shared"]
+        assert sorted(results) == [False, True]
+        # Admission slot released after the restore.
+        with runner._restore_lock:
+            assert "ws-shared" not in runner._restoring
+
+    def test_wedged_restore_admissions_defer_and_alert(self, caplog):
+        # Admission entries older than RESTORE_STALL_ALERT_SECS are alerted
+        # on but NEVER evicted: the wedged poll thread's pool slot is never
+        # released, so reclaiming its admission would just readmit a restore
+        # that can wedge another pool thread on the same cause — trading
+        # this capped degraded state (restores blocked, polling intact) for
+        # total poll-pool collapse.  New restores keep deferring; the error
+        # log is the operator's restart signal.
+        from turnstone.core.watch import RESTORE_STALL_ALERT_SECS
+
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        restore_fn = MagicMock(return_value=MagicMock())
+        runner = WatchRunner(
+            storage=storage,
+            node_id="n",
+            check_interval=0.1,
+            tool_timeout=5,
+            restore_fn=restore_fn,
+        )
+        stalled_at = time.monotonic() - RESTORE_STALL_ALERT_SECS - 1
+        with runner._restore_lock:
+            runner._restoring["wedged-1"] = stalled_at
+            runner._restoring["wedged-2"] = stalled_at  # both slots wedged
+
+        with caplog.at_level("ERROR"):
+            result = runner._dispatch_result("ws-new", {"text": "x"}, "w1")
+
+        assert result is False  # wedged capacity stays consumed → defer
+        restore_fn.assert_not_called()
+        with runner._restore_lock:
+            assert "wedged-1" in runner._restoring
+            assert "wedged-2" in runner._restoring
+        assert any("watch_runner.restore_admission_wedged" in r.message for r in caplog.records)
+
+
+class TestWatchRunnerConcurrency:
+    """The tick thread only enumerates due rows; polls run on bounded
+    daemon threads.  Pins: genuine concurrency, per-watch in-flight
+    dedup, saturation leaving rows due (not dropped), and ``stop``
+    draining in-flight polls."""
+
+    def _make_runner(self, rows: list[dict[str, Any]], **kwargs: Any) -> WatchRunner:
+        storage = MagicMock()
+        storage.update_watch.return_value = True
+        storage.list_due_watches.return_value = rows
+        return WatchRunner(
+            storage=storage,
+            node_id="test-node",
+            check_interval=0.1,
+            tool_timeout=5,
+            **kwargs,
+        )
+
+    @staticmethod
+    def _wait_in_flight_empty(runner: WatchRunner, timeout: float = 3.0) -> None:
+        def _drained() -> bool:
+            with runner._in_flight_lock:
+                return not runner._in_flight
+
+        wait_until(_drained, timeout=timeout)
+
+    def test_tick_polls_concurrently(self):
+        rows = [_watch_row(watch_id=f"w{i}", ws_id=f"ws-{i}") for i in range(3)]
+        runner = self._make_runner(rows)
+
+        all_in = threading.Event()
+        release = threading.Event()
+        barrier = threading.Barrier(3)
+
+        def fake_poll(_row: dict[str, Any]) -> None:
+            # All three poll threads must be inside simultaneously for the
+            # barrier to trip — serial execution would deadlock here (and
+            # fail via the barrier timeout instead).
+            barrier.wait(timeout=2.0)
+            all_in.set()
+            release.wait(timeout=2.0)
+
+        runner._poll_watch = fake_poll  # type: ignore[method-assign]
+        runner._tick()
+
+        assert all_in.wait(timeout=2.0), "polls did not run concurrently"
+        release.set()
+        self._wait_in_flight_empty(runner)
+
+    def test_tick_skips_in_flight_watch(self):
+        rows = [_watch_row(watch_id="w0", ws_id="ws-0")]
+        runner = self._make_runner(rows)
+        polled: list[str] = []
+        runner._poll_watch = lambda row: polled.append(row["watch_id"])  # type: ignore[method-assign]
+
+        # Simulate a slow poll from a previous tick still running.
+        with runner._in_flight_lock:
+            runner._in_flight.add("w0")
+
+        runner._tick()
+
+        assert polled == []
+        # The foreign in-flight entry was not clobbered by the skip.
+        with runner._in_flight_lock:
+            assert "w0" in runner._in_flight
+
+    def test_tick_saturation_leaves_rows_due(self):
+        rows = [_watch_row(watch_id=f"w{i}", ws_id=f"ws-{i}") for i in range(2)]
+        runner = self._make_runner(rows, max_concurrent_polls=1)
+
+        started = threading.Event()
+        release = threading.Event()
+        polled: list[str] = []
+
+        def fake_poll(row: dict[str, Any]) -> None:
+            polled.append(row["watch_id"])
+            started.set()
+            release.wait(timeout=2.0)
+
+        runner._poll_watch = fake_poll  # type: ignore[method-assign]
+        runner._tick()
+        assert started.wait(timeout=2.0)
+
+        # Only the first row got a slot this tick; the second stays due
+        # for the next tick rather than being dropped.
+        assert polled == ["w0"]
+        release.set()
+        self._wait_in_flight_empty(runner)
+
+        # Next tick (slot free again) picks up the remaining row.
+        runner._storage.list_due_watches.return_value = [rows[1]]
+        runner._tick()
+        self._wait_in_flight_empty(runner)
+        assert polled == ["w0", "w1"]
+
+    def test_stop_waits_for_in_flight_polls(self):
+        rows = [_watch_row(watch_id="w0", ws_id="ws-0")]
+        runner = self._make_runner(rows)
+
+        started = threading.Event()
+        release = threading.Event()
+
+        def fake_poll(_row: dict[str, Any]) -> None:
+            started.set()
+            release.wait(timeout=3.0)
+
+        runner._poll_watch = fake_poll  # type: ignore[method-assign]
+        runner._tick()
+        assert started.wait(timeout=2.0)
+
+        stopper = threading.Thread(target=runner.stop, daemon=True)
+        stopper.start()
+        # stop() must be draining (poll still pinned), not returned.
+        time.sleep(0.15)
+        assert stopper.is_alive(), "stop() returned while a poll was in flight"
+
+        release.set()
+        stopper.join(timeout=3.0)
+        assert not stopper.is_alive()
+        with runner._in_flight_lock:
+            assert not runner._in_flight

tests/test_watch_dispatch.py

@@ -54,9 +54,11 @@ def _make_session_for_dispatch(**kwargs: Any) -> ChatSession:
     return ChatSession(**defaults)
 
 
-def _register_runner(session: ChatSession) -> tuple[Any, Any]:
+def _register_runner(session: ChatSession, wake_fn: Any = None) -> tuple[Any, Any]:
     """Attach a minimal stub ``WatchRunner`` to *session* and return the
     ``(runner, dispatch_fn)`` pair captured by ``set_dispatch_fn``.
+    ``wake_fn`` rides through to ``set_watch_runner`` (default ``None``
+    matches the pre-wake wiring most tests here exercise).
     """
     captured: dict[str, Any] = {}
 
@@ -65,7 +67,7 @@ def _register_runner(session: ChatSession) -> tuple[Any, Any]:
             captured["fn"] = fn
 
     runner = _StubRunner()
-    session.set_watch_runner(runner)
+    session.set_watch_runner(runner, wake_fn=wake_fn)
     return runner, captured["fn"]
 
 
@@ -400,3 +402,216 @@ class TestMetadataPropagation:
         assert len(snapshot) == 1
         _nt, _text, meta = snapshot[0]
         assert meta is None
+
+
+# ---------------------------------------------------------------------------
+# Wake trigger
+# ---------------------------------------------------------------------------
+
+
+class TestWakeFn:
+    """``set_watch_runner``'s optional ``wake_fn`` fires once per enqueued
+    dispatch — AFTER the entry lands — so a watch firing on an
+    already-idle workstream (no IDLE transition for the
+    ``IdleNudgeWatcher`` to observe) can spawn the wake worker that
+    drains it.  Failures are contained: the enqueue must survive a
+    raising ``wake_fn``, because a propagated raise would abort
+    ``WatchRunner._poll_watch`` before the watch-row update commits and
+    re-fire the same reminder every subsequent tick.
+    """
+
+    def test_wake_fn_called_after_enqueue(self, tmp_db):
+        session = _make_session_for_dispatch()
+        depth_at_wake: list[int] = []
+        _runner, dispatch = _register_runner(
+            session, wake_fn=lambda: depth_at_wake.append(len(session._nudge_queue))
+        )
+
+        dispatch(_reminder("watch fired body"), "watch-1")
+
+        # Fired exactly once, and the entry was already queued when it ran
+        # — the wake worker's drain must be able to see the fresh entry.
+        assert depth_at_wake == [1]
+
+    def test_wake_fn_not_called_when_payload_sanitizes_empty(self, tmp_db):
+        """A fire whose payload strips to nothing enqueues nothing — and
+        must not wake anything either (a wake with an empty queue would
+        just spawn a worker that no-ops at the drain guard)."""
+        session = _make_session_for_dispatch()
+        wake = MagicMock()
+        _runner, dispatch = _register_runner(session, wake_fn=wake)
+
+        dispatch(_reminder("\x07\x0b\x7f"), "watch-1")
+
+        assert len(session._nudge_queue) == 0
+        wake.assert_not_called()
+
+    def test_wake_fn_exception_is_contained(self, tmp_db, caplog):
+        session = _make_session_for_dispatch()
+        wake = MagicMock(side_effect=RuntimeError("boom"))
+        _runner, dispatch = _register_runner(session, wake_fn=wake)
+
+        with caplog.at_level("WARNING"):
+            dispatch(_reminder("body"), "watch-1")  # must not raise
+
+        # Entry survived; the failure surfaced as a warning, not a raise
+        # up into the poll loop.
+        assert len(session._nudge_queue) == 1
+        assert any("watch_dispatch.wake_failed" in r.message for r in caplog.records), (
+            "expected a watch_dispatch.wake_failed warning record"
+        )
+
+
+class _RecordingRunner:
+    """Stub WatchRunner recording registration/removal order.  Mirrors the
+    production owner-checked removal semantics — the resume tail passes
+    ``owner`` and peeks ``get_dispatch_fn`` before re-registering."""
+
+    def __init__(self) -> None:
+        self.events: list[tuple[str, str]] = []
+        self.fns: dict[str, Any] = {}
+
+    def set_dispatch_fn(self, ws_id: str, fn: Any) -> None:
+        self.events.append(("set", ws_id))
+        self.fns[ws_id] = fn
+
+    def get_dispatch_fn(self, ws_id: str) -> Any:
+        return self.fns.get(ws_id)
+
+    def remove_dispatch_fn(self, ws_id: str, owner: Any = None) -> None:
+        if owner is not None and self.fns.get(ws_id) is not owner:
+            return
+        self.events.append(("remove", ws_id))
+        self.fns.pop(ws_id, None)
+
+
+class TestResumeReRegistration:
+    """A non-fork ``resume()`` rebinds ``_ws_id``; the dispatch
+    registration must FOLLOW that identity — otherwise watches stamped
+    with the adopted id never find the live session, and every fire
+    takes the restore path, spawning a duplicate auto-approved session
+    racing writes into the same conversation (CLI ``--resume`` and the
+    ``/resume`` command both hit this)."""
+
+    def _saved_ws(self, ws_id: str) -> None:
+        from turnstone.core.memory import register_workstream, save_message
+
+        register_workstream(ws_id)
+        save_message(ws_id, "user", "hi")
+
+    def test_nonfork_resume_moves_registration_to_adopted_id(self, tmp_db):
+        self._saved_ws("resume-target")
+        session = _make_session_for_dispatch()
+        old_id = session._ws_id
+        runner = _RecordingRunner()
+        session.set_watch_runner(runner, wake_fn=None)
+
+        assert session.resume("resume-target") is True
+
+        # New key live BEFORE the old key is removed — a fire during the
+        # transition can never observe an empty registry (which would
+        # divert it to the restore path).
+        assert runner.events == [
+            ("set", old_id),
+            ("set", "resume-target"),
+            ("remove", old_id),
+        ]
+        assert set(runner.fns) == {"resume-target"}
+
+    def test_fork_resume_keeps_registration(self, tmp_db):
+        self._saved_ws("fork-src")
+        session = _make_session_for_dispatch()
+        old_id = session._ws_id
+        runner = _RecordingRunner()
+        session.set_watch_runner(runner, wake_fn=None)
+
+        assert session.resume("fork-src", fork=True) is True
+
+        # Fork keeps its own identity — registration untouched.
+        assert runner.events == [("set", old_id)]
+
+    def test_resume_without_runner_is_noop(self, tmp_db):
+        # CLI --resume / restore-fn shape: resume() runs BEFORE any
+        # set_watch_runner call — nothing to re-register, nothing raises.
+        self._saved_ws("resume-bare")
+        session = _make_session_for_dispatch()
+
+        assert session.resume("resume-bare") is True
+        assert session._watch_runner is None
+
+    def test_reregistered_closure_keeps_wake_fn(self, tmp_db):
+        # The stored wake_fn rides the re-registration: a watch firing on
+        # the ADOPTED id must still wake the workstream.
+        self._saved_ws("resume-wake")
+        session = _make_session_for_dispatch()
+        runner = _RecordingRunner()
+        wake = MagicMock()
+        session.set_watch_runner(runner, wake_fn=wake)
+
+        assert session.resume("resume-wake") is True
+
+        runner.fns["resume-wake"](_reminder("watch output"), "w1")
+        assert len(session._nudge_queue) == 1
+        wake.assert_called_once()
+
+    def test_resume_does_not_steal_another_live_registration(self, tmp_db):
+        # In-session /resume of a workstream that is OPEN IN ANOTHER PANE
+        # (a degenerate two-live-sessions state): the original owner keeps
+        # its watch fires — the adopter neither clobbers the target's
+        # registration nor (on a later resume-away or close) deletes it.
+        self._saved_ws("shared-A")
+        self._saved_ws("other-C")
+        runner = _RecordingRunner()
+
+        pane_a = _make_session_for_dispatch()
+        pane_a._ws_id = "shared-A"  # pane A opened A and registered
+        pane_a.set_watch_runner(runner, wake_fn=None)
+        fn_a = runner.fns["shared-A"]
+
+        pane_b = _make_session_for_dispatch()
+        pane_b.set_watch_runner(runner, wake_fn=None)
+
+        assert pane_b.resume("shared-A") is True
+        # Pane A's registration survived the adoption…
+        assert runner.fns["shared-A"] is fn_a
+
+        assert pane_b.resume("other-C") is True
+        # …and the resume-away removed only pane B's own (absent) claim.
+        assert runner.fns["shared-A"] is fn_a
+        assert "other-C" in runner.fns
+
+    def test_new_command_moves_registration_to_fresh_id(self, tmp_db):
+        # /new is the other identity rebind: watches created AFTER it stamp
+        # the fresh id and must reach this session, while the old
+        # workstream's fires must stop landing in a conversation that no
+        # longer shows them (they divert to the restore path instead).
+        session = _make_session_for_dispatch()
+        old_id = session._ws_id
+        runner = _RecordingRunner()
+        session.set_watch_runner(runner, wake_fn=None)
+
+        # handle_command's return means "should exit" — /new never exits.
+        assert session.handle_command("/new") is False
+
+        assert session._ws_id != old_id
+        assert set(runner.fns) == {session._ws_id}
+        assert ("remove", old_id) in runner.events
+
+    def test_close_removes_only_own_registration(self, tmp_db):
+        # A watch-restore shell and a reopened pane can serve one ws_id in
+        # sequence; the shell's later teardown must not unregister the pane.
+        self._saved_ws("shared-W")
+        runner = _RecordingRunner()
+
+        shell = _make_session_for_dispatch()
+        shell._ws_id = "shared-W"
+        shell.set_watch_runner(runner, wake_fn=None)
+
+        pane = _make_session_for_dispatch()
+        pane._ws_id = "shared-W"
+        pane.set_watch_runner(runner, wake_fn=None)  # pane re-registers (last writer)
+        pane_fn = runner.fns["shared-W"]
+
+        shell.close()  # shell reaped (close_idle / eviction)
+
+        assert runner.fns.get("shared-W") is pane_fn  # pane still registered

tests/test_watch_integration.py

@@ -340,8 +340,8 @@ def test_poll_watch_terminal_fire_survives_drain(
     monkeypatch.setattr(session._nudge_queue, "enqueue", _spy_enqueue)
 
     # For the max_polls=1 case the first poll has prev_output=None and
-    # would not normally fire on output change; the max_polls branch
-    # at watch.py:412-414 still marks is_final=True so dispatch runs.
+    # would not normally fire on output change; _poll_watch's max_polls
+    # branch still marks is_final=True so dispatch runs.
     due = storage.list_due_watches("2099-01-01T00:00:00")
     matching = [r for r in due if r["watch_id"] == f"w-regression-{label}"]
     assert len(matching) == 1, f"watch row not picked up by list_due_watches: {due!r}"