fix(console): unify coordinator alias resolution across placeholder + factory

Previously /v1/api/models (home composer placeholder) and
console/session_factory.py walked separate two-/three-tier chains for
the coordinator alias.  session_factory was missing the
``model.default_alias`` tier, so admins who set the system default in
the Models tab would see it advertised but new coordinator sessions
would silently keep launching on ``registry.default``.

This commit:

- Extracts the chain into ``turnstone/console/coordinator_alias.py``.
  ``resolve_coordinator_alias`` returns the effective alias under a
  shared three-tier policy: explicit pin → ``model.default_alias`` →
  ``registry.default``.  Tier 2 is validated against
  ``registry.has_alias`` and falls through to tier 3 with a logged
  warning if unknown.  Tier 1 is intentionally passed through
  unvalidated so an explicit operator pin surfaces as 503 at
  ``registry.resolve`` rather than being silently swapped out.
- Wires both call sites through the helper.  The placeholder supplies
  an ``alias_filter`` that restricts every tier to enabled DB rows so
  the home composer never advertises a model the workstream picker
  can't actually offer; the session factory uses no filter (matches
  prior 503-on-typo behaviour for explicit pins).
- Adds direct integration tests for the session factory's chain
  (``tests/test_console_session_factory.py``) and updates the
  placeholder tests' fixture to provide a stub coord_registry, since
  the helper now requires one.

tests/test_console_available_models.py

@@ -23,7 +23,6 @@ placeholder stays correct as the precedence rules evolve.
 
 from __future__ import annotations
 
-from types import SimpleNamespace
 from typing import Any
 
 import pytest
@@ -64,11 +63,31 @@ def _seed_model(
     )
 
 
+class _StubRegistry:
+    """Mimics the surface ``resolve_coordinator_alias`` reads from
+    ``coord_registry``: ``.default`` and ``.has_alias()``.
+
+    Production wires this through ``ModelRegistry``, which in turn
+    pulls aliases from both DB rows and config.toml.  The fixture
+    mirrors the storage's enabled-row set so ``has_alias()`` agrees
+    with what the placeholder's enabled-row filter would accept —
+    without that alignment the helper rejects every tier-2 candidate
+    and the placeholder goes blank in cases that production handles
+    fine."""
+
+    def __init__(self, *, default: str, known: set[str]) -> None:
+        self.default = default
+        self._known = known
+
+    def has_alias(self, alias: str) -> bool:
+        return alias in self._known
+
+
 def _make_client(
     storage: SQLiteBackend,
     *,
     settings: dict[str, str] | None = None,
-    registry_default: str | None = None,
+    registry_default: str = "",
 ) -> TestClient:
     app = Starlette(
         routes=[Route("/v1/api/models", list_available_models)],
@@ -76,10 +95,11 @@ def _make_client(
     )
     app.state.auth_storage = storage
     app.state.config_store = _FakeConfigStore(dict(settings or {}))
-    if registry_default is not None:
-        # Mimic the live console's ``coord_registry`` shape — the handler
-        # only reads ``.default``, so a SimpleNamespace is enough.
-        app.state.coord_registry = SimpleNamespace(default=registry_default)
+    # ``coord_registry`` is always set in production after lifespan
+    # startup; mirror that here.  ``has_alias`` answers from the same
+    # enabled-rows set the handler filters against.
+    enabled = {r["alias"] for r in storage.list_model_definitions(enabled_only=True)}
+    app.state.coord_registry = _StubRegistry(default=registry_default, known=enabled)
     client = TestClient(app)
     client.headers.update({"X-Test-User": "admin", "X-Test-Perms": ""})
     return client

tests/test_console_session_factory.py

@@ -0,0 +1,219 @@
+"""``console/session_factory.py`` alias-resolution coverage.
+
+The console session factory resolves the coordinator alias through a
+three-tier chain that must stay in lockstep with the placeholder logic
+in ``console/server.py:list_available_models`` — otherwise the home
+composer advertises one alias while sessions launch on another.
+
+Tier order (highest priority first):
+
+1. Per-call ``model_alias`` arg, or the ``coordinator.model_alias``
+   ConfigStore setting (admin-pinned coordinator-specific override).
+2. ``model.default_alias`` ConfigStore setting (admin-managed system
+   default surfaced in the Models tab).
+3. ``registry.default`` (config.toml ``[model].default``, the boot-time
+   fallback).
+
+These tests pin each branch by intercepting ``registry.resolve`` —
+they short-circuit before ChatSession construction so the test never
+has to satisfy ChatSession's full kwarg contract.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+
+from tests._coord_test_helpers import _FakeConfigStore
+from turnstone.console.session_factory import build_console_session_factory
+
+
+class _StopBeforeChatSessionError(Exception):
+    """Sentinel raised by the capturing registry to short-circuit
+    factory execution after alias resolution but before ChatSession is
+    built.  The factory's outer code path is irrelevant to alias
+    resolution and would force the test to satisfy a long kwarg
+    contract for no extra coverage."""
+
+
+class _CapturingRegistry:
+    """Records the alias passed to ``resolve()`` and short-circuits.
+
+    ``has_alias`` answers from the configured known set so the
+    ``model.default_alias`` validation tier behaves realistically.
+    Mirrors the public surface ``ModelRegistry`` exposes to
+    session_factory: ``has_alias``, ``resolve``, and ``default``.
+    """
+
+    def __init__(self, *, default: str, known: set[str]) -> None:
+        self.default = default
+        self._known = known
+        self.captured_alias: str | None = None
+
+    def has_alias(self, alias: str) -> bool:
+        return alias in self._known
+
+    def resolve(self, alias: str) -> Any:
+        self.captured_alias = alias
+        raise _StopBeforeChatSessionError()
+
+
+def _build_factory(
+    *,
+    registry_default: str = "registry-default",
+    known_aliases: set[str] | None = None,
+    settings: dict[str, Any] | None = None,
+) -> tuple[Any, _CapturingRegistry]:
+    """Construct the factory with stub deps.  Returns ``(factory_callable,
+    registry)`` so tests can read back ``registry.captured_alias``."""
+
+    registry = _CapturingRegistry(
+        default=registry_default,
+        known=known_aliases if known_aliases is not None else {registry_default},
+    )
+    config_store = _FakeConfigStore(dict(settings or {}))
+    factory = build_console_session_factory(
+        registry=registry,  # type: ignore[arg-type]
+        config_store=config_store,  # type: ignore[arg-type]
+        node_id="console",
+        coord_client_factory=lambda ws_id, uid: MagicMock(),
+    )
+    return factory, registry
+
+
+def _invoke(factory: Any, **factory_kwargs: Any) -> None:
+    """Call the factory with a stub UI and absorb the sentinel.
+
+    Forwards ``factory_kwargs`` to the factory so per-call overrides
+    (e.g. ``model_alias``) can flow through.  Raises if any other
+    exception comes out — the test should fail loudly when alias
+    resolution itself errors rather than swallowing it.
+    """
+    ui = MagicMock()
+    ui._user_id = ""  # skip storage-backed username lookup branch
+    with pytest.raises(_StopBeforeChatSessionError):
+        factory(ui, **factory_kwargs)
+
+
+# ---------------------------------------------------------------------------
+# Tier 1 — explicit pin (per-call arg or coordinator.model_alias)
+# ---------------------------------------------------------------------------
+
+
+def test_per_call_model_alias_arg_wins_over_everything() -> None:
+    """The ``model_alias`` kwarg on the factory call (e.g. body field on
+    POST /workstreams/new) wins over both ConfigStore tiers and the
+    registry default."""
+    factory, registry = _build_factory(
+        known_aliases={"per-call", "coord-pin", "admin-default", "registry-default"},
+        settings={
+            "coordinator.model_alias": "coord-pin",
+            "model.default_alias": "admin-default",
+        },
+    )
+    _invoke(factory, model_alias="per-call")
+    assert registry.captured_alias == "per-call"
+
+
+def test_coordinator_model_alias_wins_when_no_per_call_override() -> None:
+    factory, registry = _build_factory(
+        known_aliases={"coord-pin", "admin-default", "registry-default"},
+        settings={
+            "coordinator.model_alias": "coord-pin",
+            "model.default_alias": "admin-default",
+        },
+    )
+    _invoke(factory)
+    assert registry.captured_alias == "coord-pin"
+
+
+def test_coordinator_model_alias_passed_through_unvalidated() -> None:
+    """Tier 1 is an *explicit* operator pin — when it's stale or typoed
+    we deliberately pass it through to ``registry.resolve`` so the
+    request layer turns it into a 503 with the alias surfaced in the
+    error.  Falling through silently would mask the misconfiguration."""
+    factory, registry = _build_factory(
+        known_aliases={"admin-default", "registry-default"},
+        settings={
+            "coordinator.model_alias": "ghost",  # unknown
+            "model.default_alias": "admin-default",
+        },
+    )
+    _invoke(factory)
+    assert registry.captured_alias == "ghost"
+
+
+def test_per_call_model_alias_arg_passed_through_unvalidated() -> None:
+    """The per-call ``model_alias`` kwarg (POST body field — the more
+    common production trigger) is the same kind of explicit pin as the
+    ConfigStore setting, so a stale value passes through to
+    ``registry.resolve`` rather than silently falling through to the
+    system default."""
+    factory, registry = _build_factory(
+        known_aliases={"registry-default"},
+        settings={"model.default_alias": "registry-default"},
+    )
+    _invoke(factory, model_alias="ghost")
+    assert registry.captured_alias == "ghost"
+
+
+# ---------------------------------------------------------------------------
+# Tier 2 — model.default_alias (admin-managed system default)
+# ---------------------------------------------------------------------------
+
+
+def test_model_default_alias_used_when_coordinator_unset() -> None:
+    """Regression for the historical drift: admin sets the system
+    default in the Models tab, the home composer advertises it, and new
+    coordinator sessions must launch on the same alias rather than
+    silently falling through to ``registry.default``."""
+    factory, registry = _build_factory(
+        known_aliases={"admin-default", "registry-default"},
+        settings={"model.default_alias": "admin-default"},
+    )
+    _invoke(factory)
+    assert registry.captured_alias == "admin-default"
+
+
+def test_unknown_model_default_alias_falls_through_to_registry_default() -> None:
+    """Tier 2 is *not* an explicit pin — operators set
+    ``model.default_alias`` once in the UI and forget about it; an alias
+    that's later disabled or typo'd should not 503 the coordinator,
+    since tier 3 (``registry.default``) is guaranteed to resolve."""
+    factory, registry = _build_factory(
+        known_aliases={"registry-default"},  # admin-default got removed
+        settings={"model.default_alias": "admin-default"},
+    )
+    _invoke(factory)
+    assert registry.captured_alias == "registry-default"
+
+
+def test_blank_model_default_alias_falls_through_to_registry_default() -> None:
+    factory, registry = _build_factory(
+        settings={"model.default_alias": ""},
+    )
+    _invoke(factory)
+    assert registry.captured_alias == "registry-default"
+
+
+# ---------------------------------------------------------------------------
+# Tier 3 — registry.default (config.toml [model].default)
+# ---------------------------------------------------------------------------
+
+
+def test_no_settings_uses_registry_default() -> None:
+    factory, registry = _build_factory()
+    _invoke(factory)
+    assert registry.captured_alias == "registry-default"
+
+
+def test_whitespace_only_coord_alias_falls_through() -> None:
+    """``"   "`` is not an explicit pin — ``.strip()`` reduces it to
+    "", which the chain should treat as unset."""
+    factory, registry = _build_factory(
+        settings={"coordinator.model_alias": "   "},
+    )
+    _invoke(factory)
+    assert registry.captured_alias == "registry-default"

turnstone/console/coordinator_alias.py

@@ -0,0 +1,88 @@
+"""Coordinator alias resolution shared by the placeholder API and the
+session factory.
+
+Both ``/v1/api/models`` (advertises the resolved default to the home
+composer) and ``console/session_factory.py:factory`` (resolves the
+alias new coordinator sessions launch on) walk the same three-tier
+chain.  Centralising it here means the tier names and the tier-2
+validation policy live once — the prior arrangement was two
+implementations coupled by a "keep these in sync" comment, which is
+exactly the drift trap that produced the historical bug where the
+home composer advertised one alias while sessions ran on another.
+
+Tiers, in priority order:
+
+  1. **Explicit pin** — per-call ``model_alias`` arg (factory only) or
+     the ``coordinator.model_alias`` ConfigStore setting.
+  2. **System default** — ``model.default_alias`` ConfigStore setting,
+     admin-managed in the Models tab.  Validated against
+     ``registry.has_alias()`` — a stale or typo'd value falls through
+     with a logged warning rather than 503ing.
+  3. **Registry default** — ``registry.default`` (config.toml
+     ``[model].default``), guaranteed by the registry to resolve.
+
+Tier 1 is intentionally passed through unvalidated by default: an
+explicit operator pin should surface as 503 at ``registry.resolve``
+when stale, not silently fall through to a different alias.  Callers
+that need stricter filtering (the placeholder API restricts to
+enabled DB rows so the home composer doesn't advertise a model the
+workstream picker can't offer) supply an ``alias_filter`` predicate
+applied to every tier.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from turnstone.core.log import get_logger
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from turnstone.core.config_store import ConfigStore
+    from turnstone.core.model_registry import ModelRegistry
+
+log = get_logger(__name__)
+
+
+def resolve_coordinator_alias(
+    *,
+    explicit: str | None,
+    config_store: ConfigStore,
+    registry: ModelRegistry,
+    alias_filter: Callable[[str], bool] | None = None,
+) -> str:
+    """Resolve the effective coordinator alias through the three tiers.
+
+    See module docstring for the full chain.  Returns the concrete
+    alias name, or ``""`` when every tier failed (rare — only when
+    ``registry.default`` itself fails the filter).
+    """
+
+    def _accept(alias: str) -> bool:
+        if not alias:
+            return False
+        return alias_filter(alias) if alias_filter is not None else True
+
+    explicit_alias = (explicit or "").strip()
+    if not explicit_alias:
+        explicit_alias = (config_store.get("coordinator.model_alias") or "").strip()
+    if _accept(explicit_alias):
+        return explicit_alias
+
+    fallback_alias = (config_store.get("model.default_alias") or "").strip()
+    if fallback_alias and not registry.has_alias(fallback_alias):
+        log.warning(
+            "coord_alias.model_default_alias_unknown alias=%r "
+            "— falling through to registry.default",
+            fallback_alias,
+        )
+        fallback_alias = ""
+    if _accept(fallback_alias):
+        return fallback_alias
+
+    registry_default = registry.default or ""
+    if _accept(registry_default):
+        return registry_default
+
+    return ""

turnstone/console/server.py

@@ -41,6 +41,7 @@ from starlette.staticfiles import StaticFiles
 from turnstone.api.console_spec import build_console_spec
 from turnstone.api.docs import make_docs_handler, make_openapi_handler
 from turnstone.console.collector import ClusterCollector
+from turnstone.console.coordinator_alias import resolve_coordinator_alias
 from turnstone.console.coordinator_client import load_task_envelope
 from turnstone.console.metrics import ConsoleMetrics
 from turnstone.console.router import ConsoleRouter
@@ -1677,44 +1678,32 @@ async def list_available_models(request: Request) -> JSONResponse:
     if cs is not None:
         default_alias = cs.get("model.default_alias") or ""
         channel_default_alias = cs.get("channels.default_model_alias") or ""
-        coordinator_default_alias = (cs.get("coordinator.model_alias") or "").strip()
         judge_default_alias = (cs.get("judge.model") or "").strip()
     enabled_aliases = {r["alias"] for r in rows}
     if default_alias and default_alias not in enabled_aliases:
         default_alias = ""
     if channel_default_alias and channel_default_alias not in enabled_aliases:
         channel_default_alias = ""
-    # Coordinator fallback chain — three tiers, in priority order:
-    #   1. explicit ``coordinator.model_alias``
-    #   2. ``model.default_alias`` (admin-managed default in the Models tab)
-    #   3. ``registry.default`` (config.toml ``[model].default``)
-    #
-    # Tier 3 mirrors what console/session_factory.py:109-110 does when
-    # ``coordinator.model_alias`` is unset (``effective_alias =
-    # explicit_alias or registry.default``).  Without it the placeholder
-    # went blank whenever operators left both ConfigStore keys unset, even
-    # though new coordinator sessions still launch on ``registry.default``.
-    #
-    # Note: this chain extends session_factory's by inserting tier 2 as a
-    # courtesy — admins who set ``model.default_alias`` in the UI expect
-    # the home composer to advertise it.  session_factory itself does NOT
-    # consult ``model.default_alias`` today, so a configuration where
-    # ``model.default_alias`` ≠ ``registry.default`` will still drift
-    # between placeholder ("X") and runtime ("Y").  Aligning that is a
-    # session_factory change, tracked separately.
-    #
+    # Coordinator default walks the standard three-tier chain (see
+    # :func:`turnstone.console.coordinator_alias.resolve_coordinator_alias`).
+    # The placeholder restricts every tier to enabled DB rows so the home
+    # composer doesn't advertise a model the workstream-creation picker
+    # can't actually offer — the session factory uses the same chain
+    # without that filter so explicit operator pins surface as 503 at
+    # ``registry.resolve`` instead of being silently swapped out.
+    coord_registry = getattr(request.app.state, "coord_registry", None)
+    if cs is not None and coord_registry is not None:
+        coordinator_default_alias = resolve_coordinator_alias(
+            explicit=cs.get("coordinator.model_alias"),
+            config_store=cs,
+            registry=coord_registry,
+            alias_filter=lambda a: a in enabled_aliases,
+        )
     # Judge falls back to the resolved coordinator alias when
     # ``judge.model`` is empty *or* not a registered alias — judge.model
     # is alias-only (matches IntentJudge.__init__), so an unknown value is
     # operator misconfiguration that the judge itself silently inherits
     # the session model on.
-    if not coordinator_default_alias or coordinator_default_alias not in enabled_aliases:
-        coordinator_default_alias = default_alias
-    if not coordinator_default_alias:
-        coord_registry = getattr(request.app.state, "coord_registry", None)
-        registry_default = getattr(coord_registry, "default", "") if coord_registry else ""
-        if registry_default and registry_default in enabled_aliases:
-            coordinator_default_alias = registry_default
     if not judge_default_alias or judge_default_alias not in enabled_aliases:
         judge_default_alias = coordinator_default_alias
     return JSONResponse(

turnstone/console/session_factory.py

@@ -21,6 +21,7 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
+from turnstone.console.coordinator_alias import resolve_coordinator_alias
 from turnstone.core.log import get_logger
 from turnstone.core.session import ChatSession
 from turnstone.core.workstream import WorkstreamKind
@@ -96,18 +97,19 @@ def build_console_session_factory(
                 f"console session factory only supports kind=COORDINATOR, got {kind!r}"
             )
 
-        # Resolve coordinator.model_alias from settings if caller didn't
-        # override.  Unset ``coordinator.model_alias`` falls back to the
-        # model registry's default alias — operators get a working
-        # coordinator on a freshly-provisioned console without an extra
-        # manual setting.  Resolve to the CONCRETE alias name
-        # (``registry.default``) rather than passing None downstream:
-        # ``ChatSession.__init__`` reads ``registry.get_provider(alias)``
-        # to pick the right provider class, and passing None makes it
-        # fall through to a generic OpenAI-compat provider — which
-        # mismatches when the default is Anthropic/Google-backed.
-        explicit_alias = model_alias or (config_store.get("coordinator.model_alias") or "").strip()
-        effective_alias = explicit_alias or registry.default
+        # Resolve to the CONCRETE alias name rather than passing None
+        # downstream: ``ChatSession.__init__`` reads
+        # ``registry.get_provider(alias)`` to pick the right provider
+        # class, and passing None makes it fall through to a generic
+        # OpenAI-compat provider — which mismatches when the default is
+        # Anthropic/Google-backed.  See
+        # :func:`turnstone.console.coordinator_alias.resolve_coordinator_alias`
+        # for the three-tier chain shared with the placeholder API.
+        effective_alias = resolve_coordinator_alias(
+            explicit=model_alias,
+            config_store=config_store,
+            registry=registry,
+        )
 
         r_client, r_model, r_cfg = registry.resolve(effective_alias)