diff --git a/tests/test_admin_model_registry_refresh.py b/tests/test_admin_model_registry_refresh.py index 5a38e1d5..255b08dd 100644 --- a/tests/test_admin_model_registry_refresh.py +++ b/tests/test_admin_model_registry_refresh.py @@ -22,6 +22,8 @@ to lock in: from __future__ import annotations +import threading +from types import SimpleNamespace from typing import Any from unittest.mock import MagicMock @@ -33,6 +35,7 @@ from starlette.testclient import TestClient from tests._coord_test_helpers import _AuthMiddleware from turnstone.console.server import ( + _maybe_bootstrap_coord_subsystem, _refresh_coord_registry, admin_create_model_definition, admin_delete_model_definition, @@ -42,6 +45,29 @@ from turnstone.console.server import ( from turnstone.core.model_registry import ModelConfig, ModelRegistry from turnstone.core.storage._sqlite import SQLiteBackend + +def _bootstrap_app(**overrides: Any) -> Any: + """Build a fake ``app`` with the ``state`` attrs the bootstrap helper + inspects. Defaults match a freshly-installed console (no coord + subsystem yet) with all required prereqs (collector, console_metrics, + config_store) populated as MagicMocks. Tests pass overrides to + suppress individual prereqs or pre-set ``coord_mgr`` etc. + """ + state_kwargs: dict[str, Any] = { + "coord_mgr": None, + "coord_adapter": None, + "coord_registry": None, + "coord_registry_error": "", + "coord_state_writer": None, + "coord_idle_observer": None, + "config_store": MagicMock(), + "collector": MagicMock(), + "console_metrics": MagicMock(), + } + state_kwargs.update(overrides) + return SimpleNamespace(state=SimpleNamespace(**state_kwargs)) + + # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @@ -213,6 +239,367 @@ def test_helper_preserves_registry_when_no_enabled_rows(storage: SQLiteBackend) assert state.coord_registry.get_config("local").model == "cached-model" +# --------------------------------------------------------------------------- +# First-row bootstrap tests — ``_maybe_bootstrap_coord_subsystem`` semantics. +# A console booted with no model rows leaves coord_mgr = None; the operator +# adding the first row at runtime must promote the subsystem to ready +# without a console restart. +# --------------------------------------------------------------------------- + + +def test_bootstrap_noop_when_coord_mgr_already_built( + storage: SQLiteBackend, monkeypatch: pytest.MonkeyPatch +) -> None: + """Idempotent fast-path — already-bootstrapped subsystem must not + re-stand-up a second SessionManager / StateWriter pair.""" + from turnstone.console import server as server_module + + _seed_model_def(storage, definition_id="m1", alias="local", model="m") + app = _bootstrap_app(coord_mgr=MagicMock()) # subsystem already built + + calls: list[Any] = [] + monkeypatch.setattr( + server_module, + "_bootstrap_coord_subsystem", + lambda *a, **kw: calls.append(a), + ) + _maybe_bootstrap_coord_subsystem(app, storage) + assert calls == [] + + +@pytest.mark.parametrize("missing_attr", ["config_store", "collector", "console_metrics"]) +def test_bootstrap_noop_when_prerequisites_missing( + storage: SQLiteBackend, + monkeypatch: pytest.MonkeyPatch, + missing_attr: str, +) -> None: + """Each strictly-required ``app.state`` attr (config_store, collector, + console_metrics) must individually short-circuit the bootstrap to a + no-op — partial init / test harnesses don't have the full set, and a + CRUD write that already landed mustn't 500 on a missing prereq.""" + from turnstone.console import server as server_module + + _seed_model_def(storage, definition_id="m1", alias="local", model="m") + app = _bootstrap_app(**{missing_attr: None}) + calls: list[Any] = [] + monkeypatch.setattr( + server_module, + "_bootstrap_coord_subsystem", + lambda *a, **kw: calls.append(a), + ) + _maybe_bootstrap_coord_subsystem(app, storage) + assert calls == [] + assert app.state.coord_mgr is None + + +def test_bootstrap_records_error_when_no_rows( + storage: SQLiteBackend, monkeypatch: pytest.MonkeyPatch +) -> None: + """All rows disabled (or none seeded) — load_model_registry raises + ValueError. Helper records the message on app.state so the + coord-endpoint 503 surfaces a current diagnosis instead of a stale + one from boot.""" + from turnstone.console import server as server_module + + app = _bootstrap_app() + calls: list[Any] = [] + monkeypatch.setattr( + server_module, + "_bootstrap_coord_subsystem", + lambda *a, **kw: calls.append(a), + ) + _maybe_bootstrap_coord_subsystem(app, storage) + assert calls == [] + assert "No model definitions found" in app.state.coord_registry_error + + +def test_bootstrap_calls_subsystem_builder_on_first_row( + storage: SQLiteBackend, monkeypatch: pytest.MonkeyPatch +) -> None: + """A row exists ⇒ helper loads the registry, hands it to the + subsystem builder, and the builder stamps it on app.state. Mirrors + the post-build invariant the real ``_bootstrap_coord_subsystem`` + establishes (coord_registry set iff coord_mgr set) so the stale + boot-time error string clears as part of the same commit step.""" + from turnstone.console import server as server_module + + _seed_model_def(storage, definition_id="m1", alias="local", model="m") + app = _bootstrap_app(coord_registry_error="stale boot-time message") + captured: dict[str, Any] = {} + + def _fake_build(app_arg: Any, _storage: Any, _cfg: Any, registry_arg: Any) -> None: + captured["app"] = app_arg + captured["registry"] = registry_arg + # Simulate the real builder's final commit step: stamp registry + # + clear stale error + set coord_mgr atomically. + app_arg.state.coord_registry = registry_arg + app_arg.state.coord_registry_error = "" + app_arg.state.coord_mgr = MagicMock() + + monkeypatch.setattr(server_module, "_bootstrap_coord_subsystem", _fake_build) + _maybe_bootstrap_coord_subsystem(app, storage) + assert captured["app"] is app + assert captured["registry"].has_alias("local") + assert app.state.coord_registry is captured["registry"] + assert app.state.coord_registry_error == "" + + +def test_bootstrap_replaces_stale_error_on_builder_failure( + storage: SQLiteBackend, monkeypatch: pytest.MonkeyPatch +) -> None: + """A builder failure after a successful registry load must not leave + the stale "no model definitions" message on app.state — that + diagnosis is demonstrably wrong (rows ARE present, the build failed + for a different reason). Replacement message must surface the + actual exception type so operators can correlate with logs.""" + from turnstone.console import server as server_module + + _seed_model_def(storage, definition_id="m1", alias="local", model="m") + app = _bootstrap_app( + coord_registry_error=( + "No model definitions found. Provide --model, configure [models.*] " + "in config.toml, or add model definitions in the admin panel." + ) + ) + + def _boom(*_a: Any, **_kw: Any) -> None: + raise RuntimeError("simulated builder failure") + + monkeypatch.setattr(server_module, "_bootstrap_coord_subsystem", _boom) + _maybe_bootstrap_coord_subsystem(app, storage) # must not raise + assert app.state.coord_mgr is None + # Stale "no models" message replaced. + assert "No model definitions found" not in app.state.coord_registry_error + # New message mentions the actual failure class so the 503 banner + # gives operators something actionable beyond "look at logs". + assert "RuntimeError" in app.state.coord_registry_error + assert "failed to initialise" in app.state.coord_registry_error + + +def test_bootstrap_tears_down_partial_state_on_builder_failure( + storage: SQLiteBackend, monkeypatch: pytest.MonkeyPatch +) -> None: + """If the builder partially stamps handles on app.state and then + raises, the helper must call the teardown path so a subsequent + retry doesn't leak a StateWriter daemon / observer subscription.""" + from turnstone.console import server as server_module + + _seed_model_def(storage, definition_id="m1", alias="local", model="m") + app = _bootstrap_app() + + state_writer = MagicMock() + idle_observer = MagicMock() + coord_adapter = MagicMock() + + def _partial_then_boom(app_arg: Any, *_a: Any, **_kw: Any) -> None: + # Mirror the real builder's stamp-immediately-after-start order: + # StateWriter spawned + stamped before SessionManager validates. + app_arg.state.coord_state_writer = state_writer + app_arg.state.coord_idle_observer = idle_observer + app_arg.state.coord_adapter = coord_adapter + raise RuntimeError("simulated mid-build failure") + + monkeypatch.setattr(server_module, "_bootstrap_coord_subsystem", _partial_then_boom) + _maybe_bootstrap_coord_subsystem(app, storage) + # Teardown ran for each partially-stamped handle. + state_writer.shutdown.assert_called_once() + idle_observer.shutdown.assert_called_once() + coord_adapter.shutdown.assert_called_once() + # And the app.state slots are reset so a retry sees a clean field. + assert app.state.coord_state_writer is None + assert app.state.coord_idle_observer is None + assert app.state.coord_adapter is None + assert app.state.coord_mgr is None + assert app.state.coord_registry is None + + +def test_real_bootstrap_stands_up_subsystem_end_to_end( + storage: SQLiteBackend, +) -> None: + """End-to-end: the real ``_bootstrap_coord_subsystem`` constructs a + working ``SessionManager`` against a real ``ConfigStore`` + real + ``ClusterCollector`` when an operator adds the first model row to + a freshly-installed console. + + This is the test that reproduces the user-reported bug — without it, + all the bootstrap helper-level tests can pass even if the real + builder never actually completes (the helper-level tests + monkeypatch the builder out). Asserts the post-bootstrap invariant + that ``_require_coord_mgr`` relies on: ``coord_mgr`` is a real + SessionManager and ``coord_registry_error`` has been cleared. + """ + from turnstone.console import server as server_module + from turnstone.console.collector import ClusterCollector + from turnstone.console.coordinator_ui import ConsoleCoordinatorUI + from turnstone.console.metrics import ConsoleMetrics + from turnstone.core.config_store import ConfigStore + from turnstone.core.session_manager import SessionManager + + _seed_model_def(storage, definition_id="m1", alias="local", model="m") + config_store = ConfigStore(storage) + # Disable the idle-cleanup daemon for this test — it has no + # stop_event hook in the bootstrap (the loop runs until process + # termination) so leaving the default 120-minute timeout would + # leak a daemon thread across every test run. + config_store.set("server.workstream_idle_timeout", 0) + # ClusterCollector is constructed but NOT started — start() spawns + # network discovery + SSE manager threads we don't need for this + # test. ensure_console_pseudo_node() (called by the bootstrap via + # start_child_event_fanout) operates on the in-memory snapshot map + # without requiring the discovery loop to be live. + collector = ClusterCollector(storage=storage) + # Snapshot ConsoleCoordinatorUI's class attrs so the test can + # restore them on teardown — the bootstrap mutates them and they + # persist across tests at process scope. + saved_coord_mgr = ConsoleCoordinatorUI._coord_mgr + saved_collector = ConsoleCoordinatorUI._collector + saved_metrics = ConsoleCoordinatorUI._console_metrics + + app = SimpleNamespace( + state=SimpleNamespace( + coord_mgr=None, + coord_adapter=None, + coord_registry=None, + coord_registry_error=( + "No model definitions found. Provide --model, configure [models.*] " + "in config.toml, or add model definitions in the admin panel." + ), + coord_state_writer=None, + coord_idle_observer=None, + config_store=config_store, + collector=collector, + console_metrics=ConsoleMetrics(), + jwt_secret="x" * 32, + console_url="http://127.0.0.1:8001", + ) + ) + + try: + _maybe_bootstrap_coord_subsystem(app, storage) + # The real builder ran and produced a working SessionManager. + assert isinstance(app.state.coord_mgr, SessionManager) + assert app.state.coord_adapter is not None + # Registry stamped with the seeded alias. + assert app.state.coord_registry is not None + assert app.state.coord_registry.has_alias("local") + # Stale boot-time error string cleared as part of the commit. + assert app.state.coord_registry_error == "" + # StateWriter daemon is alive — it's the load-bearing async + # persistence layer for SessionManager state transitions. + assert app.state.coord_state_writer is not None + # Class-level wiring on ConsoleCoordinatorUI is the path + # on_state_change / on_rename use to fan out to the dashboard. + assert ConsoleCoordinatorUI._coord_mgr is app.state.coord_mgr + assert ConsoleCoordinatorUI._collector is collector + finally: + # Tear down threads + subscriptions spawned by the bootstrap. + # ``_teardown_partial_coord_subsystem`` does the same work the + # runtime-bootstrap failure path does, so reusing it here also + # exercises that helper end-to-end. + server_module._teardown_partial_coord_subsystem(app) + # Restore ConsoleCoordinatorUI class attrs so other tests in + # the suite see them as they were before this test ran. + ConsoleCoordinatorUI._coord_mgr = saved_coord_mgr + ConsoleCoordinatorUI._collector = saved_collector + ConsoleCoordinatorUI._console_metrics = saved_metrics + + +def test_bootstrap_lock_serialises_concurrent_calls( + storage: SQLiteBackend, monkeypatch: pytest.MonkeyPatch +) -> None: + """Two simultaneous CRUD writes both seeing ``coord_mgr is None`` + must serialise via ``_COORD_BOOTSTRAP_LOCK`` and the second caller + must observe the post-build state on its inside-the-lock re-check — + so the builder runs exactly once. Without the lock + double-check, + both threads enter the build and stamp duplicate SessionManager / + StateWriter / observer triples on app.state. + + The synchronisation is deterministic, not wall-clock-based: an + instrumented lock wrapper signals when a second acquirer arrives, + so the test fails fast and reproducibly on slow CI rather than + relying on a sleep long enough to "probably" let thread 2 reach + the lock — a dependence the previous version was rightly criticised + for. + """ + from turnstone.console import server as server_module + + _seed_model_def(storage, definition_id="m1", alias="local", model="m") + app = _bootstrap_app() + build_count = 0 + count_lock = threading.Lock() + in_build = threading.Event() + release_build = threading.Event() + + def _slow_build(app_arg: Any, *_a: Any, **_kw: Any) -> None: + nonlocal build_count + with count_lock: + build_count += 1 + is_first = build_count == 1 + if is_first: + # Hold inside the build so the second thread is forced to + # queue at the lock — without the lock it would race ahead + # and increment build_count to 2. + in_build.set() + release_build.wait(timeout=2.0) + # Mirror the real builder's commit step. + app_arg.state.coord_mgr = MagicMock() + app_arg.state.coord_registry = MagicMock() + + monkeypatch.setattr(server_module, "_bootstrap_coord_subsystem", _slow_build) + + # Instrumented wrapper: delegates to a real ``threading.Lock`` so + # the production ``with _COORD_BOOTSTRAP_LOCK:`` block keeps doing + # genuine serialisation work, but counts arrivals so the main + # thread can wait deterministically until thread 2 is at the lock + # before releasing thread 1. If the production code drops the + # ``with`` block entirely, the wrapper is never entered, the + # arrival event never fires, and the assertion below times out + # with a clear error rather than the subtler false-pass a sleep + # would allow. + real_lock = threading.Lock() + arrivals_lock = threading.Lock() + arrivals = 0 + second_waiter_arrived = threading.Event() + + class _InstrumentedLock: + def __enter__(self) -> Any: + nonlocal arrivals + with arrivals_lock: + arrivals += 1 + arrival_index = arrivals + if arrival_index >= 2: + second_waiter_arrived.set() + real_lock.acquire() + return self + + def __exit__(self, *_exc: Any) -> None: + real_lock.release() + + monkeypatch.setattr(server_module, "_COORD_BOOTSTRAP_LOCK", _InstrumentedLock()) + + def _run() -> None: + _maybe_bootstrap_coord_subsystem(app, storage) + + t1 = threading.Thread(target=_run, name="bootstrap-thread-1") + t2 = threading.Thread(target=_run, name="bootstrap-thread-2") + t1.start() + assert in_build.wait(timeout=2.0), "thread 1 never entered the builder" + t2.start() + # Deterministic: block here until thread 2 has reached the lock + # (or the wait times out, signalling the lock was bypassed entirely). + assert second_waiter_arrived.wait(timeout=2.0), ( + "thread 2 never reached the lock — concurrency was not exercised, " + "production code may be skipping the lock" + ) + release_build.set() + t1.join(timeout=5.0) + t2.join(timeout=5.0) + assert not t1.is_alive() and not t2.is_alive() + assert build_count == 1, ( + f"builder ran {build_count} times — lock failed to serialise concurrent calls" + ) + + def test_helper_preserves_registry_on_reload_validation_error( storage: SQLiteBackend, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -309,6 +696,79 @@ def test_create_endpoint_refreshes_registry(storage: SQLiteBackend) -> None: assert registry.get_config("fast").model == "fast-model" +def test_create_endpoint_bootstraps_subsystem_on_fresh_install( + storage: SQLiteBackend, monkeypatch: pytest.MonkeyPatch +) -> None: + """User-visible regression: a console booted with no model rows leaves + coord_mgr unbuilt; the operator adding their first model via the + admin panel must promote the subsystem to ready (no console restart). + Before the fix, ``_refresh_coord_registry`` short-circuited on + ``coord_registry is None`` and the dashboard's 503 banner persisted + until the user restarted. + """ + from turnstone.console import server as server_module + + # Fresh-install state: registry=None, coord_mgr=None, boot-time + # error string set by the lifespan's ValueError catch. Build the + # app explicitly so the test can inspect ``app.state`` after the + # request completes (TestClient's ``.app`` attribute is typed as + # ASGIApp, which loses the ``.state`` accessor). + app = Starlette( + routes=[ + Route( + "/v1/api/admin/model-definitions", + admin_create_model_definition, + methods=["POST"], + ), + ], + middleware=[Middleware(_AuthMiddleware)], + ) + app.state.auth_storage = storage + app.state.coord_registry = None + app.state.coord_mgr = None + app.state.coord_registry_error = ( + "No model definitions found. Provide --model, configure [models.*] " + "in config.toml, or add model definitions in the admin panel." + ) + app.state.collector = MagicMock() + app.state.collector.get_all_nodes.return_value = [] + app.state.config_store = MagicMock() + app.state.console_metrics = MagicMock() + client = TestClient(app) + client.headers.update({"X-Test-User": "admin", "X-Test-Perms": "admin.models"}) + + captured: dict[str, Any] = {} + + def _fake_build(app_arg: Any, _storage: Any, _cfg: Any, registry_arg: Any) -> None: + captured["registry"] = registry_arg + # Mirror the real builder's commit step so the post-call asserts + # see the same invariant a successful real bootstrap establishes. + app_arg.state.coord_registry = registry_arg + app_arg.state.coord_registry_error = "" + app_arg.state.coord_mgr = MagicMock() + + monkeypatch.setattr(server_module, "_bootstrap_coord_subsystem", _fake_build) + + resp = client.post( + "/v1/api/admin/model-definitions", + json={ + "alias": "first", + "model": "first-model", + "provider": "openai-compatible", + "base_url": "http://localhost:9000/v1", + "api_key": "sk-x", + }, + ) + assert resp.status_code == 200, resp.text + # Bootstrap fired with a registry holding the just-added alias. + assert "registry" in captured and captured["registry"].has_alias("first") + # coord_mgr is now non-None (bootstrap completed) and the stale + # boot-time error message has been cleared so subsequent 503s + # don't lie about current state. + assert app.state.coord_mgr is not None + assert app.state.coord_registry_error == "" + + def test_update_endpoint_refreshes_registry(storage: SQLiteBackend) -> None: """PUT swaps the underlying model name behind a stable alias — the user's reported regression.""" diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 669964a9..9ce18970 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -4128,6 +4128,240 @@ def _coord_idle_cleanup_thread( log.debug("console.coord_idle_cleanup_failed", exc_info=True) +# Guards concurrent attempts to bootstrap the coord subsystem from the +# admin model-CRUD path on a freshly-installed console (no model rows at +# boot ⇒ lifespan deferred the build). Two simultaneous create-model +# requests would otherwise both see ``coord_mgr is None`` and race to +# attach two SessionManagers, two idle observers, etc., with the second +# silently overwriting the first's app.state attrs while the first's +# StateWriter / cleanup threads kept running detached. +_COORD_BOOTSTRAP_LOCK = threading.Lock() + + +def _bootstrap_coord_subsystem( + app: Starlette, + storage: Any, + config_store: Any, + coord_registry: Any, +) -> None: + """Build the console-side coord subsystem (manager, adapter, idle + observer, idle-nudge watcher, child-event fan-out, optional cleanup + thread) and stamp the resulting handles on ``app.state``. + + Shared between the lifespan startup path (when the console boots + with at least one model row in the DB) and the admin model-CRUD + path (when the operator adds the first row to a freshly-installed + console at runtime — no restart required). + """ + from turnstone.console.coordinator_adapter import CoordinatorAdapter + from turnstone.console.coordinator_client import ( + CoordinatorClient, + CoordinatorTokenManager, + ) + from turnstone.console.coordinator_idle_observer import ( + CoordinatorIdleObserver, + ) + from turnstone.console.coordinator_ui import ConsoleCoordinatorUI + from turnstone.console.session_factory import build_console_session_factory + from turnstone.core.idle_nudge_watcher import install_idle_nudge_watcher + from turnstone.core.session_manager import SessionManager + from turnstone.core.state_writer import StateWriter + + jwt_secret: str = getattr(app.state, "jwt_secret", "") + console_bind_url: str = getattr(app.state, "console_url", "") or "http://127.0.0.1:8001" + + def _ui_factory(ws: Workstream) -> ConsoleCoordinatorUI: + return ConsoleCoordinatorUI(ws_id=ws.id, user_id=ws.user_id or "") + + def _coord_client_factory(ws_id: str, user_id: str) -> CoordinatorClient: + ttl = int(config_store.get("coordinator.session_jwt_ttl_seconds")) + tm = CoordinatorTokenManager( + user_id=user_id or "system", + scopes=frozenset({"read", "write", "approve"}), + permissions=frozenset({"admin.coordinator"}), + secret=jwt_secret, + coord_ws_id=ws_id, + ttl_seconds=ttl, + ) + + def _token_factory() -> str: + return tm.token + + return CoordinatorClient( + console_base_url=console_bind_url, + storage=storage, + token_factory=_token_factory, + coord_ws_id=ws_id, + user_id=user_id, + ) + + # Pre-compute config-derived integers BEFORE any thread starts so a + # missing / non-numeric setting raises here rather than after + # ``StateWriter.start()`` has spawned a daemon we can't easily roll + # back on the runtime-bootstrap retry path. + max_active = int(config_store.get("coordinator.max_active")) + try: + idle_minutes = int(config_store.get("server.workstream_idle_timeout")) + except Exception: + idle_minutes = 0 + + coord_factory = build_console_session_factory( + registry=coord_registry, + config_store=config_store, + node_id="console", + coord_client_factory=_coord_client_factory, + ) + coord_adapter = CoordinatorAdapter( + collector=app.state.collector, + ui_factory=_ui_factory, + session_factory=coord_factory, + ) + coord_state_writer = StateWriter(storage) + coord_state_writer.start() + # Stamp the StateWriter immediately after start so a failure between + # here and the final commit (e.g. SessionManager validation) leaves + # the daemon thread reachable from ``app.state`` for + # :func:`_teardown_partial_coord_subsystem` to shut down on the + # runtime-bootstrap retry path. Without this, every retry would + # leak a fresh state-writer thread. + app.state.coord_state_writer = coord_state_writer + coord_mgr = SessionManager( + coord_adapter, + storage=storage, + max_active=max_active, + node_id=ClusterCollector.CONSOLE_PSEUDO_NODE_ID, + state_writer=coord_state_writer, + # CoordinatorAdapter implements SessionEventEmitter in full — + # every lifecycle transition fans out to the cluster collector's + # pseudo-node so the dashboard tree mirrors child state. + event_emitter=coord_adapter, + # Filter out persisted aliases that no longer resolve so a + # coordinator pinned to a since-removed alias still rehydrates + # (on the registry default) instead of 500-ing on every reopen. + model_validator=coord_registry.has_alias, + ) + # Late-bind the manager onto the adapter so + # ``_rebuild_children_registry`` / ``send`` / fan-out dispatch can + # call ``mgr.get(ws_id)``. + coord_adapter.attach(coord_mgr) + app.state.coord_mgr = coord_mgr + app.state.coord_adapter = coord_adapter + # Shared refs so ConsoleCoordinatorUI.on_state_change flows state + # transitions through the unified manager, on_rename fans out to + # the cluster dashboard, and _record_judge_metric / + # on_intent_verdict feed the console's /metrics endpoint with coord + # verdicts. + ConsoleCoordinatorUI._coord_mgr = coord_mgr + ConsoleCoordinatorUI._collector = app.state.collector + ConsoleCoordinatorUI._console_metrics = app.state.console_metrics + # Coord-side observer: when a coord goes IDLE with active children + # still running, enqueues an idle_children nudge. MUST register + # BEFORE the IdleNudgeWatcher so subscriber-fire order on the same + # IDLE event has the observer enqueueing first, then the watcher + # peeking. + coord_idle_observer = CoordinatorIdleObserver(coord_mgr, storage) + coord_idle_observer.start() + app.state.coord_idle_observer = coord_idle_observer + # Idle wake-trigger for coords. Subscribes to coord IDLE + # transitions and dispatches a synthetic empty-user-turn send when + # the coord's NudgeQueue is non-empty. + install_idle_nudge_watcher(app, coord_mgr) + # Wire the cluster-event subscription so the coordinator's SSE + # stream fans out filtered child_ws_* events. Safe to call even + # when the collector has no nodes yet — the subscription just sits + # idle until the first node event. + try: + coord_adapter.start_child_event_fanout(app.state.collector) + except Exception: + log.warning("console.coordinator_child_fanout_init_failed", exc_info=True) + # Idle cleanup: closes loaded-but-stale coords AND DB orphans left + # behind by prior console processes. The thread runs an initial + # sweep on entry (no synchronous lifespan call needed — see + # ``_coord_idle_cleanup_thread``) so cold-start cleanup doesn't + # block startup. Reuses the regular-server + # ``server.workstream_idle_timeout`` setting — the same cadence + # makes sense for both kinds and avoids a redundant config knob. + if idle_minutes > 0: + timeout_sec = float(idle_minutes * 60) + cleanup_thread = threading.Thread( + target=_coord_idle_cleanup_thread, + args=(coord_mgr, timeout_sec), + name="coord-idle-cleanup", + daemon=True, + ) + cleanup_thread.start() + app.state.coord_idle_cleanup_thread = cleanup_thread + # Final commit: stamp the registry + clear any stale boot-time error + # message under the same logical step. Callers can rely on the + # invariant that ``coord_registry`` is set iff ``coord_mgr`` is set, + # so a failed bootstrap leaves the same coherent fresh-install state + # the lifespan path produces on a registry-load failure. + app.state.coord_registry = coord_registry + app.state.coord_registry_error = "" + log.info( + "console.coordinator_mgr_ready max_active=%s", + max_active, + ) + + +def _teardown_partial_coord_subsystem(app: Any) -> None: + """Best-effort cleanup of any handles partially stamped on + ``app.state`` by a failed :func:`_bootstrap_coord_subsystem` run. + + Called from the runtime-bootstrap path's failure branch so a + subsequent CRUD retry doesn't spawn a duplicate ``StateWriter`` + daemon or re-subscribe an observer that was already wired up. + Resets ``coord_mgr`` / ``coord_adapter`` / ``coord_registry`` to + ``None`` so :func:`_require_coord_mgr` keeps returning 503 with the + builder-failure remediation message until the operator retries. + Idempotent — safe to call when nothing is partially built. + """ + from turnstone.core.idle_nudge_watcher import shutdown_idle_nudge_watchers + + state = app.state + sw = getattr(state, "coord_state_writer", None) + if sw is not None: + try: + sw.shutdown(timeout=2.0) + except Exception: + log.warning("console.coord_partial_state_writer_shutdown_failed", exc_info=True) + state.coord_state_writer = None + + obs = getattr(state, "coord_idle_observer", None) + if obs is not None: + try: + obs.shutdown() + except Exception: + log.warning("console.coord_partial_idle_observer_shutdown_failed", exc_info=True) + state.coord_idle_observer = None + + adapter = getattr(state, "coord_adapter", None) + if adapter is not None: + try: + # Tears down the ChildSource if start_child_event_fanout had + # already fired; idempotent if it hadn't. + adapter.shutdown() + except Exception: + log.warning("console.coord_partial_adapter_shutdown_failed", exc_info=True) + + # Idle nudge watchers are tracked in a list on app.state; the + # console only ever installs one (the coord watcher), so a blanket + # shutdown is safe — there is no other watcher to tear down by + # accident on a console-side app. + try: + shutdown_idle_nudge_watchers(app) + except Exception: + log.warning("console.coord_partial_idle_nudge_shutdown_failed", exc_info=True) + + state.coord_mgr = None + state.coord_adapter = None + state.coord_registry = None + # The cleanup thread is the LAST step of a successful bootstrap, so + # a partial failure can't have started one. Clear the attr defensively + # against future code-shape drift. + state.coord_idle_cleanup_thread = None + + @asynccontextmanager async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]: # Create async HTTP clients for proxy routes. Auth headers are NOT baked @@ -4327,7 +4561,6 @@ async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]: try: coord_registry = load_model_registry(storage=storage) - app.state.coord_registry = coord_registry except ValueError as exc: # No model rows configured. Endpoint returns 503 with # the error text so admin sees remediation in the UI. @@ -4335,147 +4568,19 @@ async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]: coord_registry = None if coord_registry is not None: - from turnstone.console.collector import ClusterCollector - from turnstone.console.coordinator_adapter import CoordinatorAdapter - from turnstone.console.coordinator_client import ( - CoordinatorClient, - CoordinatorTokenManager, - ) - from turnstone.console.coordinator_ui import ConsoleCoordinatorUI - from turnstone.console.session_factory import ( - build_console_session_factory, - ) - from turnstone.core.session_manager import SessionManager - - jwt_secret: str = getattr(app.state, "jwt_secret", "") - console_bind_url: str = getattr(app.state, "console_url", "") or ( - "http://127.0.0.1:8001" - ) - - def _ui_factory(ws: Workstream) -> ConsoleCoordinatorUI: - return ConsoleCoordinatorUI(ws_id=ws.id, user_id=ws.user_id or "") - - def _coord_client_factory(ws_id: str, user_id: str) -> CoordinatorClient: - ttl = int(config_store.get("coordinator.session_jwt_ttl_seconds")) - tm = CoordinatorTokenManager( - user_id=user_id or "system", - scopes=frozenset({"read", "write", "approve"}), - permissions=frozenset({"admin.coordinator"}), - secret=jwt_secret, - coord_ws_id=ws_id, - ttl_seconds=ttl, - ) - - def _token_factory() -> str: - return tm.token - - return CoordinatorClient( - console_base_url=console_bind_url, - storage=storage, - token_factory=_token_factory, - coord_ws_id=ws_id, - user_id=user_id, - ) - - coord_factory = build_console_session_factory( - registry=coord_registry, - config_store=config_store, - node_id="console", - coord_client_factory=_coord_client_factory, - ) - coord_adapter = CoordinatorAdapter( - collector=app.state.collector, - ui_factory=_ui_factory, - session_factory=coord_factory, - ) - from turnstone.core.state_writer import StateWriter - - coord_state_writer = StateWriter(storage) - coord_state_writer.start() - coord_mgr = SessionManager( - coord_adapter, - storage=storage, - max_active=int(config_store.get("coordinator.max_active")), - node_id=ClusterCollector.CONSOLE_PSEUDO_NODE_ID, - state_writer=coord_state_writer, - # CoordinatorAdapter implements SessionEventEmitter - # in full — every lifecycle transition fans out to - # the cluster collector's pseudo-node so the - # dashboard tree mirrors child state. - event_emitter=coord_adapter, - # Filter out persisted aliases that no longer resolve - # so a coordinator pinned to a since-removed alias - # still rehydrates (on the registry default) instead - # of 500-ing on every reopen. - model_validator=coord_registry.has_alias, - ) - # Late-bind the manager onto the adapter so - # ``_rebuild_children_registry`` / ``send`` / - # fan-out dispatch can call ``mgr.get(ws_id)``. - coord_adapter.attach(coord_mgr) - app.state.coord_state_writer = coord_state_writer - # Shared refs so ConsoleCoordinatorUI.on_state_change - # flows state transitions through the unified manager, - # on_rename fans out to the cluster dashboard, and - # _record_judge_metric / on_intent_verdict feed the - # console's /metrics endpoint with coord verdicts. - ConsoleCoordinatorUI._coord_mgr = coord_mgr - ConsoleCoordinatorUI._collector = app.state.collector - ConsoleCoordinatorUI._console_metrics = app.state.console_metrics - app.state.coord_mgr = coord_mgr - app.state.coord_adapter = coord_adapter - # Coord-side observer: when a coord goes IDLE with active - # children still running, enqueues an idle_children - # nudge. MUST register BEFORE the IdleNudgeWatcher so - # subscriber-fire order on the same IDLE event has the - # observer enqueueing first, then the watcher peeking. - from turnstone.console.coordinator_idle_observer import ( - CoordinatorIdleObserver, - ) - from turnstone.core.idle_nudge_watcher import install_idle_nudge_watcher - - coord_idle_observer = CoordinatorIdleObserver(coord_mgr, storage) - coord_idle_observer.start() - app.state.coord_idle_observer = coord_idle_observer - # Idle wake-trigger for coords. Subscribes to coord IDLE - # transitions and dispatches a synthetic empty-user-turn - # send when the coord's NudgeQueue is non-empty. - install_idle_nudge_watcher(app, coord_mgr) - # Wire the cluster-event subscription so the coordinator's - # SSE stream fans out filtered child_ws_* events. Safe to - # call even when the collector has no nodes yet — the - # subscription just sits idle until the first node event. - try: - coord_adapter.start_child_event_fanout(app.state.collector) - except Exception: - log.warning("console.coordinator_child_fanout_init_failed", exc_info=True) - # Idle cleanup: closes loaded-but-stale coords AND DB orphans - # left behind by prior console processes. The thread runs an - # initial sweep on entry (no synchronous lifespan call needed — - # see ``_coord_idle_cleanup_thread``) so cold-start cleanup - # doesn't block startup. Reuses the regular-server - # ``server.workstream_idle_timeout`` setting — the same cadence - # makes sense for both kinds and avoids a redundant config knob. - try: - idle_minutes = int(config_store.get("server.workstream_idle_timeout")) - except Exception: - idle_minutes = 0 - if idle_minutes > 0: - timeout_sec = float(idle_minutes * 60) - cleanup_thread = threading.Thread( - target=_coord_idle_cleanup_thread, - args=(coord_mgr, timeout_sec), - name="coord-idle-cleanup", - daemon=True, - ) - cleanup_thread.start() - app.state.coord_idle_cleanup_thread = cleanup_thread - log.info( - "console.coordinator_mgr_ready max_active=%s", - config_store.get("coordinator.max_active"), - ) + # ``_bootstrap_coord_subsystem`` stamps ``coord_registry`` + # on app.state itself as the final commit step, so a + # build failure leaves both ``coord_registry`` and + # ``coord_mgr`` ``None`` — the same coherent fresh-install + # state a registry-load failure produces above. + _bootstrap_coord_subsystem(app, storage, config_store, coord_registry) except Exception: log.warning("console.coordinator_init_failed", exc_info=True) + # The bootstrap may have partially stamped handles on + # app.state before the failure; tear them down so a daemon + # doesn't outlive a console process that gave up on + # initialising the coord subsystem. + _teardown_partial_coord_subsystem(app) yield # Shutdown @@ -9225,6 +9330,82 @@ def _refresh_coord_registry(app_state: Any, storage: Any) -> None: log.warning("console.coord_registry_refresh_shutdown_failed", exc_info=True) +def _maybe_bootstrap_coord_subsystem(app: Any, storage: Any) -> None: + """Bootstrap the coord subsystem if a freshly-installed console added + its first model row at runtime. + + On a fresh install with no model rows, the lifespan path catches the + ``ValueError`` from :func:`load_model_registry` and leaves + ``coord_mgr`` / ``coord_adapter`` / ``coord_registry`` all ``None`` + (the rest of the console still works — coord endpoints just 503 with + a remediation message). Without this helper, the operator would have + to restart the console after adding their first model definition; + :func:`_refresh_coord_registry` short-circuits when there's no + existing registry to mutate in place, so the create / update / reload + paths alone wouldn't recover. + + Idempotent — fast-paths to a no-op when the subsystem is already + built. The strictly-required ``app.state`` attributes + (``config_store``, ``collector``, ``console_metrics``) are normally + populated by the same lifespan that may have skipped the coord build; + if any are missing (e.g. unit-test shim) this is a quiet no-op so a + CRUD write that already succeeded never 500s on a bootstrap follow-on. + A failed builder run replaces the boot-time "no model definitions" + error string with a builder-failure message and tears down any partial + state so a subsequent retry doesn't leak a daemon thread. + """ + from turnstone.core.model_registry import load_model_registry + + if getattr(app.state, "coord_mgr", None) is not None: + return + config_store = getattr(app.state, "config_store", None) + collector = getattr(app.state, "collector", None) + # console_metrics is read directly (no getattr default) inside + # _bootstrap_coord_subsystem when stamping ConsoleCoordinatorUI's + # class attrs; check it here so a missing value short-circuits to a + # clean no-op instead of an AttributeError swallowed by the broad + # except below. + console_metrics = getattr(app.state, "console_metrics", None) + if config_store is None or collector is None or console_metrics is None: + # Test harness or partial init — nothing to bootstrap onto. + return + with _COORD_BOOTSTRAP_LOCK: + # Re-check inside the lock so two simultaneous CRUD writes don't + # both stand up a SessionManager / StateWriter / observer set. + if getattr(app.state, "coord_mgr", None) is not None: + return + try: + coord_registry = load_model_registry(storage=storage) + except ValueError as exc: + # Still no usable rows (e.g. all disabled). Surface the + # reason via the same channel the lifespan path uses so + # ``_require_coord_mgr``'s 503 message stays accurate. + app.state.coord_registry_error = str(exc) + return + except Exception: + log.warning("console.coord_bootstrap_load_failed", exc_info=True) + return + try: + # ``_bootstrap_coord_subsystem`` stamps ``coord_registry`` + # and clears ``coord_registry_error`` itself as the final + # commit step, so the success path needs no follow-on writes. + _bootstrap_coord_subsystem(app, storage, config_store, coord_registry) + except Exception as exc: + log.warning("console.coord_bootstrap_failed", exc_info=True) + # Tear down any partially-stamped handles so a later retry + # via the same CRUD path doesn't spawn duplicate daemons. + _teardown_partial_coord_subsystem(app) + # The "no models found" message captured at boot is + # demonstrably stale — load_model_registry just succeeded + # against N>0 rows. Replace it with a builder-specific + # diagnosis so the dashboard's 503 banner reflects the + # actual cause (operators can grep logs for the type name). + app.state.coord_registry_error = ( + f"Coordinator subsystem failed to initialise " + f"({type(exc).__name__}). See console logs and retry." + ) + + 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 @@ -9465,6 +9646,10 @@ async def admin_create_model_definition(request: Request) -> JSONResponse: ) await asyncio.to_thread(_refresh_coord_registry, request.app.state, storage) + # First-row bootstrap on a freshly-installed console: a runtime-added + # model promotes the coord subsystem from "not initialized" to ready + # without a console restart. No-op when already built. + await asyncio.to_thread(_maybe_bootstrap_coord_subsystem, request.app, storage) _emit_models_changed(request) created = storage.get_model_definition(definition_id) @@ -9629,6 +9814,7 @@ async def admin_update_model_definition(request: Request) -> JSONResponse: if updates: await asyncio.to_thread(_refresh_coord_registry, request.app.state, storage) + await asyncio.to_thread(_maybe_bootstrap_coord_subsystem, request.app, storage) _emit_models_changed(request) model_def = storage.get_model_definition(definition_id) @@ -9667,6 +9853,7 @@ async def admin_delete_model_definition(request: Request) -> JSONResponse: ) await asyncio.to_thread(_refresh_coord_registry, request.app.state, storage) + await asyncio.to_thread(_maybe_bootstrap_coord_subsystem, request.app, storage) _emit_models_changed(request) return JSONResponse({"status": "ok", "definition_id": definition_id}) @@ -9694,6 +9881,7 @@ async def admin_model_reload(request: Request) -> JSONResponse: # 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) + await asyncio.to_thread(_maybe_bootstrap_coord_subsystem, request.app, storage) _emit_models_changed(request) results = await _notify_nodes_model_reload(request)