fix(session-worker): release the slot claim when Thread.start itself fails

If thread creation raised (thread exhaustion, MemoryError), the
dispatcher had already claimed the worker slot under ws._lock — but the
flag's only clearer is _runner's finally, on a thread that never
started. The workstream then looked idle forever (no state change ever
fired) while every subsequent dispatch took the reuse path into a queue
no worker would drain, until an operator force-cancel.

Roll the claim back under the lock (identity-guarded, like _runner's
own clear, so a concurrent force-cancel's successor is never clobbered)
and re-raise. Re-raise rather than return False: callers' crash paths —
the deferred-send drain's per-iteration handler with its backoff — are
shaped for exceptions, and a False would masquerade as queue-full
backpressure and mislabel the wake gate's refusal log. worker_kind is
left stale, as documented (every reader conjoins _worker_running).

Affected every dispatch path: sends, wakes, retries, the deferred-send
drain, and workstream init.

CHANGELOG.md

@@ -175,6 +175,16 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen.
 
 ### Fixed
 
+- **A failed worker-thread spawn no longer wedges the workstream.** If
+  `Thread.start()` itself raised (thread exhaustion, out-of-memory), the
+  dispatcher had already claimed the worker slot but the flag's only
+  clearer lived in the never-started thread — the workstream looked idle
+  forever while every subsequent message queued behind a worker that
+  didn't exist, until an operator force-cancel. The claim is now rolled
+  back under the lock and the error propagates, so the workstream is
+  dispatchable again as soon as resources recover. Affected every
+  dispatch path (sends, wakes, retries, deferred-send drain, init).
+
 - **Manual `/compact` from the web UI: no phantom user turn, no frozen
   server, cancellable.** A slash command typed into the web composer no
   longer renders as a user chat bubble (it echoes as a distinct command

tests/test_session_worker.py

@@ -146,6 +146,45 @@ def test_enqueue_unexpected_exception_returns_false_logged() -> None:
     assert ws._worker_running is True
 
 
+def test_spawn_failure_releases_slot_and_reraises(monkeypatch) -> None:
+    """``Thread.start`` raising (thread exhaustion, MemoryError) must not
+    wedge the slot: the claim ``(worker_thread, _worker_running)`` taken
+    under the lock is rolled back and the exception propagates.  Without
+    the rollback the flag's only clearer is a finally on a thread that
+    never started — the workstream looks idle forever (no state change
+    ever fired) while every dispatch takes the reuse path into a queue
+    no worker will drain, until an operator force-cancel."""
+    import pytest
+
+    session = _SendSession()
+    ws = _make_ws(session)
+
+    class _ExhaustedThread(threading.Thread):
+        def start(self) -> None:
+            raise RuntimeError("can't start new thread")
+
+    class _ThreadNS:
+        # Shim only what session_worker.send touches; patching the real
+        # threading module's Thread attribute would break every other
+        # test's spawns.
+        Thread = _ExhaustedThread
+        current_thread = staticmethod(threading.current_thread)
+
+    monkeypatch.setattr(session_worker, "threading", _ThreadNS)
+    with pytest.raises(RuntimeError, match="can't start new thread"):
+        _send_message(ws, session, "doomed")
+    assert ws._worker_running is False
+    assert ws.worker_thread is None
+    assert session.send_calls == []
+
+    # The slot is genuinely reusable once resources recover.
+    monkeypatch.setattr(session_worker, "threading", threading)
+    assert _send_message(ws, session, "recovered") is True
+    ws.worker_thread.join(timeout=2.0)
+    assert session.send_calls == ["recovered"]
+    assert ws._worker_running is False
+
+
 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 —

turnstone/core/session_worker.py

@@ -145,6 +145,12 @@ def send(
         caller surfaces 429) or any other exception (logged). Falling
         through to spawn a second worker on a full queue would corrupt
         ChatSession state.
+
+    Raises:
+        Whatever ``Thread.start()`` raised when the spawn itself fails
+        (thread exhaustion, ``MemoryError``) — the slot claim is rolled
+        back first, so the workstream stays dispatchable once resources
+        recover instead of wedging behind a flag no thread will clear.
     """
     name = thread_name or f"session-worker-{ws.id[:8]}"
 
@@ -233,5 +239,27 @@ def send(
     # ``t.start()`` may run user code (worker body) before returning;
     # keep it outside the lock to avoid pinning ``ws._lock`` for the
     # full thread-creation cost.
-    t.start()
+    try:
+        t.start()
+    except Exception:
+        # Thread creation failed (RuntimeError under thread exhaustion,
+        # MemoryError): the slot was claimed under the lock above, but the
+        # flag's normal clearer is ``_runner``'s finally — on a thread
+        # that will never run.  Without this release, ``_worker_running``
+        # stays True forever on a workstream that LOOKS idle (the worker
+        # never fired a state change), every future dispatch takes the
+        # reuse path into a queue with no consumer, and only an operator
+        # force-cancel unwedges it.  Identity-guarded like ``_runner``'s
+        # clear: a concurrent force-cancel may already have cleared or
+        # replaced the slot, and this must not clobber a live successor.
+        # Re-raise rather than return False: callers' crash paths (the
+        # deferred-send drain's per-iteration handler with its backoff)
+        # are shaped for exceptions, and a False here would masquerade as
+        # queue-full backpressure.
+        with ws._lock:
+            if ws.worker_thread is t:
+                ws.worker_thread = None
+                ws._worker_running = False
+        log.exception("session_worker.spawn_failed ws=%s — slot released", ws.id[:8])
+        raise
     return True