chore: bump version to 1.5.3

Three issues from the Copilot review on PR #457:

1. SQLite race in bulk_close_stale_orphans (Copilot): the SELECT-then-
   UPDATE flow doesn't re-apply the eligibility predicates on the
   UPDATE, so a row that gets touch_workstream-bumped (or set_state-
   transitioned) between the two statements would still be flipped
   to closed.  Postgres dodges this via UPDATE...RETURNING (one atomic
   statement); SQLite needs the explicit re-application.  Fix: rebuild
   the WHERE conditions list once, apply on both SELECT and UPDATE,
   then SELECT-back by ``state='closed' AND updated=now`` to get the
   accurate closed-id list.  A row that became fresh between the two
   statements skips the UPDATE entirely.

2. SQLite IN-clause bind-parameter limit (Copilot): default 999 cap
   could be exceeded on a backlog reap (e.g. after a long outage).
   Chunked the candidate id list at 500 — same chunk size
   prune_workstreams (line 453) uses for the same reason.

3. Wall-clock-dependent test asserts (Copilot, two locations): the
   tests asserted ``updated > '2024-01-01T00:00:00'`` which is fragile
   on systems with skewed clocks or pre-2024 dates.  Replaced with
   ``updated != stale_seed`` — captures the same intent (the value
   was bumped) without depending on wall-clock date.

Two ``...``-as-no-op flags from github-code-quality were false
positives — ``...`` is the standard Python idiom for Protocol method
bodies and matches every other method in _protocol.py.  No code change.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "turnstone"
-version = "1.5.2"
+version = "1.5.3"
 description = "Multi-node AI orchestration platform with tool use, agent routing, and cluster simulation."
 readme = "README.md"
 license = "BUSL-1.1"

tests/test_console_idle_cleanup.py

@@ -0,0 +1,106 @@
+"""Tests for the console's coordinator idle-cleanup thread helper.
+
+The helper itself is a tiny loop wrapping ``mgr.close_idle``; the heavy
+lifting is in ``SessionManager.close_idle`` (covered in
+``test_session_manager.py``) and ``bulk_close_stale_orphans`` (covered
+in ``test_storage_sqlite.py``).  These tests verify the glue:
+
+- the helper runs an initial sweep BEFORE its first sleep (cold-start
+  cleanup without blocking the lifespan),
+- the helper swallows exceptions so a transient DB blip can't kill the
+  daemon thread,
+- the helper exits cleanly when ``stop_event`` is set.
+
+The ``stop_event`` parameter is exclusively for tests — production
+callers pass ``None`` and the daemon runs for process lifetime.
+"""
+
+from __future__ import annotations
+
+import threading
+from unittest.mock import patch
+
+from turnstone.console.server import _coord_idle_cleanup_thread
+
+
+class _StubMgr:
+    def __init__(
+        self, *, stop_event: threading.Event, expected_calls: int, raise_after: int = -1
+    ) -> None:
+        self.calls: list[float] = []
+        self.sleep_calls_at_each_close: list[int] = []
+        self._stop_event = stop_event
+        self._expected = expected_calls
+        self._raise_after = raise_after
+        self._sleep_count = 0
+
+    def close_idle(self, timeout_sec: float) -> list[str]:
+        # Snapshot how many sleeps preceded this close — lets the
+        # "initial sweep" test verify the first close_idle ran with
+        # zero preceding sleeps.
+        self.sleep_calls_at_each_close.append(self._sleep_count)
+        self.calls.append(timeout_sec)
+        try:
+            if 0 <= self._raise_after < len(self.calls):
+                raise RuntimeError("simulated DB blip")
+        finally:
+            # Set stop after the helper has been exercised enough,
+            # regardless of whether this call raised.
+            if len(self.calls) >= self._expected:
+                self._stop_event.set()
+        return []
+
+    def record_sleep(self, _seconds: float) -> None:
+        self._sleep_count += 1
+
+
+def _run_until_done(mgr: _StubMgr, stop_event: threading.Event, timeout_sec: float) -> None:
+    with patch("turnstone.console.server.time.sleep", mgr.record_sleep):
+        thread = threading.Thread(
+            target=_coord_idle_cleanup_thread,
+            args=(mgr, timeout_sec, stop_event),
+            daemon=True,
+        )
+        thread.start()
+        thread.join(timeout=2.0)
+        assert not thread.is_alive(), "helper failed to exit on stop_event"
+
+
+def test_coord_idle_cleanup_runs_initial_sweep_before_sleep() -> None:
+    """The first close_idle call must happen BEFORE the first time.sleep —
+    otherwise cold-start orphans wait one ``check_every`` interval (~30 min
+    on default 2h timeout) for the first reap.  Crucial because the
+    lifespan no longer does a synchronous initial sweep."""
+    stop_event = threading.Event()
+    mgr = _StubMgr(stop_event=stop_event, expected_calls=1)
+    _run_until_done(mgr, stop_event, timeout_sec=120.0)
+    assert mgr.sleep_calls_at_each_close == [0], "first close_idle should run before any sleep"
+
+
+def test_coord_idle_cleanup_calls_close_idle_each_tick() -> None:
+    stop_event = threading.Event()
+    mgr = _StubMgr(stop_event=stop_event, expected_calls=3)
+    _run_until_done(mgr, stop_event, timeout_sec=120.0)
+    assert len(mgr.calls) == 3
+    assert all(t == 120.0 for t in mgr.calls)
+
+
+def test_coord_idle_cleanup_survives_close_idle_exceptions() -> None:
+    """A transient DB error must not kill the daemon thread — the next
+    tick should still fire close_idle.  Without the try/except, a single
+    blip would silently leak orphans forever."""
+    stop_event = threading.Event()
+    mgr = _StubMgr(stop_event=stop_event, expected_calls=4, raise_after=1)
+    _run_until_done(mgr, stop_event, timeout_sec=120.0)
+    # All four calls must have fired despite calls 2-4 raising.
+    assert len(mgr.calls) == 4
+
+
+def test_coord_idle_cleanup_exits_cleanly_on_stop_event() -> None:
+    """The stop_event mechanism is the test contract; verify the thread
+    actually exits when the event is set, without needing exceptions or
+    daemon-process termination."""
+    stop_event = threading.Event()
+    mgr = _StubMgr(stop_event=stop_event, expected_calls=2)
+    _run_until_done(mgr, stop_event, timeout_sec=120.0)
+    assert stop_event.is_set()

tests/test_session_manager.py

@@ -18,13 +18,19 @@ from __future__ import annotations
 import threading
 import time
 from dataclasses import dataclass
+from datetime import UTC, datetime
 from typing import Any
 from unittest.mock import MagicMock
 
 import pytest
 
 from turnstone.core.session_manager import SessionKindAdapter, SessionManager
-from turnstone.core.workstream import Workstream, WorkstreamKind, WorkstreamState
+from turnstone.core.workstream import (
+    BULK_CLOSE_STATE_VALUES,
+    Workstream,
+    WorkstreamKind,
+    WorkstreamState,
+)
 
 # ---------------------------------------------------------------------------
 # Test fixtures
@@ -156,6 +162,8 @@ class _Row:
     kind: str
     state: str = "idle"
     parent_ws_id: str | None = None
+    updated: str = ""
+    node_id: str | None = None
 
 
 class FakeStorage:
@@ -164,8 +172,19 @@ class FakeStorage:
     def __init__(self) -> None:
         self.rows: dict[str, _Row] = {}
         self.state_updates: list[tuple[str, str]] = []
+        self.touch_calls: list[str] = []
         self.register_raises = False
         self.lock = threading.Lock()
+        # Live-services lookup target for close_idle pass 2.  Map
+        # service_type → list of live service_ids.  Tests that exercise
+        # liveness scoping populate this directly; default empty means
+        # "no peers alive" (every row unprotected by liveness).
+        self.live_services: dict[str, list[str]] = {}
+        self.list_services_raises = False
+
+    @staticmethod
+    def _now_iso() -> str:
+        return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
 
     def register_workstream(
         self,
@@ -178,6 +197,8 @@ class FakeStorage:
         parent_ws_id: str | None = None,
         skill_id: str = "",
         skill_version: int = 0,
+        state: str = "idle",
+        updated: str | None = None,
     ) -> None:
         if self.register_raises:
             raise RuntimeError("register forced failure")
@@ -188,14 +209,66 @@ class FakeStorage:
                 user_id=user_id or "",
                 name=name,
                 kind=kind_str,
+                state=state,
                 parent_ws_id=parent_ws_id,
+                updated=updated if updated is not None else self._now_iso(),
+                node_id=node_id,
             )
 
+    def touch_workstream(self, ws_id: str) -> None:
+        with self.lock:
+            self.touch_calls.append(ws_id)
+            if ws_id in self.rows:
+                self.rows[ws_id].updated = self._now_iso()
+
     def update_workstream_state(self, ws_id: str, state: str) -> None:
         with self.lock:
             self.state_updates.append((ws_id, state))
             if ws_id in self.rows:
                 self.rows[ws_id].state = state
+                self.rows[ws_id].updated = self._now_iso()
+
+    def bulk_close_stale_orphans(
+        self,
+        kind: WorkstreamKind | str,
+        cutoff: str,
+        exclude_ws_ids: list[str],
+        live_node_ids: list[str] | None = None,
+    ) -> list[str]:
+        kind_str = kind.value if isinstance(kind, WorkstreamKind) else str(kind)
+        excluded = set(exclude_ws_ids)
+        live_set = set(live_node_ids) if live_node_ids else set()
+        now = self._now_iso()
+        closed: list[str] = []
+        with self.lock:
+            for ws_id, row in self.rows.items():
+                if (
+                    row.kind == kind_str
+                    and row.state in BULK_CLOSE_STATE_VALUES
+                    and row.updated < cutoff
+                    and ws_id not in excluded
+                ):
+                    # Liveness gate: when live_node_ids was provided AND
+                    # non-empty, protect rows whose owner is in the live
+                    # set.  NULL node_id is always eligible.  When
+                    # live_node_ids is None or empty, no protection
+                    # (mirror of the real backends).
+                    if live_node_ids and row.node_id is not None and row.node_id in live_set:
+                        continue
+                    row.state = "closed"
+                    row.updated = now
+                    self.state_updates.append((ws_id, "closed"))
+                    closed.append(ws_id)
+        return closed
+
+    def list_services(self, service_type: str, max_age_seconds: int = 120) -> list[dict[str, str]]:
+        if self.list_services_raises:
+            raise RuntimeError("list_services forced failure")
+        with self.lock:
+            return [
+                {"service_id": sid, "service_type": service_type}
+                for sid in self.live_services.get(service_type, [])
+            ]
 
     def get_workstream(self, ws_id: str) -> dict[str, Any] | None:
         with self.lock:
@@ -228,6 +301,7 @@ def _make_manager(
     max_active: int = 5,
     storage: FakeStorage | None = None,
     event_emitter: Any = _EMITTER_DEFAULT,
+    node_id: str | None = None,
 ) -> tuple[SessionManager, FakeAdapter, FakeStorage]:
     """Build a SessionManager wired to a FakeAdapter for both Protocols.
 
@@ -246,6 +320,7 @@ def _make_manager(
         storage=storage,
         max_active=max_active,
         event_emitter=emitter,
+        node_id=node_id,
     )
     return mgr, adapter, storage
 
@@ -579,6 +654,24 @@ def test_open_resurrects_closed_state() -> None:
     assert ws_id in [e.ws_id for e in adapter.events_of("rehydrated")]
 
 
+def test_open_touches_workstream_on_rehydrate() -> None:
+    """Rehydrating a workstream must bump its ``updated`` so a concurrent
+    close_idle pass-2 in this same process can't clobber the freshly-loaded
+    row to ``closed`` because its DB ``updated`` is older than the cutoff.
+    The touch is best-effort (try/except in open()) but must fire on the
+    happy path."""
+    mgr, _, storage = _make_manager()
+    ws = mgr.create(user_id="u1")
+    ws_id = ws.id
+    mgr.close(ws_id)
+    storage.touch_calls.clear()  # only care about touches from rehydrate
+
+    reopened = mgr.open(ws_id)
+
+    assert reopened is not None
+    assert ws_id in storage.touch_calls
+
+
 def test_open_ignores_owner_mismatch() -> None:
     # Turnstone is a trusted-team tool; row-level ownership is
     # metadata, not an access boundary.  ``open`` no longer cares
@@ -827,6 +920,197 @@ def test_close_idle_on_empty_manager_returns_empty_list() -> None:
     assert mgr.close_idle(max_age_seconds=1.0) == []
 
 
+def test_close_idle_runs_db_orphan_pass() -> None:
+    """DB rows of this kind that aren't loaded into the manager get
+    bulk-closed when their ``updated`` is older than the cutoff.  Catches
+    the orphan-after-process-restart case the original close_idle missed."""
+    mgr, _, storage = _make_manager()
+    # Orphan rows live in storage but were never loaded via mgr.create.
+    storage.register_workstream(
+        "orphan-1",
+        kind=WorkstreamKind.INTERACTIVE,
+        updated="2020-01-01T00:00:00",
+    )
+    storage.register_workstream(
+        "orphan-2",
+        kind=WorkstreamKind.INTERACTIVE,
+        state="thinking",
+        updated="2020-01-01T00:00:00",
+    )
+
+    closed = mgr.close_idle(max_age_seconds=0.0)
+
+    assert set(closed) == {"orphan-1", "orphan-2"}
+    assert ("orphan-1", "closed") in storage.state_updates
+    assert ("orphan-2", "closed") in storage.state_updates
+    assert storage.rows["orphan-1"].state == "closed"
+    assert storage.rows["orphan-2"].state == "closed"
+
+
+def test_close_idle_excludes_loaded_workstreams_from_db_pass() -> None:
+    """A workstream loaded into memory must NOT be reaped by the DB
+    orphan pass even when its storage ``updated`` is stale — the
+    in-memory pass owns those.  Verifies the exclude_ws_ids plumbing."""
+    mgr, _, storage = _make_manager()
+    ws = mgr.create(user_id="u1")
+    # Force the storage row's ``updated`` to look stale.  In practice
+    # ``set_state`` would bump it, but we're simulating a long-running
+    # active workstream whose updated drifted older than the cutoff.
+    storage.rows[ws.id].updated = "2020-01-01T00:00:00"
+
+    # Huge timeout so the in-memory IDLE pass skips it (stays loaded).
+    closed = mgr.close_idle(max_age_seconds=10_000.0)
+
+    assert ws.id not in closed
+    assert mgr.get(ws.id) is not None
+    assert storage.rows[ws.id].state == "idle"
+
+
+def test_close_idle_filters_db_orphans_by_kind() -> None:
+    """An interactive manager's close_idle must not touch coordinator
+    rows in storage and vice versa.  Without this filter, both managers
+    would race to close each other's rows."""
+    mgr, _, storage = _make_manager()  # interactive by default
+    storage.register_workstream(
+        "coord-orphan",
+        kind=WorkstreamKind.COORDINATOR,
+        updated="2020-01-01T00:00:00",
+    )
+    storage.register_workstream(
+        "interactive-orphan",
+        kind=WorkstreamKind.INTERACTIVE,
+        updated="2020-01-01T00:00:00",
+    )
+
+    closed = mgr.close_idle(max_age_seconds=0.0)
+
+    assert "interactive-orphan" in closed
+    assert "coord-orphan" not in closed
+    assert storage.rows["coord-orphan"].state == "idle"
+    assert storage.rows["interactive-orphan"].state == "closed"
+
+
+def test_close_idle_protects_rows_owned_by_live_services() -> None:
+    """Multi-node correctness: rows whose ``node_id`` matches a service
+    with a recent heartbeat must NOT be reaped, even when *this* manager
+    is on a different node — the alive peer may legitimately have them
+    loaded.  Liveness is the rendezvous router's primitive (post-PR-#384);
+    using it here keeps reap scoping aligned with routing.
+
+    Default ``_make_manager`` uses an INTERACTIVE adapter, which derives
+    ``service_type='server'`` — so live_services seeded under "server"
+    are what the manager queries."""
+    mgr, _, storage = _make_manager()
+    storage.live_services["server"] = ["node-b"]  # only node-b is alive
+    storage.register_workstream(
+        "ours-from-dead-node",
+        node_id="node-a",  # dead pod (not in live_services)
+        kind=WorkstreamKind.INTERACTIVE,
+        updated="2020-01-01T00:00:00",
+    )
+    storage.register_workstream(
+        "theirs-still-alive",
+        node_id="node-b",
+        kind=WorkstreamKind.INTERACTIVE,
+        updated="2020-01-01T00:00:00",
+    )
+
+    closed = mgr.close_idle(max_age_seconds=0.0)
+
+    assert closed == ["ours-from-dead-node"]
+    assert storage.rows["ours-from-dead-node"].state == "closed"
+    assert storage.rows["theirs-still-alive"].state == "idle"
+
+
+def test_close_idle_protects_live_services_for_coordinator_kind() -> None:
+    """Coord-side parity: a coordinator manager derives
+    ``service_type='console'``, so live_services seeded under "console"
+    are what gets queried.  Mirrors the interactive test to ensure both
+    halves of the production wiring are exercised."""
+    coord_adapter = FakeAdapter(kind=WorkstreamKind.COORDINATOR)
+    mgr, _, storage = _make_manager(coord_adapter)
+    storage.live_services["console"] = ["console"]  # console is alive
+    storage.register_workstream(
+        "alive-console-coord",
+        node_id="console",
+        kind=WorkstreamKind.COORDINATOR,
+        updated="2020-01-01T00:00:00",
+    )
+    storage.register_workstream(
+        "dead-console-coord",
+        node_id="dead-console-instance",  # not in live set
+        kind=WorkstreamKind.COORDINATOR,
+        updated="2020-01-01T00:00:00",
+    )
+
+    closed = mgr.close_idle(max_age_seconds=0.0)
+
+    assert closed == ["dead-console-coord"]
+    assert storage.rows["alive-console-coord"].state == "idle"
+    assert storage.rows["dead-console-coord"].state == "closed"
+
+
+def test_close_idle_reaps_rows_with_null_node_id() -> None:
+    """A row with no ``node_id`` has no owner identity — age alone gates
+    the reap.  Defends against a NULL silently propagating through ``NOT
+    IN (live)`` and protecting orphans forever."""
+    mgr, _, storage = _make_manager()
+    storage.live_services["server"] = ["node-a"]
+    storage.register_workstream(
+        "no-owner",
+        node_id=None,
+        kind=WorkstreamKind.INTERACTIVE,
+        updated="2020-01-01T00:00:00",
+    )
+
+    closed = mgr.close_idle(max_age_seconds=0.0)
+
+    assert closed == ["no-owner"]
+
+
+def test_close_idle_reaps_all_orphans_when_no_peers_alive() -> None:
+    """When ``list_services`` returns an empty list (no heartbeating
+    peers), every stale orphan is unprotected and gets reaped.  This is
+    the cold-start / single-process / dead-cluster-recovery case."""
+    mgr, _, storage = _make_manager()
+    # storage.live_services["server"] left empty — no peers heartbeating
+    storage.register_workstream(
+        "any-node-1",
+        node_id="node-a",
+        kind=WorkstreamKind.INTERACTIVE,
+        updated="2020-01-01T00:00:00",
+    )
+    storage.register_workstream(
+        "any-node-2",
+        node_id="node-b",
+        kind=WorkstreamKind.INTERACTIVE,
+        updated="2020-01-01T00:00:00",
+    )
+
+    closed = mgr.close_idle(max_age_seconds=0.0)
+
+    assert set(closed) == {"any-node-1", "any-node-2"}
+
+
+def test_close_idle_skips_pass_2_when_list_services_fails() -> None:
+    """Conservative fallback: if list_services fails we can't enumerate
+    live owners safely, so pass 2 must skip rather than reap blind.  Pass
+    1 (in-memory IDLE) still runs."""
+    mgr, _, storage = _make_manager()
+    storage.list_services_raises = True
+    storage.register_workstream(
+        "would-be-orphan",
+        node_id="node-a",
+        kind=WorkstreamKind.INTERACTIVE,
+        updated="2020-01-01T00:00:00",
+    )
+
+    closed = mgr.close_idle(max_age_seconds=0.0)
+
+    assert closed == []
+    assert storage.rows["would-be-orphan"].state == "idle"
+
+
 def test_list_all_returns_creation_order() -> None:
     mgr, _, _ = _make_manager()
     a = mgr.create(user_id="u1")

tests/test_storage_sqlite.py

@@ -4,6 +4,10 @@ from __future__ import annotations
 
 from typing import Any
 
+import sqlalchemy as sa
+
+from turnstone.core.storage._schema import workstreams
+
 # -- Workstream registration ---------------------------------------------------
 
 
@@ -664,6 +668,277 @@ class TestBatchPrimitives:
         assert result == {"never-seen": 0}
 
 
+# -- bulk_close_stale_orphans --------------------------------------------------
+
+
+def _force_updated(backend: Any, ws_id: str, updated: str) -> None:
+    """Stamp a workstream row's ``updated`` column directly.
+
+    The public surface only sets ``updated`` to ``now``, which makes it
+    impossible to fabricate a stale row through register/update calls.
+    Reaches into ``backend._engine`` — same access pattern conftest uses
+    for cross-backend cleanup.
+    """
+    with backend._engine.connect() as conn:
+        conn.execute(
+            sa.update(workstreams).where(workstreams.c.ws_id == ws_id).values(updated=updated)
+        )
+        conn.commit()
+
+
+class TestBulkCloseStaleOrphans:
+    def test_closes_stale_non_terminal_rows_of_kind(self, backend):
+        backend.register_workstream("stale-idle", kind="interactive")
+        backend.register_workstream("stale-thinking", kind="interactive")
+        backend.update_workstream_state("stale-thinking", "thinking")
+        backend.register_workstream("fresh-idle", kind="interactive")
+        _force_updated(backend, "stale-idle", "2020-01-01T00:00:00")
+        _force_updated(backend, "stale-thinking", "2020-01-01T00:00:00")
+        # fresh-idle stays at registration time (effectively now)
+
+        closed = backend.bulk_close_stale_orphans(
+            "interactive", cutoff="2024-01-01T00:00:00", exclude_ws_ids=[]
+        )
+
+        assert set(closed) == {"stale-idle", "stale-thinking"}
+        rows = backend.get_workstreams_batch(["stale-idle", "stale-thinking", "fresh-idle"])
+        assert rows["stale-idle"]["state"] == "closed"
+        assert rows["stale-thinking"]["state"] == "closed"
+        assert rows["fresh-idle"]["state"] == "idle"
+
+    def test_skips_already_closed(self, backend):
+        backend.register_workstream("already-closed", kind="interactive")
+        backend.update_workstream_state("already-closed", "closed")
+        _force_updated(backend, "already-closed", "2020-01-01T00:00:00")
+
+        closed = backend.bulk_close_stale_orphans(
+            "interactive", cutoff="2024-01-01T00:00:00", exclude_ws_ids=[]
+        )
+
+        assert closed == []
+
+    def test_filters_by_kind(self, backend):
+        backend.register_workstream("interactive-stale", kind="interactive")
+        backend.register_workstream("coord-stale", kind="coordinator")
+        _force_updated(backend, "interactive-stale", "2020-01-01T00:00:00")
+        _force_updated(backend, "coord-stale", "2020-01-01T00:00:00")
+
+        closed = backend.bulk_close_stale_orphans(
+            "interactive", cutoff="2024-01-01T00:00:00", exclude_ws_ids=[]
+        )
+
+        assert closed == ["interactive-stale"]
+        rows = backend.get_workstreams_batch(["interactive-stale", "coord-stale"])
+        assert rows["interactive-stale"]["state"] == "closed"
+        assert rows["coord-stale"]["state"] == "idle"
+
+    def test_excludes_loaded_ws_ids(self, backend):
+        backend.register_workstream("ws-keep", kind="interactive")
+        backend.register_workstream("ws-close", kind="interactive")
+        _force_updated(backend, "ws-keep", "2020-01-01T00:00:00")
+        _force_updated(backend, "ws-close", "2020-01-01T00:00:00")
+
+        closed = backend.bulk_close_stale_orphans(
+            "interactive", cutoff="2024-01-01T00:00:00", exclude_ws_ids=["ws-keep"]
+        )
+
+        assert closed == ["ws-close"]
+        rows = backend.get_workstreams_batch(["ws-keep", "ws-close"])
+        assert rows["ws-keep"]["state"] == "idle"
+        assert rows["ws-close"]["state"] == "closed"
+
+    def test_empty_exclude_list_does_not_break_sql(self, backend):
+        backend.register_workstream("orphan", kind="interactive")
+        _force_updated(backend, "orphan", "2020-01-01T00:00:00")
+
+        closed = backend.bulk_close_stale_orphans(
+            "interactive", cutoff="2024-01-01T00:00:00", exclude_ws_ids=[]
+        )
+
+        assert closed == ["orphan"]
+
+    def test_no_orphans_returns_empty(self, backend):
+        backend.register_workstream("fresh", kind="interactive")
+
+        closed = backend.bulk_close_stale_orphans(
+            "interactive", cutoff="2024-01-01T00:00:00", exclude_ws_ids=[]
+        )
+
+        assert closed == []
+
+    def test_closes_all_non_terminal_states(self, backend):
+        for ws_id, state in [
+            ("o-idle", "idle"),
+            ("o-thinking", "thinking"),
+            ("o-attention", "attention"),
+            ("o-running", "running"),
+        ]:
+            backend.register_workstream(ws_id, kind="interactive")
+            if state != "idle":
+                backend.update_workstream_state(ws_id, state)
+            _force_updated(backend, ws_id, "2020-01-01T00:00:00")
+
+        closed = backend.bulk_close_stale_orphans(
+            "interactive", cutoff="2024-01-01T00:00:00", exclude_ws_ids=[]
+        )
+
+        assert set(closed) == {"o-idle", "o-thinking", "o-attention", "o-running"}
+
+    def test_bumps_updated_on_close(self, backend):
+        stale_updated = "2020-01-01T00:00:00"
+        backend.register_workstream("orphan", kind="interactive")
+        _force_updated(backend, "orphan", stale_updated)
+
+        backend.bulk_close_stale_orphans(
+            "interactive", cutoff="2024-01-01T00:00:00", exclude_ws_ids=[]
+        )
+
+        # ``updated`` must change away from the forced stale value.  Asserting
+        # inequality from the seed (rather than ``> "2024-01-01..."``) keeps
+        # the test independent of wall-clock date.
+        with backend._engine.connect() as conn:
+            row = conn.execute(
+                sa.select(workstreams.c.updated).where(workstreams.c.ws_id == "orphan")
+            ).one()
+        assert row[0] != stale_updated
+
+    def test_protects_rows_owned_by_live_services(self, backend):
+        """Liveness scoping (post-#384 rendezvous-routing world): rows
+        whose ``node_id`` matches a heartbeating service must NOT be
+        reaped, because that owner may legitimately have them loaded on
+        another worker.  Rows whose ``node_id`` matches a dead service
+        ARE eligible — that's how dead-pod orphans get reclaimed in
+        containerized deployments with dynamic hostnames."""
+        backend.register_workstream("dead-node", node_id="dead-pod-x4k2", kind="interactive")
+        backend.register_workstream("alive-node", node_id="alive-pod-y9p3", kind="interactive")
+        _force_updated(backend, "dead-node", "2020-01-01T00:00:00")
+        _force_updated(backend, "alive-node", "2020-01-01T00:00:00")
+
+        closed = backend.bulk_close_stale_orphans(
+            "interactive",
+            cutoff="2024-01-01T00:00:00",
+            exclude_ws_ids=[],
+            live_node_ids=["alive-pod-y9p3"],
+        )
+
+        assert closed == ["dead-node"]
+        rows = backend.get_workstreams_batch(["dead-node", "alive-node"])
+        assert rows["dead-node"]["state"] == "closed"
+        assert rows["alive-node"]["state"] == "idle"
+
+    def test_null_node_id_always_eligible(self, backend):
+        """A row with NULL ``node_id`` has no owner identity — age alone
+        gates the reap.  Belt-and-suspenders against ``NULL NOT IN (...)``
+        evaluating to NULL (not TRUE) and silently protecting orphans
+        forever."""
+        backend.register_workstream("no-owner", node_id=None, kind="interactive")
+        _force_updated(backend, "no-owner", "2020-01-01T00:00:00")
+
+        closed = backend.bulk_close_stale_orphans(
+            "interactive",
+            cutoff="2024-01-01T00:00:00",
+            exclude_ws_ids=[],
+            live_node_ids=["some-other-node"],
+        )
+
+        assert closed == ["no-owner"]
+
+    def test_live_node_ids_none_skips_filter(self, backend):
+        """``live_node_ids=None`` is the single-process / operator-backfill
+        mode — all rows of *kind* are eligible regardless of node_id."""
+        backend.register_workstream("node-a", node_id="node-a", kind="interactive")
+        backend.register_workstream("node-b", node_id="node-b", kind="interactive")
+        _force_updated(backend, "node-a", "2020-01-01T00:00:00")
+        _force_updated(backend, "node-b", "2020-01-01T00:00:00")
+
+        closed = backend.bulk_close_stale_orphans(
+            "interactive", cutoff="2024-01-01T00:00:00", exclude_ws_ids=[]
+        )
+
+        assert set(closed) == {"node-a", "node-b"}
+
+    def test_empty_live_node_ids_treats_all_as_dead(self, backend):
+        """Empty list ``live_node_ids=[]`` means "no nodes alive" — every
+        row's owner is unprotected.  Useful for operator scripts that
+        want to reap regardless of liveness."""
+        backend.register_workstream("any", node_id="node-a", kind="interactive")
+        _force_updated(backend, "any", "2020-01-01T00:00:00")
+
+        closed = backend.bulk_close_stale_orphans(
+            "interactive",
+            cutoff="2024-01-01T00:00:00",
+            exclude_ws_ids=[],
+            live_node_ids=[],
+        )
+
+        assert closed == ["any"]
+
+    def test_combines_live_node_ids_and_exclude_ws_ids(self, backend):
+        """Both filters stack as AND clauses on the UPDATE.  Covers the
+        full 2x2 matrix to catch a future edit that replaces an AND with
+        an OR or drops one of the filters: only the (orphan + dead-node)
+        cell should be reaped."""
+        # All four registered with the same stale ``updated``.
+        for ws_id, node in [
+            ("loaded-alive", "alive-node"),
+            ("loaded-dead", "dead-node"),
+            ("orphan-alive", "alive-node"),
+            ("orphan-dead", "dead-node"),
+        ]:
+            backend.register_workstream(ws_id, node_id=node, kind="interactive")
+            _force_updated(backend, ws_id, "2020-01-01T00:00:00")
+
+        closed = backend.bulk_close_stale_orphans(
+            "interactive",
+            cutoff="2024-01-01T00:00:00",
+            exclude_ws_ids=["loaded-alive", "loaded-dead"],
+            live_node_ids=["alive-node"],
+        )
+
+        # Only orphan-dead is unprotected by both filters.
+        assert closed == ["orphan-dead"]
+        rows = backend.get_workstreams_batch(
+            ["loaded-alive", "loaded-dead", "orphan-alive", "orphan-dead"]
+        )
+        assert rows["loaded-alive"]["state"] == "idle"
+        assert rows["loaded-dead"]["state"] == "idle"
+        assert rows["orphan-alive"]["state"] == "idle"
+        assert rows["orphan-dead"]["state"] == "closed"
+
+
+# -- touch_workstream ----------------------------------------------------------
+
+
+class TestTouchWorkstream:
+    def test_bumps_updated_only(self, backend):
+        """Used by ``open()`` on rehydrate to defend against the orphan
+        reaper clobbering a freshly-loaded row.  Must not change ``state``
+        (the open() path explicitly avoids state writes to dodge a race
+        with concurrent close())."""
+        stale_updated = "2020-01-01T00:00:00"
+        backend.register_workstream("ws-touch", kind="interactive")
+        backend.update_workstream_state("ws-touch", "closed")  # simulate prior close
+        _force_updated(backend, "ws-touch", stale_updated)
+
+        backend.touch_workstream("ws-touch")
+
+        with backend._engine.connect() as conn:
+            row = conn.execute(
+                sa.select(workstreams.c.state, workstreams.c.updated).where(
+                    workstreams.c.ws_id == "ws-touch"
+                )
+            ).one()
+        assert row[0] == "closed", "state must not be modified by touch"
+        # Compare against the forced stale value rather than a fixed calendar
+        # date so the test is independent of wall-clock time.
+        assert row[1] != stale_updated, "updated must be bumped"
+
+    def test_unknown_id_is_noop(self, backend):
+        """Touch on a missing id must not raise — open()'s exception
+        handler is best-effort."""
+        backend.touch_workstream("nonexistent")  # must not raise
+
+
 # -- Lifecycle -----------------------------------------------------------------
 
 

turnstone/__init__.py

@@ -1,3 +1,3 @@
 """turnstone - Multi-node AI orchestration platform with tool use, agent routing, and cluster simulation."""
 
-__version__ = "1.5.2"
+__version__ = "1.5.3"

turnstone/console/server.py

@@ -23,6 +23,7 @@ import queue
 import re
 import secrets
 import textwrap
+import threading
 import time
 import urllib.parse
 from concurrent.futures import ThreadPoolExecutor
@@ -89,6 +90,7 @@ if TYPE_CHECKING:
     from starlette.requests import Request
 
     from turnstone.core.session import ChatSession
+    from turnstone.core.session_manager import SessionManager
     from turnstone.core.storage._protocol import StorageBackend
 
 log = logging.getLogger("turnstone.console.server")
@@ -3687,6 +3689,50 @@ async def _verify_collector_service_scope(app: Starlette, client: httpx.AsyncCli
     )
 
 
+def _coord_idle_cleanup_thread(
+    mgr: SessionManager,
+    timeout_sec: float,
+    stop_event: threading.Event | None = None,
+) -> None:
+    """Periodically reap idle + DB-orphan coordinator workstreams.
+
+    Mirrors the regular server's ``_idle_cleanup_thread`` (turnstone/server.py)
+    but skips the rate-limiter / global-queue arms — the console doesn't have
+    those.  ``mgr.close_idle`` does the work: closes loaded IDLE rows AND
+    bulk-closes DB rows of this kind whose ``updated`` is past the cutoff
+    and which aren't currently loaded.  The latter pass catches coords left
+    behind by prior console process incarnations.
+
+    Runs an initial sweep BEFORE the first sleep so cold-start orphans are
+    reaped immediately rather than waiting one ``check_every`` interval (~30
+    min on default 2h timeout).  This intentionally diverges from the regular
+    server pattern, which has no initial sweep — the regular server runs
+    inside a normal request-handling lifecycle, the console-side coord pool
+    is a small fixed-size cache where orphans dominate the row count after
+    a cold boot.
+
+    ``stop_event`` is for tests — when set, the thread exits cleanly after
+    the next loop check.  Production callers pass ``None`` (the daemon is
+    process-lifetime).
+    """
+    check_every = min(300.0, timeout_sec / 4)
+    # Initial sweep — runs once before entering the sleep loop.
+    try:
+        mgr.close_idle(timeout_sec)
+    except Exception:
+        log.debug("console.coord_idle_cleanup_initial_failed", exc_info=True)
+    while True:
+        if stop_event is not None and stop_event.is_set():
+            return
+        time.sleep(check_every)
+        if stop_event is not None and stop_event.is_set():
+            return
+        try:
+            mgr.close_idle(timeout_sec)
+        except Exception:
+            log.debug("console.coord_idle_cleanup_failed", exc_info=True)
+
+
 @asynccontextmanager
 async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]:
     # Create async HTTP clients for proxy routes.  Auth headers are NOT baked
@@ -3992,6 +4038,27 @@ async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]:
                     coord_adapter.start_child_event_fanout(app.state.collector)
                 except Exception:
                     log.warning("console.coordinator_child_fanout_init_failed", exc_info=True)
+                # Idle cleanup: closes loaded-but-stale coords AND DB orphans
+                # left behind by prior console processes.  The thread runs an
+                # initial sweep on entry (no synchronous lifespan call needed —
+                # see ``_coord_idle_cleanup_thread``) so cold-start cleanup
+                # doesn't block startup.  Reuses the regular-server
+                # ``server.workstream_idle_timeout`` setting — the same cadence
+                # makes sense for both kinds and avoids a redundant config knob.
+                try:
+                    idle_minutes = int(config_store.get("server.workstream_idle_timeout"))
+                except Exception:
+                    idle_minutes = 0
+                if idle_minutes > 0:
+                    timeout_sec = float(idle_minutes * 60)
+                    cleanup_thread = threading.Thread(
+                        target=_coord_idle_cleanup_thread,
+                        args=(coord_mgr, timeout_sec),
+                        name="coord-idle-cleanup",
+                        daemon=True,
+                    )
+                    cleanup_thread.start()
+                    app.state.coord_idle_cleanup_thread = cleanup_thread
                 log.info(
                     "console.coordinator_mgr_ready max_active=%s",
                     config_store.get("coordinator.max_active"),

turnstone/core/session_manager.py

@@ -13,6 +13,7 @@ import contextlib
 import threading
 import time
 import uuid
+from datetime import UTC, datetime, timedelta
 from typing import TYPE_CHECKING, Any, Protocol
 
 from turnstone.core.log import get_logger
@@ -28,6 +29,22 @@ if TYPE_CHECKING:
 log = get_logger(__name__)
 
 
+# Maps each workstream kind to the ``services.service_type`` its hosting
+# process registers under.  Used by ``SessionManager.close_idle`` pass 2
+# to enumerate live peer processes for orphan-reaper liveness scoping.
+# Server processes register as ``("server", node_id, ...)`` (see
+# ``turnstone/server.py``); the console process as ``("console",
+# "console", ...)`` (see ``turnstone/console/server.py``).  Deriving from
+# kind here removes a duplicated-config footgun: any caller that builds
+# a ``SessionManager`` automatically gets the correct service_type for
+# its kind, with no risk of miswiring INTERACTIVE→"console" or vice
+# versa.
+_KIND_SERVICE_TYPE: dict[WorkstreamKind, str] = {
+    WorkstreamKind.INTERACTIVE: "server",
+    WorkstreamKind.COORDINATOR: "console",
+}
+
+
 class SessionKindAdapter(Protocol):
     """Per-kind construction + cleanup policies the shared ``SessionManager`` delegates to.
 
@@ -217,6 +234,16 @@ class SessionManager:
     def kind(self) -> WorkstreamKind:
         return self._adapter.kind
 
+    @property
+    def _service_type(self) -> str | None:
+        """``services.service_type`` this manager's hosting process registers
+        under, derived from its ``kind``.  Used by ``close_idle`` pass 2 to
+        enumerate live peer processes.  Returns ``None`` for kinds that have
+        no production service mapping (only the two existing kinds map
+        today; ``None`` would be a marker for a future kind without a
+        clustered hosting model)."""
+        return _KIND_SERVICE_TYPE.get(self.kind)
+
     @property
     def count(self) -> int:
         with self._lock:
@@ -604,6 +631,22 @@ class SessionManager:
                 # last close(). The next set_state() call syncs it
                 # naturally; writing 'idle' here could race a concurrent
                 # close() that writes 'closed' under self._lock.
+                #
+                # Bump only ``updated`` (no state write) so this row's
+                # timestamp is fresh against the orphan-reaper cutoff —
+                # otherwise a concurrent close_idle pass-2 in this same
+                # process could clobber a freshly-rehydrated row whose
+                # ``updated`` is older than the cutoff.  The pure-
+                # timestamp write is safe against concurrent close()
+                # because close still wins on the state column.
+                try:
+                    self._storage.touch_workstream(ws_id)
+                except Exception:
+                    log.debug(
+                        "session_mgr.touch_workstream_failed ws=%s",
+                        ws_id[:8],
+                        exc_info=True,
+                    )
                 if self._event_emitter is not None:
                     self._event_emitter.emit_rehydrated(ws)
                 return ws
@@ -804,15 +847,52 @@ class SessionManager:
     def close_idle(self, max_age_seconds: float) -> list[str]:
         """Close IDLE workstreams inactive for more than ``max_age_seconds``.
 
-        Returns the list of closed ws_ids. Unlike the old WSM version,
-        this does NOT skip the last workstream — the default-startup
-        relic is gone, callers can handle the 0-workstream case.
+        Two-pass shape:
+
+        - Pass 1 (in-memory): close loaded ``IDLE`` rows whose
+          ``ws.last_active`` (monotonic) is past timeout.  Closes only
+          ``IDLE`` so legitimately-attentive rows (waiting for user
+          response) stay live.
+        - Pass 2 (DB orphans): bulk-close DB rows of this manager's
+          kind whose ``updated`` is past the wall-clock cutoff and
+          which are not currently loaded.  This catches workstreams
+          left behind by prior process incarnations — a process crash
+          /restart leaves rows in non-terminal states forever
+          otherwise.  Closes ``idle/thinking/attention/running``
+          because any matching row is by definition not loaded by any
+          live process and cannot be in a live interaction.
+
+          **Liveness scoping** (the rendezvous router's primitive
+          since PR #384): when ``self._service_type`` resolves to a
+          known service type — both production kinds do — pass 2
+          calls ``storage.list_services`` to enumerate peer processes
+          with recent heartbeats and protects rows whose ``node_id``
+          matches a live ``service_id`` from reap, even when *this*
+          manager is on a different node.  This is essential for
+          containerized deployments with dynamic hostnames: dead-pod
+          rows fall out of the live set after the heartbeat window
+          and become reapable; alive-pod rows stay protected as long
+          as the owner heartbeats.  A future kind with no service
+          registration would resolve ``_service_type`` to ``None``
+          and skip the live-services lookup (single-process / CLI).
+
+          **Conservative fallback**: if ``list_services`` raises,
+          pass 2 is skipped entirely this tick — never reap when
+          liveness state is unknown.  Pass 1 still runs.  Next tick
+          retries the lookup.
+
+        Returns the combined list of closed ws_ids (in-memory first,
+        then DB orphans).  Pass 1 emits ``ws_closed``; pass 2 does
+        not, because never-loaded rows have no SSE listeners
+        expecting them.
 
         Atomic pop per victim under ``self._lock`` (bug-5): a pending
         tool result can flip state IDLE→RUNNING between the snapshot
         and the close, so the state test + pop must run together.
         Batches every pop under one ``self._lock`` acquisition (perf-5)
-        rather than locking once per victim.
+        rather than locking once per victim.  The DB pass runs OUTSIDE
+        ``self._lock`` — only a brief lock to snapshot loaded keys —
+        so a slow UPDATE doesn't block create/get/set_state.
         """
         now = time.monotonic()
         popped: list[Workstream] = []
@@ -853,6 +933,63 @@ class SessionManager:
             if self._event_emitter is not None:
                 self._event_emitter.emit_closed(ws.id, name=ws.name)
             closed_ids.append(ws.id)
+
+        # Pass 2: reap DB orphans of this kind older than the cutoff.
+        # Snapshot loaded keys under self._lock briefly so a concurrent
+        # create/load doesn't get its row clobbered by the UPDATE; release
+        # before the DB call.
+        #
+        # Liveness scoping uses ``services.last_heartbeat`` — the same
+        # primitive the rendezvous router (PR #384) uses for routing.  A
+        # row's ``node_id`` is stamped at create time and never updated;
+        # in containerized deployments with dynamic hostnames the dead
+        # pod's ``node_id`` points at a service that's no longer
+        # heartbeating, so the row falls through to reap.  Conversely,
+        # rows whose ``node_id`` matches a heartbeating service are
+        # protected even when *this* manager is on a different node —
+        # the alive peer may legitimately have them loaded.
+        with self._lock:
+            loaded = list(self._workstreams.keys())
+        cutoff = (datetime.now(UTC) - timedelta(seconds=max_age_seconds)).strftime(
+            "%Y-%m-%dT%H:%M:%S"
+        )
+        live_node_ids: list[str] | None = None
+        skip_pass_2 = False
+        if self._service_type is not None:
+            try:
+                live_services = self._storage.list_services(self._service_type)
+                live_node_ids = [
+                    str(svc["service_id"]) for svc in live_services if svc.get("service_id")
+                ]
+            except Exception:
+                # Conservative fallback: skip pass 2 entirely this tick
+                # so we can't accidentally reap rows whose owners we
+                # failed to enumerate.  Next tick retries.
+                log.debug(
+                    "session_mgr.list_services_failed kind=%s",
+                    self.kind.value,
+                    exc_info=True,
+                )
+                skip_pass_2 = True
+        orphans: list[str] = []
+        if not skip_pass_2:
+            try:
+                orphans = self._storage.bulk_close_stale_orphans(
+                    self.kind, cutoff, loaded, live_node_ids=live_node_ids
+                )
+            except Exception:
+                log.debug(
+                    "session_mgr.bulk_close_orphans_failed kind=%s",
+                    self.kind.value,
+                    exc_info=True,
+                )
+        if orphans:
+            log.info(
+                "session_mgr.bulk_close_orphans count=%d kind=%s",
+                len(orphans),
+                self.kind.value,
+            )
+        closed_ids.extend(orphans)
         return closed_ids
 
     def _close_if_idle_locked(self, ws_id: str) -> Workstream | None:

turnstone/core/storage/_postgresql.py

@@ -97,7 +97,7 @@ from turnstone.core.storage._utils import sanitize_text
 from turnstone.core.storage._utils import (
     scan_skill_content as _scan_skill_content,
 )
-from turnstone.core.workstream import WorkstreamKind
+from turnstone.core.workstream import BULK_CLOSE_STATE_VALUES, WorkstreamKind
 
 log = get_logger(__name__)
 
@@ -595,6 +595,57 @@ class PostgreSQLBackend:
             )
             conn.commit()
 
+    def bulk_close_stale_orphans(
+        self,
+        kind: WorkstreamKind | str,
+        cutoff: str,
+        exclude_ws_ids: list[str],
+        live_node_ids: list[str] | None = None,
+    ) -> list[str]:
+        norm_kind = WorkstreamKind(kind).value
+        now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
+        stmt = (
+            sa.update(workstreams)
+            .where(
+                workstreams.c.kind == norm_kind,
+                workstreams.c.state.in_(BULK_CLOSE_STATE_VALUES),
+                workstreams.c.updated < cutoff,
+            )
+            .values(state="closed", updated=now)
+            .returning(workstreams.c.ws_id)
+        )
+        # Protect rows whose owning process is still heartbeating in the
+        # services table (rendezvous router's liveness primitive).  NULL
+        # node_id rows have no owner identity — always eligible.  The
+        # ``and live_node_ids`` short-circuits both ``None`` (skip the
+        # filter entirely — single-process / operator backfill) and ``[]``
+        # (no nodes alive — every row unprotected, no extra predicate
+        # needed since absence equals match-all).
+        if live_node_ids is not None and live_node_ids:
+            stmt = stmt.where(
+                sa.or_(
+                    workstreams.c.node_id.is_(None),
+                    ~workstreams.c.node_id.in_(live_node_ids),
+                )
+            )
+        if exclude_ws_ids:
+            # Skip ``NOT IN ()`` when nothing to exclude — keeps the SQL clean
+            # and avoids SQLAlchemy's empty-collection warning.
+            stmt = stmt.where(~workstreams.c.ws_id.in_(exclude_ws_ids))
+        with self._conn() as conn:
+            result = conn.execute(stmt)
+            ids = [row[0] for row in result]
+            conn.commit()
+            return ids
+
+    def touch_workstream(self, ws_id: str) -> None:
+        now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
+        with self._conn() as conn:
+            conn.execute(
+                sa.update(workstreams).where(workstreams.c.ws_id == ws_id).values(updated=now)
+            )
+            conn.commit()
+
     def update_workstream_name(self, ws_id: str, name: str) -> None:
         now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
         with self._conn() as conn:

turnstone/core/storage/_protocol.py

@@ -431,6 +431,68 @@ class StorageBackend(Protocol):
         """Update a workstream's state and bump updated timestamp."""
         ...
 
+    def bulk_close_stale_orphans(
+        self,
+        kind: WorkstreamKind | str,
+        cutoff: str,
+        exclude_ws_ids: list[str],
+        live_node_ids: list[str] | None = None,
+    ) -> list[str]:
+        """Close DB-side workstream rows of *kind* whose state is in
+        ``BULK_CLOSE_STATE_VALUES`` and whose ``updated`` is lex-older than
+        *cutoff*, excluding rows currently loaded in memory.  Sets
+        ``state='closed'`` and bumps ``updated``.  Returns the list of ws_ids
+        actually transitioned.
+
+        ``cutoff`` is a UTC ``YYYY-MM-DDTHH:MM:SS`` string matching the on-disk
+        format ``update_workstream_state`` writes — lex compare is safe for
+        same-offset timestamps.  Empty ``exclude_ws_ids`` means no exclusion.
+
+        ``live_node_ids`` is the set of ``services.service_id`` values whose
+        ``last_heartbeat`` is recent (i.e. owning processes still alive);
+        rows whose ``node_id`` matches one of these are protected because
+        their owning process may legitimately have them loaded on another
+        worker.  ``None`` skips the filter entirely (single-process / tests
+        / operator backfill).  Empty list ``[]`` treats every node as dead —
+        useful when operator scripts want to reap regardless of liveness.
+
+        Rows with ``NULL`` ``node_id`` are always eligible: they have no
+        meaningful owner identity, so age alone gates the reap.
+
+        Liveness scoping replaces an earlier ``node_id == self`` heuristic.
+        That heuristic broke in the post-rendezvous-routing world (PR #384):
+        ``workstreams.node_id`` is stamped at create time and never updated,
+        so dead-pod orphans in containerized deployments with dynamic
+        hostnames couldn't be reclaimed.  ``services.last_heartbeat`` is the
+        rendezvous router's authoritative liveness primitive — using it here
+        keeps reap scoping aligned with routing.
+
+        Asymmetric with ``SessionManager.close_idle``'s in-memory pass on
+        purpose: that pass closes only ``IDLE`` (legitimately-attentive rows
+        stay), this method closes the broader ``BULK_CLOSE_STATE_VALUES`` set
+        because any row matching here is by definition not loaded by any
+        live process and cannot be in a live interaction.
+        """
+        ...
+
+    def touch_workstream(self, ws_id: str) -> None:
+        """Bump a workstream row's ``updated`` timestamp without touching its
+        state.
+
+        Used by ``SessionManager.open()`` on cold rehydrate so a freshly-
+        loaded row's ``updated`` can't be older than the orphan-reaper cutoff
+        — protects against a same-process race where a parallel
+        ``close_idle`` pass-2 snapshots loaded keys after the storage read
+        but before the in-memory install.  Distinct from
+        ``update_workstream_state(ws_id, current_state)`` because the
+        rehydrate path explicitly avoids a state write (see the
+        ``open()`` no-DB-state-flip-on-resurrect comment): a state write
+        could race a concurrent ``close()`` and resurrect a closed row.
+        Bumping only ``updated`` is safe — close still wins on the state
+        column.
+        """
+        ...
+
     def update_workstream_name(self, ws_id: str, name: str) -> None:
         """Update a workstream's display name."""
         ...

turnstone/core/storage/_sqlite.py

@@ -97,7 +97,7 @@ from turnstone.core.storage._utils import sanitize_text
 from turnstone.core.storage._utils import (
     scan_skill_content as _scan_skill_content,
 )
-from turnstone.core.workstream import WorkstreamKind
+from turnstone.core.workstream import BULK_CLOSE_STATE_VALUES, WorkstreamKind
 
 log = get_logger(__name__)
 
@@ -707,6 +707,87 @@ class SQLiteBackend:
             )
             conn.commit()
 
+    def bulk_close_stale_orphans(
+        self,
+        kind: WorkstreamKind | str,
+        cutoff: str,
+        exclude_ws_ids: list[str],
+        live_node_ids: list[str] | None = None,
+    ) -> list[str]:
+        norm_kind = WorkstreamKind(kind).value
+        now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
+        # SQLite has no RETURNING precedent in this file — do SELECT-then-
+        # UPDATE in one transaction, with the SAME WHERE predicates re-applied
+        # to the UPDATE.  Re-application defends against a same-process race:
+        # ``SessionManager.open()`` calls ``touch_workstream`` between the
+        # SELECT and the UPDATE could have bumped a row's ``updated`` past
+        # ``cutoff`` (or ``set_state`` could have flipped its state out of
+        # the bulk-close set).  Without the re-applied WHERE the UPDATE
+        # closes those rows anyway; with it, the UPDATE skips rows that
+        # became ineligible after the SELECT and the row stays open.
+        # Chunked through ``_in_chunks`` so the ``IN`` clause never exceeds
+        # SQLite's bind-parameter limit (default 999) on a large reap.
+        candidate_conditions = [
+            workstreams.c.kind == norm_kind,
+            workstreams.c.state.in_(BULK_CLOSE_STATE_VALUES),
+            workstreams.c.updated < cutoff,
+        ]
+        if live_node_ids is not None and live_node_ids:
+            # Protect rows owned by heartbeating services.  NULL node_id is
+            # always eligible.  Empty list means "no nodes alive" — every
+            # row is unprotected; the absence of this predicate is
+            # equivalent to "match all," so we just skip it.
+            candidate_conditions.append(
+                sa.or_(
+                    workstreams.c.node_id.is_(None),
+                    ~workstreams.c.node_id.in_(live_node_ids),
+                )
+            )
+        if exclude_ws_ids:
+            candidate_conditions.append(~workstreams.c.ws_id.in_(exclude_ws_ids))
+        select_stmt = sa.select(workstreams.c.ws_id).where(*candidate_conditions)
+        closed: list[str] = []
+        # Match the chunk size used by ``prune_workstreams`` (line 453) — keeps
+        # ``IN`` clauses well below SQLite's default 999-bind-param limit even
+        # on very large reaps.
+        chunk_size = 500
+        with self._conn() as conn:
+            candidate_ids = [row[0] for row in conn.execute(select_stmt)]
+            for i in range(0, len(candidate_ids), chunk_size):
+                chunk = candidate_ids[i : i + chunk_size]
+                # Re-apply the eligibility predicates on the UPDATE so a row
+                # that became fresh between the SELECT and the UPDATE is not
+                # clobbered.  Then SELECT back by ``state='closed' AND updated=now``
+                # to determine which rows actually transitioned this commit —
+                # the returned list reflects reality even when re-application
+                # filters out some candidates.
+                conn.execute(
+                    sa.update(workstreams)
+                    .where(workstreams.c.ws_id.in_(chunk), *candidate_conditions)
+                    .values(state="closed", updated=now)
+                )
+                actually_closed = [
+                    row[0]
+                    for row in conn.execute(
+                        sa.select(workstreams.c.ws_id).where(
+                            workstreams.c.ws_id.in_(chunk),
+                            workstreams.c.state == "closed",
+                            workstreams.c.updated == now,
+                        )
+                    )
+                ]
+                closed.extend(actually_closed)
+            conn.commit()
+            return closed
+
+    def touch_workstream(self, ws_id: str) -> None:
+        now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
+        with self._conn() as conn:
+            conn.execute(
+                sa.update(workstreams).where(workstreams.c.ws_id == ws_id).values(updated=now)
+            )
+            conn.commit()
+
     def update_workstream_name(self, ws_id: str, name: str) -> None:
         now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
         with self._conn() as conn:

turnstone/core/storage/migrations/versions/048_workstream_reaper_index.py

@@ -0,0 +1,88 @@
+"""Add a partial composite index for the orphan-reaper query.
+
+``StorageBackend.bulk_close_stale_orphans`` (introduced alongside the
+workstream-lifecycle leak fix) runs every ``min(300s, idle_timeout/4)``
+on every server and console process.  Its WHERE shape is:
+
+    WHERE kind = ?
+      AND state IN ('idle', 'thinking', 'attention', 'running')
+      AND updated < ?
+      AND (node_id IS NULL OR node_id NOT IN (alive_service_ids))
+
+At current scale (low-thousands of workstream rows) the existing single-
+column indexes are sufficient — ``idx_workstreams_state`` prunes to the
+non-closed subset, and the planner filters the rest sequentially.  At
+100k+ rows that filter becomes a tablescan-shaped cost on the reaper's
+periodic run.
+
+A **partial** index covering only ``BULK_CLOSE_STATE_VALUES`` rows
+matches the reaper's query exactly while staying tiny — closed rows
+(typically 95%+ of the table per empirical diagnosis) and ``error``
+rows are excluded, so the index is roughly 5% the size a full multi-
+column index would be.  Write amplification only kicks in for
+transitions that touch one of the four covered states.
+
+Column order ``(kind, updated)``:
+
+- ``kind`` first because the reaper always supplies it as an equality
+  predicate; partitions the partial index into interactive vs
+  coordinator subtrees.
+- ``updated`` last so the range comparison rides the trailing column —
+  classic composite-index pattern for ``WHERE eq AND range``.
+
+``node_id`` is intentionally NOT in the index.  The reaper's predicate
+on it is ``NOT IN (small list)`` against an unbounded-cardinality
+column, which planners don't index well; including it would just add
+write cost for negligible read benefit.
+
+PostgreSQL uses ``CREATE INDEX CONCURRENTLY`` so the build is
+non-blocking on a live system; SQLite has no concurrent concept and
+the table-level write lock already serializes, so a plain
+``CREATE INDEX`` is fine.
+
+Revision ID: 048
+Revises: 047
+Create Date: 2026-04-30
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "048"
+down_revision = "047"
+branch_labels = None
+depends_on = None
+
+
+_REAPER_PARTIAL_WHERE = "state IN ('idle', 'thinking', 'attention', 'running')"
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    dialect = bind.dialect.name
+
+    if dialect == "postgresql":
+        with op.get_context().autocommit_block():
+            op.execute(
+                "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_workstreams_reaper "
+                "ON workstreams (kind, updated) "
+                f"WHERE {_REAPER_PARTIAL_WHERE}"
+            )
+    else:
+        op.create_index(
+            "idx_workstreams_reaper",
+            "workstreams",
+            ["kind", "updated"],
+            sqlite_where=sa.text(_REAPER_PARTIAL_WHERE),
+        )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    dialect = bind.dialect.name
+
+    if dialect == "postgresql":
+        with op.get_context().autocommit_block():
+            op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_workstreams_reaper")
+    else:
+        op.drop_index("idx_workstreams_reaper", table_name="workstreams")

turnstone/core/workstream.py

@@ -77,6 +77,25 @@ class WorkstreamState(enum.Enum):
     ERROR = "error"  # last operation failed
 
 
+# States the orphan reaper (``SessionManager.close_idle`` pass 2 +
+# ``StorageBackend.bulk_close_stale_orphans``) is allowed to flip to
+# ``closed`` for rows past the staleness cutoff.  Excludes ``ERROR``
+# deliberately — error rows are user-investigatable and shouldn't be
+# auto-reaped — and excludes ``CLOSED`` (terminal).  Centralized here
+# so the storage backends and FakeStorage all agree; if a new transient
+# state is added to ``WorkstreamState``, deciding whether it joins
+# this set is part of the change rather than an after-the-fact
+# audit across three files.
+BULK_CLOSE_STATE_VALUES: frozenset[str] = frozenset(
+    {
+        WorkstreamState.IDLE.value,
+        WorkstreamState.THINKING.value,
+        WorkstreamState.RUNNING.value,
+        WorkstreamState.ATTENTION.value,
+    }
+)
+
+
 # ---------------------------------------------------------------------------
 # Workstream dataclass
 # ---------------------------------------------------------------------------

uv.lock

@@ -2533,7 +2533,7 @@ wheels = [
 
 [[package]]
 name = "turnstone"
-version = "1.5.2"
+version = "1.5.3"
 source = { editable = "." }
 dependencies = [
     { name = "alembic" },