chore: bump version to 1.5.1

The pre-existing comment said pending_approval_detail "rides on
every ws_state event" — that overstated the case.  The node-side
emit is gated on ``_pending_approval is not None`` so the field is
absent on the steady-state broadcast and possibly null on a node
mid-rolling-upgrade.  The handleChildState fallback already
handles both cases; only the comment was wrong.

pyproject.toml

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

tests/test_admin_model_registry_refresh.py

@@ -0,0 +1,397 @@
+"""Console-side coord_registry auto-refresh on model-definition CRUD + reload.
+
+The console builds ``app.state.coord_registry`` once at lifespan startup
+and the coordinator session factory closes over that exact instance.
+Without these refresh hooks, an admin who edits a model definition
+through the UI sees the DB change immediately but coordinator sessions
+keep calling the prior model name — the on-disk truth diverges from the
+in-process registry until the console is restarted.
+
+These tests cover both the helper (``_refresh_coord_registry``)
+and the four wired endpoints (create / update / delete / explicit reload)
+to lock in:
+
+- in-place mutation: ``coord_registry`` object identity is preserved
+  across refreshes (factory closure must not be invalidated);
+- failure isolation: a load or reload failure leaves the existing
+  registry intact rather than tearing down a working coordinator;
+- no-op safety: the helper short-circuits when ``coord_registry`` is
+  ``None`` so a coord-less console (no model rows at boot) doesn't
+  500 on routine model-definition CRUD.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+from starlette.applications import Starlette
+from starlette.middleware import Middleware
+from starlette.routing import Route
+from starlette.testclient import TestClient
+
+from tests._coord_test_helpers import _AuthMiddleware
+from turnstone.console.server import (
+    _refresh_coord_registry,
+    admin_create_model_definition,
+    admin_delete_model_definition,
+    admin_model_reload,
+    admin_update_model_definition,
+)
+from turnstone.core.model_registry import ModelConfig, ModelRegistry
+from turnstone.core.storage._sqlite import SQLiteBackend
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def storage(tmp_path: Any) -> SQLiteBackend:
+    return SQLiteBackend(str(tmp_path / "models.db"))
+
+
+def _seed_model_def(
+    storage: SQLiteBackend,
+    *,
+    definition_id: str,
+    alias: str,
+    model: str,
+    base_url: str = "http://localhost:8000/v1",
+    enabled: bool = True,
+) -> None:
+    """Insert a model definition row directly via the storage API."""
+    storage.create_model_definition(
+        definition_id=definition_id,
+        alias=alias,
+        model=model,
+        provider="openai-compatible",
+        base_url=base_url,
+        api_key="sk-test",
+        context_window=8192,
+        capabilities="{}",
+        enabled=enabled,
+        created_by="admin",
+    )
+
+
+def _make_config(alias: str, model: str) -> ModelConfig:
+    return ModelConfig(
+        alias=alias,
+        base_url="http://localhost:8000/v1",
+        api_key="sk-test",
+        model=model,
+        context_window=8192,
+        provider="openai-compatible",
+        source="db",
+    )
+
+
+def _make_registry(
+    *,
+    alias: str = "local",
+    model: str = "old-model",
+    extras: dict[str, str] | None = None,
+) -> ModelRegistry:
+    """Build a real ModelRegistry seeded with ``alias`` (the default) plus
+    any ``extras`` (alias → model).  ``ModelRegistry.__init__`` rejects an
+    empty model dict so tests that exercise the helper need at least one
+    entry; pass ``extras`` for multi-alias scenarios (e.g. delete-by-alias).
+    """
+    configs = {alias: _make_config(alias, model)}
+    for extra_alias, extra_model in (extras or {}).items():
+        configs[extra_alias] = _make_config(extra_alias, extra_model)
+    return ModelRegistry(configs, default=alias)
+
+
+class _AppState:
+    """Shim mirroring Starlette's ``app.state`` for direct helper tests."""
+
+    coord_registry: ModelRegistry | None = None
+
+
+# ---------------------------------------------------------------------------
+# Helper-level tests — ``_refresh_coord_registry`` semantics
+# ---------------------------------------------------------------------------
+
+
+def test_helper_rebuilds_registry_from_db(storage: SQLiteBackend) -> None:
+    """Helper pulls the latest DB rows into the existing registry."""
+    _seed_model_def(storage, definition_id="m1", alias="local", model="new-model")
+    state = _AppState()
+    state.coord_registry = _make_registry(alias="local", model="old-model")
+
+    _refresh_coord_registry(state, storage)
+
+    assert state.coord_registry is not None
+    assert state.coord_registry.get_config("local").model == "new-model"
+
+
+def test_helper_preserves_object_identity(storage: SQLiteBackend) -> None:
+    """The factory closes over the registry object — refresh must mutate
+    in place rather than swap the attribute."""
+    _seed_model_def(storage, definition_id="m1", alias="local", model="new-model")
+    state = _AppState()
+    state.coord_registry = _make_registry()
+    before = id(state.coord_registry)
+
+    _refresh_coord_registry(state, storage)
+
+    assert id(state.coord_registry) == before
+
+
+def test_helper_noop_when_coord_registry_none(storage: SQLiteBackend) -> None:
+    """Console boot with no model rows leaves coord_registry = None.
+    The helper must not 500 in that state — CRUD that lands the FIRST
+    row would otherwise fail before the operator can recover."""
+    _seed_model_def(storage, definition_id="m1", alias="local", model="m")
+    state = _AppState()
+    state.coord_registry = None
+
+    _refresh_coord_registry(state, storage)  # must not raise
+
+    assert state.coord_registry is None
+
+
+def test_helper_preserves_registry_when_load_fails(
+    storage: SQLiteBackend, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """An unexpected error from ``load_model_registry`` (e.g. config.toml
+    parse failure, programming bug) must not tear down a working
+    registry — log + leave the existing instance intact."""
+    state = _AppState()
+    state.coord_registry = _make_registry(alias="local", model="old-model")
+
+    def _boom(**_kw: Any) -> ModelRegistry:
+        raise RuntimeError("simulated loader failure")
+
+    monkeypatch.setattr("turnstone.core.model_registry.load_model_registry", _boom)
+    _refresh_coord_registry(state, storage)
+
+    assert state.coord_registry is not None
+    assert state.coord_registry.get_config("local").model == "old-model"
+
+
+def test_helper_preserves_registry_when_strict_load_fails(
+    storage: SQLiteBackend, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """``load_model_registry`` normally swallows storage read errors and
+    would return a config.toml-only registry on a transient DB outage —
+    applying that via ``reload()`` would silently drop every DB-sourced
+    alias.  The helper passes ``strict=True`` so the loader re-raises
+    instead, the helper's outer except catches it, and the existing
+    registry survives intact."""
+    _seed_model_def(storage, definition_id="m1", alias="local", model="db-model")
+    state = _AppState()
+    state.coord_registry = _make_registry(alias="local", model="db-model")
+
+    def _broken(**_kw: Any) -> Any:
+        raise RuntimeError("simulated transient DB outage")
+
+    monkeypatch.setattr(storage, "list_model_definitions", _broken)
+    _refresh_coord_registry(state, storage)
+
+    assert state.coord_registry is not None
+    # Existing registry untouched — strict=True surfaced the storage
+    # error to the helper before the loader's silent fallback could
+    # produce a truncated registry for reload().
+    assert state.coord_registry.get_config("local").model == "db-model"
+
+
+def test_helper_preserves_registry_when_no_enabled_rows(storage: SQLiteBackend) -> None:
+    """All rows disabled/deleted: ModelRegistry.__init__ rejects an empty
+    model dict (raises ValueError).  Helper must catch and preserve the
+    existing registry so coord stays usable while admin restores rows."""
+    _seed_model_def(storage, definition_id="m1", alias="local", model="m", enabled=False)
+    state = _AppState()
+    state.coord_registry = _make_registry(alias="local", model="cached-model")
+
+    _refresh_coord_registry(state, storage)
+
+    assert state.coord_registry is not None
+    assert state.coord_registry.get_config("local").model == "cached-model"
+
+
+def test_helper_preserves_registry_on_reload_validation_error(
+    storage: SQLiteBackend, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A reload that raises mid-mutation (e.g. validation guard) must
+    leave the existing registry instance functional."""
+    _seed_model_def(storage, definition_id="m1", alias="local", model="new-model")
+    state = _AppState()
+    state.coord_registry = _make_registry(alias="local", model="old-model")
+
+    def _broken_reload(*_a: Any, **_kw: Any) -> None:
+        raise ValueError("simulated reload validation failure")
+
+    monkeypatch.setattr(state.coord_registry, "reload", _broken_reload)
+    _refresh_coord_registry(state, storage)
+
+    # Existing registry still reachable; the broken reload was a no-op
+    # at the public-facing level.
+    assert state.coord_registry is not None
+    assert state.coord_registry.get_config("local").model == "old-model"
+
+
+# ---------------------------------------------------------------------------
+# Endpoint-level integration tests — verify wiring
+# ---------------------------------------------------------------------------
+
+
+def _make_client(storage: SQLiteBackend, registry: ModelRegistry | None) -> TestClient:
+    """Build a TestClient wired to the four model-definition endpoints.
+
+    Uses the shared header-driven ``_AuthMiddleware`` from
+    ``tests/_coord_test_helpers``; default headers below grant
+    ``admin.models`` permission so the endpoint gate passes.
+    """
+    app = Starlette(
+        routes=[
+            Route(
+                "/v1/api/admin/model-definitions",
+                admin_create_model_definition,
+                methods=["POST"],
+            ),
+            Route(
+                "/v1/api/admin/model-definitions/reload",
+                admin_model_reload,
+                methods=["POST"],
+            ),
+            Route(
+                "/v1/api/admin/model-definitions/{definition_id}",
+                admin_update_model_definition,
+                methods=["PUT"],
+            ),
+            Route(
+                "/v1/api/admin/model-definitions/{definition_id}",
+                admin_delete_model_definition,
+                methods=["DELETE"],
+            ),
+        ],
+        middleware=[Middleware(_AuthMiddleware)],
+    )
+    app.state.auth_storage = storage
+    app.state.coord_registry = registry
+    # Reload endpoint also touches these — stub them so the test focuses
+    # on the registry-refresh behaviour without dragging in a full
+    # collector / proxy_client wiring.
+    app.state.collector = MagicMock()
+    app.state.collector.get_all_nodes.return_value = []
+    app.state.proxy_client = MagicMock()
+    app.state.config_store = MagicMock()
+    client = TestClient(app)
+    client.headers.update({"X-Test-User": "admin", "X-Test-Perms": "admin.models"})
+    return client
+
+
+def test_create_endpoint_refreshes_registry(storage: SQLiteBackend) -> None:
+    """POST /api/admin/model-definitions bumps the in-process registry
+    so newly-spawned coord sessions see the new alias immediately."""
+    # Pre-existing alias (registry needs at least one row)
+    _seed_model_def(storage, definition_id="m1", alias="local", model="m")
+    registry = _make_registry(alias="local", model="m")
+    client = _make_client(storage, registry)
+
+    resp = client.post(
+        "/v1/api/admin/model-definitions",
+        json={
+            "alias": "fast",
+            "model": "fast-model",
+            "provider": "openai-compatible",
+            "base_url": "http://localhost:9000/v1",
+            "api_key": "sk-x",
+            "context_window": 4096,
+        },
+    )
+    assert resp.status_code == 200, resp.text
+    assert registry.has_alias("fast")
+    assert registry.get_config("fast").model == "fast-model"
+
+
+def test_update_endpoint_refreshes_registry(storage: SQLiteBackend) -> None:
+    """PUT swaps the underlying model name behind a stable alias — the
+    user's reported regression."""
+    _seed_model_def(storage, definition_id="m1", alias="local", model="old-model")
+    registry = _make_registry(alias="local", model="old-model")
+    client = _make_client(storage, registry)
+
+    resp = client.put(
+        "/v1/api/admin/model-definitions/m1",
+        json={"model": "new-model"},
+    )
+    assert resp.status_code == 200, resp.text
+    assert registry.get_config("local").model == "new-model"
+
+
+def test_update_endpoint_skips_refresh_on_empty_body(
+    storage: SQLiteBackend, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """An empty PUT body must skip the registry refresh — the
+    ``if updates:`` gate exists because ``load_model_registry`` is
+    non-trivial and a no-op refresh on every PUT would burn cycles
+    rebuilding state that hasn't changed.  Spy on the helper to lock
+    the gate down: a regression that drops the conditional would
+    register a call here and trip the assertion.
+    """
+    from turnstone.console import server as server_module
+
+    _seed_model_def(storage, definition_id="m1", alias="local", model="locked-in")
+    registry = _make_registry(alias="local", model="locked-in")
+    client = _make_client(storage, registry)
+
+    calls: list[tuple[Any, Any]] = []
+
+    def _spy(app_state: Any, storage: Any) -> None:
+        calls.append((app_state, storage))
+
+    monkeypatch.setattr(server_module, "_refresh_coord_registry", _spy)
+
+    resp = client.put("/v1/api/admin/model-definitions/m1", json={})
+    assert resp.status_code == 200, resp.text
+    assert calls == []  # gate held: empty body did not trigger a refresh
+
+
+def test_delete_endpoint_refreshes_registry(storage: SQLiteBackend) -> None:
+    """DELETE drops the alias from the in-process registry too — a
+    coord session that tried to resolve the deleted alias would
+    otherwise hit a stale cached client."""
+    _seed_model_def(storage, definition_id="m1", alias="local", model="m")
+    _seed_model_def(storage, definition_id="m2", alias="extra", model="x")
+    registry = _make_registry(alias="local", model="m", extras={"extra": "x"})
+    client = _make_client(storage, registry)
+
+    resp = client.delete("/v1/api/admin/model-definitions/m2")
+    assert resp.status_code == 200, resp.text
+    assert not registry.has_alias("extra")
+    assert registry.has_alias("local")  # default alias unaffected
+
+
+def test_reload_endpoint_refreshes_registry(
+    storage: SQLiteBackend, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The explicit reload button must refresh the console's own
+    registry — until this PR it only fanned out to nodes."""
+    _seed_model_def(storage, definition_id="m1", alias="local", model="initial")
+    registry = _make_registry(alias="local", model="initial")
+    client = _make_client(storage, registry)
+
+    # Bypass the CRUD endpoints to mimic an out-of-band DB change (e.g.
+    # an operator psql session) and verify the explicit reload path
+    # still pulls the change in.
+    storage.update_model_definition("m1", model="reloaded-model")
+
+    # Stub the async cluster fan-out helpers — they require a fully-wired
+    # collector / proxy_client which is orthogonal to the helper under test.
+    async def _noop_publish(_request: Any) -> None:
+        return None
+
+    async def _noop_notify(_request: Any) -> dict[str, Any]:
+        return {}
+
+    monkeypatch.setattr("turnstone.console.server._publish_config_change", _noop_publish)
+    monkeypatch.setattr("turnstone.console.server._notify_nodes_model_reload", _noop_notify)
+
+    resp = client.post("/v1/api/admin/model-definitions/reload")
+    assert resp.status_code == 200, resp.text
+    assert registry.get_config("local").model == "reloaded-model"

tests/test_console.py

@@ -314,6 +314,48 @@ class TestCollectorSnapshot:
         assert event["ws_id"] == "ws1"
         assert event["state"] == "running"
 
+    def test_apply_snapshot_state_change_forwards_pending_approval_detail(self):
+        """Reconnect-via-snapshot is the resync path after every console
+        restart or network blip.  Without forwarding the field here,
+        a child sitting in approval-pending across the gap renders as
+        ``activity_state=approval`` with no buttons until the next
+        state change — broken UX during the most common re-sync event."""
+        c = _make_collector()
+        c._nodes["node-a"] = NodeSnapshot(
+            node_id="node-a",
+            server_url="http://a:8080",
+            workstreams={"ws1": {"id": "ws1", "name": "same", "state": "idle"}},
+        )
+        q: queue.Queue[dict] = queue.Queue()
+        c.register_listener(q)
+
+        detail = {
+            "items": [{"call_id": "c1", "header": "tool x"}],
+            "judge_pending": False,
+        }
+        c._apply_snapshot(
+            "node-a",
+            {
+                "type": "node_snapshot",
+                "node_id": "node-a",
+                "workstreams": [
+                    {
+                        "id": "ws1",
+                        "name": "same",
+                        "state": "running",
+                        "activity_state": "approval",
+                        "pending_approval_detail": detail,
+                    }
+                ],
+                "health": {},
+                "aggregate": {},
+            },
+        )
+
+        event = q.get_nowait()
+        assert event["type"] == "cluster_state"
+        assert event["pending_approval_detail"] == detail
+
     def test_apply_snapshot_skips_empty_id_workstream(self):
         c = _make_collector()
         c._nodes["node-a"] = NodeSnapshot(node_id="node-a", server_url="http://a:8080")
@@ -359,6 +401,40 @@ class TestCollectorDelta:
         # Verify in-memory state was updated
         assert c._nodes["node-a"].workstreams["ws1"]["state"] == "running"
 
+    def test_apply_delta_ws_state_forwards_pending_approval_detail(self):
+        """The rich approval payload now travels on the cluster bus so
+        coord tabs can render inline approve/deny buttons in lockstep
+        with the activity_state transition.  Collector must forward
+        the field verbatim — the adapter does the child-routing on
+        top, but the bus carries the data."""
+        c = _make_collector()
+        c._nodes["node-a"] = NodeSnapshot(
+            node_id="node-a",
+            server_url="http://a:8080",
+            workstreams={"ws1": {"id": "ws1", "name": "test", "state": "idle"}},
+        )
+        q: queue.Queue[dict] = queue.Queue()
+        c.register_listener(q)
+
+        detail = {
+            "items": [{"call_id": "c1", "header": "tool x"}],
+            "judge_pending": False,
+        }
+        c._apply_delta(
+            "node-a",
+            {
+                "type": "ws_state",
+                "ws_id": "ws1",
+                "state": "running",
+                "activity_state": "approval",
+                "pending_approval_detail": detail,
+            },
+        )
+
+        event = q.get_nowait()
+        assert event["type"] == "cluster_state"
+        assert event["pending_approval_detail"] == detail
+
     def test_apply_delta_ws_created(self):
         c = _make_collector()
         c._nodes["node-a"] = NodeSnapshot(node_id="node-a", server_url="http://a:8080")

tests/test_coordinator_adapter.py

@@ -558,3 +558,54 @@ class TestCoordinatorAdapterDispatchChildEvent:
             }
         )
         assert recorder.enqueued[0]["ws_id"] == "coord-a"
+
+    def test_dispatch_cluster_state_forwards_pending_approval_detail(self) -> None:
+        """The rich approval payload now rides on child_ws_state directly so
+        the browser can mutate liveBadgeCache without a separate live-bulk
+        fetch.  Drift here means the inline approve/deny buttons would
+        regress to chasing the dashboard cache (the load-storm pattern
+        Shape A is unwinding)."""
+        adapter, recorder, _ = self._setup()
+        with adapter._children_lock:
+            adapter._merge_child_ids_locked("coord-a", ["child-a1"])
+        detail = {
+            "items": [{"call_id": "c1", "header": "tool x"}],
+            "judge_pending": False,
+        }
+        adapter._dispatch_child_event(
+            {
+                "type": "cluster_state",
+                "ws_id": "child-a1",
+                "state": "running",
+                "activity_state": "approval",
+                "pending_approval_detail": detail,
+            }
+        )
+        assert len(recorder.enqueued) == 1
+        payload = recorder.enqueued[0]
+        assert payload["type"] == "child_ws_state"
+        assert payload["activity_state"] == "approval"
+        assert payload["pending_approval_detail"] == detail
+
+    def test_dispatch_cluster_state_pending_approval_detail_none_passes_through(
+        self,
+    ) -> None:
+        """Missing pending_approval_detail (no approval pending, or pre-fix
+        node mid-rolling-upgrade) must forward as None — not raise, not
+        omit — so the browser's handleChildState treats it as "no SSE-
+        supplied detail, fall back to cached value"."""
+        adapter, recorder, _ = self._setup()
+        with adapter._children_lock:
+            adapter._merge_child_ids_locked("coord-a", ["child-a1"])
+        adapter._dispatch_child_event(
+            {
+                "type": "cluster_state",
+                "ws_id": "child-a1",
+                "state": "running",
+                "activity_state": "tool",
+            }
+        )
+        assert len(recorder.enqueued) == 1
+        payload = recorder.enqueued[0]
+        assert "pending_approval_detail" in payload
+        assert payload["pending_approval_detail"] is None

tests/test_coordinator_page.py

@@ -152,3 +152,95 @@ def test_coordinator_js_exposes_inline_approval_helpers():
     # any prior denial.  bug-1 / bug-3 from the second /review pass.
     assert "Denied by user" in body
     assert "callOutcomes" in body
+
+
+def test_coordinator_js_handle_child_state_reads_sse_pending_approval_detail():
+    """Lock the Shape A behavior change: child_ws_state SSE events now
+    carry ``pending_approval_detail`` directly so the browser mutates
+    ``liveBadgeCache`` without firing an urgent live-bulk fetch on
+    every activity_state transition into/out of approval.  A refactor
+    that re-introduces the urgent-fetch path on routine transitions
+    (or drops the SSE-source merge guard in flushLiveFetches) would
+    re-open the load-storm pattern this PR is fixing.
+
+    Structural assertions (regex against multi-line source) — symbol-
+    presence alone wouldn't catch a guard that keeps the names but
+    inverts the comparison or drops the ``prev.live`` check.  This
+    codebase has no JS test framework, so locking the guard's shape
+    here is the next-best thing to a behavioral test."""
+    import re
+    from pathlib import Path
+
+    coord_js = Path(__file__).resolve().parent.parent / (
+        "turnstone/console/static/coordinator/coordinator.js"
+    )
+    body = coord_js.read_text(encoding="utf-8")
+
+    # handleChildState now reads the SSE-supplied detail.
+    assert "ev.pending_approval_detail" in body
+    # The pre-fix urgent-fetch on activity_state transitions is
+    # gone (the 409 retry path keeps its own ``{ urgent: true }``
+    # for stale-call_id refresh — that's a different scenario).
+    assert "enteredApproval" not in body
+    assert "leftApproval" not in body
+
+    # SSE-authoritative window constant is defined and used.
+    assert re.search(r"\bconst\s+SSE_AUTHORITATIVE_MS\s*=\s*\d+", body), (
+        "SSE_AUTHORITATIVE_MS constant must be defined as a numeric literal"
+    )
+
+    # handleChildState writes sseUpdatedAt = Date.now() into the cache
+    # entry it sets.  This is the SSE-source tag; without it, the
+    # merge guard in flushLiveFetches has nothing to gate on.
+    assert re.search(
+        r"sseUpdatedAt:\s*Date\.now\(\)",
+        body,
+    ), "handleChildState must write sseUpdatedAt: Date.now() onto liveBadgeCache entries"
+
+    # flushLiveFetches' merge guard structure: SSE-set pending_approval
+    # / _detail wins over a stale bulk-poll snapshot when (live) AND
+    # (prev exists) AND (prev.sseUpdatedAt set) AND (within window)
+    # AND (prev.live exists).  Inverting the comparison or dropping
+    # any of these guards reopens the clobber bug.
+    merge_guard = re.search(
+        r"if\s*\(\s*live\s*&&\s*prev\s*&&\s*prev\.sseUpdatedAt\s*&&\s*"
+        r"now\s*-\s*prev\.sseUpdatedAt\s*<\s*SSE_AUTHORITATIVE_MS\s*&&\s*"
+        r"prev\.live\s*\)",
+        body,
+    )
+    assert merge_guard is not None, (
+        "flushLiveFetches merge guard must be the conjunction "
+        "(live && prev && prev.sseUpdatedAt && now - prev.sseUpdatedAt < "
+        "SSE_AUTHORITATIVE_MS && prev.live).  An inverted comparison or "
+        "missing prev.live check would let a stale bulk-poll clobber a "
+        "fresh SSE-set approval."
+    )
+
+    # The merge body must preserve BOTH pending_approval and
+    # pending_approval_detail from prev — preserving only one would
+    # render a row with a phantom badge but no buttons (or vice versa).
+    merge_body = re.search(
+        r"mergedLive\s*=\s*Object\.assign\(\s*\{\}\s*,\s*live\s*,\s*\{"
+        r"[^}]*pending_approval:\s*prev\.live\.pending_approval[^}]*"
+        r"pending_approval_detail:\s*prev\.live\.pending_approval_detail",
+        body,
+    )
+    assert merge_body is not None, (
+        "Merge body must preserve both pending_approval AND "
+        "pending_approval_detail from prev.live — preserving only one "
+        "creates a half-rendered approval row."
+    )
+
+    # flushLiveFetches must forward sseUpdatedAt onto the new cache
+    # entry so the SSE-source tag survives the bulk-poll write back —
+    # without this, every bulk-poll resets the window and the next
+    # late-arriving poll silently clobbers.
+    assert re.search(
+        r"sseUpdatedAt:\s*prev\s*\?\s*prev\.sseUpdatedAt",
+        body,
+    ), (
+        "flushLiveFetches must forward prev.sseUpdatedAt onto the new "
+        "cache entry (preserving the SSE-source window across bulk-poll "
+        "cycles) — without this, the second bulk-poll after an SSE "
+        "transition silently clobbers."
+    )

tests/test_model_registry.py

@@ -770,16 +770,76 @@ class TestRegistryReload:
         assert reg.has_alias("b")
         assert reg.default == "b"
 
-    def test_reload_clears_clients(self) -> None:
-        models = {"a": ModelConfig("a", "http://x/v1", "key", "m")}
+    def test_reload_keeps_clients_when_connection_target_unchanged(self) -> None:
+        """Selective teardown: a model edit that leaves base_url / api_key /
+        provider intact (e.g. admin tweaks the underlying ``model`` name or
+        ``temperature``) keeps the cached HTTP client warm — no need to
+        re-establish TLS+pool when the endpoint is the same."""
+        models = {"a": ModelConfig("a", "http://x/v1", "key", "m1", provider="openai")}
         reg = ModelRegistry(models=models, default="a")
-        # Force client creation
         reg.get_client("a")
-        assert "a" in reg._clients
+        client_before = reg._clients["a"]
+        provider_before = reg.get_provider("a")
+
+        # Same endpoint (base_url, api_key, provider), only ``model`` changed.
+        new_models = {"a": ModelConfig("a", "http://x/v1", "key", "m2", provider="openai")}
+        reg.reload(new_models, "a")
+
+        assert "a" in reg._clients
+        assert reg._clients["a"] is client_before
+        assert "a" in reg._providers
+        assert reg._providers["a"] is provider_before
+
+    def test_reload_drops_client_when_base_url_changes(self) -> None:
+        """A ``base_url`` change drops the cached client (different
+        endpoint = new connection) but keeps the cached provider —
+        ``LLMProvider`` is keyed only on the provider string, which
+        didn't change."""
+        models = {"a": ModelConfig("a", "http://x/v1", "key", "m", provider="openai")}
+        reg = ModelRegistry(models=models, default="a")
+        reg.get_client("a")
+        provider_before = reg.get_provider("a")
+
+        new_models = {"a": ModelConfig("a", "http://y/v1", "key", "m", provider="openai")}
+        reg.reload(new_models, "a")
 
-        # Reload with same models — clients should be cleared
-        reg.reload(dict(models), "a")
         assert "a" not in reg._clients
+        assert "a" in reg._providers
+        assert reg._providers["a"] is provider_before
+
+    def test_reload_drops_provider_when_provider_string_changes(self) -> None:
+        """A provider-type swap (e.g. openai → anthropic) drops both the
+        client AND the provider so the next resolve picks up the right
+        ``LLMProvider`` implementation against the new SDK."""
+        models = {"a": ModelConfig("a", "http://x/v1", "key", "m", provider="openai")}
+        reg = ModelRegistry(models=models, default="a")
+        reg.get_client("a")
+        reg.get_provider("a")
+
+        new_models = {"a": ModelConfig("a", "http://x/v1", "key", "m", provider="anthropic")}
+        reg.reload(new_models, "a")
+
+        assert "a" not in reg._clients
+        assert "a" not in reg._providers
+
+    def test_reload_drops_clients_for_removed_aliases(self) -> None:
+        """Aliases removed from the registry must release their cached
+        clients — otherwise a deleted endpoint's connection pool would
+        outlive the alias indefinitely."""
+        models = {
+            "a": ModelConfig("a", "http://x/v1", "key", "m"),
+            "b": ModelConfig("b", "http://y/v1", "key", "m"),
+        }
+        reg = ModelRegistry(models=models, default="a")
+        reg.get_client("a")
+        reg.get_client("b")
+
+        # Drop "b" entirely.
+        new_models = {"a": ModelConfig("a", "http://x/v1", "key", "m")}
+        reg.reload(new_models, "a")
+
+        assert "a" in reg._clients  # unchanged endpoint, kept warm
+        assert "b" not in reg._clients
 
     def test_reload_validates_default(self) -> None:
         models_a = {"a": ModelConfig("a", "x", "x", "m")}

tests/test_webui_content.py

@@ -155,3 +155,87 @@ class TestContentAccumulation:
         assert len(idle_events) == 1
         # Content should be capped, not contain everything
         assert len(idle_events[0]["content"]) <= _MAX_TURN_CONTENT_CHARS + 1024
+
+
+class TestPendingApprovalDetailGate:
+    """The Shape A SSE plumbing carries ``pending_approval_detail`` on the
+    ``ws_state`` event so the coord tree UI can render inline approve/deny
+    buttons in lockstep with the activity_state transition.  The gate
+    (``if self._pending_approval is not None``) keeps the per-broadcast
+    serializer cost off the common no-approval-pending path — these tests
+    lock both branches down."""
+
+    def test_state_broadcast_omits_field_when_no_approval_pending(self):
+        """Common case: no approval pending → field absent from event so the
+        per-broadcast verdict-cache deepcopy in
+        ``serialize_pending_approval_detail`` never runs.  A regression
+        that drops the gate would silently 10x the cost of every state
+        broadcast in the steady state."""
+        ui = _make_ui()
+        assert ui._pending_approval is None
+        ui._broadcast_state("running")
+
+        events = _drain_global()
+        running_events = [e for e in events if e.get("state") == "running"]
+        assert len(running_events) == 1
+        assert "pending_approval_detail" not in running_events[0]
+
+    def test_state_broadcast_includes_field_when_approval_pending(self):
+        """When an approval is pending the broadcast must carry the rich
+        payload — the coord tree UI reads it directly to render inline
+        approve/deny buttons.  Without this, a coord browser would have
+        to chase a separate ``cluster/ws/live`` fetch on every
+        activity_state transition (the load-storm pattern Shape A is
+        unwinding)."""
+        ui = _make_ui()
+        # Mirror the shape ``pause_for_approval`` writes (session_ui_base
+        # lines 576-580) — items with call_id + header is the minimum
+        # the serializer needs to project.
+        ui._pending_approval = {
+            "type": "approve_request",
+            "items": [
+                {
+                    "call_id": "c1",
+                    "header": "tool x",
+                    "func_args": "{}",
+                    "intent_summary": "do x",
+                    "needs_approval": True,
+                }
+            ],
+            "judge_pending": False,
+        }
+        ui._broadcast_state("attention")
+
+        events = _drain_global()
+        attn = [e for e in events if e.get("state") == "attention"]
+        assert len(attn) == 1
+        # Field present and structurally sound — the serializer's
+        # full shape is covered by tests/test_session_ui_base.py;
+        # here we only need to confirm the gate fires and the
+        # serializer's output is what lands on the event.
+        assert "pending_approval_detail" in attn[0]
+        detail = attn[0]["pending_approval_detail"]
+        assert detail is not None
+        assert detail.get("items")
+        assert detail["items"][0]["call_id"] == "c1"
+
+    def test_field_cleared_after_approval_resolves(self):
+        """Once ``_pending_approval`` is cleared, subsequent state
+        broadcasts must drop the field again — without this, the
+        browser would render stale approve/deny buttons until the
+        next bulk-poll TTL window expired."""
+        ui = _make_ui()
+        ui._pending_approval = {
+            "type": "approve_request",
+            "items": [{"call_id": "c1", "header": "x"}],
+            "judge_pending": False,
+        }
+        ui._broadcast_state("attention")
+        _drain_global()  # discard the with-detail event
+
+        ui._pending_approval = None
+        ui._broadcast_state("running")
+        events = _drain_global()
+        running = [e for e in events if e.get("state") == "running"]
+        assert len(running) == 1
+        assert "pending_approval_detail" not in running[0]

turnstone/__init__.py

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

turnstone/console/collector.py

@@ -498,6 +498,7 @@ class ClusterCollector:
                         "kind": WorkstreamKind.from_raw(new_w.get("kind")),
                         "parent_ws_id": new_w.get("parent_ws_id"),
                         "activity_state": new_w.get("activity_state", ""),
+                        "pending_approval_detail": new_w.get("pending_approval_detail"),
                     }
                 )
             old_name = old_ws.get("title", "") or old_ws.get("name", "")
@@ -557,6 +558,18 @@ class ClusterCollector:
                         ws["kind"] = data["kind"]
                     if "parent_ws_id" in data:
                         ws["parent_ws_id"] = data["parent_ws_id"]
+                    # ``pending_approval_detail`` overwrites (no
+                    # ``ws.get`` fallback): the node's broadcast gate
+                    # on ``_pending_approval is not None`` means the
+                    # field is absent from ``data`` exactly when no
+                    # approval is pending — falling back to the cached
+                    # value would resurrect a stale detail after the
+                    # approval resolved.  Without this assignment the
+                    # cached ``node.workstreams`` dict served by
+                    # ``get_node_detail`` / ``get_snapshot`` between
+                    # reconciliations would render stale approve/deny
+                    # buttons on closed approvals.
+                    ws["pending_approval_detail"] = data.get("pending_approval_detail")
                     pending_events.append(
                         {
                             "type": "cluster_state",
@@ -568,6 +581,7 @@ class ClusterCollector:
                             "kind": WorkstreamKind.from_raw(ws.get("kind")),
                             "parent_ws_id": ws.get("parent_ws_id"),
                             "activity_state": ws.get("activity_state", ""),
+                            "pending_approval_detail": data.get("pending_approval_detail"),
                         }
                     )
 

turnstone/console/coordinator_adapter.py

@@ -685,11 +685,13 @@ class CoordinatorAdapter:
                     "tokens": event.get("tokens", 0),
                     "node_id": event.get("node_id", ""),
                     # activity_state lets the JS detect approval-state
-                    # transitions and fire urgent live-bulk fetches so
-                    # inline approve/deny buttons render in lockstep
-                    # with the child entering attention (instead of
-                    # waiting up to 5s for the next TTL window).
+                    # transitions; pending_approval_detail rides on
+                    # the same event so the browser can mutate
+                    # liveBadgeCache directly and render inline
+                    # approve/deny buttons in lockstep with the
+                    # transition, no separate dashboard refetch.
                     "activity_state": event.get("activity_state", ""),
+                    "pending_approval_detail": event.get("pending_approval_detail"),
                 }
             elif etype == "ws_closed":
                 child_event = {

turnstone/console/coordinator_client.py

@@ -81,13 +81,13 @@ WAIT_MAX_TIMEOUT: float = 600.0
 WAIT_POLL_INTERVAL: float = 0.5
 
 # Per-ws cap on the inline ``message`` field bundled into wait_for_workstream
-# results.  Sized so a fan-out of 32 children at the cap is ~192 KiB of
+# results.  Sized so a fan-out of 32 children at the cap is ~320 KiB of
 # tool output — large but not catastrophic on commercial models, and
 # typical waits run with a handful of children.  Truncation is from the
 # END (the lead is usually more informative than the tail) and sets a
 # ``truncated=True`` flag so the model can opt into a follow-up read if
 # the trailing bytes matter.
-WAIT_MESSAGE_MAX_BYTES: int = 6 * 1024
+WAIT_MESSAGE_MAX_BYTES: int = 10 * 1024
 
 # How many tail messages ``wait_for_workstream`` reads when extracting a
 # child's last assistant turn.  The conversation tail almost always

turnstone/console/server.py

@@ -2453,7 +2453,7 @@ def _require_admin_coordinator(
     )
 
 
-def _resolve_coordinator_or_404(
+async def _resolve_coordinator_or_404(
     request: Request,
     coord_mgr: Any,
     storage: Any,
@@ -2484,7 +2484,11 @@ def _resolve_coordinator_or_404(
         if storage is None:
             return None, miss
         try:
-            row = storage.get_workstream(ws_id)
+            # Cold-cache path (every console restart, eviction, console
+            # proxy hop) — offload the sync DB call so the coord
+            # children/tasks handlers don't block the event loop on the
+            # same DB the rest of the handler is unblocking.
+            row = await asyncio.to_thread(storage.get_workstream, ws_id)
         except Exception:
             log.debug("resolve_coordinator.storage_failed ws=%s", ws_id[:8], exc_info=True)
             return None, miss
@@ -2910,7 +2914,7 @@ async def coordinator_children(request: Request) -> JSONResponse:
     if not _VALID_WS_ID_RE.match(ws_id):
         return JSONResponse({"error": "invalid ws_id"}, status_code=400)
     user_id = _auth_user_id(request)
-    _ws, err404 = _resolve_coordinator_or_404(request, coord_mgr, storage, ws_id, user_id)
+    _ws, err404 = await _resolve_coordinator_or_404(request, coord_mgr, storage, ws_id, user_id)
     if err404 is not None:
         return err404
 
@@ -2918,7 +2922,8 @@ async def coordinator_children(request: Request) -> JSONResponse:
     # the full child subtree.  ``user_id`` stays on each row as
     # metadata, not a filter.
     try:
-        raw = storage.list_workstreams(
+        raw = await asyncio.to_thread(
+            storage.list_workstreams,
             limit=_CHILDREN_PAGE_LIMIT + 1,
             parent_ws_id=ws_id,
             kind=None,
@@ -3033,7 +3038,7 @@ async def coordinator_metrics(request: Request) -> JSONResponse:
     if not _VALID_WS_ID_RE.match(ws_id):
         return JSONResponse({"error": "invalid ws_id"}, status_code=400)
     user_id = _auth_user_id(request)
-    _ws, err404 = _resolve_coordinator_or_404(request, coord_mgr, storage, ws_id, user_id)
+    _ws, err404 = await _resolve_coordinator_or_404(request, coord_mgr, storage, ws_id, user_id)
     if err404 is not None:
         return err404
 
@@ -3233,7 +3238,7 @@ async def _resolve_coord_session(
     if not _VALID_WS_ID_RE.match(ws_id):
         return JSONResponse({"error": "invalid ws_id"}, status_code=400)
     user_id = _auth_user_id(request)
-    ws, err404 = _resolve_coordinator_or_404(request, coord_mgr, storage, ws_id, user_id)
+    ws, err404 = await _resolve_coordinator_or_404(request, coord_mgr, storage, ws_id, user_id)
     if err404 is not None:
         return err404
     if ws is None or ws.session is None:
@@ -3523,11 +3528,11 @@ async def coordinator_tasks(request: Request) -> JSONResponse:
     if not _VALID_WS_ID_RE.match(ws_id):
         return JSONResponse({"error": "invalid ws_id"}, status_code=400)
     user_id = _auth_user_id(request)
-    _ws, err404 = _resolve_coordinator_or_404(request, coord_mgr, storage, ws_id, user_id)
+    _ws, err404 = await _resolve_coordinator_or_404(request, coord_mgr, storage, ws_id, user_id)
     if err404 is not None:
         return err404
 
-    envelope, _corrupt = load_task_envelope(storage, ws_id)
+    envelope, _corrupt = await asyncio.to_thread(load_task_envelope, storage, ws_id)
     return JSONResponse(envelope)
 
 
@@ -3750,6 +3755,18 @@ async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]:
     audit_exec = ThreadPoolExecutor(max_workers=4, thread_name_prefix="coord-audit")
     app.state.audit_executor = audit_exec
     _set_audit_executor(audit_exec)
+    # Dedicated executor for coord SSE queue polling, mirroring the
+    # interactive-side ``sse_executor`` in ``turnstone/server.py``.
+    # Each coord ``events`` SSE listener parks a thread on
+    # ``client_queue.get(timeout=5)`` for the connection lifetime.
+    # Without this pool, those parks land on Python's default
+    # ThreadPoolExecutor (~min(32, cpu_count+4)) and compete with
+    # every other ``asyncio.to_thread`` caller — a few coord tabs
+    # against a multi-child workstream are enough to stall new
+    # request handlers.
+    app.state.coord_sse_executor = ThreadPoolExecutor(
+        max_workers=200, thread_name_prefix="coord-sse"
+    )
     # Populate the router's services cache if a router is configured
     _router: ConsoleRouter | None = getattr(app.state, "router", None)
     if _router is not None:
@@ -4029,6 +4046,15 @@ async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]:
     if audit_exec_shutdown is not None:
         _set_audit_executor(None)
         audit_exec_shutdown.shutdown(wait=True)
+    # Drain the coord SSE pool AFTER ``coord_adapter.shutdown()`` above:
+    # adapter shutdown deregisters listeners so no new events handlers
+    # arrive at this pool; in-flight ``client_queue.get`` futures
+    # already running are bounded by their 5s timeout and finish
+    # naturally. ``cancel_futures=True`` discards any queued-but-not-
+    # started futures so we don't block lifespan teardown on them.
+    coord_sse_exec_shutdown = getattr(app.state, "coord_sse_executor", None)
+    if coord_sse_exec_shutdown is not None:
+        coord_sse_exec_shutdown.shutdown(wait=True, cancel_futures=True)
 
 
 # ---------------------------------------------------------------------------
@@ -7996,6 +8022,82 @@ async def _collect_model_status(
     return {nid: models for nid, models in results if models is not None}
 
 
+def _refresh_coord_registry(app_state: Any, storage: Any) -> None:
+    """Rebuild ``app_state.coord_registry`` in place from DB model definitions.
+
+    The console-side coordinator session factory closes over the
+    ``coord_registry`` instance built at lifespan startup
+    (see this module's lifespan setup and ``console/session_factory.py``).
+    Replacing the attribute would orphan the closure — new sessions would
+    still resolve through the stale object.  Mutating in place via
+    ``ModelRegistry.reload()`` preserves identity, so:
+
+    - new coordinator sessions see the new state at create-time;
+    - active coordinator sessions auto-pick up the swap at next ``send()``
+      via ``ChatSession._refresh_model_from_registry`` (the per-send
+      check compares ``cfg.model`` against ``self.model`` and re-resolves
+      on mismatch).
+
+    Errors are logged + swallowed.  The DB write that triggered this
+    refresh has already succeeded, and the explicit reload button
+    remains the user-facing recovery path.  Validation failures
+    (e.g. admin deleted the alias that ``registry.default`` points at)
+    leave the existing registry intact rather than tearing down a
+    working coordinator.
+    """
+    from turnstone.core.model_registry import load_model_registry
+
+    existing = getattr(app_state, "coord_registry", None)
+    if existing is None:
+        # Lifespan didn't build a coord_registry (no DB model rows at boot)
+        # — the entire coord subsystem stayed uninitialized, so a console
+        # restart is required after the operator adds the first row.
+        return
+    try:
+        # ``strict=True`` so a transient DB read error surfaces here.
+        # Without it, the loader degrades to a config.toml-only registry
+        # and ``existing.reload()`` would silently drop every DB-sourced
+        # alias.
+        new_registry = load_model_registry(storage=storage, strict=True)
+    except ValueError as exc:
+        # ModelRegistry.__init__ raises ValueError for several distinct
+        # config issues — empty models, default/fallback/agent/plan/task
+        # alias not present in the loaded set.  Log the actual reason so
+        # operators can tell "no enabled rows" from "default alias typo
+        # in config.toml".  Existing registry stays in place either way.
+        log.warning("console.coord_registry_refresh_skipped reason=%s", exc)
+        return
+    except Exception:
+        log.warning("console.coord_registry_refresh_load_failed", exc_info=True)
+        return
+    try:
+        existing.reload(
+            new_registry.models,
+            new_registry.default,
+            new_registry.fallback,
+            new_registry.agent_model,
+            plan_model=new_registry.plan_model,
+            task_model=new_registry.task_model,
+            plan_effort=new_registry.plan_effort,
+            task_effort=new_registry.task_effort,
+        )
+    except Exception:
+        log.warning("console.coord_registry_refresh_reload_failed", exc_info=True)
+    finally:
+        # Defensive — load_model_registry doesn't eagerly create clients
+        # (ModelRegistry.__init__ leaves _clients/_providers empty; they
+        # populate lazily on first resolve), so shutdown() iterates empty
+        # dicts in practice.  Kept against the day the loader grows
+        # eager-init or a future caller pre-warms the throwaway, and
+        # wrapped because shutdown() in finally would otherwise escape
+        # after a successful in-place reload — surfacing as 500 with the
+        # registry actually mutated and the audit row recording success.
+        try:
+            new_registry.shutdown()
+        except Exception:
+            log.warning("console.coord_registry_refresh_shutdown_failed", exc_info=True)
+
+
 async def _notify_nodes_model_reload(request: Request) -> dict[str, Any]:
     """Tell all nodes to re-read model definitions from DB and rebuild registry."""
     collector: ClusterCollector = request.app.state.collector
@@ -8232,6 +8334,8 @@ async def admin_create_model_definition(request: Request) -> JSONResponse:
         ip,
     )
 
+    await asyncio.to_thread(_refresh_coord_registry, request.app.state, storage)
+
     created = storage.get_model_definition(definition_id)
     if created is None:
         return JSONResponse(
@@ -8389,6 +8493,9 @@ async def admin_update_model_definition(request: Request) -> JSONResponse:
         ip,
     )
 
+    if updates:
+        await asyncio.to_thread(_refresh_coord_registry, request.app.state, storage)
+
     model_def = storage.get_model_definition(definition_id)
     return JSONResponse(_mask_model_secrets(model_def or {}))
 
@@ -8424,6 +8531,8 @@ async def admin_delete_model_definition(request: Request) -> JSONResponse:
         ip,
     )
 
+    await asyncio.to_thread(_refresh_coord_registry, request.app.state, storage)
+
     return JSONResponse({"status": "ok", "definition_id": definition_id})
 
 
@@ -8443,6 +8552,13 @@ async def admin_model_reload(request: Request) -> JSONResponse:
     # before they rebuild their model registries.
     await _publish_config_change(request)
 
+    # Refresh the console's own coord_registry first.  The node fan-out
+    # below carries DB→nodes propagation; the console hosts coordinator
+    # sessions itself and must mutate its in-process registry too —
+    # otherwise the coord LLM keeps calling the prior model name even
+    # after a successful reload.
+    await asyncio.to_thread(_refresh_coord_registry, request.app.state, storage)
+
     results = await _notify_nodes_model_reload(request)
     return JSONResponse({"status": "ok", "results": results})
 
@@ -10196,6 +10312,14 @@ def create_app(
         # tombstones are non-resurrectable.
         saved_state_filter="closed",
         saved_loaded_lookup=_coord_saved_loaded_lookup,
+        # Isolate coord SSE polling on its own 200-thread pool so a
+        # handful of coord tabs (each parking a thread on
+        # ``client_queue.get``) can't starve the default executor and
+        # stall every other ``asyncio.to_thread`` caller (storage,
+        # router, audit). Mirrors the interactive endpoint's
+        # ``sse_executor_lookup`` wiring on ``interactive_endpoint_config``
+        # in ``turnstone/server.py``.
+        sse_executor_lookup=lambda request: request.app.state.coord_sse_executor,
     )
     coord_workstream_routes: list[Any] = []
     register_session_routes(

turnstone/console/static/coordinator/coordinator.js

@@ -2085,6 +2085,14 @@
   const TERMINAL_CHILD_STATES = new Set(["closed", "deleted"]);
   const LIVE_BADGE_TTL_MS = 5000;
   const LIVE_BADGE_DEBOUNCE_MS = 250;
+  // After handleChildState mutates liveBadgeCache from a child_ws_state
+  // SSE event, a bulk-poll landing within this window must NOT
+  // overwrite the SSE-supplied pending_approval / _detail fields with
+  // its own (potentially stale) snapshot — the upstream node
+  // /dashboard cache has its own ~2s TTL so a poll right after a
+  // transition can carry pre-transition state.  3s covers the worst
+  // case (upstream TTL + console TTL minus a margin).
+  const SSE_AUTHORITATIVE_MS = 3000;
   // Debounce window for /tasks refreshes triggered by ``tasks``
   // tool_result SSE events.  Without it, a model that runs
   // ``add → list`` (or any back-to-back mutation pair) double-fetches
@@ -2192,9 +2200,11 @@
     // Inline approve/deny block \u2014 shown only when the live block
     // carries pending_approval_detail (the rich payload added by the
     // server-side dashboard projection).  A "\u2691 approval" badge alone
-    // means the child is in attention state but the rich detail hasn't
-    // arrived yet (urgent live-bulk fetch is in flight); the row gets
-    // re-rendered when it lands.
+    // means the child is in attention state but the rich detail
+    // hasn't arrived on the cache yet \u2014 a rare cross-version race
+    // (e.g. a node mid-rolling-upgrade emitted ws_state without
+    // pending_approval_detail before this PR landed).  The next SSE
+    // tick or the 5s TTL bulk-poll catches up and re-renders.
     if (cached && cached.live && cached.live.pending_approval_detail) {
       const detail = cached.live.pending_approval_detail;
       const block = renderApprovalBlock(child, detail);
@@ -3066,12 +3076,35 @@
           ? results[id]
           : null;
         const wasDenied = denied.indexOf(id) !== -1;
+        const prev = liveBadgeCache.get(id);
+        // SSE-set pending_approval / _detail wins over a stale
+        // bulk-poll snapshot for SSE_AUTHORITATIVE_MS after the
+        // SSE update.  Without this guard, a poll landing right
+        // after a child_ws_state transition can clobber freshly-
+        // mutated approval state with pre-transition data from
+        // the upstream /dashboard cache (which has its own ~2s
+        // TTL).  Other fields (tokens, context_ratio) still track
+        // the bulk response — only the approval surface is gated.
+        let mergedLive = live;
+        if (
+          live &&
+          prev &&
+          prev.sseUpdatedAt &&
+          now - prev.sseUpdatedAt < SSE_AUTHORITATIVE_MS &&
+          prev.live
+        ) {
+          mergedLive = Object.assign({}, live, {
+            pending_approval: prev.live.pending_approval,
+            pending_approval_detail: prev.live.pending_approval_detail,
+          });
+        }
         liveBadgeCache.set(id, {
-          live: live,
+          live: mergedLive,
           fetched: now,
           // Denied ids are permission/identity misses — mark permanent
           // so SSE state ticks on those rows don't retry every window.
           permanent: wasDenied,
+          sseUpdatedAt: prev ? prev.sseUpdatedAt || 0 : 0,
         });
         const row = childrenTreeEl.querySelector(
           '.ch-row[data-ws-id="' + cssEscape(id) + '"]',
@@ -3092,10 +3125,12 @@
       const isPermanent = e && /HTTP 403/.test(e.message || "");
       const now = Date.now();
       ids.forEach((id) => {
+        const prev = liveBadgeCache.get(id);
         liveBadgeCache.set(id, {
           live: null,
           fetched: now,
           permanent: isPermanent,
+          sseUpdatedAt: prev ? prev.sseUpdatedAt || 0 : 0,
         });
       });
       if (!isPermanent) console.warn("flushLiveFetches failed", e);
@@ -3135,7 +3170,6 @@
       ws_id: childId,
       name: "",
     };
-    const prevActivity = existing.activity_state || "";
     existing.state = ev.state || existing.state;
     existing.activity_state =
       typeof ev.activity_state === "string"
@@ -3144,32 +3178,60 @@
     if (ev.node_id) existing.node_id = ev.node_id;
     childrenState.set(childId, existing);
     _touchChild(childId);
-    renderChildren();
-    // Do NOT invalidateLiveBadge on routine state ticks — that defeats
-    // the 5s TTL cache and devolves rate-limiting to the 250ms
-    // debouncer, hitting cluster_ws_detail ~4 req/s per chatty child.
-    // The TTL check in scheduleLiveFetch will refresh the badge on its
-    // own schedule; identity-changing events (created/rename/closed)
-    // still invalidate below.
-    //
-    // Two activity_state transitions warrant an *urgent* (TTL-bypassing)
-    // fetch so the row carries pending_approval_detail in lockstep with
-    // the child's true state:
-    //   - "" / "tool" / "thinking" → "approval"  (need rich payload now
-    //     so the inline approve/deny buttons can render)
-    //   - "approval" → anything else            (need to drop the
-    //     stale payload so the buttons disappear; without this the
-    //     5s TTL leaves stale buttons on a row whose approval was
-    //     resolved elsewhere — e.g. the child's own UI tab)
-    const enteredApproval =
-      existing.activity_state === "approval" && prevActivity !== "approval";
-    const leftApproval =
-      prevActivity === "approval" && existing.activity_state !== "approval";
-    if (enteredApproval || leftApproval) {
-      scheduleLiveFetch(childId, { urgent: true });
+    // pending_approval_detail rides on the ws_state event when an
+    // approval is pending (see turnstone/server.py
+    // WebUI._broadcast_state — gated on ``_pending_approval is not
+    // None``, so absent on the steady state and possibly null on a
+    // node mid-rolling-upgrade).  When present we mutate
+    // liveBadgeCache directly here — inline approve/deny buttons
+    // render in lockstep with the activity_state transition without
+    // a separate live-bulk fetch.  The cache entry is tagged
+    // ``sseUpdatedAt`` so a bulk-poll landing within
+    // SSE_AUTHORITATIVE_MS preserves the SSE-supplied fields
+    // (the upstream /dashboard cache's ~2s TTL would otherwise
+    // clobber a fresh transition with pre-transition state).
+    const pendingApproval = existing.activity_state === "approval";
+    const evDetail =
+      ev.pending_approval_detail !== undefined
+        ? ev.pending_approval_detail
+        : null;
+    const cached = liveBadgeCache.get(childId);
+    // When pending, prefer SSE-supplied detail.  If SSE didn't
+    // carry it (rare race; e.g. a node mid-rolling-upgrade), keep
+    // any existing cached detail rather than blanking the row —
+    // the next bulk-poll catches up.  When not pending, hard-clear.
+    let nextDetail;
+    if (pendingApproval) {
+      nextDetail =
+        evDetail !== null
+          ? evDetail
+          : cached && cached.live
+            ? cached.live.pending_approval_detail
+            : null;
     } else {
-      scheduleLiveFetch(childId);
+      nextDetail = null;
     }
+    const nextLive = Object.assign({}, (cached && cached.live) || {}, {
+      pending_approval: pendingApproval,
+      pending_approval_detail: nextDetail,
+    });
+    liveBadgeCache.set(childId, {
+      live: nextLive,
+      // Preserve prior bulk-poll fetched timestamp so a fresh SSE
+      // tick doesn't artificially extend the 5s TTL gate in
+      // scheduleLiveFetch — the bulk-poll still drives slower-
+      // moving fields (tokens, context_ratio) on its own schedule.
+      fetched: cached ? cached.fetched : 0,
+      permanent: !!(cached && cached.permanent),
+      sseUpdatedAt: Date.now(),
+    });
+    renderChildren();
+    // Do NOT invalidateLiveBadge on routine state ticks — that
+    // defeats the 5s TTL cache and devolves rate-limiting to the
+    // 250ms debouncer.  The TTL check in scheduleLiveFetch handles
+    // refresh cadence for slower-moving fields; identity-changing
+    // events (created/rename/closed) still invalidate below.
+    scheduleLiveFetch(childId);
   }
 
   function handleChildClosed(ev) {

turnstone/core/memory.py

@@ -432,7 +432,7 @@ LAST_ERROR_CONFIG_KEY = "last_error"
 # such error per workstream would bloat workstream_config and the model
 # prompt the coord LLM ingests on inspect.  1024 chars matches the
 # practical "useful for triage" length while staying well under the
-# WAIT_MESSAGE_MAX_BYTES (6 KiB) cap so the truncate happens here at
+# WAIT_MESSAGE_MAX_BYTES (10 KiB) cap so the truncate happens here at
 # write time, not later at the wait surface.
 LAST_ERROR_MAX_LEN = 1024
 

turnstone/core/model_registry.py

@@ -230,6 +230,7 @@ class ModelRegistry:
         if task_model and task_model not in models:
             raise ValueError(f"Task model '{task_model}' not found in registry")
         with self._client_lock:
+            old_models = self._models
             self._models = dict(models)
             self.default = default
             self.fallback = list(fallback) if fallback else []
@@ -238,11 +239,32 @@ class ModelRegistry:
             self.task_model = task_model
             self.plan_effort = plan_effort
             self.task_effort = task_effort
-            for client in self._clients.values():
-                if hasattr(client, "close"):
-                    client.close()
-            self._clients.clear()
-            self._providers.clear()
+            # Selective teardown — close + drop only clients whose
+            # connection target actually changed (alias removed, or
+            # base_url / api_key / provider differs).  Keeps connection
+            # pools warm for the common admin-edit case where only
+            # ``model`` / ``temperature`` / ``context_window`` changed.
+            for alias, client in list(self._clients.items()):
+                old_cfg = old_models.get(alias)
+                new_cfg = self._models.get(alias)
+                if (
+                    new_cfg is None
+                    or old_cfg is None
+                    or old_cfg.base_url != new_cfg.base_url
+                    or old_cfg.api_key != new_cfg.api_key
+                    or old_cfg.provider != new_cfg.provider
+                ):
+                    if hasattr(client, "close"):
+                        client.close()
+                    del self._clients[alias]
+            # Providers are keyed on alias but only depend on
+            # ``cfg.provider`` — drop only when the provider string
+            # changed or the alias was removed.
+            for alias in list(self._providers.keys()):
+                old_cfg = old_models.get(alias)
+                new_cfg = self._models.get(alias)
+                if new_cfg is None or old_cfg is None or old_cfg.provider != new_cfg.provider:
+                    del self._providers[alias]
 
     def shutdown(self) -> None:
         """Close all cached client connections."""
@@ -300,6 +322,7 @@ def load_model_registry(
     context_window: int = 32768,
     provider: str = "openai",
     storage: Any | None = None,
+    strict: bool = False,
 ) -> ModelRegistry:
     """Build a ModelRegistry from CLI args, ``config.toml``, and database.
 
@@ -317,6 +340,15 @@ def load_model_registry(
        ``[model].plan_effort``, ``[model].task_effort`` control routing.
        ``plan_model``/``task_model`` override ``agent_model`` per sub-agent
        role; both fall back to it when unset.
+
+    ``strict``: when True, a storage read failure during the DB-rows step
+    re-raises instead of degrading to a config.toml-only registry.
+    Callers that hot-reload an existing registry need this so a transient
+    DB outage doesn't silently drop every DB-sourced alias when the
+    truncated result is applied via ``ModelRegistry.reload``.  Callers
+    that build a fresh registry from scratch (CLI, lifespan startup) want
+    the default behaviour — boot succeeds with a config-only fallback
+    rather than crashing on a flaky DB.
     """
     import json as _json
 
@@ -370,6 +402,8 @@ def load_model_registry(
                     server_compat=row_server_compat,
                 )
         except Exception:
+            if strict:
+                raise
             log.warning("Failed to load model definitions from storage", exc_info=True)
 
     # 2. Build configs from [models.*] sections (overrides DB for same alias)

turnstone/prompts/base_coordinator.md

@@ -1,8 +1,8 @@
 You are a coordinator on a small, focused infrastructure team.  Your role is to orchestrate work across the cluster: you decompose a user's request into tasks, spawn child workstreams on appropriate nodes with the right skills, monitor their progress, synthesise their results, and surface the outcome back to the user.
 
-You do not edit files, run shell commands, browse the web, or manipulate the codebase directly.  Children do that.  Your job is to pick the right child, give it a well-formed brief, and keep the plan coherent while multiple children run in parallel.
+You do not edit files, run shells, or browse the web — children do. You pick the right child, give a well-formed brief, and keep the plan coherent while multiple children run.
 
-You think in plans: a tasks entry, a child to own it, a way to know when it's done.  When a child reports back, you read what it said, decide whether the goal is met, and either close it out, push a follow-up message, or spawn another child to cover the gap.
+You think in plans: enumerate the independent units of work, spawn one child per unit, run them in parallel by default. Sequential only when one child's output feeds the next. When a child reports back, you decide whether the goal is met, then close it out, push a follow-up, or spawn another child to cover the gap.
 
 You are precise about what you delegate.  A child gets the minimum context it needs — skill, initial_message, maybe a node_id.  You don't paste whole files into its prompt; children have their own tools for that.
 

turnstone/prompts/tools_coordinator.md

@@ -1,7 +1,5 @@
 TOOL PATTERNS:
 
-You are a coordinator.  You do not edit files, run shell commands, or browse the web directly.  You delegate work by spawning child workstreams on cluster nodes, monitoring their progress, and synthesising their results.  Every tool below is in your schema; nothing else is.
-
 Discover available capacity → list_nodes / list_skills:
    list_nodes(filters={'capability': 'gpu'})
    list_skills(category='engineering')
@@ -10,11 +8,11 @@ Delegate a task → spawn_workstream:
    spawn_workstream(initial_message='audit auth.py for CSRF handling', name='csrf-audit')
    spawn_workstream(initial_message='compare FastAPI vs Starlette for async websockets', target_node='flat-blck-io_43a3')
 
-Fan out to multiple children in one approval → spawn_batch (up to 10):
+Fan out across independent inputs → spawn_batch:
    spawn_batch(children=[
-     {'initial_message': 'benchmark A'},
-     {'initial_message': 'benchmark B'},
-     {'initial_message': 'prototype the winner'},
+     {'initial_message': 'top stories on Hacker News'},
+     {'initial_message': 'top stories on Lobsters'},
+     {'initial_message': 'top stories on r/programming'},
    ])
 
 Check on a child → inspect_workstream:
@@ -24,8 +22,9 @@ Wait for spawned children to finish → wait_for_workstream (PREFER over busy-po
    wait_for_workstream(ws_ids=['a1b2c3d4'], timeout=120)
    wait_for_workstream(ws_ids=['a1b2c3d4', 'e5f6g7h8', 'i9j0k1l2'], mode='all', timeout=300)
 
-Push a follow-up message to a running child → send_to_workstream:
+Push a follow-up message to a child → send_to_workstream (mid-run nudge, or course-correct a child that drifted off-brief):
    send_to_workstream(ws_id='a1b2c3d4', message='also capture the test-coverage delta')
+   send_to_workstream(ws_id='a1b2c3d4', message='stop — you are editing auth_legacy.py, the active path is auth.py')
 
 List what you've spawned → list_workstreams:
    list_workstreams()
@@ -38,19 +37,10 @@ Wind a child down → close_workstream (soft; session stops, storage kept) or de
    close_workstream(ws_id='a1b2c3d4', reason='task complete')
    delete_workstream(ws_id='a1b2c3d4')
 
-Wind all direct children down at once → close_all_children (soft-close cascade, single approval):
+Wind all direct children down at once → close_all_children (soft-close cascade):
    close_all_children(reason='batch complete, synthesising results')
 
 Plan and track work → tasks (your scratchpad; children don't see it):
    tasks(action='add', title='audit auth.py for CSRF')
    tasks(action='update', task_id='t_03', status='in_progress')
-   tasks(action='list')
    tasks(action='remove', task_id='t_03')
-
-## Workflow shape
-
-Prefer: tasks to plan → spawn_workstream to delegate → wait_for_workstream to block on completion → inspect_workstream to read the final message → synthesise → close_workstream.
-
-Each repeated `inspect_workstream` poll costs a full assistant turn (+ judge + tokens); a single `wait_for_workstream` absorbs the wait at one call + one result. The cost gap widens fast on fan-outs of 3+ children.
-
-If a user asks you to "edit X" or "run Y", spawn a child and delegate — the coordinator's tool schema doesn't include file or shell access by design.

turnstone/server.py

@@ -194,6 +194,18 @@ class WebUI(SessionUIBase):
             }
             if state == "idle":
                 event["content"] = payload["content"]
+            # Coord tree-UI renders inline approve/deny buttons off
+            # ``pending_approval_detail``; carrying it on the
+            # state-change broadcast lets the cluster bus update those
+            # buttons in lockstep with ``activity_state`` instead of
+            # forcing the browser to chase a separate dashboard fetch.
+            # Gated on existence so we don't pay the serializer's
+            # per-broadcast verdict-cache deepcopy on the common
+            # no-approval-pending path.
+            if self._pending_approval is not None:
+                detail = self.serialize_pending_approval_detail()
+                if detail is not None:
+                    event["pending_approval_detail"] = detail
             try:
                 WebUI._global_queue.put_nowait(event)
             except queue.Full:
@@ -823,6 +835,16 @@ def _build_node_snapshot(app_state: Any) -> dict[str, Any]:
         title = ""
         if ws.session:
             title = get_workstream_display_name(ws.session.ws_id) or ""
+        # ``pending_approval_detail`` mirrors the dashboard handler's
+        # projection so the console collector's reconnect-via-snapshot
+        # path (``_reconcile_node``) can carry the rich approval payload
+        # across reconnects — without it, a child sitting in approval-
+        # pending across a console restart or network blip would render
+        # with no buttons until the next state change. Same data, same
+        # ``read`` scope as ``/v1/api/dashboard``.
+        approval_detail: dict[str, Any] | None = None
+        if ui is not None and hasattr(ui, "serialize_pending_approval_detail"):
+            approval_detail = ui.serialize_pending_approval_detail()
         ws_list.append(
             {
                 "id": ws.id,
@@ -839,6 +861,7 @@ def _build_node_snapshot(app_state: Any) -> dict[str, Any]:
                 "kind": ws.kind,
                 "parent_ws_id": ws.parent_ws_id,
                 "user_id": ws.user_id,
+                "pending_approval_detail": approval_detail,
             }
         )
     return {
@@ -3323,8 +3346,9 @@ def create_app(
         # ``sse_executor`` so SSE polling stayed isolated from
         # every other ``asyncio.to_thread`` caller in the process
         # (storage, router, audit). Restore that isolation under
-        # the lifted contract — coord wires ``None`` and falls
-        # back to the default executor.
+        # the lifted contract. The console's coord endpoint wires
+        # its own ``coord_sse_executor`` on the same lookup hook —
+        # see ``turnstone/console/server.py``.
         sse_executor_lookup=lambda request: request.app.state.sse_executor,
         create_supports_attachments=True,
         create_supports_user_id_override=True,
@@ -3544,6 +3568,11 @@ def main() -> None:
         default=8080,
         help="Port to listen on (default: 8080)",
     )
+    parser.add_argument(
+        "--skip-permissions",
+        action="store_true",
+        help="Auto-approve all tool calls (no confirmation prompts)",
+    )
     # MCP config path is bootstrap-critical (needed before ConfigStore for tool loading)
     parser.add_argument(
         "--mcp-config",
@@ -3971,7 +4000,7 @@ def main() -> None:
         ws = manager.create(user_id="", name="resumed")
         if not isinstance(ws.ui, WebUI):
             raise TypeError(f"Expected WebUI, got {type(ws.ui).__name__}")
-        if config_store.get("tools.skip_permissions"):
+        if args.skip_permissions or config_store.get("tools.skip_permissions"):
             ws.ui.auto_approve = True
         assert ws.session is not None
         ws.session.set_watch_runner(
@@ -4009,7 +4038,7 @@ def main() -> None:
         _advertise_host = args.host if args.host not in ("0.0.0.0", "::") else socket.gethostname()
         _advertise_url = f"http://{_advertise_host}:{args.port}"
 
-    _skip_perms = config_store.get("tools.skip_permissions")
+    _skip_perms = args.skip_permissions or config_store.get("tools.skip_permissions")
     app = create_app(
         workstreams=manager,
         global_queue=global_queue,

turnstone/tools/wait_for_workstream.json

@@ -1,6 +1,6 @@
 {
   "name": "wait_for_workstream",
-  "description": "Block until one or all named child workstreams reach a terminal state. Prefer this over busy-polling inspect_workstream after a fan-out: the tool absorbs the wait, so you get one call + one result regardless of duration. Returns `{results: {ws_id: {state, tokens, updated, message, truncated}}, elapsed, complete, mode}` — `results` is keyed by ws_id (not top-level), `complete` is true when the wait condition fired before the timeout. `message` is the child's last assistant text for `idle`/`error`, a short sentinel for `closed`/`denied`, and `null` for non-terminal rows so a follow-up read knows which children still need work; capped at 6 KiB UTF-8, with `truncated=true` when the cap fires (call inspect_workstream for the rest). Real terminal states: `idle`, `error`, `closed`, `deleted` (the last is unobservable since hard-delete cascades the row out of storage). `mode='any'` returns when the first child hits a real terminal — a `denied` id alone never satisfies the condition, so a typo'd / foreign / nonexistent id can't false-positive a wait. `mode='all'` returns once every id has settled (real terminal OR denied). Cross-tenant guard: only the coordinator's own children (or itself) are visible; everything else is reported `state='denied'`. Capped at 32 ws_ids and 600s; both overflows error rather than silently truncating.",
+  "description": "Block until one or all named child workstreams reach a terminal state. Prefer this over busy-polling inspect_workstream after a fan-out: the tool absorbs the wait, so you get one call + one result regardless of duration. Returns `{results: {ws_id: {state, tokens, updated, message, truncated}}, elapsed, complete, mode}` — `results` is keyed by ws_id (not top-level), `complete` is true when the wait condition fired before the timeout. `message` is the child's last assistant text for `idle`/`error`, a short sentinel for `closed`/`denied`, and `null` for non-terminal rows so a follow-up read knows which children still need work; capped at 10 KiB UTF-8, with `truncated=true` when the cap fires (call inspect_workstream for the rest). Real terminal states: `idle`, `error`, `closed`, `deleted` (the last is unobservable since hard-delete cascades the row out of storage). `mode='any'` returns when the first child hits a real terminal — a `denied` id alone never satisfies the condition, so a typo'd / foreign / nonexistent id can't false-positive a wait. `mode='all'` returns once every id has settled (real terminal OR denied). Cross-tenant guard: only the coordinator's own children (or itself) are visible; everything else is reported `state='denied'`. Capped at 32 ws_ids and 600s; both overflows error rather than silently truncating.",
   "parameters": {
     "type": "object",
     "properties": {

uv.lock

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