diff --git a/sdk/typescript/src/types.ts b/sdk/typescript/src/types.ts index bde9a07b..dbfcbb33 100644 --- a/sdk/typescript/src/types.ts +++ b/sdk/typescript/src/types.ts @@ -284,7 +284,6 @@ export interface CreateSkillResourceRequest { export interface BackendStatus { status: string; - circuit_state: string; } export interface WorkstreamCounts { diff --git a/tests/test_console.py b/tests/test_console.py index 453392f8..7e34325d 100644 --- a/tests/test_console.py +++ b/tests/test_console.py @@ -418,13 +418,12 @@ class TestCollectorDelta: c._nodes["node-a"] = NodeSnapshot( node_id="node-a", server_url="http://a:8080", - health={"status": "ok", "backend": {"status": "up", "circuit_state": "closed"}}, + health={"status": "ok", "backend": {"status": "up"}}, ) - c._apply_delta("node-a", {"type": "health_changed", "circuit_state": "open"}) + c._apply_delta("node-a", {"type": "health_changed", "backend_status": "degraded"}) health = c._nodes["node-a"].health - assert health["backend"]["circuit_state"] == "open" assert health["backend"]["status"] == "down" assert health["status"] == "degraded" diff --git a/tests/test_healthcheck.py b/tests/test_healthcheck.py index ecc2edfd..783e0bd8 100644 --- a/tests/test_healthcheck.py +++ b/tests/test_healthcheck.py @@ -1,4 +1,4 @@ -"""Tests for turnstone.core.healthcheck — backend health monitor with circuit breaker.""" +"""Tests for turnstone.core.healthcheck — passive backend health tracking.""" from __future__ import annotations @@ -10,39 +10,16 @@ import pytest if TYPE_CHECKING: from collections.abc import Generator -from turnstone.core.healthcheck import BackendHealthMonitor, CircuitState +from turnstone.core.healthcheck import BackendHealthTracker, HealthTrackerRegistry # --------------------------------------------------------------------------- -# CircuitState enum +# Fixtures # --------------------------------------------------------------------------- -class TestCircuitState: - def test_closed(self) -> None: - assert CircuitState.CLOSED.value == "closed" - - def test_open(self) -> None: - assert CircuitState.OPEN.value == "open" - - def test_half_open(self) -> None: - assert CircuitState.HALF_OPEN.value == "half_open" - - -# --------------------------------------------------------------------------- -# BackendHealthMonitor -# --------------------------------------------------------------------------- - - -@pytest.fixture -def mock_client() -> MagicMock: - client = MagicMock() - client.models.list.return_value.data = [MagicMock(id="test-model")] - return client - - @pytest.fixture def mock_metrics() -> Generator[MagicMock]: - """Patch the metrics singleton so set_backend_status / set_circuit_state exist.""" + """Patch the metrics singleton so set_backend_status exists.""" m = MagicMock() with ( patch("turnstone.core.healthcheck.metrics", m, create=True), @@ -51,240 +28,185 @@ def mock_metrics() -> Generator[MagicMock]: yield m -def _make_monitor( - client: MagicMock, - failure_threshold: int = 3, - cooldown: float = 60.0, -) -> BackendHealthMonitor: - return BackendHealthMonitor( - client=client, - probe_interval=1.0, - probe_timeout=1.0, - failure_threshold=failure_threshold, - cooldown=cooldown, - ) +def _make_tracker(failure_threshold: int = 3) -> BackendHealthTracker: + return BackendHealthTracker(failure_threshold=failure_threshold) -class TestBackendHealthMonitor: - def test_starts_closed(self, mock_client: MagicMock) -> None: - mon = _make_monitor(mock_client) - assert mon.circuit_state == CircuitState.CLOSED - assert mon.is_healthy is True +# --------------------------------------------------------------------------- +# BackendHealthTracker +# --------------------------------------------------------------------------- - def test_record_failure_increments( - self, mock_client: MagicMock, mock_metrics: MagicMock - ) -> None: - """Failures below threshold do not open the circuit.""" - mon = _make_monitor(mock_client, failure_threshold=5) + +class TestBackendHealthTracker: + def test_starts_healthy(self) -> None: + t = _make_tracker() + assert t.is_healthy is True + assert t.is_degraded is False + assert t.consecutive_failures == 0 + + def test_failures_below_threshold(self, mock_metrics: MagicMock) -> None: + """Failures below threshold do not degrade.""" + t = _make_tracker(failure_threshold=5) for _ in range(4): - mon.record_failure() - assert mon.circuit_state == CircuitState.CLOSED + t.record_failure() + assert t.is_healthy is True + assert t.consecutive_failures == 4 - def test_opens_after_threshold(self, mock_client: MagicMock, mock_metrics: MagicMock) -> None: - mon = _make_monitor(mock_client, failure_threshold=3) + def test_degrades_at_threshold(self, mock_metrics: MagicMock) -> None: + t = _make_tracker(failure_threshold=3) for _ in range(3): - mon.record_failure() - assert mon.circuit_state == CircuitState.OPEN - assert mon.is_healthy is False + t.record_failure() + assert t.is_degraded is True + assert t.is_healthy is False - def test_should_reject_when_open(self, mock_client: MagicMock, mock_metrics: MagicMock) -> None: - mon = _make_monitor(mock_client, failure_threshold=1, cooldown=9999.0) - mon.record_failure() - assert mon.circuit_state == CircuitState.OPEN - assert mon.acquire_request_permit() is False + def test_stays_degraded_on_more_failures(self, mock_metrics: MagicMock) -> None: + t = _make_tracker(failure_threshold=2) + for _ in range(5): + t.record_failure() + assert t.is_degraded is True + assert t.consecutive_failures == 5 - @patch("turnstone.core.healthcheck.time") - def test_half_open_after_cooldown( - self, mock_time: MagicMock, mock_client: MagicMock, mock_metrics: MagicMock - ) -> None: - """After cooldown elapses, should_allow_request transitions to HALF_OPEN.""" - t = 1000.0 - mock_time.monotonic.return_value = t + def test_success_clears_degraded(self, mock_metrics: MagicMock) -> None: + t = _make_tracker(failure_threshold=2) + t.record_failure() + t.record_failure() + assert t.is_degraded is True + t.record_success() + assert t.is_healthy is True + assert t.consecutive_failures == 0 - mon = _make_monitor(mock_client, failure_threshold=1, cooldown=60.0) - # Override _last_state_change to use our mocked time - mon._last_state_change = t - mon.record_failure() - assert mon.circuit_state == CircuitState.OPEN + def test_success_resets_failure_count(self, mock_metrics: MagicMock) -> None: + t = _make_tracker(failure_threshold=5) + for _ in range(4): + t.record_failure() + t.record_success() + assert t.consecutive_failures == 0 + # Should need 5 more failures to degrade + for _ in range(4): + t.record_failure() + assert t.is_healthy is True - # Advance past cooldown - mock_time.monotonic.return_value = t + 61.0 - assert mon.acquire_request_permit() is True - assert mon.circuit_state == CircuitState.HALF_OPEN # type: ignore[comparison-overlap] + def test_state_changed_callback_on_degrade(self, mock_metrics: MagicMock) -> None: + events: list[str] = [] + t = BackendHealthTracker(failure_threshold=2, on_state_changed=events.append) + t.record_failure() + assert events == [] + t.record_failure() + assert events == ["degraded"] - def test_success_resets(self, mock_client: MagicMock, mock_metrics: MagicMock) -> None: - """record_success resets failures and closes circuit from any state.""" - mon = _make_monitor(mock_client, failure_threshold=1) - mon.record_failure() - assert mon.circuit_state == CircuitState.OPEN + def test_state_changed_callback_on_recover(self, mock_metrics: MagicMock) -> None: + events: list[str] = [] + t = BackendHealthTracker(failure_threshold=1, on_state_changed=events.append) + t.record_failure() + assert events == ["degraded"] + t.record_success() + assert events == ["degraded", "healthy"] - mon.record_success() - assert mon.circuit_state == CircuitState.CLOSED # type: ignore[comparison-overlap] - assert mon.is_healthy is True - # Internal counter should be reset - assert mon._consecutive_failures == 0 + def test_no_callback_when_already_degraded(self, mock_metrics: MagicMock) -> None: + """Extra failures after degraded don't fire again.""" + events: list[str] = [] + t = BackendHealthTracker(failure_threshold=1, on_state_changed=events.append) + t.record_failure() + t.record_failure() + t.record_failure() + assert events == ["degraded"] # only once - def test_should_allow_when_closed(self, mock_client: MagicMock) -> None: - mon = _make_monitor(mock_client) - assert mon.acquire_request_permit() is True + def test_no_callback_when_already_healthy(self, mock_metrics: MagicMock) -> None: + """Success while healthy doesn't fire.""" + events: list[str] = [] + t = BackendHealthTracker(failure_threshold=3, on_state_changed=events.append) + t.record_success() + t.record_success() + assert events == [] - def test_half_open_allows_only_one_request( - self, mock_client: MagicMock, mock_metrics: MagicMock - ) -> None: - """HALF_OPEN permits exactly one probe; subsequent callers are blocked.""" - mon = _make_monitor(mock_client, failure_threshold=1) - mon.record_failure() - assert mon.circuit_state == CircuitState.OPEN + def test_no_direct_metrics_calls(self) -> None: + """Tracker does not touch metrics — the server callback handles it.""" + t = _make_tracker(failure_threshold=1) + t.record_failure() + t.record_success() + # No assertion on metrics — the tracker delegates metric updates + # to the server-level callback via on_state_changed - # Force into HALF_OPEN with permit - with mon._lock: - mon._state = CircuitState.HALF_OPEN - mon._half_open_permit = True - # First caller gets through - assert mon.acquire_request_permit() is True - # Second caller is blocked - assert mon.acquire_request_permit() is False - # Third caller is also blocked - assert mon.acquire_request_permit() is False +# --------------------------------------------------------------------------- +# HealthTrackerRegistry +# --------------------------------------------------------------------------- - def test_half_open_success_reopens_to_all( - self, mock_client: MagicMock, mock_metrics: MagicMock - ) -> None: - """After probe succeeds in HALF_OPEN, circuit closes and all requests pass.""" - mon = _make_monitor(mock_client, failure_threshold=1) - mon.record_failure() - with mon._lock: - mon._state = CircuitState.HALF_OPEN - mon._half_open_permit = False # permit already consumed - # Probe succeeds - mon.record_success() - assert mon.circuit_state == CircuitState.CLOSED # type: ignore[comparison-overlap] - # All callers pass now - assert mon.acquire_request_permit() is True - assert mon.acquire_request_permit() is True +class TestHealthTrackerRegistry: + def test_same_backend_shares_tracker(self, mock_metrics: MagicMock) -> None: + """Two aliases on the same (provider, base_url) share a tracker.""" + reg = HealthTrackerRegistry(failure_threshold=5) + t1 = reg.get_tracker("openai", "https://api.openai.com/v1") + t2 = reg.get_tracker("openai", "https://api.openai.com/v1") + assert t1 is t2 - def test_half_open_failure_blocks_all( - self, mock_client: MagicMock, mock_metrics: MagicMock - ) -> None: - """After probe fails in HALF_OPEN, circuit reopens and all requests blocked.""" - mon = _make_monitor(mock_client, failure_threshold=1, cooldown=9999.0) - mon.record_failure() - with mon._lock: - mon._state = CircuitState.HALF_OPEN - mon._half_open_permit = False + def test_different_backends_independent(self, mock_metrics: MagicMock) -> None: + """Different (provider, base_url) pairs get independent trackers.""" + reg = HealthTrackerRegistry(failure_threshold=5) + t_cloud = reg.get_tracker("openai", "https://api.openai.com/v1") + t_local = reg.get_tracker("openai-compatible", "http://localhost:8000/v1") + assert t_cloud is not t_local - # Probe fails - mon.record_failure() - assert mon.circuit_state == CircuitState.OPEN - assert mon.acquire_request_permit() is False + def test_trailing_slash_normalized(self, mock_metrics: MagicMock) -> None: + """Trailing slashes on base_url are normalized away.""" + reg = HealthTrackerRegistry(failure_threshold=5) + t1 = reg.get_tracker("openai", "https://api.openai.com/v1/") + t2 = reg.get_tracker("openai", "https://api.openai.com/v1") + assert t1 is t2 - def test_half_open_failure_reopens( - self, mock_client: MagicMock, mock_metrics: MagicMock - ) -> None: - """A failure in HALF_OPEN re-opens the circuit immediately.""" - mon = _make_monitor(mock_client, failure_threshold=1) - mon.record_failure() - assert mon.circuit_state == CircuitState.OPEN + def test_degraded_isolation(self, mock_metrics: MagicMock) -> None: + """Degrading one backend does not affect another.""" + reg = HealthTrackerRegistry(failure_threshold=2) + t_cloud = reg.get_tracker("openai", "https://api.openai.com/v1") + t_local = reg.get_tracker("openai-compatible", "http://localhost:8000/v1") + # Degrade the cloud tracker + t_cloud.record_failure() + t_cloud.record_failure() + assert t_cloud.is_degraded is True + # Local should be unaffected + assert t_local.is_healthy is True - # Force into HALF_OPEN - with mon._lock: - mon._state = CircuitState.HALF_OPEN - mon._update_metrics() + def test_get_tracker_for_alias(self, mock_metrics: MagicMock) -> None: + """get_tracker_for_alias looks up by model config's backend.""" + from turnstone.core.model_registry import ModelConfig, ModelRegistry - # Another failure should reopen - mon.record_failure() - assert mon.circuit_state == CircuitState.OPEN + models = { + "cloud": ModelConfig( + "cloud", "https://api.openai.com/v1", "sk", "gpt-4o", provider="openai" + ), + "local": ModelConfig( + "local", "http://localhost:8000/v1", "x", "qwen", provider="openai-compatible" + ), + } + model_reg = ModelRegistry(models=models, default="cloud") - def test_probe_success_closes(self, mock_client: MagicMock, mock_metrics: MagicMock) -> None: - """A successful probe closes the circuit.""" - mon = _make_monitor(mock_client, failure_threshold=1) - mon.record_failure() - assert mon.circuit_state == CircuitState.OPEN + reg = HealthTrackerRegistry(failure_threshold=5) + # No tracker created yet — should return None + assert reg.get_tracker_for_alias(model_reg, "cloud") is None - # Simulate probe success - assert mon._probe_once() is True - mon.record_success() - assert mon.circuit_state == CircuitState.CLOSED # type: ignore[comparison-overlap] + # Create a tracker for the cloud backend + t = reg.get_tracker("openai", "https://api.openai.com/v1") + assert reg.get_tracker_for_alias(model_reg, "cloud") is t - def test_probe_failure_opens(self, mock_client: MagicMock, mock_metrics: MagicMock) -> None: - """Enough probe failures open the circuit.""" - mock_client.with_options.return_value.models.list.side_effect = ConnectionError("down") - mon = _make_monitor(mock_client, failure_threshold=2) + # Local alias should still return None (no tracker for that backend) + assert reg.get_tracker_for_alias(model_reg, "local") is None - assert mon._probe_once() is False - mon.record_failure() - assert mon.circuit_state == CircuitState.CLOSED # only 1 failure - - assert mon._probe_once() is False - mon.record_failure() - assert mon.circuit_state == CircuitState.OPEN # type: ignore[comparison-overlap] - - def test_probe_loop_autonomous_recovery( - self, mock_client: MagicMock, mock_metrics: MagicMock - ) -> None: - """_probe_loop transitions OPEN → HALF_OPEN → CLOSED without user requests.""" - # Use very short intervals so the test is fast - mon = BackendHealthMonitor( - client=mock_client, - probe_interval=0.05, - probe_timeout=1.0, - failure_threshold=1, - cooldown=0.1, + def test_state_changed_callback(self, mock_metrics: MagicMock) -> None: + """on_state_changed fires with backend key and state.""" + events: list[tuple[str, str]] = [] + reg = HealthTrackerRegistry( + failure_threshold=2, + on_state_changed=lambda backend, state: events.append((backend, state)), ) - # Trip the circuit - mon.record_failure() - assert mon.circuit_state == CircuitState.OPEN + t = reg.get_tracker("openai", "https://api.openai.com/v1") + t.record_failure() + t.record_failure() # triggers degraded + assert len(events) == 1 + assert events[0][0] == "openai:https://api.openai.com/v1" + assert events[0][1] == "degraded" - # Backend is healthy — probe_once will succeed - mock_client.with_options.return_value.models.list.return_value = MagicMock() - - # Start the probe loop and wait for autonomous recovery - mon.start() - try: - import time - - deadline = time.monotonic() + 5.0 - while mon.circuit_state != CircuitState.CLOSED and time.monotonic() < deadline: - time.sleep(0.05) - assert mon.circuit_state == CircuitState.CLOSED - # User requests should flow again without anyone calling acquire_request_permit - assert mon.acquire_request_permit() is True - finally: - mon.stop() - if mon._thread: - mon._thread.join(timeout=2.0) - - def test_probe_loop_no_user_permit_during_probe( - self, mock_client: MagicMock, mock_metrics: MagicMock - ) -> None: - """While background probe is in HALF_OPEN, user requests are blocked.""" - mon = BackendHealthMonitor( - client=mock_client, - probe_interval=0.05, - probe_timeout=1.0, - failure_threshold=1, - cooldown=0.1, - ) - mon.record_failure() - assert mon.circuit_state == CircuitState.OPEN - - # Force into HALF_OPEN as the probe loop would - with mon._lock: - mon._state = CircuitState.HALF_OPEN - mon._half_open_permit = False # probe consumes it - - # User requests should be blocked — only the probe gets through - assert mon.acquire_request_permit() is False - - def test_stop_thread(self, mock_client: MagicMock) -> None: - """stop() signals the probe loop to exit.""" - mon = _make_monitor(mock_client) - mon.start() - assert mon._thread is not None - assert mon._thread.is_alive() - - mon.stop() - mon._thread.join(timeout=3.0) - assert not mon._thread.is_alive() + def test_backend_key_static(self) -> None: + """backend_key is a static method returning normalized tuple.""" + key = HealthTrackerRegistry.backend_key("anthropic", "https://api.anthropic.com/") + assert key == ("anthropic", "https://api.anthropic.com") diff --git a/tests/test_model_registry.py b/tests/test_model_registry.py index f85cd328..3707a63f 100644 --- a/tests/test_model_registry.py +++ b/tests/test_model_registry.py @@ -952,71 +952,113 @@ class TestExtractContextWindow: m.model_dump.return_value = {} assert _extract_context_window(m, "openai") is None + # Model-change detection via active probes was removed. + # Backend health is now tracked passively (see test_healthcheck.py). -class TestHealthMonitorModelChange: - def test_model_change_fires_callback(self) -> None: - from turnstone.core.healthcheck import BackendHealthMonitor - changes: list[tuple[str, int | None]] = [] +# --------------------------------------------------------------------------- +# load_model_registry — DB-only startup (no CLI model) +# --------------------------------------------------------------------------- - def on_change(model_id: str, ctx: int | None) -> None: - changes.append((model_id, ctx)) - client = MagicMock() - monitor = BackendHealthMonitor( - client=client, - provider="openai", - initial_model="model-a", - on_model_changed=on_change, +class TestLoadModelRegistryDBOnly: + """Tests for starting the server with models defined only in DB/config, + without any CLI --model argument.""" + + def test_db_only_no_cli_model(self) -> None: + """Registry builds from DB models when model='' (no CLI model).""" + storage = _MockStorage( + [ + { + "alias": "cloud", + "model": "gpt-5", + "provider": "openai", + "base_url": "https://api.openai.com/v1", + "api_key": "sk-test", + "context_window": 128000, + "capabilities": "{}", + "enabled": True, + }, + ] ) + with patch("turnstone.core.model_registry.load_config", return_value={}): + reg = load_model_registry(model="", storage=storage) + assert reg.count == 1 + assert reg.has_alias("cloud") + # "cloud" should be picked as default since "default" doesn't exist + assert reg.default == "cloud" - # Simulate probe returning a different model - resp = MagicMock() - m = MagicMock() - m.id = "model-b" - m.model_dump.return_value = {"max_model_len": 131072} - resp.data = [m] - - monitor._check_model_change(resp) - assert len(changes) == 1 - assert changes[0] == ("model-b", 131072) - assert monitor._last_detected_model == "model-b" - - def test_same_model_no_callback(self) -> None: - from turnstone.core.healthcheck import BackendHealthMonitor - - changes: list[tuple[str, int | None]] = [] - - def on_change(model_id: str, ctx: int | None) -> None: - changes.append((model_id, ctx)) - - client = MagicMock() - monitor = BackendHealthMonitor( - client=client, - provider="openai", - initial_model="model-a", - on_model_changed=on_change, + def test_db_only_with_config_default(self) -> None: + """Config [model].default is respected when it matches a DB alias.""" + storage = _MockStorage( + [ + { + "alias": "fast", + "model": "gpt-4o-mini", + "provider": "openai", + "base_url": "https://api.openai.com/v1", + "api_key": "sk-test", + "context_window": 128000, + "capabilities": "{}", + "enabled": True, + }, + { + "alias": "smart", + "model": "gpt-5", + "provider": "openai", + "base_url": "https://api.openai.com/v1", + "api_key": "sk-test", + "context_window": 128000, + "capabilities": "{}", + "enabled": True, + }, + ] ) + fake_cfg: dict[str, Any] = {"model": {"default": "smart"}} + with patch("turnstone.core.model_registry.load_config", return_value=fake_cfg): + reg = load_model_registry(model="", storage=storage) + assert reg.default == "smart" - resp = MagicMock() - m = MagicMock() - m.id = "model-a" - m.model_dump.return_value = {} - resp.data = [m] + def test_config_toml_only_no_cli_model(self) -> None: + """Registry builds from config.toml [models.*] when model=''.""" + fake_cfg: dict[str, Any] = { + "models": { + "local": { + "model": "qwen3-32b", + "base_url": "http://localhost:8000/v1", + "api_key": "dummy", + }, + }, + } + with patch("turnstone.core.model_registry.load_config", return_value=fake_cfg): + reg = load_model_registry(model="") + assert reg.count == 1 + assert reg.default == "local" - monitor._check_model_change(resp) - assert len(changes) == 0 + def test_no_models_anywhere_raises(self) -> None: + """ValueError when no models from CLI, config, or DB.""" + with ( + patch("turnstone.core.model_registry.load_config", return_value={}), + pytest.raises(ValueError, match="No model definitions found"), + ): + load_model_registry(model="") - def test_no_callback_configured(self) -> None: - from turnstone.core.healthcheck import BackendHealthMonitor - - client = MagicMock() - monitor = BackendHealthMonitor(client=client, initial_model="model-a") - - resp = MagicMock() - m = MagicMock() - m.id = "model-b" - resp.data = [m] - - # Should not raise - monitor._check_model_change(resp) + def test_no_default_entry_created_when_model_empty(self) -> None: + """When model='', no 'default' alias is created from CLI args.""" + storage = _MockStorage( + [ + { + "alias": "cloud", + "model": "gpt-5", + "provider": "openai", + "base_url": "https://api.openai.com/v1", + "api_key": "sk-test", + "context_window": 128000, + "capabilities": "{}", + "enabled": True, + }, + ] + ) + with patch("turnstone.core.model_registry.load_config", return_value={}): + reg = load_model_registry(model="", storage=storage) + assert not reg.has_alias("default") diff --git a/tests/test_server_live.py b/tests/test_server_live.py index c577f141..bfe8a16d 100644 --- a/tests/test_server_live.py +++ b/tests/test_server_live.py @@ -756,7 +756,6 @@ class TestServerHealthMetrics: data = json.loads(body) assert "backend" in data assert data["backend"]["status"] in ("up", "down") - assert data["backend"]["circuit_state"] in ("closed", "open", "half_open") def test_metrics_contains_sse_connections(self): _, _, body = self._get("/metrics") @@ -770,9 +769,10 @@ class TestServerHealthMetrics: _, _, body = self._get("/metrics") assert "turnstone_backend_up" in body - def test_metrics_contains_circuit_state(self): + def test_metrics_no_circuit_state(self): + """Circuit state metric was removed (passive health tracking only).""" _, _, body = self._get("/metrics") - assert "turnstone_circuit_state" in body + assert "turnstone_circuit_state" not in body def test_metrics_contains_eviction_counter(self): _, _, body = self._get("/metrics") diff --git a/turnstone/api/server_schemas.py b/turnstone/api/server_schemas.py index 13e1b604..a93632b7 100644 --- a/turnstone/api/server_schemas.py +++ b/turnstone/api/server_schemas.py @@ -145,7 +145,6 @@ class ListSavedWorkstreamsResponse(BaseModel): class BackendStatus(BaseModel): status: str = Field(examples=["up", "down"]) - circuit_state: str = Field(examples=["closed", "open", "half_open"]) class WorkstreamCounts(BaseModel): diff --git a/turnstone/console/collector.py b/turnstone/console/collector.py index 7bbdfcb5..010b8e26 100644 --- a/turnstone/console/collector.py +++ b/turnstone/console/collector.py @@ -524,15 +524,14 @@ class ClusterCollector: pending_events.append({"type": "ws_rename", "ws_id": ws_id, "name": name}) elif etype == "health_changed": - # Update the health dict's circuit state in-place - circuit = data.get("circuit_state", "") - if circuit: + # Update the health dict's backend status in-place + bstatus = data.get("backend_status", "") + if bstatus: if not node.health: node.health = {} backend = node.health.setdefault("backend", {}) - backend["circuit_state"] = circuit - backend["status"] = "up" if circuit == "closed" else "down" - node.health["status"] = "ok" if circuit == "closed" else "degraded" + backend["status"] = "up" if bstatus == "healthy" else "down" + node.health["status"] = "ok" if bstatus == "healthy" else "degraded" # Not forwarded to cluster SSE — next snapshot refreshes UI elif etype == "aggregate": diff --git a/turnstone/console/server.py b/turnstone/console/server.py index d978db66..29e0f679 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -5201,7 +5201,31 @@ async def admin_list_model_definitions(request: Request) -> JSONResponse: } ) - return JSONResponse({"models": result}) + # Include the effective default alias so the UI can highlight it. + # Prefer ConfigStore override, fall back to config.toml [model].default, + # then validate against the actual enabled model list (same fallback + # rules as load_model_registry). + configured_default = "" + cs = getattr(request.app.state, "config_store", None) + if cs: + configured_default = cs.get("model.default_alias") or "" + if not configured_default: + from turnstone.core.config import load_config as _load_cfg + + configured_default = _load_cfg().get("model", {}).get("default", "default") + + enabled_aliases = [m["alias"] for m in result if m.get("alias") and m.get("enabled", True)] + enabled_set = set(enabled_aliases) + if configured_default in enabled_set: + default_alias = configured_default + elif "default" in enabled_set: + default_alias = "default" + elif enabled_aliases: + default_alias = enabled_aliases[0] + else: + default_alias = "" + + return JSONResponse({"models": result, "default_alias": default_alias}) async def admin_create_model_definition(request: Request) -> JSONResponse: diff --git a/turnstone/console/static/admin.js b/turnstone/console/static/admin.js index ee24eb3b..bab6082f 100644 --- a/turnstone/console/static/admin.js +++ b/turnstone/console/static/admin.js @@ -4098,6 +4098,7 @@ function _pollInstallStatus(serverId, serverName, attempt) { // --------------------------------------------------------------------------- var _modelDefs = []; +var _modelDefaultAlias = ""; var _modelCreateTrap = null; var _modelCreateTrigger = null; @@ -4109,6 +4110,7 @@ function loadAdminModels() { }) .then(function (data) { _modelDefs = data.models || []; + _modelDefaultAlias = data.default_alias || ""; _renderModels(_modelDefs); }) .catch(function () { @@ -4161,7 +4163,8 @@ function _renderModels(items) { row.className = "admin-row models-grid " + rowClass; row.setAttribute("role", "listitem"); - // Alias + source badge + // Alias + source badge + default badge + var isDefault = m.alias === _modelDefaultAlias; var colAlias = document.createElement("span"); colAlias.className = "admin-col"; colAlias.textContent = m.alias; @@ -4172,6 +4175,13 @@ function _renderModels(items) { badge.textContent = isConfig ? "config" : "db"; colAlias.appendChild(document.createTextNode(" ")); colAlias.appendChild(badge); + if (isDefault) { + var defBadge = document.createElement("span"); + defBadge.className = "scope-badge scope-default"; + defBadge.textContent = "default"; + colAlias.appendChild(document.createTextNode(" ")); + colAlias.appendChild(defBadge); + } row.appendChild(colAlias); // Model ID @@ -4210,11 +4220,21 @@ function _renderModels(items) { // Actions var colActions = document.createElement("span"); colActions.className = "admin-col"; + if (!isDefault && m.enabled) { + var defBtn = document.createElement("button"); + defBtn.className = "admin-btn-action"; + defBtn.textContent = "set default"; + defBtn.setAttribute("data-model-set-default", m.alias); + defBtn.setAttribute("aria-label", "Set " + m.alias + " as default model"); + defBtn.setAttribute("title", "Set " + m.alias + " as default model"); + colActions.appendChild(defBtn); + } if (!isConfig) { var editBtn = document.createElement("button"); editBtn.className = "admin-btn-action"; editBtn.textContent = "edit"; editBtn.setAttribute("data-model-edit", m.definition_id); + editBtn.setAttribute("title", "Edit " + m.alias); colActions.appendChild(editBtn); var delBtn = document.createElement("button"); @@ -4222,6 +4242,7 @@ function _renderModels(items) { delBtn.textContent = "del"; delBtn.setAttribute("data-model-delete", m.definition_id); delBtn.setAttribute("data-model-alias", m.alias); + delBtn.setAttribute("title", "Delete " + m.alias); colActions.appendChild(delBtn); } row.appendChild(colActions); @@ -4230,6 +4251,30 @@ function _renderModels(items) { } // Bind event handlers + el.querySelectorAll("[data-model-set-default]").forEach(function (btn) { + btn.addEventListener("click", function () { + var alias = this.getAttribute("data-model-set-default"); + var self = this; + self.disabled = true; + self.textContent = "setting\u2026"; + authFetch("/v1/api/admin/settings/model.default_alias", { + method: "PUT", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ value: alias }), + }) + .then(function (r) { + if (!r.ok) throw new Error(); + showToast("Default model set to " + alias); + _flagModelSyncPending(); + loadAdminModels(); + }) + .catch(function () { + showToast("Failed to set default model"); + self.disabled = false; + self.textContent = "set default"; + }); + }); + }); el.querySelectorAll("[data-model-edit]").forEach(function (btn) { btn.addEventListener("click", function () { showEditModelModal(this.getAttribute("data-model-edit")); diff --git a/turnstone/console/static/app.js b/turnstone/console/static/app.js index 76758485..d02b56d9 100644 --- a/turnstone/console/static/app.js +++ b/turnstone/console/static/app.js @@ -653,25 +653,21 @@ function buildNodeRow(node) { '%">' : ""; - var circuitTitle = ""; + var healthTitle = ""; if (node.health && node.health.backend) { - circuitTitle = - "backend: " + - node.health.backend.status + - ", circuit: " + - node.health.backend.circuit_state; + healthTitle = "backend: " + node.health.backend.status; } var degradedBadge = isDegraded ? 'degraded' : ""; row.innerHTML = '' + diff --git a/turnstone/console/static/style.css b/turnstone/console/static/style.css index 313e2911..09c6ff07 100644 --- a/turnstone/console/static/style.css +++ b/turnstone/console/static/style.css @@ -2196,6 +2196,7 @@ h3.skill-spec-heading { font-size: inherit; margin-block: 0; } /* -- MCP source badges ---------------------------------------------------- */ .scope-config{color:var(--magenta);border-color:rgba(192,132,252,.25)} +.scope-default{color:var(--yellow);border-color:rgba(251,191,36,.3)} .scope-manual{color:var(--cyan);border-color:rgba(103,232,249,.2)} .scope-registry{color:var(--green);border-color:rgba(52,211,153,.2)} @@ -2357,12 +2358,13 @@ h3.skill-spec-heading { font-size: inherit; margin-block: 0; } } /* -- Models grid --------------------------------------------------------- */ -.models-grid{grid-template-columns:1.2fr 1.2fr 80px 90px 80px 120px;gap:0 6px} +.models-grid{grid-template-columns:1.2fr 1.2fr 80px 90px 80px 160px;gap:0 6px} @media(max-width:700px){ - .models-grid{grid-template-columns:1fr 80px 120px} + .models-grid{grid-template-columns:1fr 80px 160px} .models-grid .admin-col:nth-child(2), .models-grid .admin-col:nth-child(3), .models-grid .admin-col:nth-child(4){display:none} + .models-grid .admin-col:last-child{white-space:normal;display:flex;flex-wrap:wrap;gap:2px} } /* Model status indicators */ diff --git a/turnstone/core/config.py b/turnstone/core/config.py index 40f9b872..0b8a5f81 100644 --- a/turnstone/core/config.py +++ b/turnstone/core/config.py @@ -133,10 +133,7 @@ _CONFIG_MAP: dict[str, dict[str, str]] = { "trusted_proxies": "ratelimit_trusted_proxies", }, "health": { - "backend_probe_interval": "health_probe_interval", - "backend_probe_timeout": "health_probe_timeout", - "circuit_breaker_threshold": "circuit_breaker_threshold", - "circuit_breaker_cooldown": "circuit_breaker_cooldown", + "failure_threshold": "health_failure_threshold", }, "database": { "backend": "db_backend", diff --git a/turnstone/core/healthcheck.py b/turnstone/core/healthcheck.py index 64f194b5..f3c3c541 100644 --- a/turnstone/core/healthcheck.py +++ b/turnstone/core/healthcheck.py @@ -1,10 +1,13 @@ -"""Background LLM backend health monitor with circuit breaker.""" +"""Per-backend health tracking via passive success/failure recording. + +No active probing or circuit breakers — backends are marked *degraded* +after a configurable number of consecutive failures and recover +automatically when a request succeeds. +""" from __future__ import annotations -import enum import threading -import time from typing import TYPE_CHECKING, Any from turnstone.core.log import get_logger @@ -12,77 +15,39 @@ from turnstone.core.log import get_logger if TYPE_CHECKING: from collections.abc import Callable - from openai import OpenAI - log = get_logger(__name__) -class CircuitState(enum.Enum): - CLOSED = "closed" - OPEN = "open" - HALF_OPEN = "half_open" +# --------------------------------------------------------------------------- +# Per-backend health tracker +# --------------------------------------------------------------------------- -class BackendHealthMonitor: - """Monitors LLM backend health via periodic probes and passive failure tracking. +class BackendHealthTracker: + """Tracks LLM backend health via passive success/failure recording. - Circuit breaker state machine: - CLOSED -- backend responding, all requests pass - OPEN -- backend unreachable, fast-fail for cooldown period - HALF_OPEN -- cooldown expired, next probe decides + State machine:: + + healthy --(N consecutive failures)--> degraded + degraded --(any success)-------------> healthy + + Requests are **never blocked** — the degraded flag is advisory + (used for observability and fallback ordering). """ def __init__( self, - client: OpenAI, - probe_interval: float = 30.0, - probe_timeout: float = 5.0, failure_threshold: int = 5, - cooldown: float = 60.0, - *, - provider: str = "openai", - initial_model: str = "", - on_model_changed: Callable[[str, int | None], None] | None = None, on_state_changed: Callable[[str], None] | None = None, ) -> None: - self._client = client - self._probe_interval = probe_interval - self._probe_timeout = probe_timeout self._failure_threshold = failure_threshold - self._cooldown = cooldown - - # Model change detection - self._provider = provider - self._last_detected_model = initial_model - self._on_model_changed = on_model_changed self._on_state_changed = on_state_changed self._lock = threading.Lock() - self._state = CircuitState.CLOSED + self._degraded = False self._consecutive_failures = 0 - self._last_state_change = time.monotonic() - # Set True on OPEN→HALF_OPEN; consumed by first acquire_request_permit() call - self._half_open_permit = False - self._stop_event = threading.Event() - self._thread: threading.Thread | None = None - - # ------------------------------------------------------------------ - # Lifecycle - # ------------------------------------------------------------------ - - def start(self) -> None: - """Start background probe daemon thread.""" - self._thread = threading.Thread(target=self._probe_loop, daemon=True) - self._thread.start() - - def stop(self) -> None: - """Signal the probe thread to stop.""" - self._stop_event.set() - - # ------------------------------------------------------------------ - # Passive tracking (called by request path) - # ------------------------------------------------------------------ + # -- passive tracking ---------------------------------------------------- def _fire_state_callback(self, state_val: str | None) -> None: """Fire on_state_changed callback outside the lock.""" @@ -93,188 +58,121 @@ class BackendHealthMonitor: log.debug("on_state_changed callback error", exc_info=True) def record_success(self) -> None: - """Called on successful LLM call. Resets failure count, closes circuit.""" + """Called on successful LLM call. Clears degraded state.""" state_to_dispatch: str | None = None with self._lock: self._consecutive_failures = 0 - if self._state != CircuitState.CLOSED: - prev = self._state - self._state = CircuitState.CLOSED - self._half_open_permit = False - self._last_state_change = time.monotonic() - log.info("Circuit breaker CLOSED (was %s): backend recovered", prev.value) - self._update_metrics() - state_to_dispatch = self._state.value + if self._degraded: + self._degraded = False + log.info("Backend recovered (was degraded)") + state_to_dispatch = "healthy" self._fire_state_callback(state_to_dispatch) def record_failure(self) -> None: - """Called on LLM call failure. May open circuit.""" + """Called on LLM call failure. May mark backend as degraded.""" state_to_dispatch: str | None = None with self._lock: self._consecutive_failures += 1 - if self._state == CircuitState.HALF_OPEN: - # Probe failed in HALF_OPEN — re-open immediately - self._state = CircuitState.OPEN - self._half_open_permit = False - self._last_state_change = time.monotonic() - log.warning("Circuit breaker OPEN: probe failed in HALF_OPEN") - self._update_metrics() - state_to_dispatch = self._state.value - elif ( - self._state == CircuitState.CLOSED - and self._consecutive_failures >= self._failure_threshold - ): - self._state = CircuitState.OPEN - self._last_state_change = time.monotonic() + if not self._degraded and self._consecutive_failures >= self._failure_threshold: + self._degraded = True log.warning( - "Circuit breaker OPEN: %d consecutive failures", + "Backend degraded: %d consecutive failures", self._consecutive_failures, ) - self._update_metrics() - state_to_dispatch = self._state.value + state_to_dispatch = "degraded" self._fire_state_callback(state_to_dispatch) - # ------------------------------------------------------------------ - # Query helpers - # ------------------------------------------------------------------ + # -- query helpers ------------------------------------------------------- @property def is_healthy(self) -> bool: with self._lock: - return self._state == CircuitState.CLOSED + return not self._degraded @property - def circuit_state(self) -> CircuitState: + def is_degraded(self) -> bool: with self._lock: - return self._state + return self._degraded - def acquire_request_permit(self) -> bool: - """Consume one request permit if available. - - Returns True when the caller may proceed. In HALF_OPEN, only one probe - request is allowed — subsequent callers are blocked until the probe - completes (via ``record_success`` or ``record_failure``). - """ + @property + def consecutive_failures(self) -> int: with self._lock: - if self._state == CircuitState.OPEN: - if (time.monotonic() - self._last_state_change) >= self._cooldown: - self._state = CircuitState.HALF_OPEN - self._half_open_permit = False # consumed by this caller - self._last_state_change = time.monotonic() - log.info("Circuit breaker HALF_OPEN: cooldown elapsed, one probe permitted") - self._update_metrics() - return True # this caller is the probe - return False - if self._state == CircuitState.HALF_OPEN: - # Only one probe request allowed; subsequent callers block - if self._half_open_permit: - self._half_open_permit = False - return True - return False - return True # CLOSED + return self._consecutive_failures - # ------------------------------------------------------------------ - # Background probe - # ------------------------------------------------------------------ - def _probe_loop(self) -> None: - """Background: probe backend every interval. +# --------------------------------------------------------------------------- +# Per-backend health tracker registry +# --------------------------------------------------------------------------- - An initial jitter (derived from the PID) staggers probes across - cluster nodes so they don't all hit the LLM backend at once. + +class HealthTrackerRegistry: + """Manages per-backend health trackers keyed by ``(provider, base_url)``. + + Two model aliases that point at the same backend share a single + :class:`BackendHealthTracker`. Aliases on different backends get + independent trackers. + + Thread-safe. Trackers are created eagerly at startup (or on model + reload) — never lazily from the request path. + """ + + def __init__( + self, + failure_threshold: int = 5, + on_state_changed: Callable[[str, str], None] | None = None, + ) -> None: + self._failure_threshold = failure_threshold + # callback(backend_key_str, state_value) + self._on_state_changed = on_state_changed + self._trackers: dict[tuple[str, str], BackendHealthTracker] = {} + self._lock = threading.Lock() + + # -- key helpers --------------------------------------------------------- + + @staticmethod + def backend_key(provider: str, base_url: str) -> tuple[str, str]: + """Normalize a ``(provider, base_url)`` pair for use as a dict key.""" + return (provider, base_url.rstrip("/")) + + # -- tracker lifecycle --------------------------------------------------- + + def get_tracker( + self, + provider: str, + base_url: str, + ) -> BackendHealthTracker: + """Get or create a tracker for the given backend. Thread-safe.""" + key = self.backend_key(provider, base_url) + with self._lock: + if key not in self._trackers: + outer = self._on_state_changed + + def _state_cb(state: str, _k: tuple[str, str] = key) -> None: + if outer: + outer(f"{_k[0]}:{_k[1]}", state) + + tracker = BackendHealthTracker( + failure_threshold=self._failure_threshold, + on_state_changed=_state_cb, + ) + self._trackers[key] = tracker + log.info("Health tracker created for backend %s:%s", key[0], key[1]) + return self._trackers[key] + + def get_tracker_for_alias( + self, + registry: Any, + alias: str, + ) -> BackendHealthTracker | None: + """Look up the tracker for a model alias, if one exists. + + Returns ``None`` if the alias is unknown or no tracker has been + created for its backend yet. """ - import os - - # Deterministic per-process jitter: spread across half the interval - jitter = ((os.getpid() * 2654435761) & 0x7FFFFFFF) / 0x7FFFFFFF * (self._probe_interval / 2) - self._stop_event.wait(jitter) - while not self._stop_event.is_set(): - self._stop_event.wait(self._probe_interval) - if self._stop_event.is_set(): - break - # When circuit is OPEN, only probe after cooldown expires. - with self._lock: - if self._state == CircuitState.OPEN: - elapsed = time.monotonic() - self._last_state_change - remaining = self._cooldown - elapsed - if remaining > 0: - # Wait precisely for cooldown rather than skipping - # a full probe_interval (which could overshoot). - self._lock.release() - try: - self._stop_event.wait(remaining) - finally: - self._lock.acquire() - if self._stop_event.is_set(): - break - # Transition to HALF_OPEN for the probe. The background - # probe itself is the single HALF_OPEN request — keep - # _half_open_permit False so concurrent user requests - # are blocked until the probe completes. - self._state = CircuitState.HALF_OPEN - self._half_open_permit = False - self._last_state_change = time.monotonic() - log.info("Circuit breaker HALF_OPEN: cooldown elapsed, probing") - self._update_metrics() - success = self._probe_once() - if success: - self.record_success() - else: - self.record_failure() - - def _probe_once(self) -> bool: - """Single probe: call ``client.models.list()``. Returns True on success.""" try: - resp = self._client.with_options(timeout=self._probe_timeout).models.list() - self._check_model_change(resp) - return True - except Exception: - return False - - def _check_model_change(self, resp: Any) -> None: - """Compare detected model against last known and fire callback if changed.""" - if not self._on_model_changed or not resp.data: - return - try: - from turnstone.core.model_registry import ( - _extract_context_window, - _select_best_model, - ) - - all_ids = [m.id for m in resp.data] - selected = _select_best_model(all_ids, self._provider) - if selected == self._last_detected_model: - return - model_obj = next((m for m in resp.data if m.id == selected), None) - ctx = _extract_context_window(model_obj, self._provider) if model_obj else None - log.info( - "Backend model changed: %s -> %s (ctx=%s)", - self._last_detected_model, - selected, - ctx, - ) - self._last_detected_model = selected - self._on_model_changed(selected, ctx) - except Exception: - log.debug("Model change check failed", exc_info=True) - - # ------------------------------------------------------------------ - # Metrics - # ------------------------------------------------------------------ - - def _update_metrics(self) -> None: - """Push circuit-breaker state to metrics collector. - - Called with *self._lock* held. State-change callbacks are dispatched - by the callers (``record_success`` / ``record_failure``) after the - lock is released, not by this method. - """ - from turnstone.core.metrics import metrics - - metrics.set_backend_status(self._state == CircuitState.CLOSED) - state_int = { - CircuitState.CLOSED: 0, - CircuitState.OPEN: 1, - CircuitState.HALF_OPEN: 2, - } - metrics.set_circuit_state(state_int[self._state]) + cfg = registry.get_config(alias) + except (ValueError, KeyError): + return None + key = self.backend_key(cfg.provider, cfg.base_url) + with self._lock: + return self._trackers.get(key) diff --git a/turnstone/core/metrics.py b/turnstone/core/metrics.py index a38c1d76..53cf59c1 100644 --- a/turnstone/core/metrics.py +++ b/turnstone/core/metrics.py @@ -29,7 +29,6 @@ class MetricsCollector: self._context_ratio: float = 0.0 self._sse_connections: int = 0 # gauge: active SSE connections self._backend_up: bool = True # gauge: 1 if up, 0 if down - self._circuit_state: int = 0 # gauge: 0=closed, 1=open, 2=half_open # counters (continued) self._ratelimit_rejects: int = 0 # counter: total 429 responses self._evictions: int = 0 # counter: workstreams evicted @@ -101,11 +100,6 @@ class MetricsCollector: with self._lock: self._backend_up = up - def set_circuit_state(self, state: int) -> None: - """0=closed, 1=open, 2=half_open.""" - with self._lock: - self._circuit_state = state - def record_eviction(self) -> None: with self._lock: self._evictions += 1 @@ -172,7 +166,6 @@ class MetricsCollector: sse_connections = self._sse_connections ratelimit_rejects = self._ratelimit_rejects backend_up = self._backend_up - circuit_state = self._circuit_state evictions = self._evictions judge_verdicts = dict(self._judge_verdicts) judge_latency = dict(self._judge_latency) @@ -277,13 +270,6 @@ class MetricsCollector: 1 if backend_up else 0, ) - # turnstone_circuit_state - gauge( - "turnstone_circuit_state", - "Circuit breaker state (0=closed, 1=open, 2=half_open)", - circuit_state, - ) - # turnstone_workstreams_evicted_total counter( "turnstone_workstreams_evicted_total", diff --git a/turnstone/core/model_registry.py b/turnstone/core/model_registry.py index 4be3a99d..8ee9f5b9 100644 --- a/turnstone/core/model_registry.py +++ b/turnstone/core/model_registry.py @@ -212,9 +212,9 @@ def _resolve_openai_provider(provider: str, base_url: str) -> str: def load_model_registry( - base_url: str, - api_key: str, - model: str, + base_url: str = "", + api_key: str = "", + model: str = "", context_window: int = 32768, provider: str = "openai", storage: Any | None = None, @@ -296,8 +296,9 @@ def load_model_registry( ) # 3. Ensure a "default" entry from CLI args (only if not already defined - # by config.toml or DB — those take precedence) - if "default" not in configs: + # by config.toml or DB — those take precedence, and only when a CLI + # model was actually provided) + if "default" not in configs and model: configs["default"] = ModelConfig( alias="default", base_url=base_url, @@ -307,11 +308,24 @@ def load_model_registry( provider=_resolve_openai_provider(provider, base_url), ) + if not configs: + raise ValueError( + "No model definitions found. Provide --model, configure [models.*] " + "in config.toml, or add model definitions in the admin panel." + ) + # Determine default alias default_alias = model_section.get("default", "default") if default_alias not in configs: - log.warning("Configured default model '%s' not found, using 'default'", default_alias) - default_alias = "default" + if "default" in configs: + default_alias = "default" + else: + default_alias = next(iter(configs)) + log.info( + "No '%s' model alias; using '%s' as default", + model_section.get("default", "default"), + default_alias, + ) # Fallback chain fallback_raw = model_section.get("fallback", []) diff --git a/turnstone/core/session.py b/turnstone/core/session.py index d1179884..b45756d3 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -98,7 +98,7 @@ if TYPE_CHECKING: from collections.abc import Iterator from turnstone.core.config_store import ConfigStore - from turnstone.core.healthcheck import BackendHealthMonitor + from turnstone.core.healthcheck import BackendHealthTracker, HealthTrackerRegistry from turnstone.core.judge import IntentJudge, JudgeConfig from turnstone.core.mcp_client import MCPClientManager from turnstone.core.model_registry import ModelConfig, ModelRegistry @@ -288,7 +288,7 @@ class ChatSession: mcp_client: MCPClientManager | None = None, registry: ModelRegistry | None = None, model_alias: str | None = None, - health_monitor: BackendHealthMonitor | None = None, + health_registry: HealthTrackerRegistry | None = None, node_id: str | None = None, ws_id: str | None = None, tool_search: str = "auto", @@ -307,7 +307,7 @@ class ChatSession: self.model = model self._registry = registry self._model_alias = model_alias - self._health_monitor = health_monitor + self._health_registry = health_registry # Resolve provider for the current model self._provider: LLMProvider = ( registry.get_provider(model_alias) @@ -1401,43 +1401,90 @@ class ChatSession: _MAX_RETRIES = 3 _RETRY_BASE_DELAY = 1.0 # seconds + def _get_health_tracker(self) -> BackendHealthTracker | None: + """Get the health tracker for this session's current backend. + + Uses a read-only lookup — only returns trackers that were already + created eagerly at startup or during model reload. + + Returns ``None`` when no health registry is configured, the model + alias is unknown, or no tracker exists for this backend yet. + """ + if not self._health_registry or not self._registry or not self._model_alias: + return None + return self._health_registry.get_tracker_for_alias(self._registry, self._model_alias) + def _create_stream_with_retry(self, msgs: list[dict[str, Any]]) -> Iterator[StreamChunk]: """Create a streaming request with retry on transient errors. If all retries fail and a fallback chain is configured, tries each - fallback model in order before giving up. Checks the circuit breaker - before attempting a call — fast-fails when the backend is unreachable. + fallback model in order before giving up. Records success/failure + on the per-backend health tracker for observability. """ - # Circuit breaker check — fast-fail if backend is known to be down - if self._health_monitor and not self._health_monitor.acquire_request_permit(): - raise ConnectionError("Backend unreachable (circuit breaker open)") + tracker = self._get_health_tracker() try: result = self._try_stream(self.client, self.model, msgs) - if self._health_monitor: - self._health_monitor.record_success() + if tracker: + tracker.record_success() return result except Exception as primary_err: - if self._health_monitor: - self._health_monitor.record_failure() + if tracker: + tracker.record_failure() if not self._registry or not self._registry.fallback: raise - # Try each fallback model. Fallbacks may use different backends; - # we intentionally do NOT call record_success/failure for fallbacks — - # recovery of the primary backend is detected by the background probe. + # Try each fallback model. Prefer non-degraded backends first, + # but still try degraded ones as a last resort. + degraded_fallbacks: list[str] = [] for alias in self._registry.fallback: if alias == self._model_alias: continue - try: - fb_client, fb_model, _ = self._registry.resolve(alias) - fb_provider = self._registry.get_provider(alias) - self.ui.on_info(f"[Primary model failed, falling back to {alias}]") - return self._try_stream(fb_client, fb_model, msgs, provider=fb_provider) - except Exception as fb_err: - self.ui.on_info(f"[Fallback {alias} also failed: {fb_err}]") - continue + # Skip degraded backends on the first pass + if self._health_registry: + fb_tracker = self._health_registry.get_tracker_for_alias(self._registry, alias) + if fb_tracker and fb_tracker.is_degraded: + degraded_fallbacks.append(alias) + continue + stream = self._try_fallback(alias, msgs) + if stream is not None: + return stream + # Second pass: try degraded backends as last resort + for alias in degraded_fallbacks: + self.ui.on_info(f"[Fallback {alias} is degraded, trying anyway]") + stream = self._try_fallback(alias, msgs) + if stream is not None: + return stream raise primary_err + def _try_fallback(self, alias: str, msgs: list[dict[str, Any]]) -> Iterator[StreamChunk] | None: + """Attempt a single fallback model. Returns stream or None. + + Records success/failure on the fallback's health tracker so + the two-pass ordering (healthy-first, then degraded) learns + across request cycles. + + Caller must ensure ``self._registry`` is not ``None``. + """ + assert self._registry is not None + fb_tracker = ( + self._health_registry.get_tracker_for_alias(self._registry, alias) + if self._health_registry + else None + ) + try: + fb_client, fb_model, _ = self._registry.resolve(alias) + fb_provider = self._registry.get_provider(alias) + self.ui.on_info(f"[Primary model failed, falling back to {alias}]") + result = self._try_stream(fb_client, fb_model, msgs, provider=fb_provider) + if fb_tracker: + fb_tracker.record_success() + return result + except Exception as fb_err: + if fb_tracker: + fb_tracker.record_failure() + self.ui.on_info(f"[Fallback {alias} also failed: {fb_err}]") + return None + def _try_stream( self, client: Any, diff --git a/turnstone/core/settings_registry.py b/turnstone/core/settings_registry.py index 940b922d..96678198 100644 --- a/turnstone/core/settings_registry.py +++ b/turnstone/core/settings_registry.py @@ -43,6 +43,17 @@ def _build_registry() -> dict[str, SettingDef]: "model", help="Which AI model to use for conversations. Leave empty to use the provider's default.", ), + SettingDef( + "model.default_alias", + "str", + "", + "Default model alias for new sessions (empty = use config.toml [model].default)", + "model", + help="Which named model alias to use for new sessions. When empty, falls back to " + "the [model].default setting in config.toml (which defaults to 'default'). " + "Change this at runtime to switch all new sessions to a different model " + "without restarting.", + ), SettingDef( "model.temperature", "float", @@ -335,43 +346,15 @@ def _build_registry() -> dict[str, SettingDef]: ), # -- health --------------------------------------------------------- SettingDef( - "health.backend_probe_interval", - "int", - 30, - "Backend health probe interval in seconds", - "health", - min_value=5, - help="How often to check whether the AI model backend (e.g. OpenAI API) is reachable.", - ), - SettingDef( - "health.backend_probe_timeout", + "health.failure_threshold", "int", 5, - "Backend health probe timeout in seconds", + "Consecutive failures before backend is marked degraded", "health", min_value=1, - ), - SettingDef( - "health.circuit_breaker_threshold", - "int", - 5, - "Consecutive failures before circuit opens", - "health", - min_value=1, - help="If the AI backend fails this many times in a row, the circuit breaker trips " - "and stops sending requests for a cooldown period. This prevents cascading failures " - "and wasted API calls when the backend is down.", - reference_url="https://martinfowler.com/bliki/CircuitBreaker.html", - ), - SettingDef( - "health.circuit_breaker_cooldown", - "int", - 60, - "Seconds before half-open retry", - "health", - min_value=5, - help="After the circuit breaker trips, wait this long before sending a single test " - "request to see if the backend has recovered.", + help="If the AI backend fails this many times in a row, it is marked as degraded. " + "Degraded backends are deprioritised in the fallback chain but requests are never " + "blocked. The backend recovers automatically when a request succeeds.", ), # -- judge ---------------------------------------------------------- SettingDef( diff --git a/turnstone/sdk/events.py b/turnstone/sdk/events.py index 0dbffccd..6b5d11b2 100644 --- a/turnstone/sdk/events.py +++ b/turnstone/sdk/events.py @@ -313,10 +313,10 @@ class NodeSnapshotEvent(ClusterEvent): @dataclass class HealthChangedEvent(ClusterEvent): - """Circuit breaker state transition on a server node.""" + """Backend health state transition on a server node.""" type: str = "health_changed" - circuit_state: str = "" + backend_status: str = "" # "healthy" or "degraded" @dataclass diff --git a/turnstone/server.py b/turnstone/server.py index a506a71c..c8ac0268 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -1214,8 +1214,20 @@ def _build_health_dict(app_state: Any) -> dict[str, Any]: mgr: WorkstreamManager = app_state.workstreams wss = mgr.list_all() states = _count_ws_states(wss) - monitor = getattr(app_state, "health_monitor", None) - backend_ok = monitor.is_healthy if monitor else True + health_reg = getattr(app_state, "health_registry", None) + registry = getattr(app_state, "registry", None) + tracker = None + if health_reg and registry: + # Prefer ConfigStore runtime override, fall back to registry default + config_store = getattr(app_state, "config_store", None) + effective_alias = None + if config_store: + effective_alias = config_store.get("model.default_alias") or None + if effective_alias: + tracker = health_reg.get_tracker_for_alias(registry, effective_alias) + if tracker is None: + tracker = health_reg.get_tracker_for_alias(registry, registry.default) + backend_ok = tracker.is_healthy if tracker else True data: dict[str, Any] = { "status": "ok" if backend_ok else "degraded", "version": __version__, @@ -1226,7 +1238,6 @@ def _build_health_dict(app_state: Any) -> dict[str, Any]: "workstreams": {"total": len(wss), **states}, "backend": { "status": "up" if backend_ok else "down", - "circuit_state": monitor.circuit_state.value if monitor else "closed", }, } mc = getattr(app_state, "mcp_client", None) @@ -2149,10 +2160,18 @@ def internal_model_reload(request: Request) -> JSONResponse: provider=cli_args["provider"], storage=get_storage(), ) + # Allow runtime override of the default alias via ConfigStore + effective_default = new_registry.default + cs = getattr(request.app.state, "config_store", None) + if cs: + cs_alias = cs.get("model.default_alias") + if cs_alias and cs_alias in new_registry.models: + effective_default = cs_alias + try: registry.reload( new_registry.models, - new_registry.default, + effective_default, new_registry.fallback, new_registry.agent_model, ) @@ -2160,6 +2179,14 @@ def internal_model_reload(request: Request) -> JSONResponse: return JSONResponse({"status": "error", "reason": str(exc)}, status_code=422) finally: new_registry.shutdown() + + # Ensure health trackers exist for any newly-added backends + health_reg = getattr(request.app.state, "health_registry", None) + if health_reg: + for alias in registry.list_aliases(): + cfg = registry.get_config(alias) + health_reg.get_tracker(provider=cfg.provider, base_url=cfg.base_url) + return JSONResponse({"status": "ok", "aliases": registry.list_aliases()}) @@ -2211,13 +2238,51 @@ async def internal_migrate(request: Request) -> JSONResponse: # --------------------------------------------------------------------------- -def _emit_health_changed(circuit_state: str, gq: queue.Queue[dict[str, Any]]) -> None: +def _emit_health_changed( + status: str, gq: queue.Queue[dict[str, Any]], app_state: Any = None +) -> None: """Push a health_changed event onto the global SSE queue. - Called from the BackendHealthMonitor callback on circuit breaker transitions. + Called from the BackendHealthTracker callback on state transitions. + *status* is ``"healthy"`` or ``"degraded"``. + + Also updates the global ``turnstone_backend_up`` metric using the + effective default backend's health (not the backend that triggered + this callback, which may be a non-default fallback). """ + if app_state is not None: + _update_backend_metric(app_state) with contextlib.suppress(queue.Full): - gq.put_nowait({"type": "health_changed", "circuit_state": circuit_state}) + gq.put_nowait( + { + "type": "health_changed", + "backend_status": status, + } + ) + + +def _update_backend_metric(app_state: Any) -> None: + """Update ``turnstone_backend_up`` from the effective default's tracker. + + Called on any backend state change. Only the effective default + backend drives this global metric — fallback backend transitions + do not affect it. + """ + health_reg = getattr(app_state, "health_registry", None) + registry = getattr(app_state, "registry", None) + if not health_reg or not registry: + return + config_store = getattr(app_state, "config_store", None) + effective = None + if config_store: + effective = config_store.get("model.default_alias") or None + tracker = None + if effective: + tracker = health_reg.get_tracker_for_alias(registry, effective) + if tracker is None: + tracker = health_reg.get_tracker_for_alias(registry, registry.default) + if tracker is not None: + _metrics.set_backend_status(tracker.is_healthy) def _aggregate_emitter_thread( @@ -2420,8 +2485,7 @@ async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]: await tls_client.stop_renewal() if app.state.watch_runner: app.state.watch_runner.stop() - if app.state.health_monitor: - app.state.health_monitor.stop() + # health_registry is stateless (no background threads) — nothing to stop if app.state.mcp_client: app.state.mcp_client.shutdown() if app.state.registry: @@ -2462,7 +2526,7 @@ def create_app( skip_permissions: bool, jwt_secret: str = "", auth_storage: Any = None, - health_monitor: Any = None, + health_registry: Any = None, rate_limiter: Any = None, mcp_client: Any = None, mcp_ref: list[Any] | None = None, @@ -2542,7 +2606,7 @@ def create_app( app.state.skip_permissions = skip_permissions app.state.jwt_secret = jwt_secret app.state.auth_storage = auth_storage - app.state.health_monitor = health_monitor + app.state.health_registry = health_registry app.state.rate_limiter = rate_limiter app.state.mcp_client = mcp_client app.state.mcp_ref = mcp_ref @@ -2735,10 +2799,10 @@ def main() -> None: model, detected_ctx = detect_model(client, provider=provider_name, fatal=False) if model is None: - # LLM backend unreachable — start with a placeholder model name. - # The health monitor will report degraded and the circuit breaker - # will prevent requests until the backend comes up. - model = "unavailable" + # LLM backend unreachable — no CLI model specified. + # Set empty so load_model_registry skips the CLI "default" + # entry and relies on DB / config.toml models instead. + model = "" # Use detected context window, fall back to ConfigStore override or 32768 cfg_ctx = config_store.get("model.context_window") @@ -2763,6 +2827,11 @@ def main() -> None: storage=_get_storage(), ) + # Apply runtime default alias override from ConfigStore (if set) + cs_default_alias = config_store.get("model.default_alias") + if cs_default_alias and registry.has_alias(cs_default_alias): + registry.reload(registry.models, cs_default_alias, registry.fallback, registry.agent_model) + # Initialize MCP client (connects to configured MCP servers, if any) from turnstone.core.mcp_client import create_mcp_client @@ -2776,56 +2845,33 @@ def main() -> None: # including ones created by internal_mcp_reload after startup. _mcp_ref: list[Any] = [mcp_client] - # Backend health monitor with circuit breaker - from turnstone.core.healthcheck import BackendHealthMonitor - - def _handle_model_change(new_model_id: str, new_ctx: int | None) -> None: - """Called from health probe thread when backend model changes.""" - cli_args = getattr(getattr(app, "state", None), "cli_model_args", None) - if not cli_args or cli_args.get("_user_specified_model"): - return - old_model = cli_args["model"] - ctx = new_ctx or cli_args["context_window"] - log.info("Backend model changed: %s -> %s (ctx=%s)", old_model, new_model_id, ctx) - new_reg = None - try: - new_reg = load_model_registry( - base_url=cli_args["base_url"], - api_key=cli_args["api_key"], - model=new_model_id, - context_window=ctx, - provider=cli_args["provider"], - storage=get_storage(), - ) - registry.reload(new_reg.models, new_reg.default, new_reg.fallback, new_reg.agent_model) - # Update cli_model_args only after successful reload - cli_args["model"] = new_model_id - cli_args["context_window"] = ctx - except Exception: - log.warning("Model change reload failed", exc_info=True) - finally: - if new_reg is not None: - new_reg.shutdown() + # Per-backend passive health tracking (no active probes / circuit breakers) + from turnstone.core.healthcheck import HealthTrackerRegistry # Set up global event queue for state-change broadcasts (created early so - # the health monitor callback can reference it). + # the health tracker callback can reference it). global_queue: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=10000) global_listeners: list[queue.Queue[dict[str, Any]]] = [] global_listeners_lock = threading.Lock() WebUI._global_queue = global_queue - health_monitor = BackendHealthMonitor( - client=client, - probe_interval=config_store.get("health.backend_probe_interval"), - probe_timeout=config_store.get("health.backend_probe_timeout"), - failure_threshold=config_store.get("health.circuit_breaker_threshold"), - cooldown=config_store.get("health.circuit_breaker_cooldown"), - provider=provider_name, - initial_model=model, - on_model_changed=_handle_model_change, - on_state_changed=lambda state: _emit_health_changed(state, global_queue), + # Mutable ref so the health callback can access app.state after app + # creation (same pattern as _mcp_ref). + _app_ref: list[Any] = [None] + + health_registry = HealthTrackerRegistry( + failure_threshold=config_store.get("health.failure_threshold"), + on_state_changed=lambda _backend, state: _emit_health_changed( + state, global_queue, _app_ref[0].state if _app_ref[0] else None + ), ) - health_monitor.start() + + # Eagerly create trackers for all registered backends. Sessions use + # read-only lookups (get_tracker_for_alias) and never create trackers + # on the hot path, so every backend must be registered here. + for _alias in registry.list_aliases(): + _cfg = registry.get_config(_alias) + health_registry.get_tracker(provider=_cfg.provider, base_url=_cfg.base_url) # Per-IP rate limiter from turnstone.core.ratelimit import RateLimiter @@ -2875,6 +2921,17 @@ def main() -> None: ) # Session factory — captures shared config (including config_store for hot-reload) + def _effective_default_alias() -> str: + """Return the runtime-effective default model alias. + + Checks ConfigStore for a ``model.default_alias`` override first, + then falls back to the registry's static default. + """ + cs_alias: str = config_store.get("model.default_alias") + if cs_alias and registry.has_alias(cs_alias): + return cs_alias + return registry.default + def session_factory( ui: SessionUI | None, model_alias: str | None = None, @@ -2884,6 +2941,9 @@ def main() -> None: client_type: str = "", ) -> ChatSession: assert ui is not None + # Resolve the effective alias once and use it consistently + # for both client resolution and ChatSession.model_alias. + model_alias = model_alias or _effective_default_alias() r_client, r_model, r_cfg = registry.resolve(model_alias) # Read MCP client from shared ref — may have been replaced after startup # by internal_mcp_reload (Sync to Nodes) when no --mcp-config was passed. @@ -2924,8 +2984,8 @@ def main() -> None: tool_truncation=config_store.get("tools.truncation"), mcp_client=live_mcp_client, registry=registry, - model_alias=model_alias or registry.default, - health_monitor=health_monitor, + model_alias=model_alias, + health_registry=health_registry, node_id=_node_id, ws_id=ws_id, tool_search=config_store.get("tools.search"), @@ -3048,7 +3108,7 @@ def main() -> None: skip_permissions=_skip_perms, jwt_secret=jwt_secret, auth_storage=get_storage(), - health_monitor=health_monitor, + health_registry=health_registry, rate_limiter=rate_limiter, mcp_client=mcp_client, mcp_ref=_mcp_ref, @@ -3062,6 +3122,9 @@ def main() -> None: advertise_url=_advertise_url, ) + # Wire app ref so health callbacks can access app.state for metrics + _app_ref[0] = app + # Store CLI model args for hot-reload (internal_model_reload reads these) app.state.cli_model_args = { "base_url": base_url, @@ -3083,9 +3146,8 @@ def main() -> None: log.info("MCP tools: %d from %d server(s)", len(mcp_tools), mcp_client.server_count) mcp_client.set_storage(get_storage()) log.info( - "Health monitor: probe every %ss, circuit breaker threshold=%s", - config_store.get("health.backend_probe_interval"), - config_store.get("health.circuit_breaker_threshold"), + "Health tracking: failure_threshold=%s", + config_store.get("health.failure_threshold"), ) if rate_limiter.enabled: log.info( diff --git a/turnstone/ui/static/app.js b/turnstone/ui/static/app.js index 35753cad..f77ed484 100644 --- a/turnstone/ui/static/app.js +++ b/turnstone/ui/static/app.js @@ -2152,15 +2152,14 @@ function pollHealth() { var el = document.getElementById("health-indicator"); if (!el) return; if (data.status === "degraded") { - el.textContent = "backend down"; + el.textContent = "backend degraded"; el.className = "health-degraded"; el.title = - "Circuit: " + - ((data.backend && data.backend.circuit_state) || "unknown"); + "Backend: " + ((data.backend && data.backend.status) || "unknown"); el.setAttribute( "aria-label", - "Backend degraded. Circuit: " + - ((data.backend && data.backend.circuit_state) || "unknown"), + "Backend degraded: " + + ((data.backend && data.backend.status) || "unknown"), ); } else { el.textContent = ""; @@ -2825,12 +2824,7 @@ function updateDashFooter(agg) { parts.push(formatUptime(agg.uptime_seconds) + " uptime"); statsEl.textContent = parts.join(" \u00b7 "); if (_lastHealth && _lastHealth.status === "degraded") { - statsEl.textContent += - " \u00b7 backend down (circuit " + - (_lastHealth.backend && _lastHealth.backend.circuit_state - ? _lastHealth.backend.circuit_state - : "unknown") + - ")"; + statsEl.textContent += " \u00b7 backend degraded"; } }