diff --git a/solstone/think/providers/local_budget.py b/solstone/think/providers/local_budget.py index 94a2876c9..6fb2e54d7 100644 --- a/solstone/think/providers/local_budget.py +++ b/solstone/think/providers/local_budget.py @@ -16,6 +16,9 @@ from solstone.think.providers.local import ContextBudgetExceeded LOG = logging.getLogger(__name__) _SAFETY_MARGIN_TOKENS = 256 +# Conservative per-image upper estimate; images inform endpoint completion +# clamps but are never dropped or truncated by the fitting helper. +_ESTIMATED_IMAGE_TOKENS = 2500 _OUTPUT_RESERVE_DIVISOR = 4 _CHARS_PER_TOKEN = 3 _TOKENIZE_TIMEOUT_S = 5.0 @@ -100,8 +103,9 @@ def fit_contents( max_output_tokens: int, *, count: Callable[[str], int], + window: int | None = None, ) -> tuple[Any, dict | None]: - budget = compute_input_budget(max_output_tokens) + budget = compute_input_budget(max_output_tokens, window) if isinstance(contents, str): block = contents diff --git a/solstone/think/providers/local_endpoint.py b/solstone/think/providers/local_endpoint.py index cde4c9234..477a9bf79 100644 --- a/solstone/think/providers/local_endpoint.py +++ b/solstone/think/providers/local_endpoint.py @@ -5,12 +5,15 @@ from __future__ import annotations +import json import logging +import time from collections.abc import Callable, Iterator from dataclasses import dataclass from typing import Any from solstone.think.journal_config import read_journal_config +from solstone.think.providers.shared import _CONTEXT_WINDOW_PATTERNS, _contains_any LOG = logging.getLogger(__name__) @@ -28,6 +31,13 @@ _REASON_COPY_BY_CODE = { "local_endpoint_contract_failed": LOCAL_ENDPOINT_CONTRACT_COPY, } _DEFAULT_BYO_PARALLEL_SLOTS = 2 +ENDPOINT_SERVED_CONTEXT_WINDOW_CONFIG_KEY = "served_context_window" +ENDPOINT_SERVED_CONTEXT_WINDOW_MIN_TOKENS = 2048 +ENDPOINT_SERVED_WINDOW_CACHE_TTL_S = 300.0 +ENDPOINT_MODELS_TIMEOUT_S = 2.5 +ENDPOINT_ERROR_BODY_CAP_CHARS = 4096 + +_SERVED_WINDOW_CACHE: dict[tuple[str, str], tuple[float, int | None]] = {} @dataclass(frozen=True) @@ -37,7 +47,9 @@ class LocalEndpoint: Bundled endpoints always carry ``parallel_slots=None``; bundled capacity is live server state and this field is inert. Confidential BYO endpoints also carry ``None`` and are ungoverned. Non-confidential BYO endpoints carry the - resolved ``int >= 1`` client-side slot count. + resolved ``int >= 1`` client-side slot count. ``is_confidential`` marks BYO + endpoints routed through the confidential forwarder; bundled endpoints are + not confidential endpoints even when confidential service config is present. """ base_url: str @@ -45,6 +57,7 @@ class LocalEndpoint: credential: str | None is_bundled: bool parallel_slots: int | None = None + is_confidential: bool = False def normalize_local_endpoint_url(raw_url: str) -> str: @@ -105,20 +118,44 @@ def _configured_byo_parallel_slots(local_config: dict[str, Any]) -> int: return raw -def resolve_local_endpoint_from_config(config: dict[str, Any]) -> LocalEndpoint: - """Resolve local provider traffic from an already-read journal config.""" - +def _local_provider_config(config: dict[str, Any]) -> dict[str, Any]: providers_config = config.get("providers", {}) local_config: Any = {} if isinstance(providers_config, dict): local_config = providers_config.get("local", {}) - if not isinstance(local_config, dict): - local_config = {} + return local_config if isinstance(local_config, dict) else {} + + +def _configured_served_context_window(local_config: dict[str, Any]) -> int | None: + if ENDPOINT_SERVED_CONTEXT_WINDOW_CONFIG_KEY not in local_config: + return None + + raw = local_config.get(ENDPOINT_SERVED_CONTEXT_WINDOW_CONFIG_KEY) + if ( + not isinstance(raw, int) + or isinstance(raw, bool) + or raw < ENDPOINT_SERVED_CONTEXT_WINDOW_MIN_TOKENS + ): + LOG.warning( + "Invalid providers.local.%s in journal config: %r - falling through " + "to endpoint discovery", + ENDPOINT_SERVED_CONTEXT_WINDOW_CONFIG_KEY, + raw, + ) + return None + return raw + + +def resolve_local_endpoint_from_config(config: dict[str, Any]) -> LocalEndpoint: + """Resolve local provider traffic from an already-read journal config.""" + + local_config = _local_provider_config(config) endpoint_url = str(local_config.get("endpoint_url") or "").strip() served_model_id = str(local_config.get("served_model_id") or "").strip() if endpoint_url and served_model_id: credential = local_config.get("credential") or None + is_confidential = confidential_provenance_block(config) is not None return LocalEndpoint( base_url=normalize_local_endpoint_url(endpoint_url), served_model_id=served_model_id, @@ -126,11 +163,12 @@ def resolve_local_endpoint_from_config(config: dict[str, Any]) -> LocalEndpoint: is_bundled=False, parallel_slots=( None - if confidential_provenance_block(config) is not None + if is_confidential else _configured_byo_parallel_slots(local_config) ), + is_confidential=is_confidential, ) - return LocalEndpoint("", "", None, is_bundled=True) + return LocalEndpoint("", "", None, is_bundled=True, is_confidential=False) def resolve_local_endpoint() -> LocalEndpoint: @@ -139,6 +177,80 @@ def resolve_local_endpoint() -> LocalEndpoint: return resolve_local_endpoint_from_config(read_journal_config()) +def reset_endpoint_served_window_cache() -> None: + """Clear the process-local endpoint served-window discovery cache.""" + + _SERVED_WINDOW_CACHE.clear() + + +def resolve_endpoint_served_window(endpoint: LocalEndpoint) -> int | None: + """Resolve the endpoint's served context window without mutating journal state.""" + + if endpoint.is_bundled: + return None + + local_config = _local_provider_config(read_journal_config()) + override = _configured_served_context_window(local_config) + if override is not None: + return override + + now = time.monotonic() + cache_key = (endpoint.base_url, endpoint.served_model_id) + cached = _SERVED_WINDOW_CACHE.get(cache_key) + if cached is not None: + cached_at, value = cached + if now - cached_at < ENDPOINT_SERVED_WINDOW_CACHE_TTL_S: + return value + + value = _discover_endpoint_served_window(endpoint) + _SERVED_WINDOW_CACHE[cache_key] = (now, value) + if endpoint.is_confidential and value is None: + LOG.warning( + "Could not resolve served context window for confidential local " + "endpoint %s", + endpoint.served_model_id, + ) + return value + + +def _discover_endpoint_served_window(endpoint: LocalEndpoint) -> int | None: + import httpx + + from solstone.think.services.spp_transport import confidential_egress_base_url + + try: + base_url = ( + confidential_egress_base_url(endpoint.base_url) + if endpoint.is_confidential + else endpoint.base_url + ) + kwargs: dict[str, Any] = {"timeout": ENDPOINT_MODELS_TIMEOUT_S} + if endpoint.credential: + kwargs["headers"] = {"Authorization": f"Bearer {endpoint.credential}"} + response = httpx.get(f"{base_url}/v1/models", **kwargs) + response.raise_for_status() + data = response.json() + except Exception: + LOG.debug( + "Could not discover local endpoint served context window", exc_info=True + ) + return None + + if not isinstance(data, dict): + return None + models = data.get("data") + if not isinstance(models, list): + return None + for item in models: + if not isinstance(item, dict) or item.get("id") != endpoint.served_model_id: + continue + max_model_len = item.get("max_model_len") + if isinstance(max_model_len, int) and not isinstance(max_model_len, bool): + return max_model_len + return None + return None + + def probe_local_endpoint( endpoint: LocalEndpoint, timeout_s: float = 1.0, @@ -218,6 +330,68 @@ def is_byo_capacity_error(exc: BaseException) -> bool: return bool(_BYO_CAPACITY_EXC_NAMES & names) +def _payload_text(payload: Any, credential: str | None) -> str | None: + if payload is None: + return None + redacted = redact_event_payload(payload, credential) + if isinstance(redacted, str): + text = redacted + elif isinstance(redacted, dict | list | tuple): + try: + text = json.dumps(redacted, default=str) + except Exception: + text = str(redacted) + else: + text = str(redacted) + if not text: + return None + return text[:ENDPOINT_ERROR_BODY_CAP_CHARS] + + +def _candidate_exception_texts( + item: BaseException, + credential: str | None, +) -> Iterator[str]: + body = _payload_text(getattr(item, "body", None), credential) + if body: + yield body + + response = getattr(item, "response", None) + response_text = _payload_text(getattr(response, "text", None), credential) + if response_text: + yield response_text + response_json = getattr(response, "json", None) + if callable(response_json): + try: + json_text = _payload_text(response_json(), credential) + except Exception: + json_text = None + if json_text: + yield json_text + + message = _payload_text(getattr(item, "message", None), credential) + if message: + yield message + + item_text = _payload_text(str(item), credential) + if item_text: + yield item_text + + +def byo_exception_matches_context_window( + exc: BaseException, + *, + credential: str | None = None, +) -> bool: + """True when a BYO exception chain carries a context-window-shaped body.""" + + for item in _exception_chain(exc): + for text in _candidate_exception_texts(item, credential): + if _contains_any(text.lower(), _CONTEXT_WINDOW_PATTERNS): + return True + return False + + def classify_byo_cogitate_error(exc: BaseException) -> str | None: """Return a BYO local-endpoint reason code for known OpenHands/LiteLLM errors.""" @@ -225,6 +399,8 @@ def classify_byo_cogitate_error(exc: BaseException) -> str | None: names = {type(item).__name__ for item in chain} statuses = {_status_code(item) for item in chain} + if byo_exception_matches_context_window(exc): + return "context_window_exceeded" if 400 in statuses or "BadRequestError" in names: return "local_endpoint_contract_failed" if is_byo_network_error(exc) or 500 in statuses or "InternalServerError" in names: @@ -345,9 +521,15 @@ def wrap_on_event_redacting( __all__ = [ + "ENDPOINT_ERROR_BODY_CAP_CHARS", + "ENDPOINT_MODELS_TIMEOUT_S", + "ENDPOINT_SERVED_CONTEXT_WINDOW_CONFIG_KEY", + "ENDPOINT_SERVED_CONTEXT_WINDOW_MIN_TOKENS", + "ENDPOINT_SERVED_WINDOW_CACHE_TTL_S", "LOCAL_ENDPOINT_CONTRACT_COPY", "LOCAL_ENDPOINT_UNREACHABLE_COPY", "LocalEndpoint", + "byo_exception_matches_context_window", "classify_byo_cogitate_error", "confidential_fingerprint_provenance_block", "confidential_provenance_block", @@ -359,6 +541,8 @@ __all__ = [ "redact_exception_credential", "redact_event_payload", "redact_local_endpoint_credential", + "reset_endpoint_served_window_cache", + "resolve_endpoint_served_window", "resolve_local_endpoint", "resolve_local_endpoint_from_config", "wrap_on_event_redacting", diff --git a/solstone/think/providers/shared.py b/solstone/think/providers/shared.py index 803682740..79912feef 100644 --- a/solstone/think/providers/shared.py +++ b/solstone/think/providers/shared.py @@ -164,6 +164,7 @@ _CONTEXT_WINDOW_PATTERNS = ( "context size has been exceeded", "exceeds the context window", "maximum context length", + "longer than the model's context length", "context length exceeded", ) @@ -418,6 +419,7 @@ class GenerateResult(TypedDict, total=False): input_budget: Optional[ dict ] # Out-of-band truncation metadata when the bundled-local input was clipped + endpoint_budget: Optional[dict] # Endpoint served-window clamp facts when known inference: Optional[dict] # Content-free local inference timing/admission record diff --git a/tests/test_local_endpoint.py b/tests/test_local_endpoint.py index f6337b970..91d400bc9 100644 --- a/tests/test_local_endpoint.py +++ b/tests/test_local_endpoint.py @@ -59,6 +59,7 @@ def test_resolve_local_endpoint_active_requires_url_and_model( assert endpoint.base_url == "http://h:8080" assert endpoint.served_model_id == "model" assert endpoint.credential is None + assert endpoint.is_confidential is False def test_resolve_local_endpoint_carries_placeholder_credential(monkeypatch): @@ -79,6 +80,7 @@ def test_resolve_local_endpoint_carries_placeholder_credential(monkeypatch): assert endpoint.is_bundled is False assert endpoint.credential == "test-token-PLACEHOLDER" assert endpoint.parallel_slots == 2 + assert endpoint.is_confidential is False @pytest.mark.parametrize( @@ -149,6 +151,7 @@ def test_resolve_local_endpoint_bundled_ignores_stray_parallel_slots( assert endpoint.is_bundled is True assert endpoint.parallel_slots is None + assert endpoint.is_confidential is False assert "providers.local.parallel_slots" not in caplog.text @@ -175,6 +178,7 @@ def test_resolve_local_endpoint_confidential_ignores_stray_parallel_slots( assert endpoint.is_bundled is False assert endpoint.parallel_slots is None + assert endpoint.is_confidential is True assert "providers.local.parallel_slots" not in caplog.text @@ -328,6 +332,18 @@ def test_classify_byo_cogitate_error_contract_by_status_or_name(): ) +def test_byo_exception_matches_context_window_caps_body_excerpt(): + exc = BadRequestError("bad request") + exc.body = ( + "x" * (local_endpoint.ENDPOINT_ERROR_BODY_CAP_CHARS + 1) + + "longer than the model's context length" + ) + assert local_endpoint.byo_exception_matches_context_window(exc) is False + + exc.body = "longer than the model's context length" + assert local_endpoint.byo_exception_matches_context_window(exc) is True + + @pytest.mark.parametrize( "inner", [ diff --git a/tests/test_local_endpoint_window.py b/tests/test_local_endpoint_window.py new file mode 100644 index 000000000..f93f2401e --- /dev/null +++ b/tests/test_local_endpoint_window.py @@ -0,0 +1,222 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import itertools + +import httpx +import pytest + +from solstone.think.providers import local_endpoint +from tests.test_local import _FAKE_MODELS_BODY + + +def _endpoint( + *, + model: str = "Qwen/Qwen3.5-4B", + credential: str | None = None, + confidential: bool = False, +) -> local_endpoint.LocalEndpoint: + return local_endpoint.LocalEndpoint( + base_url="http://byo.example/openai", + served_model_id=model, + credential=credential, + is_bundled=False, + is_confidential=confidential, + ) + + +def _config(local_config: dict | None = None) -> dict: + return {"providers": {"local": local_config or {}}} + + +def _response(url: str, status_code: int, text: str) -> httpx.Response: + return httpx.Response( + status_code, + request=httpx.Request("GET", url), + text=text, + ) + + +@pytest.fixture(autouse=True) +def _reset_cache(): + local_endpoint.reset_endpoint_served_window_cache() + yield + local_endpoint.reset_endpoint_served_window_cache() + + +def test_resolve_endpoint_served_window_discovers_max_model_len(monkeypatch): + calls = [] + monkeypatch.setattr(local_endpoint, "read_journal_config", lambda: _config()) + + def fake_get(url, **kwargs): + calls.append({"url": url, **kwargs}) + return _response(url, 200, _FAKE_MODELS_BODY) + + monkeypatch.setattr(httpx, "get", fake_get) + + assert ( + local_endpoint.resolve_endpoint_served_window( + _endpoint(credential="test-token-PLACEHOLDER") + ) + == 16384 + ) + assert calls == [ + { + "url": "http://byo.example/openai/v1/models", + "timeout": local_endpoint.ENDPOINT_MODELS_TIMEOUT_S, + "headers": {"Authorization": "Bearer test-token-PLACEHOLDER"}, + } + ] + + +@pytest.mark.parametrize( + "body", + [ + '{"object":"list","data":[{"id":"Qwen/Qwen3.5-4B"}]}', + '{"object":"list","data":[{"id":"Qwen/Qwen3.5-4B","max_model_len":"16384"}]}', + '{"object":"list","data":[{"id":"other-model","max_model_len":16384}]}', + ], +) +def test_resolve_endpoint_served_window_returns_none_for_bad_model_shape( + monkeypatch, + body, +): + monkeypatch.setattr(local_endpoint, "read_journal_config", lambda: _config()) + monkeypatch.setattr(httpx, "get", lambda url, **_kwargs: _response(url, 200, body)) + + assert local_endpoint.resolve_endpoint_served_window(_endpoint()) is None + + +def test_resolve_endpoint_served_window_override_wins_without_network(monkeypatch): + monkeypatch.setattr( + local_endpoint, + "read_journal_config", + lambda: _config({"served_context_window": 32768}), + ) + + def fail_get(*_args, **_kwargs): + raise AssertionError("override should skip discovery") + + monkeypatch.setattr(httpx, "get", fail_get) + + assert local_endpoint.resolve_endpoint_served_window(_endpoint()) == 32768 + + +@pytest.mark.parametrize("raw", [True, False, 2047, 1.5, "16384", [], {}]) +def test_resolve_endpoint_served_window_invalid_override_warns_and_discovers( + monkeypatch, + caplog, + raw, +): + monkeypatch.setattr( + local_endpoint, + "read_journal_config", + lambda: _config({"served_context_window": raw}), + ) + monkeypatch.setattr( + httpx, + "get", + lambda url, **_kwargs: _response(url, 200, _FAKE_MODELS_BODY), + ) + + assert local_endpoint.resolve_endpoint_served_window(_endpoint()) == 16384 + assert ( + f"Invalid providers.local.served_context_window in journal config: {raw!r} - " + "falling through to endpoint discovery" + ) in caplog.text + + +def test_resolve_endpoint_served_window_confidential_none_warns(monkeypatch, caplog): + calls = [] + monkeypatch.setattr(local_endpoint, "read_journal_config", lambda: _config()) + monkeypatch.setattr( + "solstone.think.services.spp_transport.confidential_egress_base_url", + lambda base_url: ( + "http://127.0.0.1:4567" + if base_url == "http://byo.example/openai" + else base_url + ), + ) + + def fake_get(url, **kwargs): + calls.append({"url": url, **kwargs}) + return _response( + url, + 200, + '{"object":"list","data":[{"id":"other","max_model_len":16384}]}', + ) + + monkeypatch.setattr(httpx, "get", fake_get) + + assert ( + local_endpoint.resolve_endpoint_served_window(_endpoint(confidential=True)) + is None + ) + assert calls[0]["url"] == "http://127.0.0.1:4567/v1/models" + assert ( + "Could not resolve served context window for confidential local endpoint" + in (caplog.text) + ) + + +def test_resolve_endpoint_served_window_cache_ttl_and_override(monkeypatch): + bodies = iter( + [ + _FAKE_MODELS_BODY, + ( + '{"object":"list","data":[{"id":"Qwen/Qwen3.5-4B",' + '"max_model_len":32768}]}' + ), + ] + ) + calls = [] + current_config = {"value": _config()} + times = itertools.chain( + [ + 100.0, + 101.0, + 100.0 + local_endpoint.ENDPOINT_SERVED_WINDOW_CACHE_TTL_S + 1.0, + ], + itertools.repeat(999.0), + ) + monkeypatch.setattr( + local_endpoint, + "read_journal_config", + lambda: current_config["value"], + ) + monkeypatch.setattr(local_endpoint.time, "monotonic", lambda: next(times)) + + def fake_get(url, **kwargs): + calls.append({"url": url, **kwargs}) + return _response(url, 200, next(bodies)) + + monkeypatch.setattr(httpx, "get", fake_get) + endpoint = _endpoint() + + assert local_endpoint.resolve_endpoint_served_window(endpoint) == 16384 + assert local_endpoint.resolve_endpoint_served_window(endpoint) == 16384 + current_config["value"] = _config({"served_context_window": 24576}) + assert local_endpoint.resolve_endpoint_served_window(endpoint) == 24576 + current_config["value"] = _config() + assert local_endpoint.resolve_endpoint_served_window(endpoint) == 32768 + assert len(calls) == 2 + + +@pytest.mark.parametrize( + "fake_get", + [ + lambda url, **_kwargs: (_ for _ in ()).throw(httpx.ConnectError("down")), + lambda url, **_kwargs: (_ for _ in ()).throw(httpx.ReadTimeout("slow")), + lambda url, **_kwargs: _response(url, 500, "server error"), + lambda url, **_kwargs: _response(url, 200, "{"), + lambda url, **_kwargs: _response(url, 200, '{"object":"list"}'), + lambda url, **_kwargs: _response(url, 200, '{"data":{}}'), + ], +) +def test_resolve_endpoint_served_window_failures_return_none(monkeypatch, fake_get): + monkeypatch.setattr(local_endpoint, "read_journal_config", lambda: _config()) + monkeypatch.setattr(httpx, "get", fake_get) + + assert local_endpoint.resolve_endpoint_served_window(_endpoint()) is None diff --git a/tests/test_provider_error_classification.py b/tests/test_provider_error_classification.py index 695a41d06..c20ddb61e 100644 --- a/tests/test_provider_error_classification.py +++ b/tests/test_provider_error_classification.py @@ -1,6 +1,8 @@ # SPDX-License-Identifier: AGPL-3.0-only # Copyright (c) 2026 sol pbc +from pathlib import Path + import pytest from solstone.think.providers import openhands @@ -75,6 +77,22 @@ def test_classifies_litellm_bad_request_unrelated_unknown(): assert classify_provider_error(exc, "local") != "context_window_exceeded" +def test_context_reason_codes_are_registered_with_existing_owner_copy(): + from solstone.convey import provider_readiness + from solstone.think.providers.shared import RUNTIME_REASON_CODES + + projection = provider_readiness.chat_reason_projection() + chat_reasons = Path("solstone/convey/static/chat_reasons.js").read_text( + encoding="utf-8" + ) + + for reason_code in ("context_window_exceeded", "context_budget_exceeded"): + assert reason_code in RUNTIME_REASON_CODES + assert reason_code in provider_readiness.mapped_reason_codes() + assert reason_code in projection + assert f'"{reason_code}"' in chat_reasons + + def test_classifies_max_turns_exhausted(): from solstone.think.cogitate_policy import MaxTurnsExhausted