chore: bump version to 1.6.9

- The guard snapshotted live threads by `Thread.ident`, but idents are
  recycled after a thread exits — a new leaked thread reusing an exited
  thread's ident would be mistaken for pre-existing and missed (false
  negative). Snapshot the Thread OBJECTS and compare by identity instead.
- Fix the `serve` fixture docstring: the factory returns the ephemeral
  port, not the server.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "turnstone"
-version = "1.6.8"
+version = "1.6.9"
 description = "Multi-node AI orchestration platform with tool use, agent routing, and cluster simulation."
 readme = "README.md"
 license = "Apache-2.0"
@@ -95,7 +95,10 @@ include = [
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]
-markers = ["live: requires a running LLM backend"]
+markers = [
+    "live: requires a running LLM backend",
+    "allow_thread_leak: test intentionally leaves a background thread running (opts out of the leaked-thread guard)",
+]
 filterwarnings = [
     # mcp v1 deprecates streamablehttp_client for an entry point whose call
     # shape only settles in v2 — adoption rides the deliberate v2 migration

tests/conftest.py

@@ -1,17 +1,118 @@
 from __future__ import annotations
 
+import asyncio
+import contextlib
+import logging
 import os
+import threading
+import time
 from typing import TYPE_CHECKING, Any
 from unittest.mock import MagicMock
 
 import pytest
 
+
+def stop_loop_thread(loop: asyncio.AbstractEventLoop, thread: threading.Thread) -> None:
+    """Fully tear down a ``loop.run_forever``-in-a-thread test loop.
+
+    Shuts the loop's default executor down ON the loop (joining its worker
+    threads — the ``asyncio_N`` threads that otherwise leak past the test),
+    then stops the loop, joins the thread, and closes the loop. Use in the
+    ``finally`` of a background-loop fixture so nothing outlives the test.
+    """
+    with contextlib.suppress(Exception):
+        asyncio.run_coroutine_threadsafe(loop.shutdown_default_executor(), loop).result(timeout=5)
+    loop.call_soon_threadsafe(loop.stop)
+    thread.join(timeout=5)
+    with contextlib.suppress(Exception):
+        loop.close()
+
+
+def serve_until_exit(server: Any) -> None:
+    """Run a uvicorn ``Server`` on a fresh event loop until it exits.
+
+    The thread target for an in-thread test upstream: when ``server.serve()``
+    returns (the fixture set ``server.should_exit`` / ``force_exit``), the loop
+    is closed so it doesn't leak past the fixture.
+    """
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    try:
+        loop.run_until_complete(server.serve())
+    finally:
+        # Cancel + drain anything the app left pending (e.g. sse_starlette's
+        # shutdown watcher) so loop.close() doesn't warn "Task was destroyed
+        # but it is pending".
+        pending = asyncio.all_tasks(loop)
+        for task in pending:
+            task.cancel()
+        if pending:
+            with contextlib.suppress(Exception):
+                loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
+        loop.close()
+
+
 if TYPE_CHECKING:
+    from collections.abc import Iterator
+
     from turnstone.core.mcp_client import MCPClientManager, StaticServerState
     from turnstone.core.mcp_crypto import MCPTokenCipher
     from turnstone.core.oidc import OIDCConfig
 
 
+# A background daemon (e.g. title generation) can log into pytest's per-test
+# capture as it is torn down — a benign "I/O operation on closed file" handler
+# error. Don't let the logging module turn that race into noisy stderr
+# tracebacks. (Process-global, test-only — product runtime keeps the default.)
+logging.raiseExceptions = False
+
+
+# Threads a test leaves running after teardown bleed into LATER tests' captured
+# output (the "I/O operation on closed file" heisenbug) and, worse, can wedge
+# the whole run (a leaked event loop / server that never stops). This grace
+# lets a legitimately-finishing quick daemon settle before we judge a leak.
+_THREAD_LEAK_GRACE = 5.0
+
+
+@pytest.fixture(autouse=True)
+def _no_leaked_threads(request: pytest.FixtureRequest) -> Iterator[None]:
+    """Fail a test that leaves a background thread running past teardown.
+
+    Snapshots the live threads at setup; at teardown, gives any NEW thread a
+    short grace to finish, then fails listing those still alive — so a leak is
+    caught here instead of as a heisenbug days later. Opt out with
+    ``@pytest.mark.allow_thread_leak`` (e.g. module-scoped servers in the live
+    suite).
+    """
+    if request.node.get_closest_marker("allow_thread_leak"):
+        yield
+        return
+    # Snapshot the Thread OBJECTS, not their idents: Thread.ident is recycled
+    # after a thread exits, so an ident-based snapshot could mistake a new
+    # leaked thread (reusing an exited thread's ident) for a pre-existing one.
+    before = set(threading.enumerate())
+    yield
+    main = threading.main_thread()
+    current = threading.current_thread()
+    # One deadline shared across all joined threads — a deliberate TOTAL
+    # teardown budget (not per-thread), so a pathological test can't stall
+    # teardown by N×grace. A genuine never-stopping leak exhausts it and fails.
+    deadline = time.monotonic() + _THREAD_LEAK_GRACE
+    leaked = []
+    for t in threading.enumerate():
+        if t in before or t is main or t is current or not t.is_alive():
+            continue
+        t.join(timeout=max(0.0, deadline - time.monotonic()))
+        if t.is_alive():
+            leaked.append(t.name)
+    if leaked:
+        pytest.fail(
+            f"test left background threads running after teardown: {leaked}. "
+            "Stop them in teardown (shut down servers / close event loops / join "
+            "threads), or mark @pytest.mark.allow_thread_leak if intentional."
+        )
+
+
 def make_mcp_token_cipher() -> MCPTokenCipher:
     """Build a single-key MCP token cipher for tests.
 

tests/test_coordinator_adapter.py

@@ -85,6 +85,59 @@ def test_emit_created_calls_collector_with_coord_fields() -> None:
     )
 
 
+def test_emit_created_seeds_resolved_display_name(tmp_path: Any) -> None:
+    """The collector seed uses the resolved display name (alias > title >
+    name), not the synthetic ``ws.name``.  A coordinator carrying a
+    persisted LLM auto-title (written by ``update_workstream_title``) then
+    shows that title in the live cluster tree instead of reverting to
+    ``ws-xxxx``.  Regression guard for the adapter half of the
+    coordinator-title-persistence fix — the server-side ``_coordinator_rows``
+    half is pinned in test_coordinator_endpoints.py."""
+    from turnstone.core.storage import init_storage, reset_storage
+
+    reset_storage()
+    backend = init_storage("sqlite", path=str(tmp_path / "adapter.db"), run_migrations=False)
+    try:
+        # Titled coordinator → the title surfaces over the placeholder name.
+        backend.register_workstream(
+            "coord-1",
+            node_id="console",
+            user_id="u1",
+            name="ws-c0c0",
+            kind=WorkstreamKind.COORDINATOR,
+        )
+        backend.update_workstream_title("coord-1", "Investigate the title bug")
+        adapter, collector = _make_adapter()
+        adapter.emit_created(_make_ws(name="ws-c0c0"))
+        assert (
+            collector.emit_console_ws_created.call_args.kwargs["name"]
+            == "Investigate the title bug"
+        )
+
+        # A user alias outranks the auto-title (alias > title > name).
+        assert backend.set_workstream_alias("coord-1", "Pinned name")
+        collector.emit_console_ws_created.reset_mock()
+        adapter._fanout_console_ws_created(_make_ws(name="ws-c0c0"))
+        assert collector.emit_console_ws_created.call_args.kwargs["name"] == "Pinned name"
+    finally:
+        reset_storage()
+
+
+def test_coord_display_name_skips_uninitialized_storage() -> None:
+    """_coord_display_name runs on a lifecycle-event path and must NOT trip
+    get_storage()'s SQLite auto-init (a stray .turnstone.db in the CWD) when
+    storage isn't initialized — it falls back to the placeholder ws.name and
+    leaves storage untouched."""
+    from turnstone.console.coordinator_adapter import _coord_display_name
+    from turnstone.core.storage import is_storage_initialized, reset_storage
+
+    reset_storage()
+    assert not is_storage_initialized()
+    assert _coord_display_name(_make_ws(name="ws-abcd")) == "ws-abcd"
+    # The resolution did not auto-initialize storage as a side effect.
+    assert not is_storage_initialized()
+
+
 def test_emit_state_calls_collector_state() -> None:
     """Post-rich-payload, emit_state passes tokens / context_ratio /
     activity / activity_state / content kwargs read from ws.ui's

tests/test_coordinator_endpoints.py

@@ -65,8 +65,10 @@ from turnstone.core.session_routes import (
     make_history_handler,
     make_list_handler,
     make_open_handler,
+    make_refresh_title_handler,
     make_saved_handler,
     make_send_handler,
+    make_set_title_handler,
 )
 from turnstone.core.workstream import WorkstreamKind
 
@@ -204,6 +206,16 @@ def _make_client(
                 ),
                 methods=["POST"],
             ),
+            Route(
+                "/v1/api/workstreams/{ws_id}/refresh-title",
+                make_refresh_title_handler(_coord_endpoint_config),
+                methods=["POST"],
+            ),
+            Route(
+                "/v1/api/workstreams/{ws_id}/title",
+                make_set_title_handler(_coord_endpoint_config),
+                methods=["POST"],
+            ),
             Route(
                 "/v1/api/workstreams/{ws_id}/history",
                 make_history_handler(_coord_endpoint_config),
@@ -370,6 +382,114 @@ def test_unresolvable_alias_returns_503(storage):
 _COORD_HEADERS = {"X-Test-User": "user-1", "X-Test-Perms": "admin.coordinator"}
 
 
+# ---------------------------------------------------------------------------
+# Title verbs — refresh-title (LLM regenerate) + set title (manual alias),
+# ported to coordinators via the lifted make_refresh_title_handler /
+# make_set_title_handler factories so both kinds share one body.
+# ---------------------------------------------------------------------------
+
+
+def test_coord_refresh_title_triggers_regeneration(storage):
+    mgr = _build_mgr(storage)
+    ws = mgr.create(user_id="user-1", name="c1")
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+    resp = client.post(f"/v1/api/workstreams/{ws.id}/refresh-title", headers=_COORD_HEADERS)
+    assert resp.status_code == 200
+    # The lifted handler resolves the current display name and asks the
+    # live session to regenerate a (different) title in the background.
+    ws.session.request_title_refresh.assert_called_once_with("c1")
+
+
+def test_coord_refresh_title_requires_operator_permission(storage):
+    mgr = _build_mgr(storage)
+    ws = mgr.create(user_id="user-1", name="c1")
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+    resp = client.post(
+        f"/v1/api/workstreams/{ws.id}/refresh-title",
+        headers={"X-Test-User": "user-1", "X-Test-Perms": "read"},
+    )
+    assert resp.status_code == 403
+    ws.session.request_title_refresh.assert_not_called()
+
+
+def test_coord_refresh_title_unknown_ws_404(storage):
+    mgr = _build_mgr(storage)
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+    resp = client.post(
+        "/v1/api/workstreams/" + ("0" * 32) + "/refresh-title", headers=_COORD_HEADERS
+    )
+    assert resp.status_code == 404
+
+
+def test_coord_set_title_stores_alias_and_broadcasts(storage):
+    from turnstone.core.memory import get_workstream_display_name
+
+    mgr = _build_mgr(storage)
+    ws = mgr.create(user_id="user-1", name="c1")
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+    resp = client.post(
+        f"/v1/api/workstreams/{ws.id}/title",
+        json={"title": "Nightly migration sweep"},
+        headers=_COORD_HEADERS,
+    )
+    assert resp.status_code == 200
+    assert resp.json()["title"] == "Nightly migration sweep"
+    # Stored as the alias (outranks the auto-title) ...
+    assert get_workstream_display_name(ws.id) == "Nightly migration sweep"
+    # ... and broadcast live to the dashboard via the session UI.
+    ws.session.ui.on_rename.assert_called_once_with("Nightly migration sweep")
+
+
+def test_coord_set_title_empty_400(storage):
+    mgr = _build_mgr(storage)
+    ws = mgr.create(user_id="user-1", name="c1")
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+    resp = client.post(
+        f"/v1/api/workstreams/{ws.id}/title", json={"title": "   "}, headers=_COORD_HEADERS
+    )
+    assert resp.status_code == 400
+
+
+def test_coord_set_title_alias_conflict_409(storage):
+    mgr = _build_mgr(storage)
+    first = mgr.create(user_id="user-1", name="c1")
+    second = mgr.create(user_id="user-1", name="c2")
+    storage.set_workstream_alias(first.id, "taken")
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+    resp = client.post(
+        f"/v1/api/workstreams/{second.id}/title", json={"title": "taken"}, headers=_COORD_HEADERS
+    )
+    assert resp.status_code == 409
+
+
+def test_coord_set_title_rejects_unowned_ws_404(storage):
+    """An admin.coordinator operator can't rename a workstream the coord
+    manager doesn't own (here a cross-kind interactive row) via the coord
+    /title route: set_workstream_alias is a global kind-unscoped UPDATE, so
+    the handler 404s on the in-memory coord lookup BEFORE writing — no
+    silent 200, no cross-kind alias write."""
+    from turnstone.core.memory import get_workstream_display_name
+
+    mgr = _build_mgr(storage)
+    # An interactive-kind row in storage, NOT held by coord_mgr.
+    storage.register_workstream(
+        "i" * 32,
+        node_id="node-1",
+        user_id="user-1",
+        name="interactive-ws",
+        kind=WorkstreamKind.INTERACTIVE,
+    )
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+    resp = client.post(
+        f"/v1/api/workstreams/{'i' * 32}/title",
+        json={"title": "hijacked"},
+        headers=_COORD_HEADERS,
+    )
+    assert resp.status_code == 404
+    # The interactive ws's display name is untouched — the alias write never fired.
+    assert get_workstream_display_name("i" * 32) == "interactive-ws"
+
+
 def test_active_list_row_shape_includes_unified_fields(storage):
     """Stage 2 list-verb-lift parity regression — coord active-list row
     carries the always-include fields (ws_id, name, state, kind,
@@ -2433,6 +2553,54 @@ def test_coordinator_rows_persisted_cluster_wide(storage):
         assert {r["name"] for r in rows} == {"alice-closed", "bob-closed", "orphan-closed"}
 
 
+def test_coordinator_rows_surface_persisted_title(storage):
+    """Regression for the coordinator-title-persistence bug.
+
+    The LLM auto-title (``update_workstream_title``) and the user alias
+    (``set_workstream_alias``) live only in ``workstreams.title`` /
+    ``workstreams.alias``.  ``_coordinator_rows`` must resolve the
+    display name ``alias > title > name`` from the persisted row for BOTH
+    lanes — the in-memory ``ws.name`` is the synthetic ``ws-xxxx``
+    placeholder.  Before the fix the read path hardcoded ``title=""`` and
+    used ``ws.name`` / the ``name`` column, so a generated title was
+    written but never read back: it reverted to ``ws-xxxx`` on every
+    dashboard refresh."""
+    from turnstone.console.server import _coordinator_rows
+    from turnstone.core.workstream import WorkstreamKind
+
+    mgr = _build_mgr(storage)
+
+    # In-memory lane: a LIVE coordinator titled after creation.  The
+    # manager assigned the placeholder ``ws.name``; the title is in the DB.
+    live = mgr.create(user_id="alice", name="ws-abcd")
+    storage.update_workstream_title(live.id, "Refactor the auth layer")
+
+    # Persisted lane: a closed coordinator (evicted from the manager)
+    # carrying BOTH a title and a user alias — the alias must win.
+    storage.register_workstream(
+        "f" * 32,
+        node_id="console",
+        user_id="bob",
+        name="ws-f0f0",
+        state="closed",
+        kind=WorkstreamKind.COORDINATOR,
+        parent_ws_id=None,
+    )
+    storage.update_workstream_title("f" * 32, "auto-generated title")
+    assert storage.set_workstream_alias("f" * 32, "Bob's pinned name")
+
+    request = _persisted_rows_request(storage, mgr, "alice", frozenset({"read"}))
+    rows = {r["id"]: r for r in _coordinator_rows(request)}
+
+    live_row = rows[live.id]
+    assert live_row["name"] == "Refactor the auth layer"
+    assert live_row["title"] == "Refactor the auth layer"
+
+    closed_row = rows["f" * 32]
+    assert closed_row["name"] == "Bob's pinned name"  # alias > title > name
+    assert closed_row["title"] == "auto-generated title"
+
+
 # ---------------------------------------------------------------------------
 # Stage 2 P1.5 — coord attachment surface parity with interactive
 # ---------------------------------------------------------------------------

tests/test_docker_healthcheck.py

@@ -52,13 +52,32 @@ class _Handler(http.server.BaseHTTPRequestHandler):
         pass
 
 
-def _serve(handler_cls, ssl_context: ssl.SSLContext | None = None) -> int:
-    """Start a daemon-thread HTTP(S) server on an ephemeral port."""
-    httpd = http.server.HTTPServer(("127.0.0.1", 0), handler_cls)
-    if ssl_context is not None:
-        httpd.socket = ssl_context.wrap_socket(httpd.socket, server_side=True)
-    threading.Thread(target=httpd.serve_forever, daemon=True).start()
-    return httpd.server_address[1]
+@pytest.fixture
+def serve():
+    """Factory that starts an HTTP(S) server on an ephemeral port and returns
+    that port.
+
+    Every server it starts is shut down + its serve_forever thread joined at
+    teardown, so the thread never outlives the test (which would otherwise bleed
+    into a later test's captured output / leak the listener).
+    """
+    started: list[tuple[http.server.HTTPServer, threading.Thread]] = []
+
+    def _factory(handler_cls, ssl_context: ssl.SSLContext | None = None) -> int:
+        httpd = http.server.HTTPServer(("127.0.0.1", 0), handler_cls)
+        if ssl_context is not None:
+            httpd.socket = ssl_context.wrap_socket(httpd.socket, server_side=True)
+        thread = threading.Thread(target=httpd.serve_forever, daemon=True)
+        thread.start()
+        started.append((httpd, thread))
+        return httpd.server_address[1]
+
+    yield _factory
+
+    for httpd, thread in started:
+        httpd.shutdown()  # break the serve_forever loop
+        httpd.server_close()  # release the listening socket
+        thread.join(timeout=5)
 
 
 @pytest.fixture
@@ -90,29 +109,29 @@ def mtls_setup(tmp_path):
 # ── Plain HTTP (mTLS disabled — the default deployment) ─────────────────────
 
 
-def test_plain_http_ok():
+def test_plain_http_ok(serve):
     """Default path: plain probe succeeds, PEM dir never consulted."""
-    port = _serve(_Handler)
+    port = serve(_Handler)
     result = run_healthcheck(f"http://127.0.0.1:{port}/health")
     assert result.returncode == 0, result.stderr
 
 
-def test_plain_http_degraded_is_healthy():
+def test_plain_http_degraded_is_healthy(serve):
     """'degraded' (backend down, server up) still counts as container-healthy."""
 
     class Degraded(_Handler):
         payload = {"status": "degraded"}
 
-    port = _serve(Degraded)
+    port = serve(Degraded)
     result = run_healthcheck(f"http://127.0.0.1:{port}/health")
     assert result.returncode == 0, result.stderr
 
 
-def test_plain_http_bad_status_fails():
+def test_plain_http_bad_status_fails(serve):
     class Bad(_Handler):
         payload = {"status": "error"}
 
-    port = _serve(Bad)
+    port = serve(Bad)
     result = run_healthcheck(f"http://127.0.0.1:{port}/health")
     assert result.returncode == 1
     assert "unhealthy payload" in result.stderr
@@ -128,40 +147,40 @@ def test_server_down_fails():
 # ── mTLS (tls.enabled) ───────────────────────────────────────────────────────
 
 
-def test_mtls_probe_with_pem_dir(mtls_setup):
+def test_mtls_probe_with_pem_dir(mtls_setup, serve):
     """The regression case: mTLS node + plain-HTTP probe URL.
 
     The plain attempt is rejected at the socket; the script must fall back
     to HTTPS with the node cert as client cert and report healthy."""
     pem_root, server_ctx = mtls_setup
-    port = _serve(_Handler, ssl_context=server_ctx)
+    port = serve(_Handler, ssl_context=server_ctx)
     result = run_healthcheck(f"http://127.0.0.1:{port}/health", pem_root=pem_root)
     assert result.returncode == 0, result.stderr
 
 
-def test_mtls_probe_without_pems_fails(mtls_setup):
+def test_mtls_probe_without_pems_fails(mtls_setup, serve):
     """mTLS node but no PEM material on disk: the probe must fail."""
     _, server_ctx = mtls_setup
-    port = _serve(_Handler, ssl_context=server_ctx)
+    port = serve(_Handler, ssl_context=server_ctx)
     result = run_healthcheck(f"http://127.0.0.1:{port}/health", pem_root=None)
     assert result.returncode == 1
     assert "Health check failed" in result.stderr
 
 
-def test_mtls_unhealthy_payload_fails(mtls_setup):
+def test_mtls_unhealthy_payload_fails(mtls_setup, serve):
     """A reachable mTLS server with a bad payload is still unhealthy."""
     pem_root, server_ctx = mtls_setup
 
     class Bad(_Handler):
         payload = {"status": "error"}
 
-    port = _serve(Bad, ssl_context=server_ctx)
+    port = serve(Bad, ssl_context=server_ctx)
     result = run_healthcheck(f"http://127.0.0.1:{port}/health", pem_root=pem_root)
     assert result.returncode == 1
     assert "unhealthy payload" in result.stderr
 
 
-def test_mtls_incomplete_pem_dir_fails(mtls_setup, tmp_path):
+def test_mtls_incomplete_pem_dir_fails(mtls_setup, tmp_path, serve):
     """A PEM dir missing the key is skipped, not half-used."""
     _, server_ctx = mtls_setup
     incomplete = tmp_path / "incomplete-root"
@@ -170,7 +189,7 @@ def test_mtls_incomplete_pem_dir_fails(mtls_setup, tmp_path):
     (d / "fullchain.pem").write_text("not a cert")
     (d / "ca.pem").write_text("not a cert")
 
-    port = _serve(_Handler, ssl_context=server_ctx)
+    port = serve(_Handler, ssl_context=server_ctx)
     result = run_healthcheck(f"http://127.0.0.1:{port}/health", pem_root=incomplete)
     assert result.returncode == 1
 

tests/test_mcp_pool_auth_integration.py

@@ -38,7 +38,7 @@ import uvicorn
 from mcp.server.fastmcp import FastMCP
 from starlette.middleware.base import BaseHTTPMiddleware
 
-from tests.conftest import make_mcp_token_cipher
+from tests.conftest import make_mcp_token_cipher, serve_until_exit, stop_loop_thread
 from turnstone.core.mcp_client import MCPClientManager
 from turnstone.core.mcp_crypto import MCPTokenStore
 from turnstone.core.mcp_oauth import TokenLookupResult
@@ -187,7 +187,14 @@ def _build_server(port: int, behaviour: dict[str, Any]) -> uvicorn.Server:
 
     app = mcp.streamable_http_app()
     app.add_middleware(BehaviorMiddleware, behaviour=behaviour)
-    config = uvicorn.Config(app, host="127.0.0.1", port=port, log_level="warning", access_log=False)
+    config = uvicorn.Config(
+        app,
+        host="127.0.0.1",
+        port=port,
+        log_level="warning",
+        access_log=False,
+        timeout_graceful_shutdown=0,  # don't block teardown on a held-open stream
+    )
     return uvicorn.Server(config)
 
 
@@ -215,9 +222,7 @@ def upstream():
     server = _build_server(port, behaviour)
 
     def _run() -> None:
-        loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
-        loop.run_until_complete(server.serve())
+        serve_until_exit(server)
 
     t = threading.Thread(target=_run, daemon=True, name="phase6-upstream")
     t.start()
@@ -225,7 +230,11 @@ def upstream():
         _wait_ready(port)
         yield f"http://127.0.0.1:{port}/mcp", behaviour
     finally:
+        # should_exit alone triggers a GRACEFUL shutdown that can wait forever
+        # on a held-open streamable-http stream; force_exit skips that wait so
+        # serve() returns and the upstream thread doesn't leak past the test.
         server.should_exit = True
+        server.force_exit = True
         t.join(timeout=5)
 
 
@@ -311,8 +320,7 @@ def running_loop_mgr():
 
         with contextlib.suppress(Exception):
             asyncio.run_coroutine_threadsafe(_drain(mgr), loop).result(timeout=2)
-        loop.call_soon_threadsafe(loop.stop)
-        thread.join(timeout=2)
+        stop_loop_thread(loop, thread)
 
 
 # ---------------------------------------------------------------------------

tests/test_mcp_pool_auth_introspection.py

@@ -34,7 +34,7 @@ from unittest.mock import AsyncMock, MagicMock
 import httpx
 import pytest
 
-from tests.conftest import make_mcp_token_cipher
+from tests.conftest import make_mcp_token_cipher, stop_loop_thread
 from turnstone.core.mcp_client import (
     MCPClientManager,
     _AuthCapture,
@@ -131,8 +131,7 @@ def running_loop_mgr():
 
         with contextlib.suppress(Exception):
             asyncio.run_coroutine_threadsafe(_drain(mgr), loop).result(timeout=2)
-        loop.call_soon_threadsafe(loop.stop)
-        thread.join(timeout=2)
+        stop_loop_thread(loop, thread)
 
 
 def _run_on_loop(loop: asyncio.AbstractEventLoop, coro: Any) -> Any:

tests/test_mcp_pool_auth_prompt_integration.py

@@ -26,7 +26,7 @@ import uvicorn
 from mcp.server.fastmcp import FastMCP
 from starlette.middleware.base import BaseHTTPMiddleware
 
-from tests.conftest import make_mcp_token_cipher
+from tests.conftest import make_mcp_token_cipher, serve_until_exit, stop_loop_thread
 from turnstone.core.mcp_client import MCPClientManager
 from turnstone.core.mcp_crypto import MCPTokenStore
 from turnstone.core.mcp_oauth import TokenLookupResult
@@ -130,7 +130,14 @@ def _build_server(port: int, behaviour: dict[str, Any]) -> uvicorn.Server:
 
     app = mcp.streamable_http_app()
     app.add_middleware(BehaviorMiddleware, behaviour=behaviour)
-    config = uvicorn.Config(app, host="127.0.0.1", port=port, log_level="warning", access_log=False)
+    config = uvicorn.Config(
+        app,
+        host="127.0.0.1",
+        port=port,
+        log_level="warning",
+        access_log=False,
+        timeout_graceful_shutdown=0,  # don't block teardown on a held-open stream
+    )
     return uvicorn.Server(config)
 
 
@@ -152,9 +159,7 @@ def upstream():
     server = _build_server(port, behaviour)
 
     def _run() -> None:
-        loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
-        loop.run_until_complete(server.serve())
+        serve_until_exit(server)
 
     t = threading.Thread(target=_run, daemon=True, name="phase7b-prompt-upstream")
     t.start()
@@ -162,7 +167,11 @@ def upstream():
         _wait_ready(port)
         yield f"http://127.0.0.1:{port}/mcp", behaviour
     finally:
+        # should_exit alone triggers a GRACEFUL shutdown that can wait forever
+        # on a held-open streamable-http stream; force_exit skips that wait so
+        # serve() returns and the upstream thread doesn't leak past the test.
         server.should_exit = True
+        server.force_exit = True
         t.join(timeout=5)
 
 
@@ -248,8 +257,7 @@ def running_loop_mgr():
 
         with contextlib.suppress(Exception):
             asyncio.run_coroutine_threadsafe(_drain(mgr), loop).result(timeout=2)
-        loop.call_soon_threadsafe(loop.stop)
-        thread.join(timeout=2)
+        stop_loop_thread(loop, thread)
 
 
 def _seed_pool_prompt_map(

tests/test_mcp_pool_auth_resource_integration.py

@@ -26,7 +26,7 @@ import uvicorn
 from mcp.server.fastmcp import FastMCP
 from starlette.middleware.base import BaseHTTPMiddleware
 
-from tests.conftest import make_mcp_token_cipher
+from tests.conftest import make_mcp_token_cipher, serve_until_exit, stop_loop_thread
 from turnstone.core.mcp_client import MCPClientManager
 from turnstone.core.mcp_crypto import MCPTokenStore
 from turnstone.core.mcp_oauth import TokenLookupResult
@@ -139,7 +139,14 @@ def _build_server(port: int, behaviour: dict[str, Any]) -> uvicorn.Server:
 
     app = mcp.streamable_http_app()
     app.add_middleware(BehaviorMiddleware, behaviour=behaviour)
-    config = uvicorn.Config(app, host="127.0.0.1", port=port, log_level="warning", access_log=False)
+    config = uvicorn.Config(
+        app,
+        host="127.0.0.1",
+        port=port,
+        log_level="warning",
+        access_log=False,
+        timeout_graceful_shutdown=0,  # don't block teardown on a held-open stream
+    )
     return uvicorn.Server(config)
 
 
@@ -161,9 +168,7 @@ def upstream():
     server = _build_server(port, behaviour)
 
     def _run() -> None:
-        loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
-        loop.run_until_complete(server.serve())
+        serve_until_exit(server)
 
     t = threading.Thread(target=_run, daemon=True, name="phase7b-resource-upstream")
     t.start()
@@ -171,7 +176,11 @@ def upstream():
         _wait_ready(port)
         yield f"http://127.0.0.1:{port}/mcp", behaviour
     finally:
+        # should_exit alone triggers a GRACEFUL shutdown that can wait forever
+        # on a held-open streamable-http stream; force_exit skips that wait so
+        # serve() returns and the upstream thread doesn't leak past the test.
         server.should_exit = True
+        server.force_exit = True
         t.join(timeout=5)
 
 
@@ -257,8 +266,7 @@ def running_loop_mgr():
 
         with contextlib.suppress(Exception):
             asyncio.run_coroutine_threadsafe(_drain(mgr), loop).result(timeout=2)
-        loop.call_soon_threadsafe(loop.stop)
-        thread.join(timeout=2)
+        stop_loop_thread(loop, thread)
 
 
 def _seed_pool_resource_map(

tests/test_mcp_user_pool.py

@@ -25,7 +25,7 @@ from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
-from tests.conftest import make_mcp_token_cipher
+from tests.conftest import make_mcp_token_cipher, stop_loop_thread
 from turnstone.core.mcp_client import MCPClientManager, PoolEntryState
 from turnstone.core.mcp_crypto import MCPTokenStore
 from turnstone.core.storage._sqlite import SQLiteBackend
@@ -123,8 +123,7 @@ def running_loop_mgr():
 
         with contextlib.suppress(Exception):
             asyncio.run_coroutine_threadsafe(_drain(mgr), loop).result(timeout=2)
-        loop.call_soon_threadsafe(loop.stop)
-        thread.join(timeout=2)
+        stop_loop_thread(loop, thread)
 
 
 def _run_on_loop(loop: asyncio.AbstractEventLoop, coro: Any) -> Any:

tests/test_session.py

@@ -150,6 +150,27 @@ def _send_with_mocks(session, responses, mock_execute, **extra_patches):
         yield save_msg
 
 
+def _capturing_thread_cls():
+    """Return a no-op ``threading.Thread`` stand-in plus the list it records
+    each constructed thread's ``target`` into.
+
+    Patched over ``session.threading.Thread`` so a test can assert WHICH
+    callable was scheduled (e.g. ``_generate_title``) without the thread
+    actually running — ``start()`` is a no-op, so no background LLM call
+    fires.
+    """
+    started: list = []
+
+    class _CaptureThread:
+        def __init__(self, *a, target=None, **kw):
+            started.append(target)
+
+        def start(self):
+            pass
+
+    return _CaptureThread, started
+
+
 def _user_pending(session) -> list[tuple[str, str]]:
     """Return user-channel queued nudges as ``(type, text)`` tuples.
 
@@ -1010,6 +1031,66 @@ class TestTitleRetry:
         # Restore for cleanup
         session._ws_id = original_ws_id
 
+    def test_title_fires_after_send_not_after_tool_free_turn(self, tmp_db):
+        """Auto-title fires right after the user turn is recorded, BEFORE
+        tools run — it no longer waits for a tool-call-free assistant
+        turn.  Coordinators spend nearly every turn in tool calls and may
+        never reach that terminal text turn, so the old end-of-turn
+        trigger almost never fired for them (the timing half of the
+        coordinator-title bug)."""
+        session = _make_session()
+        assert session._title_generated is False
+        # The assistant's opening turn is ALL tool calls — under the old
+        # trigger no title would generate until a later text-only turn.
+        responses = [
+            {
+                "role": "assistant",
+                "content": "working",
+                "tool_calls": [
+                    {
+                        "id": "c1",
+                        "type": "function",
+                        "function": {"name": "echo", "arguments": "{}"},
+                    }
+                ],
+            },
+            {"role": "assistant", "content": "done"},
+        ]
+        capture_cls, started = _capturing_thread_cls()
+
+        def mock_execute(_tool_calls):
+            # The title must already be scheduled by the time tools run.
+            assert session._title_generated is True
+            return [("c1", "ok")], None
+
+        with (
+            _send_with_mocks(session, responses, mock_execute),
+            patch("turnstone.core.session.threading.Thread", capture_cls),
+        ):
+            session.send("refactor the auth layer")
+
+        assert session._title_generated is True
+        assert session._generate_title in started
+
+    def test_title_not_generated_for_blank_or_wake_send(self, tmp_db):
+        """Blank input and synthetic wake sends don't burn the one-shot
+        auto-title — ``_generate_title`` needs first-user-message text,
+        and a wake carries none."""
+        capture_cls, started = _capturing_thread_cls()
+
+        def mock_execute(_tool_calls):
+            return [], None
+
+        for user_input, kwargs in (("   ", {}), ("a real message", {"from_wake": True})):
+            session = _make_session()
+            with (
+                _send_with_mocks(session, [{"role": "assistant", "content": "ok"}], mock_execute),
+                patch("turnstone.core.session.threading.Thread", capture_cls),
+            ):
+                session.send(user_input, **kwargs)
+            assert session._generate_title not in started
+            assert session._title_generated is False
+
 
 class TestLiveConfigUpdate:
     """ConfigStore-backed sessions pick up settings changes at point-of-use."""

tests/test_shell_js.py

@@ -602,6 +602,27 @@ def test_step7_tab_menu_wired_per_persona() -> None:
     )
 
 
+def test_coordinator_tab_menu_enables_title_verbs() -> None:
+    """Coordinators carry LLM/auto titles like interactive workstreams, so
+    their tab dropdown must surface Refresh/Edit title — convTabMenu's
+    ``titleVerbs`` block, POSTed to the console-origin coord
+    ``refresh-title`` / ``title`` routes via the base-aware lane (default
+    base ""). Scoped to the coordinator registerType block so it can't
+    pass on the interactive pane's long-standing ``titleVerbs``."""
+    shell = _SHELL_JS.read_text(encoding="utf-8")
+    start = shell.index('registerType("coordinator"')
+    tail = shell[start:]
+    nxt = tail.find("registerType(", 1)  # bound at the next pane registration
+    coord_block = tail[:nxt] if nxt != -1 else tail
+    assert "pane._ctl.closeSession()" in coord_block, (
+        "sanity: the extracted block is the coordinator pane"
+    )
+    assert "convTabMenu(" in coord_block, "the coordinator pane must wire a tab menu"
+    assert "titleVerbs: true" in coord_block, (
+        "the coordinator tab menu must enable titleVerbs (Refresh/Edit title)"
+    )
+
+
 def test_tab_menu_base_aware_verb_lane() -> None:
     """Lifecycle round 2: a proxied interactive pane's tab menu must act on the
     pane's OWN transport base, not the console origin — the globals lane only

tests/test_workstream_endpoints.py

@@ -31,16 +31,17 @@ from turnstone.core.session_routes import (
     make_export_handler,
     make_history_handler,
     make_open_handler,
+    make_refresh_title_handler,
     make_retry_handler,
     make_rewind_handler,
+    make_set_title_handler,
 )
 from turnstone.core.storage._sqlite import SQLiteBackend
 from turnstone.core.workstream import WorkstreamKind
 from turnstone.server import (
+    _interactive_tenant_check,
     delete_workstream_endpoint,
     list_interface_settings,
-    refresh_workstream_title,
-    set_workstream_title,
     update_interface_setting,
 )
 
@@ -113,6 +114,18 @@ def delete_client(_inject_storage):
 
 @pytest.fixture
 def title_client(_inject_storage):
+    # Build the lifted refresh/set-title handlers the same way server.py
+    # wires the interactive bundle — same SessionEndpointConfig
+    # (manager_lookup + _interactive_tenant_check) so the tests exercise
+    # the production resolution path (mgr fast-path → storage ownership).
+    mock_mgr = MagicMock()
+    cfg = SessionEndpointConfig(
+        permission_gate=None,
+        manager_lookup=lambda _r: (mock_mgr, None),
+        tenant_check=_interactive_tenant_check,
+        not_found_label="Workstream not found",
+        audit_action_prefix="workstream",
+    )
     app = Starlette(
         routes=[
             Mount(
@@ -120,12 +133,12 @@ def title_client(_inject_storage):
                 routes=[
                     Route(
                         "/api/workstreams/{ws_id}/title",
-                        set_workstream_title,
+                        make_set_title_handler(cfg),
                         methods=["POST"],
                     ),
                     Route(
                         "/api/workstreams/{ws_id}/refresh-title",
-                        refresh_workstream_title,
+                        make_refresh_title_handler(cfg),
                         methods=["POST"],
                     ),
                 ],
@@ -133,7 +146,6 @@ def title_client(_inject_storage):
         ],
         middleware=[Middleware(_InjectAuthMiddleware)],
     )
-    mock_mgr = MagicMock()
     app.state.workstreams = mock_mgr
     return TestClient(app), mock_mgr
 

turnstone/__init__.py

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

turnstone/console/collector.py

@@ -94,6 +94,11 @@ class ClusterCollector:
         self._nodes: dict[str, NodeSnapshot] = {}
         self._running = False
         self._threads: list[threading.Thread] = []
+        # Wakes the discovery loop out of its inter-scan sleep so ``stop()``
+        # can join it promptly instead of blocking up to ``discovery_interval``
+        # (a long interval would otherwise leave the thread sleeping past
+        # join's timeout — a leaked background thread).
+        self._discovery_wake = threading.Event()
 
         # SSE fan-out to browser clients
         self._listeners: list[queue.Queue[dict[str, Any]]] = []
@@ -135,6 +140,9 @@ class ClusterCollector:
     def start(self) -> None:
         """Start background threads."""
         self._running = True
+        # Clear the shutdown wake so a restarted collector (stop() set it) sleeps
+        # the full interval again instead of busy-spinning the discovery loop.
+        self._discovery_wake.clear()
         # Subscribe to the ``services`` channel for reactive node discovery.
         # NOTIFY-driven wake-ups bring new-node visibility from up-to-60 s
         # (next discovery tick) down to ~500 ms on Postgres; the 60 s
@@ -161,6 +169,7 @@ class ClusterCollector:
         its ``finally`` cleanup (cancel tasks, close AsyncClient).
         """
         self._running = False
+        self._discovery_wake.set()  # wake the discovery loop out of its sleep
         if self._notify_unsubscribe is not None:
             with contextlib.suppress(Exception):
                 self._notify_unsubscribe()
@@ -417,7 +426,9 @@ class ClusterCollector:
                 pass  # already logged by storage layer
             except Exception:
                 log.exception("Node discovery error")
-            time.sleep(self._discovery_interval)
+            # Interruptible inter-scan sleep — ``stop()`` sets the event to
+            # wake us immediately instead of blocking out the full interval.
+            self._discovery_wake.wait(self._discovery_interval)
 
     def _discover_nodes(self) -> None:
         """Query the service registry and update the node map."""

turnstone/console/coordinator_adapter.py

@@ -23,6 +23,8 @@ from turnstone.core.child_event_bus import ChildEventBus
 from turnstone.core.child_source import ClusterChildSource
 from turnstone.core.children_registry import ChildrenRegistry
 from turnstone.core.log import get_logger
+from turnstone.core.memory import get_workstream_display_name, get_workstream_display_names
+from turnstone.core.storage import is_storage_initialized
 from turnstone.core.workstream import Workstream, WorkstreamKind, WorkstreamState
 
 if TYPE_CHECKING:
@@ -37,6 +39,28 @@ if TYPE_CHECKING:
 log = get_logger(__name__)
 
 
+def _coord_display_name(ws: Workstream) -> str:
+    """Resolve a coordinator's display name (``alias > title > name``).
+
+    ``ws.name`` is the synthetic ``ws-xxxx`` placeholder; the persisted
+    auto-title (``update_workstream_title``) and user alias live only in
+    the DB.  Seeding the collector with the resolved name means a
+    rehydrated coordinator shows its title in the live cluster tree
+    immediately, rather than reverting to ``ws-xxxx`` until a (for
+    coordinators, rarely-firing) ``on_rename`` event arrives.
+
+    Skips the DB read when storage isn't initialized: this runs on a
+    lifecycle-event path, and a display-name resolution must never trip
+    ``get_storage``'s SQLite auto-init side effect (a stray
+    ``.turnstone.db``) before the host has called ``init_storage`` (the
+    real cluster always does so at startup — this only bites early /
+    test call paths). The placeholder ``ws.name`` is the right fallback.
+    """
+    if not is_storage_initialized():
+        return ws.name
+    return get_workstream_display_name(ws.id) or ws.name
+
+
 class CoordinatorAdapter:
     """Bridges SessionManager to the console's coordinator transport."""
 
@@ -132,7 +156,7 @@ class CoordinatorAdapter:
         try:
             self._collector.emit_console_ws_created(
                 ws.id,
-                name=ws.name,
+                name=_coord_display_name(ws),
                 user_id=ws.user_id,
                 kind=ws.kind.value,
                 state=ws.state.value,
@@ -466,11 +490,16 @@ class CoordinatorAdapter:
         # creates happened before the collector was wired up and their
         # rows never showed on the snapshot. (Coord-specific — interactive
         # has no analogous pseudo-node.)
-        for ws in mgr.list_all():
+        coords = mgr.list_all()
+        # One round-trip for every coordinator's display name instead of a
+        # per-``ws`` ``_coord_display_name`` lookup (N+1); cold path, but
+        # the bulk helper is right there.
+        seed_names = get_workstream_display_names([ws.id for ws in coords])
+        for ws in coords:
             try:
                 collector.emit_console_ws_created(
                     ws.id,
-                    name=ws.name,
+                    name=seed_names.get(ws.id) or ws.name,
                     user_id=ws.user_id or "",
                     kind=WorkstreamKind.COORDINATOR.value,
                     state=ws.state.value,

turnstone/console/server.py

@@ -56,6 +56,7 @@ from turnstone.core.auth import (
     require_permission,
 )
 from turnstone.core.deadline import DeadlineExceededError, run_with_deadline
+from turnstone.core.memory import get_workstream_display_names
 from turnstone.core.rendezvous import NoAvailableNodeError
 from turnstone.core.session_replay import session_replay_preamble
 from turnstone.core.session_routes import (
@@ -75,9 +76,11 @@ from turnstone.core.session_routes import (
     make_history_handler,
     make_list_handler,
     make_open_handler,
+    make_refresh_title_handler,
     make_retry_handler,
     make_rewind_handler,
     make_send_handler,
+    make_set_title_handler,
     make_unified_saved_handler,
     register_coord_verbs,
     register_session_routes,
@@ -853,6 +856,14 @@ def _coordinator_rows(request: Request) -> list[dict[str, Any]]:
     In-memory wins on ws_id conflict so live state stays authoritative
     for active sessions.
 
+    Display name resolves ``alias > title > name`` from the persisted
+    row for BOTH lanes.  ``ws.name`` on the in-memory Workstream is the
+    synthetic ``ws-xxxx`` placeholder; the LLM auto-title
+    (``update_workstream_title``) and the user alias
+    (``set_workstream_alias``) live only in the DB, so without the
+    persisted lookup the live lane would show ``ws-xxxx`` and the
+    auto-title would never survive a dashboard refresh.
+
     Trusted-team visibility (post-#400): the cluster dashboard shows
     every coordinator regardless of caller identity; ``user_id`` is
     surfaced on each row as display metadata.
@@ -870,6 +881,61 @@ def _coordinator_rows(request: Request) -> list[dict[str, Any]]:
         val = getattr(sess, name, "") if sess else ""
         return val if isinstance(val, str) else ""
 
+    # Persisted coordinator rows serve two purposes: (1) surface
+    # closed / error / deleted coordinators the manager has already
+    # evicted from ``self._workstreams``, and (2) supply the persisted
+    # display name (``alias > title > name``) for the LIVE coordinators
+    # too — ``ws.name`` is the synthetic placeholder.  Cluster-wide
+    # (trusted-team visibility).  Indexed by ws_id so both lanes resolve
+    # the same way.
+    storage = getattr(request.app.state, "auth_storage", None)
+    persisted: list[Any] = []
+    if storage is not None:
+        try:
+            persisted = storage.list_workstreams(
+                kind=WorkstreamKind.COORDINATOR,
+                user_id=None,
+                limit=200,
+            )
+        except Exception:
+            log.debug("cluster_workstreams.coord_persisted_failed", exc_info=True)
+            persisted = []
+    # SQLAlchemy Row — access via _mapping so future SELECT reorders /
+    # new columns don't silently corrupt the projection (per the
+    # storage-protocol guidance on list_workstreams).  Test doubles must
+    # expose the same ._mapping attribute.
+    meta: dict[str, Any] = {}
+    for row in persisted:
+        m = row._mapping
+        rid = m.get("ws_id") or ""
+        if rid:
+            meta[rid] = m
+
+    # Live coordinators resolve their display name through the bulk
+    # helper keyed on their EXACT ids (one round-trip, no row cap) rather
+    # than the ``limit=200`` ``meta`` map: a live coord that has dropped
+    # below the 200-row ``updated DESC`` window would otherwise revert to
+    # its synthetic ``ws.name``.  Closed/evicted rows (the persisted lane
+    # below) already carry alias/title in their own ``_mapping``.
+    live_display = get_workstream_display_names([ws.id for ws in wss]) if wss else {}
+
+    def _display_name(ws_id: str, fallback: str) -> str:
+        m = meta.get(ws_id)
+        if m is None:
+            return fallback
+        return m.get("alias") or m.get("title") or m.get("name") or fallback
+
+    def _title(ws_id: str) -> str:
+        # Best-effort: the secondary ``title`` field is sourced from the
+        # ``limit=200`` ``meta`` map, so a live coord outside that window
+        # reports ``""`` here. The user-visible ``name`` stays correct
+        # (resolved via the uncapped ``live_display`` above, and the UI
+        # renders ``title || name``); the empty title is harmless and the
+        # window is unreachable in practice (live coords are bounded by
+        # ``max_active`` and sort to the top of ``updated DESC``).
+        m = meta.get(ws_id)
+        return str(m.get("title") or "") if m is not None else ""
+
     rows: list[dict[str, Any]] = []
     seen: set[str] = set()
     for ws in wss:
@@ -877,9 +943,9 @@ def _coordinator_rows(request: Request) -> list[dict[str, Any]]:
         rows.append(
             {
                 "id": ws.id,
-                "name": ws.name,
+                "name": live_display.get(ws.id) or ws.name,
                 "state": ws.state.value,
-                "title": "",
+                "title": _title(ws.id),
                 "node": "console",
                 "server_url": "",
                 "model": _str_sess_attr(sess, "model"),
@@ -896,30 +962,7 @@ def _coordinator_rows(request: Request) -> list[dict[str, Any]]:
         )
         seen.add(ws.id)
 
-    # Second lane — persisted coordinator rows, used to surface
-    # closed / error / deleted coordinators the manager has already
-    # evicted from ``self._workstreams``.  Cluster-wide (trusted-team
-    # visibility).
-    storage = getattr(request.app.state, "auth_storage", None)
-    if storage is None:
-        return rows
-    try:
-        persisted = storage.list_workstreams(
-            kind=WorkstreamKind.COORDINATOR,
-            user_id=None,
-            limit=200,
-        )
-    except Exception:
-        log.debug("cluster_workstreams.coord_persisted_failed", exc_info=True)
-        return rows
-
     for row in persisted:
-        # SQLAlchemy Row — access via _mapping so future SELECT reorders
-        # / new columns don't silently corrupt the projection (per the
-        # storage-protocol guidance on list_workstreams).  Test doubles
-        # must expose the same ._mapping attribute; positional indexing
-        # was removed because it hard-coded column offsets that drift
-        # with migrations.
         m = row._mapping
         row_id = m.get("ws_id") or ""
         if not row_id or row_id in seen:
@@ -928,9 +971,9 @@ def _coordinator_rows(request: Request) -> list[dict[str, Any]]:
         rows.append(
             {
                 "id": row_id,
-                "name": m.get("name") or f"coord-{row_id[:4]}",
+                "name": _display_name(row_id, f"coord-{row_id[:4]}"),
                 "state": str(m.get("state") or "idle"),
-                "title": "",
+                "title": _title(row_id),
                 "node": "console",
                 "server_url": "",
                 "model": "",
@@ -12979,6 +13022,8 @@ def create_app(
                 audit_emit=_audit_close_coordinator,
                 supports_close_reason=False,
             ),
+            refresh_title=make_refresh_title_handler(coord_endpoint_config),  # lifted: shared body
+            set_title=make_set_title_handler(coord_endpoint_config),  # lifted: shared body
             send=make_send_handler(coord_endpoint_config),  # lifted: shared body (P1.5)
             dequeue=make_dequeue_handler(coord_endpoint_config),  # lifted: shared body
             approve=make_approve_handler(coord_endpoint_config),  # lifted: shared body

turnstone/core/session.py

@@ -2425,10 +2425,15 @@ class ChatSession:
         ws_id = self._ws_id  # Capture before async work
         log.info("ws.title.gen_start", ws_id=ws_id[:8])
         try:
-            # Gather first user message and first assistant reply
+            # Gather first user message and first assistant reply.
+            # Snapshot ``self.messages`` (C-level atomic copy under the
+            # GIL): this runs in a background thread that may now fire
+            # while the main ``send`` loop is still streaming and
+            # appending turns, so iterating the live list directly could
+            # raise "list changed size during iteration".
             user_msg = ""
             asst_msg = ""
-            for m in self.messages:
+            for m in list(self.messages):
                 content = m.text  # joins text blocks; multipart attachments contribute none
                 if m.role is Role.USER and not user_msg:
                     user_msg = content[:300]
@@ -4150,6 +4155,25 @@ class ChatSession:
         # legacy per-message ``_reminders`` side-channel splice.
         self._emit_pending_user_nudges()
 
+        # Auto-title from the opening user message — fire NOW rather than
+        # waiting for the assistant's final tool-call-free turn.  The old
+        # trigger sat in the ``not tool_calls`` branch of the loop below;
+        # coordinators spend nearly every turn in tool calls and may never
+        # reach that terminal text turn, so the title almost never
+        # generated for them.  Gate on a real user message: synthetic wake
+        # sends carry no content and ``_generate_title`` would no-op on the
+        # empty/attachment-only case anyway (it needs first-user-message
+        # text).  Concurrency: this background thread runs alongside the
+        # streaming turn started below, but safely — it snapshots
+        # ``self.messages`` for iteration, and the only UI it touches is
+        # ``on_aux_usage`` (storage/metrics, no ``_ws_lock`` state) and
+        # ``on_rename`` (queue/locked fan-out), both documented
+        # auxiliary-thread-safe on ``SessionUIBase``; the provider + client
+        # handle concurrent requests (the same path ``task_agent`` uses).
+        if not self._title_generated and user_input.strip() and not from_wake:
+            self._title_generated = True
+            threading.Thread(target=self._generate_title, daemon=True).start()
+
         # A fresh session composed its system prefix at __init__ with an empty
         # history, so memory selection fell back to recency (no query, no rerank).
         # Recompose once the first real user message exists so the opening turn
@@ -4292,10 +4316,6 @@ class ChatSession:
                         self._compact_messages(auto=True)
                         # Update status bar with post-compaction token counts
                         self._print_status_line()
-                    # Auto-title session after first exchange
-                    if not self._title_generated:
-                        self._title_generated = True
-                        threading.Thread(target=self._generate_title, daemon=True).start()
                     # Flush any queued messages that weren't injected
                     # (no tool calls → no advisory seam to inject at).
                     # If anything drained, the model hasn't seen those

turnstone/core/session_routes.py

@@ -493,9 +493,8 @@ class SharedSessionVerbHandlers:
     """Bundle of HTTP handler callables for verbs both kinds expose.
 
     All handlers are optional; ``None`` skips that route. One bundle
-    describes either kind — coord omits ``delete`` / ``refresh_title``
-    / ``set_title`` / attachments; interactive populates every
-    interaction verb post-Stage-2.
+    describes either kind — coord omits ``delete``; interactive
+    populates every interaction verb post-Stage-2.
     """
 
     # Listing
@@ -972,6 +971,122 @@ def make_close_handler(
     return close
 
 
+def make_refresh_title_handler(cfg: SessionEndpointConfig) -> Handler:
+    """Lifted body for ``POST {prefix}/{ws_id}/refresh-title``.
+
+    Regenerates the workstream title via a background LLM call
+    (:meth:`ChatSession.request_title_refresh`). Both kinds share the
+    auth → mgr → ws-lookup → request sequence; the session must be live
+    in memory (``mgr.get``, not ``open``) since the refresh runs on the
+    loaded :class:`ChatSession`. The current display name is passed so
+    the generator is steered toward a *different* title on a manual
+    refresh.
+    """
+
+    async def refresh_title(request: Request) -> Response:
+        import asyncio
+
+        from turnstone.core.memory import get_workstream_display_name
+
+        if cfg.permission_gate is not None:
+            err = cfg.permission_gate(request)
+            if err is not None:
+                return err
+        mgr_opt, err503 = cfg.manager_lookup(request)
+        if err503 is not None:
+            return err503
+        # See ``make_approve_handler`` for the cast rationale.
+        mgr = cast("SessionManager", mgr_opt)
+        ws_id = request.path_params.get("ws_id", "")
+
+        if cfg.tenant_check is not None:
+            err_tenant = await asyncio.to_thread(cfg.tenant_check, request, ws_id, mgr)
+            if err_tenant is not None:
+                return err_tenant
+
+        ws = mgr.get(ws_id)
+        if ws is None or ws.session is None:
+            return JSONResponse({"error": cfg.not_found_label}, status_code=404)
+
+        current_title = await asyncio.to_thread(get_workstream_display_name, ws_id) or ""
+        ws.session.request_title_refresh(current_title)
+        return JSONResponse({"status": "ok"})
+
+    return refresh_title
+
+
+def make_set_title_handler(cfg: SessionEndpointConfig) -> Handler:
+    """Lifted body for ``POST {prefix}/{ws_id}/title``.
+
+    Sets a user-chosen title manually. Stored as the workstream *alias*
+    so it outranks the LLM auto-title in the display fallback chain
+    (``alias > title > name``). Both kinds share the auth → validate →
+    ``set_workstream_alias`` → ``on_rename`` sequence. Returns 409 when
+    the name collides with another workstream's alias.
+
+    Behavior matches the pre-lift interactive handler: the alias is set
+    against storage regardless of whether the session is loaded (so a
+    saved/closed workstream can still be renamed), and the live
+    ``on_rename`` broadcast fires only when the session is in memory.
+    """
+
+    async def set_title(request: Request) -> Response:
+        import asyncio
+
+        from turnstone.core.memory import set_workstream_alias
+        from turnstone.core.web_helpers import read_json_or_400
+
+        if cfg.permission_gate is not None:
+            err = cfg.permission_gate(request)
+            if err is not None:
+                return err
+        mgr_opt, err503 = cfg.manager_lookup(request)
+        if err503 is not None:
+            return err503
+        mgr = cast("SessionManager", mgr_opt)
+        ws_id = request.path_params.get("ws_id", "")
+        if not ws_id:
+            return JSONResponse({"error": "ws_id is required"}, status_code=400)
+
+        if cfg.tenant_check is not None:
+            err_tenant = await asyncio.to_thread(cfg.tenant_check, request, ws_id, mgr)
+            if err_tenant is not None:
+                return err_tenant
+
+        # Resolve the workstream BEFORE writing the alias. ``set_workstream_alias``
+        # is a global, kind-unscoped UPDATE keyed on ``ws_id`` alone (it returns
+        # True even on a 0-row match), so a kind that has no ``tenant_check``
+        # storage gate (coord — the in-memory manager is its existence + kind
+        # authority) must 404 here, or an operator could rename a workstream this
+        # manager doesn't own (e.g. an interactive ws via the coord route) and a
+        # bogus id would silently 200. Interactive keeps ``tenant_check`` as its
+        # existence gate, so this stays skipped there and a non-loaded
+        # saved/closed ws still renames.
+        ws = mgr.get(ws_id)
+        if cfg.tenant_check is None and ws is None:
+            return JSONResponse({"error": cfg.not_found_label}, status_code=404)
+
+        body = await read_json_or_400(request)
+        if isinstance(body, JSONResponse):
+            return body
+        title = str(body.get("title", "")).strip()
+        if not title:
+            return JSONResponse({"error": "title is required"}, status_code=400)
+        title = title[:80]
+
+        if not await asyncio.to_thread(set_workstream_alias, ws_id, title):
+            return JSONResponse(
+                {"error": "That name is already used by another workstream"},
+                status_code=409,
+            )
+
+        if ws is not None and ws.session is not None and ws.session.ui is not None:
+            ws.session.ui.on_rename(title)
+        return JSONResponse({"status": "ok", "title": title})
+
+    return set_title
+
+
 CancelAuditEmitter = Callable[
     ["Request", str, "Workstream", bool],
     None,

turnstone/core/session_ui_base.py

@@ -201,8 +201,18 @@ class SessionUIBase:
     methods (and the approval blocking helpers that live on
     subclasses); HTTP handlers drive ``_register_listener`` /
     ``_unregister_listener`` / ``resolve_approval`` from the event
-    loop. All shared state is guarded by ``_listeners_lock`` or
-    ``threading.Event`` primitives.
+    loop. All shared state is guarded by ``_listeners_lock`` /
+    ``_ws_lock`` or ``threading.Event`` primitives.
+
+    Two ``on_*`` methods are additionally safe to call from a
+    *concurrent* auxiliary thread (e.g. background title generation in
+    ``ChatSession._generate_title``, or ``task_agent`` sub-agents), even
+    while the worker thread is mid-stream: :meth:`on_aux_usage` (a
+    storage ``usage_event`` write + thread-safe metric counters — it
+    touches none of the ``_ws_lock``-guarded inflight state
+    :meth:`on_status`/token writers mutate) and :meth:`on_rename` (a
+    queue / locked fan-out). Keep those two free of unguarded
+    ``_ws_*`` writes so the auxiliary-thread guarantee holds.
     """
 
     def __init__(self, ws_id: str = "", user_id: str = "") -> None:

turnstone/core/storage/__init__.py

@@ -8,6 +8,7 @@ from turnstone.core.storage._registry import (
     StorageUnavailableError,
     get_storage,
     init_storage,
+    is_storage_initialized,
     reset_storage,
 )
 
@@ -17,5 +18,6 @@ __all__ = [
     "StorageUnavailableError",
     "get_storage",
     "init_storage",
+    "is_storage_initialized",
     "reset_storage",
 ]

turnstone/core/storage/_postgresql.py

@@ -1052,6 +1052,12 @@ class PostgreSQLBackend:
                     workstreams.c.skill_id,
                     workstreams.c.skill_version,
                     workstreams.c.user_id,
+                    # Appended after ``user_id`` so positional fallbacks in
+                    # consumers (``_coord_children_row`` et al.) that index
+                    # up to row[9] stay valid; ``_coordinator_rows`` reads
+                    # these by name to surface the persisted display title.
+                    workstreams.c.title,
+                    workstreams.c.alias,
                 )
                 .order_by(workstreams.c.updated.desc())
                 .limit(limit)

turnstone/core/storage/_protocol.py

@@ -679,7 +679,9 @@ class StorageBackend(Protocol):
         Returns a list of SQLAlchemy ``Row`` objects.  **Prefer dict access
         via ``row._mapping[<col>]``**; positional indexing is brittle against
         future SELECT reorders and against new columns appearing in the
-        tail (the select currently ends with ``user_id``).
+        tail (the select currently ends with ``user_id, title, alias`` —
+        ``title``/``alias`` were appended after ``user_id`` so existing
+        positional fallbacks that index up to row[9] stay valid).
         """
         ...
 

turnstone/core/storage/_registry.py

@@ -124,6 +124,19 @@ def get_storage() -> StorageBackend:
     return _storage
 
 
+def is_storage_initialized() -> bool:
+    """Return True when the storage singleton has been initialized.
+
+    Lets callers on lifecycle / early-startup paths consult storage
+    without tripping :func:`get_storage`'s SQLite auto-init side effect
+    (which would create ``.turnstone.db`` in the CWD). Use this to guard
+    a best-effort read that should simply be skipped before the host has
+    called :func:`init_storage` — never as a substitute for the explicit
+    init the app's startup performs.
+    """
+    return _storage is not None
+
+
 def reset_storage() -> None:
     """Close and clear the storage backend singleton (for tests)."""
     global _storage

turnstone/core/storage/_sqlite.py

@@ -1209,6 +1209,12 @@ class SQLiteBackend:
                     workstreams.c.skill_id,
                     workstreams.c.skill_version,
                     workstreams.c.user_id,
+                    # Appended after ``user_id`` so positional fallbacks in
+                    # consumers (``_coord_children_row`` et al.) that index
+                    # up to row[9] stay valid; ``_coordinator_rows`` reads
+                    # these by name to surface the persisted display title.
+                    workstreams.c.title,
+                    workstreams.c.alias,
                 )
                 .order_by(workstreams.c.updated.desc())
                 .limit(limit)

turnstone/server.py

@@ -77,10 +77,12 @@ from turnstone.core.session_routes import (
     make_history_handler,
     make_list_handler,
     make_open_handler,
+    make_refresh_title_handler,
     make_retry_handler,
     make_rewind_handler,
     make_saved_handler,
     make_send_handler,
+    make_set_title_handler,
     register_session_routes,
 )
 from turnstone.core.session_ui_base import (
@@ -2309,74 +2311,6 @@ async def delete_workstream_endpoint(request: Request) -> JSONResponse:
         return JSONResponse({"error": "Delete failed"}, status_code=500)
 
 
-async def refresh_workstream_title(request: Request, ws_id: str = "") -> JSONResponse:
-    """POST /v1/api/workstreams/{ws_id}/refresh-title — regenerate workstream title via LLM."""
-    from turnstone.core.log import get_logger
-    from turnstone.core.memory import get_workstream_display_name
-
-    log = get_logger(__name__)
-    ws_id = request.path_params.get("ws_id", "")
-    log.info("ws.title.refresh_requested", ws_id=ws_id[:8] if ws_id else "empty")
-    mgr = request.app.state.workstreams
-    _owner, err = _require_ws_access(request, ws_id, mgr=mgr)
-    if err:
-        return err
-    ws = mgr.get(ws_id)
-    if not ws or not ws.session:
-        log.warning(
-            "ws.title.refresh_failed",
-            ws_id=ws_id[:8] if ws_id else "empty",
-            reason="workstream_not_found",
-        )
-        return JSONResponse({"error": "Workstream not found or not active"}, status_code=404)
-    # Fetch current title so the LLM can generate something different
-    current_title = get_workstream_display_name(ws_id) or ""
-    log.info("ws.title.refresh_triggered", ws_id=ws_id[:8], current_title=current_title[:50])
-    ws.session.request_title_refresh(current_title)
-    return JSONResponse({"status": "ok"})
-
-
-async def set_workstream_title(request: Request, ws_id: str = "") -> JSONResponse:
-    """POST /v1/api/workstreams/{ws_id}/title — set workstream title manually.
-
-    Stores the user-chosen title as the workstream *alias* so it takes
-    priority over the LLM auto-generated title in the display name
-    fallback chain (alias -> title -> name).
-    """
-    from turnstone.core.log import get_logger
-    from turnstone.core.memory import set_workstream_alias
-    from turnstone.core.web_helpers import read_json_or_400
-
-    log = get_logger(__name__)
-    ws_id = request.path_params.get("ws_id", "")
-    log.info("ws.title.set_requested", ws_id=ws_id[:8] if ws_id else "empty")
-    if not ws_id:
-        return JSONResponse({"error": "ws_id is required"}, status_code=400)
-    mgr = request.app.state.workstreams
-    _owner, err = _require_ws_access(request, ws_id, mgr=mgr)
-    if err:
-        return err
-    body = await read_json_or_400(request)
-    if isinstance(body, JSONResponse):
-        return body
-    title = str(body.get("title", "")).strip()
-    if not title:
-        return JSONResponse({"error": "title is required"}, status_code=400)
-    title = title[:80]
-    if not set_workstream_alias(ws_id, title):
-        log.warning("ws.title.set_alias_conflict", ws_id=ws_id[:8], title=title[:50])
-        return JSONResponse(
-            {"error": "That name is already used by another workstream"},
-            status_code=409,
-        )
-    log.info("ws.title.set_alias_updated", ws_id=ws_id[:8])
-    ws = mgr.get(ws_id)
-    if ws and ws.session and ws.session.ui:
-        ws.session.ui.on_rename(title)
-    log.info("ws.title.set_success", ws_id=ws_id[:8], title=title)
-    return JSONResponse({"status": "ok", "title": title})
-
-
 def _auth_user_id(request: Request) -> str:
     """Return the authenticated user's id (empty string when absent).
 
@@ -3959,6 +3893,8 @@ def create_app(
     history_handler = make_history_handler(interactive_endpoint_config)
     export_handler = make_export_handler(interactive_endpoint_config)
     detail_handler = make_detail_handler(interactive_endpoint_config)
+    refresh_title_handler = make_refresh_title_handler(interactive_endpoint_config)
+    set_title_handler = make_set_title_handler(interactive_endpoint_config)
     v1_routes: list[Any] = [
         Route("/api/events/global", global_events_sse),
     ]
@@ -3973,8 +3909,8 @@ def create_app(
             detail=detail_handler,  # lifted: shared body (interactive feature gain)
             open=open_handler,  # lifted: shared body
             close=close_handler,  # lifted: shared body
-            refresh_title=refresh_workstream_title,
-            set_title=set_workstream_title,
+            refresh_title=refresh_title_handler,  # lifted: shared body
+            set_title=set_title_handler,  # lifted: shared body
             send=send_handler,  # lifted: shared body (P1.5)
             dequeue=dequeue_handler,  # lifted (P1.5) — DELETE /send
             approve=approve_handler,  # lifted: shared body

turnstone/shared_static/shell.js

@@ -837,6 +837,11 @@ async function mountShell() {
         });
         pane.tabMenu = () =>
           convTabMenu(pane, pm, id, {
+            // Coordinators carry titles like interactive workstreams now:
+            // surface Refresh/Edit title. The default base ("") targets the
+            // console origin, where the coord refresh-title / title routes
+            // are mounted (same base coordinator.js posts every verb to).
+            titleVerbs: true,
             closeSession: () => {
               if (pane._ctl && pane._ctl.closeSession) pane._ctl.closeSession();
             },

uv.lock

@@ -2437,7 +2437,7 @@ wheels = [
 
 [[package]]
 name = "turnstone"
-version = "1.6.8"
+version = "1.6.9"
 source = { editable = "." }
 dependencies = [
     { name = "alembic" },