# SPDX-License-Identifier: AGPL-3.0-only
# Copyright (c) 2026 sol pbc

from __future__ import annotations

import asyncio
import base64
import copy
import importlib
import json
import logging
import sys
import traceback
import types
from pathlib import Path
from types import SimpleNamespace

import pytest

from solstone.think.models import (
    DEFAULT_MODEL_BY_PROVIDER,
    LOCAL_MODEL,
    get_model_provider,
)
from solstone.think.providers.artifact_proof import ReadinessOutcome
from solstone.think.responsiveness import NON_RESPONSIVE_REASON_CODE
from solstone.think.schema_prep import SCHEMA_TRUNCATE_KEY
from solstone.think.talents import TalentHookError


@pytest.fixture(autouse=True)
def _isolate_local_admission(monkeypatch, tmp_path):
    from solstone.think.providers import local_admission

    monkeypatch.setattr(
        local_admission,
        "_admission_dir",
        lambda: tmp_path / "local-inference-admission",
    )
    monkeypatch.setattr(local_admission, "record_local_inference", lambda _record: None)


@pytest.fixture(autouse=True)
def _default_endpoint_models_unknown(monkeypatch):
    import httpx

    from solstone.think.providers import local_endpoint

    local_endpoint.reset_endpoint_served_window_cache()

    def fake_get(url, **_kwargs):
        return httpx.Response(
            404,
            request=httpx.Request("GET", url),
            text="not found",
        )

    monkeypatch.setattr(httpx, "get", fake_get)
    yield
    local_endpoint.reset_endpoint_served_window_cache()


def _provider():
    providers_pkg = importlib.import_module("solstone.think.providers")
    if hasattr(providers_pkg, "local_budget"):
        delattr(providers_pkg, "local_budget")
    sys.modules.pop("solstone.think.providers.local_budget", None)
    return importlib.reload(importlib.import_module("solstone.think.providers.local"))


def _schema_keyword_paths(schema, keywords):
    found = []

    def walk(node, path="$"):
        if isinstance(node, dict):
            for key, value in node.items():
                child_path = f"{path}/{key}"
                if key in keywords:
                    found.append(child_path)
                walk(value, child_path)
        elif isinstance(node, list):
            for index, item in enumerate(node):
                walk(item, f"{path}[{index}]")

    walk(schema)
    return found


def _local_response(finish_reason):
    return {
        "choices": [
            {
                "message": {"content": "ok"},
                "finish_reason": finish_reason,
            }
        ]
    }


def _request_json_payload(url, body):
    import httpx

    request = httpx.Request("POST", url, json=body)
    decoded = request.content.decode("utf-8")
    return json.loads(decoded), decoded, request.content


def _has_surrogate_codepoint(text: str) -> bool:
    return any(0xD800 <= ord(char) <= 0xDFFF for char in text)


def _assert_no_surrogate_codepoint(text: str) -> None:
    assert not _has_surrogate_codepoint(text)


class _ChatResponse:
    def __init__(self, text: str = "hello") -> None:
        self.text = text

    def raise_for_status(self):
        return None

    def json(self):
        return {
            "choices": [
                {
                    "message": {"content": self.text},
                    "finish_reason": "stop",
                }
            ],
        }


_FAKE_MODELS_BODY = (
    '{"object":"list","data":[{"id":"Qwen/Qwen3.5-4B","object":"model",'
    '"created":1784825047,"owned_by":"sglang","root":"Qwen/Qwen3.5-4B",'
    '"parent":null,"max_model_len":16384}]}'
)
_FAKE_F1_COMPLETION_OVERFLOW_BODY = (
    '{"object":"error","message":"Requested token count exceeds the model\'s '
    "maximum context length of 16384 tokens. You requested a total of 16397 "
    "tokens: 13 tokens from the input messages and 16384 tokens for the "
    "completion. Please reduce the number of tokens in the input messages or "
    'the completion to fit within the limit.","type":"BadRequestError",'
    '"param":null,"code":400}'
)
_FAKE_F2_PROMPT_OVERFLOW_BODY = (
    '{"object":"error","message":"The input (18010 tokens) is longer than '
    'the model\'s context length (16384 tokens).","type":"BadRequestError",'
    '"param":null,"code":400}'
)
_FAKE_REJECT_WINDOW = 16384


class _FakeRejectEndpoint:
    def __init__(
        self,
        *,
        prompt_tokens: int = 13,
        models_body: str = _FAKE_MODELS_BODY,
        completion_overflow_body: str = _FAKE_F1_COMPLETION_OVERFLOW_BODY,
        prompt_overflow_body: str = _FAKE_F2_PROMPT_OVERFLOW_BODY,
        force_completion_overflow: bool = False,
        force_prompt_overflow: bool = False,
        force_non_context_400: bool = False,
        force_second_context_400: bool = False,
    ) -> None:
        self.prompt_tokens = prompt_tokens
        self.models_body = models_body
        self.completion_overflow_body = completion_overflow_body
        self.prompt_overflow_body = prompt_overflow_body
        self.force_completion_overflow = force_completion_overflow
        self.force_prompt_overflow = force_prompt_overflow
        self.force_non_context_400 = force_non_context_400
        self.force_second_context_400 = force_second_context_400
        self.gets: list[dict] = []
        self.posts: list[dict] = []

    @property
    def max_tokens(self) -> list[int]:
        return [int(post["json"]["max_tokens"]) for post in self.posts]

    def install(self, monkeypatch) -> None:
        import httpx

        monkeypatch.setattr(httpx, "get", self.get)
        monkeypatch.setattr(httpx, "post", self.post)

        fake_endpoint = self

        class AsyncClient:
            async def __aenter__(self):
                return self

            async def __aexit__(self, *_args):
                return None

            async def post(self, url, **kwargs):
                return fake_endpoint.post(url, **kwargs)

        monkeypatch.setattr(httpx, "AsyncClient", AsyncClient)

    def get(self, url, **kwargs):
        import httpx

        self.gets.append({"url": url, **kwargs})
        request = httpx.Request("GET", url)
        if str(url).endswith("/v1/models"):
            return httpx.Response(200, request=request, text=self.models_body)
        return httpx.Response(404, request=request, text="not found")

    def post(self, url, **kwargs):
        import httpx

        request = httpx.Request("POST", url)
        if str(url).endswith("/tokenize"):
            content = str((kwargs.get("json") or {}).get("content") or "")
            return httpx.Response(
                200,
                request=request,
                json={"tokens": list(range(max(1, len(content) // 3)))},
            )
        if not str(url).endswith("/v1/chat/completions"):
            raise AssertionError(f"unexpected local provider URL: {url}")

        body = kwargs["json"]
        self.posts.append({"url": url, **kwargs})
        if self.force_non_context_400:
            return httpx.Response(400, request=request, text="invalid temperature")
        if self.force_prompt_overflow or self.prompt_tokens >= _FAKE_REJECT_WINDOW:
            return httpx.Response(
                400,
                request=request,
                text=self.prompt_overflow_body,
            )
        if (
            self.force_completion_overflow
            or self.force_second_context_400
            and len(self.posts) > 1
            or self.prompt_tokens + int(body["max_tokens"]) > _FAKE_REJECT_WINDOW
        ):
            return httpx.Response(
                400,
                request=request,
                text=self.completion_overflow_body,
            )
        return httpx.Response(
            200,
            request=request,
            json={
                "choices": [
                    {
                        "message": {"content": "ok"},
                        "finish_reason": "stop",
                    }
                ],
                "usage": {
                    "prompt_tokens": self.prompt_tokens,
                    "completion_tokens": int(body["max_tokens"]),
                    "total_tokens": self.prompt_tokens + int(body["max_tokens"]),
                },
            },
        )


def _http_status_error(body: str):
    import httpx

    request = httpx.Request("POST", "http://byo.example/openai/v1/chat/completions")
    response = httpx.Response(400, request=request, text=body)
    return httpx.HTTPStatusError("bad request", request=request, response=response)


def test_local_model_prefix_maps_to_provider():
    assert get_model_provider(LOCAL_MODEL) == "local"


def test_local_model_specs():
    provider = _provider()

    assert set(provider.LOCAL_MODEL_SPECS) == {LOCAL_MODEL}
    spec = provider.LOCAL_MODEL_SPECS[LOCAL_MODEL]
    assert spec.repo == "unsloth/Qwen3.5-4B-GGUF"
    assert spec.filename == "Qwen3.5-4B-Q4_K_M.gguf"
    assert (
        spec.sha256
        == "00fe7986ff5f6b463e62455821146049db6f9313603938a70800d1fb69ef11a4"
    )
    assert spec.size_bytes == 2740937888
    assert spec.min_ram_bytes == 8 * 1024**3
    assert spec.mmproj_filename == "mmproj-F16.gguf"
    assert (
        spec.mmproj_sha256
        == "cd88edcf8d031894960bb0c9c5b9b7e1fea6ebee02b9f7ce925a00d12891f864"
    )
    assert spec.mmproj_size_bytes == 672423616


def test_local_provider_defaults_and_registry():
    from solstone.think.providers import PROVIDER_METADATA, PROVIDER_REGISTRY

    assert DEFAULT_MODEL_BY_PROVIDER["local"] == LOCAL_MODEL
    assert PROVIDER_REGISTRY["local"] == "solstone.think.providers.local"
    assert PROVIDER_METADATA["local"] == {
        "label": "Local (on-device)",
        "env_key": "",
    }


def test_context_budget_exceeded_classifies_by_reason_code():
    provider = _provider()

    assert (
        provider.classify_provider_error(
            provider.ContextBudgetExceeded("too large"), "local"
        )
        == "context_budget_exceeded"
    )


@pytest.mark.parametrize(
    ("raw", "expected"),
    [
        ("stop", "stop"),
        ("length", "max_tokens"),
        ("max_tokens", "max_tokens"),
        ("content_filter", "content_filter"),
    ],
)
def test_parse_response_normalizes_known_finish_reasons(raw, expected):
    provider = _provider()

    result = provider._parse_response(_local_response(raw))

    assert result["finish_reason"] == expected


@pytest.mark.parametrize("raw", [None, "", "weird", "tool_calls", "function_call"])
def test_parse_response_fails_closed_on_bad_finish_reasons(raw):
    provider = _provider()

    with pytest.raises(provider.LocalProviderError) as exc_info:
        provider._parse_response(_local_response(raw))

    assert exc_info.value.reason_code == "provider_response_invalid"


def test_list_models_returns_specs():
    models = _provider().list_models("local")

    assert [model["model"] for model in models] == [LOCAL_MODEL]
    assert models[0]["min_ram_bytes"] == 8 * 1024**3


def test_validate_key_uses_tiny_generate(monkeypatch):
    provider = _provider()
    calls = []

    def fake_generate(*args, **kwargs):
        calls.append((args, kwargs))
        return {"text": "OK"}

    monkeypatch.setattr(provider, "run_generate", fake_generate)

    assert provider.validate_key("local", "") == {"valid": True}
    assert calls[0][0] == ("Say OK",)
    assert calls[0][1]["model"] == LOCAL_MODEL
    assert calls[0][1]["max_output_tokens"] == 8


def test_run_generate_posts_to_loopback(monkeypatch):
    provider = _provider()
    served_model_id = (
        "/Users/sol/.cache/huggingface/hub/"
        "models--mlx-community--Qwen3.5-9B/snapshots/abc123"
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: SimpleNamespace(
            port=4321,
            base_url="http://127.0.0.1:4321",
            served_model_id=served_model_id,
        ),
    )
    captured = {}

    class Response:
        def raise_for_status(self):
            return None

        def json(self):
            return {
                "model": served_model_id,
                "choices": [
                    {
                        "message": {"content": "hello"},
                        "finish_reason": "stop",
                    }
                ],
                "usage": {
                    "prompt_tokens": 3,
                    "completion_tokens": 2,
                    "total_tokens": 5,
                },
            }

    def fake_post(url, json, timeout):
        captured.update({"url": url, "json": json, "timeout": timeout})
        return Response()

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    result = provider.run_generate("hello", model=LOCAL_MODEL, max_output_tokens=16)

    assert captured["url"] == "http://127.0.0.1:4321/v1/chat/completions"
    assert captured["json"]["model"] == served_model_id
    assert captured["json"]["messages"] == [{"role": "user", "content": "hello"}]
    assert captured["json"]["max_tokens"] == 16
    assert captured["json"]["chat_template_kwargs"] == {"enable_thinking": False}
    assert captured["json"]["top_p"] == 0.8
    assert captured["json"]["top_k"] == 20
    assert captured["json"]["min_p"] == 0.0
    assert captured["json"]["presence_penalty"] == 1.5
    assert result["text"] == "hello"
    assert result["model"] == LOCAL_MODEL
    assert result["usage"] == {
        "input_tokens": 3,
        "output_tokens": 2,
        "total_tokens": 5,
    }


def test_run_generate_emits_chat_completions_image_url(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: SimpleNamespace(
            port=4321,
            base_url="http://127.0.0.1:4321",
            served_model_id=LOCAL_MODEL,
        ),
    )
    png = b"\x89PNG\r\n\x1a\npayload"
    captured = {}

    class Response:
        def raise_for_status(self):
            return None

        def json(self):
            return {
                "model": LOCAL_MODEL,
                "choices": [
                    {
                        "message": {"content": "ok"},
                        "finish_reason": "stop",
                    }
                ],
            }

    def fake_post(url, json, timeout):
        captured.update({"url": url, "json": json, "timeout": timeout})
        return Response()

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    provider.run_generate(["look", png], model=LOCAL_MODEL)

    assert captured["json"]["messages"] == [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "look"},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "data:image/png;base64,"
                        + base64.b64encode(png).decode("ascii")
                    },
                },
            ],
        }
    ]


def test_run_generate_bundled_clips_oversized_text_block(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: SimpleNamespace(
            port=4321,
            base_url="http://127.0.0.1:4321",
            served_model_id=LOCAL_MODEL,
        ),
    )
    from solstone.think.providers import local_budget

    monkeypatch.setattr(local_budget, "count_tokens", lambda text, _base_url: len(text))
    chunks = [
        "## 2026-06-23 09:00:00 - 09:05:00\n",
        "### Transcript\noldest " + ("o" * 5000) + "\n",
        "### Screen Activity\nmiddle " + ("m" * 5000) + "\n",
        "## 2026-06-23 09:05:00 - 09:10:00\n",
        "### Transcript\nrecent " + ("r" * 5000) + "\n",
        "### Screen Activity\nlatest " + ("l" * 5000) + "\n",
    ]
    big_block = "".join(chunks)
    captured = {}

    class Response:
        def raise_for_status(self):
            return None

        def json(self):
            return {
                "model": LOCAL_MODEL,
                "choices": [{"message": {"content": "ok"}, "finish_reason": "stop"}],
            }

    def fake_post(url, json, timeout):
        captured.update({"url": url, "json": json, "timeout": timeout})
        return Response()

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    schema = {"type": "object"}
    result = provider.run_generate(
        [big_block, "talent prompt"],
        model=LOCAL_MODEL,
        max_output_tokens=8192 * 6,
        system_instruction="system",
        json_schema=schema,
    )

    assert captured["json"]["messages"][0] == {"role": "system", "content": "system"}
    user_message = captured["json"]["messages"][1]["content"]
    assert local_budget.TRUNCATION_MARKER in user_message
    assert "oldest " not in user_message
    assert "latest " in user_message
    assert "talent prompt" in user_message
    assert len(user_message) < len(big_block)
    assert captured["json"]["response_format"]["json_schema"]["schema"] == schema
    assert result["input_budget"]["clipped"] is True


def test_run_generate_does_not_mutate_caller_schema(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: SimpleNamespace(
            port=4321,
            base_url="http://127.0.0.1:4321",
            served_model_id=LOCAL_MODEL,
        ),
    )
    captured = {}

    class Response:
        def raise_for_status(self):
            return None

        def json(self):
            return {
                "model": LOCAL_MODEL,
                "choices": [{"message": {"content": "ok"}, "finish_reason": "stop"}],
            }

    def fake_post(url, json, timeout):
        captured.update({"url": url, "json": json, "timeout": timeout})
        return Response()

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    schema = {
        "type": "object",
        "properties": {
            "timestamp": {
                "type": "string",
                "pattern": r"^\d{2}:\d{2}:\d{2}$",
                "minLength": 8,
                "maxLength": 8,
                SCHEMA_TRUNCATE_KEY: True,
            },
            "slots": {"type": "array", "items": {"type": "string"}},
        },
    }
    original_schema = copy.deepcopy(schema)

    provider.run_generate("hello", model=LOCAL_MODEL, json_schema=schema)

    posted_schema = captured["json"]["response_format"]["json_schema"]["schema"]
    unsupported = {"pattern", "minLength", "maxLength", SCHEMA_TRUNCATE_KEY}
    assert _schema_keyword_paths(posted_schema, unsupported) == []
    assert posted_schema["properties"]["slots"]["maxItems"] == 192
    assert schema == original_schema
    assert schema["properties"]["timestamp"]["pattern"] == r"^\d{2}:\d{2}:\d{2}$"
    assert schema["properties"]["timestamp"]["minLength"] == 8
    assert schema["properties"]["timestamp"]["maxLength"] == 8
    assert schema["properties"]["timestamp"][SCHEMA_TRUNCATE_KEY] is True
    assert sorted(_schema_keyword_paths(schema, unsupported)) == [
        "$/properties/timestamp/maxLength",
        "$/properties/timestamp/minLength",
        "$/properties/timestamp/pattern",
        f"$/properties/timestamp/{SCHEMA_TRUNCATE_KEY}",
    ]
    assert "maxItems" not in schema["properties"]["slots"]


def test_prepare_local_schema_bounds_arrays_only_and_preserves_input():
    provider = _provider()
    schema = {
        "type": "object",
        "properties": {
            "items": {
                "type": "array",
                "minItems": 1,
                "items": {
                    "type": "string",
                    "pattern": r"^\d+$",
                    "minLength": 1,
                    "maxLength": 5,
                },
            },
            "nullable_items": {
                "type": ["array", "null"],
                "items": {"type": "string"},
            },
            "prebounded": {
                "type": "array",
                "maxItems": 7,
                "items": {"type": "string"},
            },
            "status": {"type": "string", "enum": ["open", "closed"]},
            "empty": {"type": "null"},
            "name": {"type": "string"},
            "score": {"type": "number", "minimum": 0, "maximum": 10},
        },
    }
    original_schema = copy.deepcopy(schema)

    prepared = provider._prepare_local_schema(schema)

    assert not hasattr(provider, "_normalize_schema_patterns")
    assert schema == original_schema
    assert prepared["properties"]["items"]["maxItems"] == 192
    assert prepared["properties"]["items"]["minItems"] == 1
    assert prepared["properties"]["nullable_items"]["maxItems"] == 192
    assert prepared["properties"]["prebounded"]["maxItems"] == 7
    assert prepared["properties"]["items"]["items"] == {"type": "string"}
    assert prepared["properties"]["status"] == schema["properties"]["status"]
    assert prepared["properties"]["empty"] == schema["properties"]["empty"]
    assert prepared["properties"]["name"] == schema["properties"]["name"]
    assert prepared["properties"]["score"]["minimum"] == 0
    assert prepared["properties"]["score"]["maximum"] == 10

    assert _schema_keyword_paths(prepared, {"pattern", "minLength", "maxLength"}) == []


def test_prepare_local_schema_skips_json_literals_and_bounds_schema_nodes():
    provider = _provider()
    schema = {
        "type": "object",
        "properties": {
            "literal": {
                "enum": [
                    {"type": "array", "pattern": r"^\d+$", "maxLength": 12},
                ],
            },
            "fixed": {
                "const": {"type": "array", "pattern": r"^\d+$", "maxLength": 12},
            },
            "code": {
                "type": "string",
                "pattern": r"^\d+$",
                "minLength": 1,
                "maxLength": 12,
            },
            "type": {"type": "array", "items": {"type": "string"}},
        },
    }

    prepared = provider._prepare_local_schema(schema)

    assert prepared["properties"]["literal"]["enum"] == [
        {"type": "array", "pattern": r"^\d+$", "maxLength": 12},
    ]
    assert "maxItems" not in prepared["properties"]["literal"]["enum"][0]
    assert prepared["properties"]["fixed"]["const"] == {
        "type": "array",
        "pattern": r"^\d+$",
        "maxLength": 12,
    }
    assert "maxItems" not in prepared["properties"]["fixed"]["const"]
    assert prepared["properties"]["code"] == {"type": "string"}
    assert prepared["properties"]["type"]["maxItems"] == 192


def test_prepare_local_schema_bounds_array_schema_with_enum():
    provider = _provider()
    schema = {
        "type": "array",
        "enum": [["a"], ["b"]],
        "items": {"type": "string"},
    }

    prepared = provider._prepare_local_schema(schema)

    assert prepared["maxItems"] == 192
    assert prepared["enum"] == [["a"], ["b"]]


def test_run_generate_bundled_non_overflow_keeps_body_unmarked(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: SimpleNamespace(
            port=4321,
            base_url="http://127.0.0.1:4321",
            served_model_id=LOCAL_MODEL,
        ),
    )
    from solstone.think.providers import local_budget

    monkeypatch.setattr(local_budget, "count_tokens", lambda text, _base_url: len(text))
    small_block = "## Segment\n### Transcript\nsmall\n"
    captured = {}

    class Response:
        def raise_for_status(self):
            return None

        def json(self):
            return {
                "model": LOCAL_MODEL,
                "choices": [{"message": {"content": "ok"}, "finish_reason": "stop"}],
            }

    def fake_post(url, json, timeout):
        captured.update({"url": url, "json": json, "timeout": timeout})
        return Response()

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    result = provider.run_generate(
        [small_block, "talent prompt"],
        model=LOCAL_MODEL,
        max_output_tokens=1024,
        system_instruction="system",
    )

    assert captured["json"]["messages"][1]["content"] == (
        small_block + "\ntalent prompt"
    )
    assert (
        local_budget.TRUNCATION_MARKER not in captured["json"]["messages"][1]["content"]
    )
    assert "input_budget" not in result


def test_run_generate_bundled_preserved_exceeds_budget_skips_post(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: SimpleNamespace(
            port=4321,
            base_url="http://127.0.0.1:4321",
            served_model_id=LOCAL_MODEL,
        ),
    )
    from solstone.think.providers import local_budget

    monkeypatch.setattr(local_budget, "count_tokens", lambda text, _base_url: len(text))

    def fake_post(*_args, **_kwargs):
        raise AssertionError("httpx.post not expected")

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    with pytest.raises(provider.ContextBudgetExceeded) as exc:
        provider.run_generate(
            "## Segment\n### Transcript\nsmall\n",
            model=LOCAL_MODEL,
            max_output_tokens=8192 * 6,
            system_instruction="s" * 13000,
        )

    assert exc.value.reason_code == "context_budget_exceeded"


def test_run_generate_bundled_context_rejection_backstop(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: SimpleNamespace(
            port=4321,
            base_url="http://127.0.0.1:4321",
            served_model_id=LOCAL_MODEL,
        ),
    )
    from solstone.think.providers import local_budget

    monkeypatch.setattr(local_budget, "count_tokens", lambda text, _base_url: len(text))

    def fake_post(url, json, timeout):
        del json, timeout
        request = httpx.Request("POST", url)
        return httpx.Response(
            400,
            request=request,
            json={
                "error": {
                    "type": "exceed_context_size_error",
                    "message": (
                        "request (17 tokens) exceeds the available context size "
                        "(16 tokens), try increasing it"
                    ),
                    "n_prompt_tokens": 17,
                    "n_ctx": 16,
                }
            },
        )

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    with pytest.raises(provider.ContextBudgetExceeded) as exc:
        provider.run_generate("hello", model=LOCAL_MODEL, max_output_tokens=16)

    assert exc.value.reason_code == "context_budget_exceeded"


def test_run_generate_bundled_context_rejection_backstop_alt_phrasing(monkeypatch):
    # llama-server emits this after post-admission unified-KV exhaustion; the
    # fitted prompt is not proven too long, so this is transient capacity.
    provider = _provider()
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: SimpleNamespace(
            port=4321,
            base_url="http://127.0.0.1:4321",
            served_model_id=LOCAL_MODEL,
        ),
    )
    from solstone.think.providers import local_budget

    monkeypatch.setattr(local_budget, "count_tokens", lambda text, _base_url: len(text))

    def fake_post(url, json, timeout):
        del json, timeout
        request = httpx.Request("POST", url)
        return httpx.Response(
            500,
            request=request,
            json={
                "error": {
                    "type": "server_error",
                    "message": "Context size has been exceeded.",
                }
            },
        )

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    with pytest.raises(provider.LocalProviderError) as exc:
        provider.run_generate("hello", model=LOCAL_MODEL, max_output_tokens=16)

    assert type(exc.value).__name__ == "LocalCapacityExhausted"
    assert exc.value.reason_code == "local_capacity_exhausted"


def test_run_generate_bundled_context_rejection_missing_type_is_capacity(
    monkeypatch,
):
    provider = _provider()
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: SimpleNamespace(
            port=4321,
            base_url="http://127.0.0.1:4321",
            served_model_id=LOCAL_MODEL,
        ),
    )
    from solstone.think.providers import local_budget

    monkeypatch.setattr(local_budget, "count_tokens", lambda text, _base_url: len(text))

    def fake_post(url, json, timeout):
        del json, timeout
        request = httpx.Request("POST", url)
        return httpx.Response(
            500,
            request=request,
            json={"error": {"message": "Context size has been exceeded."}},
        )

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    with pytest.raises(provider.LocalProviderError) as exc:
        provider.run_generate("hello", model=LOCAL_MODEL, max_output_tokens=16)

    assert type(exc.value).__name__ == "LocalCapacityExhausted"
    assert exc.value.reason_code == "local_capacity_exhausted"


def test_run_agenerate_bundled_capacity_rejection_matches_sync(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _bundled_endpoint)
    _patch_bundled_server(monkeypatch)

    class Response:
        text = '{"error":{"type":"server_error","message":"Context size has been exceeded."}}'

        def raise_for_status(self):
            request = httpx.Request(
                "POST",
                "http://127.0.0.1:4321/v1/chat/completions",
            )
            response = httpx.Response(500, request=request, text=self.text)
            raise httpx.HTTPStatusError(
                "server error",
                request=request,
                response=response,
            )

    class AsyncClient:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *_args):
            return None

        async def post(self, *_args, **_kwargs):
            return Response()

    import httpx

    monkeypatch.setattr(httpx, "AsyncClient", AsyncClient)

    with pytest.raises(provider.LocalProviderError) as exc:
        asyncio.run(provider.run_agenerate("hello", model=LOCAL_MODEL))

    assert type(exc.value).__name__ == "LocalCapacityExhausted"
    assert exc.value.reason_code == "local_capacity_exhausted"


def test_run_generate_bundled_capacity_rejection_records_retry_telemetry(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _bundled_endpoint)
    _patch_bundled_server(monkeypatch)

    from solstone.think.providers import local_admission

    records: list[dict] = []
    monkeypatch.setattr(local_admission, "record_local_inference", records.append)

    def fake_post(url, json, timeout):
        del json, timeout
        request = httpx.Request("POST", url)
        return httpx.Response(
            500,
            request=request,
            json={
                "error": {
                    "type": "server_error",
                    "message": "Context size has been exceeded.",
                }
            },
        )

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    with pytest.raises(provider.LocalProviderError):
        provider.run_generate(
            "private prompt text",
            model=LOCAL_MODEL,
            max_output_tokens=16,
        )
    with pytest.raises(provider.LocalProviderError):
        provider.run_generate(
            "private prompt text",
            model=LOCAL_MODEL,
            max_output_tokens=16,
            inference_retry_index=1,
            local_exclusive_admission=True,
        )

    assert len(records) == 2
    assert [record["retry_index"] for record in records] == [0, 1]
    for record in records:
        assert record["reason_code"] == "local_capacity_exhausted"
        assert record["outcome"] == "error"
        serialized = json.dumps(record, sort_keys=True)
        assert "private prompt text" not in serialized
        assert "Context size has been exceeded." not in serialized
        assert "server_error" not in serialized


def test_openhands_local_llm_kwargs(monkeypatch):
    from solstone.think.providers import local_server, openhands

    captured = {}
    served_model_id = (
        "/Users/sol/.cache/huggingface/hub/"
        "models--mlx-community--Qwen3.5-9B/snapshots/abc123"
    )

    class FakeLLM:
        def __init__(self, **kwargs):
            captured.update(kwargs)

    sdk_module = types.ModuleType("openhands.sdk")
    sdk_module.LLM = FakeLLM
    monkeypatch.setitem(sys.modules, "openhands.sdk", sdk_module)
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: SimpleNamespace(port=9876, served_model_id=served_model_id),
    )

    llm = openhands._build_llm(
        "local",
        LOCAL_MODEL,
        endpoint=_bundled_endpoint(),
    )

    assert isinstance(llm, FakeLLM)
    assert captured == {
        "model": f"openai/{served_model_id}",
        "base_url": "http://127.0.0.1:9876/v1",
        "api_key": "EMPTY",
        "native_tool_calling": False,
        "timeout": openhands.LLM_TIMEOUT_S,
        "num_retries": openhands.LLM_NUM_RETRIES,
        "max_input_tokens": local_server.LOCAL_MIN_CONTEXT_TOKENS,
        "max_output_tokens": openhands._local_output_reserve_tokens(
            local_server.LOCAL_MIN_CONTEXT_TOKENS
        ),
        "input_cost_per_token": 0,
        "output_cost_per_token": 0,
        "litellm_extra_body": {"chat_template_kwargs": {"enable_thinking": False}},
    }
    capable_tier = local_server.select_server_tier(24576)
    assert capable_tier.context_tokens == 32768
    assert captured["max_input_tokens"] == 16384
    assert captured["max_input_tokens"] != capable_tier.context_tokens
    assert "chat_template_kwargs" not in captured
    assert openhands._prefixed_model("local", LOCAL_MODEL) == f"openai/{LOCAL_MODEL}"


def _byo_endpoint(
    credential: str | None = "test-token-PLACEHOLDER",
    parallel_slots: int | None = None,
):
    from solstone.think.providers.local_endpoint import (
        LocalEndpoint,
        normalize_local_endpoint_url,
    )

    return LocalEndpoint(
        base_url=normalize_local_endpoint_url("http://byo.example/openai/v1/"),
        served_model_id="served-model",
        credential=credential,
        is_bundled=False,
        parallel_slots=parallel_slots,
    )


def _qwen_byo_endpoint():
    from solstone.think.providers.local_endpoint import (
        LocalEndpoint,
        normalize_local_endpoint_url,
    )

    return LocalEndpoint(
        base_url=normalize_local_endpoint_url("http://byo.example/openai/v1/"),
        served_model_id="Qwen/Qwen3.5-4B",
        credential="test-token-PLACEHOLDER",
        is_bundled=False,
    )


def _bundled_endpoint():
    from solstone.think.providers.local_endpoint import LocalEndpoint

    return LocalEndpoint("", "", None, is_bundled=True)


def _patch_bundled_server(
    monkeypatch,
    *,
    window: int | None = None,
    slots: int | None = None,
    profile: str | None = None,
    served_model_id: str = LOCAL_MODEL,
):
    from solstone.think import utils
    from solstone.think.providers import local_server

    window = local_server.LOCAL_MIN_CONTEXT_TOKENS if window is None else window
    if slots is None:
        slots = local_server._slots_from_launched_tier(window) or 1
    if profile is None:
        profile = "capable" if slots == 2 else "floor"
    local_server.reset_parallel_slots_cache()

    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: SimpleNamespace(
            port=4321,
            base_url="http://127.0.0.1:4321",
            served_model_id=served_model_id,
        ),
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_server.read_server_capacity",
        lambda: local_server.ServerCapacity(slots, "test", profile),
    )
    monkeypatch.setattr(utils, "read_service_port", lambda service: 4321)
    monkeypatch.setattr(
        local_server,
        "read_server_context_props",
        lambda port: local_server.ServerContextProps(
            n_ctx=window * slots,
            total_slots=slots,
        ),
    )


def _launched_context_tokens(plan) -> int:
    from solstone.think import supervisor

    cmd = supervisor._build_local_llama_cmd(plan, 4321)
    return int(cmd[cmd.index("-c") + 1])


def _budget_count_tokens(text: str, _base_url: str | None = None) -> int:
    if text.startswith("b'budget-image"):
        return 0
    return len(text)


def _block_that_trims_to_budget(budget_tokens: int) -> str:
    from solstone.think.providers import local_budget

    marker_tokens = len(local_budget.TRUNCATION_MARKER + "\n\n")
    keep_prefix = "## keep\n"
    keep_tokens = budget_tokens - marker_tokens
    assert keep_tokens > len(keep_prefix)
    dropped = "## drop\n" + ("o" * budget_tokens) + "\n"
    kept = keep_prefix + ("k" * (keep_tokens - len(keep_prefix)))
    return dropped + kept


def _run_bundled_generate_capture(
    monkeypatch,
    *,
    contents,
    window: int,
    slots: int,
    max_output_tokens: int,
    profile: str | None = None,
):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _bundled_endpoint)
    _patch_bundled_server(
        monkeypatch,
        window=window,
        slots=slots,
        profile=profile,
    )

    from solstone.think.providers import local_budget

    monkeypatch.setattr(local_budget, "count_tokens", _budget_count_tokens)
    monkeypatch.setattr(
        provider,
        "encode_image_part",
        lambda _part: ("image/png", "encoded"),
    )
    captured = {}

    def fake_post(url, json, timeout):
        del timeout
        if str(url).endswith("/v1/chat/completions"):
            captured["body"] = json
            return _ChatResponse("ok")
        raise AssertionError(f"unexpected local provider URL: {url}")

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    result = provider.run_generate(
        contents,
        model=LOCAL_MODEL,
        max_output_tokens=max_output_tokens,
    )
    return result, captured, provider


@pytest.mark.parametrize(
    ("prompt", "expected"),
    [
        ("pre\ud83dpost", "pre\ufffdpost"),
        ("pre\ude00post", "pre\ufffdpost"),
        ("pre\ud83d\ude00post", "pre\U0001f600post"),
        ("pre\ud83dX\ude00post", "pre\ufffdX\ufffdpost"),
        ("pre\U0001f600post", "pre\U0001f600post"),
    ],
    ids=[
        "unpaired-high",
        "unpaired-low",
        "surrogate-pair",
        "separated-pair",
        "valid-non-bmp",
    ],
)
def test_run_generate_endpoint_unknown_window_serializes_normalized_surrogate_text(
    monkeypatch,
    prompt,
    expected,
):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)
    monkeypatch.setattr(
        provider,
        "resolve_endpoint_served_window",
        lambda _endpoint: None,
        raising=False,
    )
    captured = {}

    def fake_post(url, **kwargs):
        payload, decoded, content = _request_json_payload(url, kwargs["json"])
        captured.update({"payload": payload, "decoded": decoded, "content": content})
        return _ChatResponse("ok")

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    provider.run_generate(prompt, model=LOCAL_MODEL, max_output_tokens=7)

    message = captured["payload"]["messages"][0]["content"]
    assert message == expected
    _assert_no_surrogate_codepoint(message)
    assert captured["decoded"].encode("utf-8") == captured["content"]
    if expected == "pre\U0001f600post":
        assert b"pre\xf0\x9f\x98\x80post" in captured["content"]


def test_run_agenerate_endpoint_unknown_window_serializes_normalized_surrogate_text(
    monkeypatch,
):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)
    monkeypatch.setattr(
        provider,
        "resolve_endpoint_served_window",
        lambda _endpoint: None,
        raising=False,
    )
    captured = {}

    class AsyncClient:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *_args):
            return None

        async def post(self, url, **kwargs):
            payload, decoded, content = _request_json_payload(url, kwargs["json"])
            captured.update(
                {"payload": payload, "decoded": decoded, "content": content}
            )
            return _ChatResponse("ok")

    import httpx

    monkeypatch.setattr(httpx, "AsyncClient", AsyncClient)

    asyncio.run(
        provider.run_agenerate(
            "pre\ud83dpost",
            model=LOCAL_MODEL,
            max_output_tokens=7,
        )
    )

    message = captured["payload"]["messages"][0]["content"]
    assert message == "pre\ufffdpost"
    _assert_no_surrogate_codepoint(message)
    assert captured["decoded"].encode("utf-8") == captured["content"]


def test_run_generate_endpoint_known_window_normalizes_before_fit_and_serialization(
    monkeypatch,
):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)
    monkeypatch.setattr(
        provider,
        "resolve_endpoint_served_window",
        lambda _endpoint: 4096,
        raising=False,
    )

    from solstone.think.providers import local_budget

    original_fit_contents = local_budget.fit_contents
    captured_fit = {}

    def spy_fit_contents(
        contents, system_instruction, max_output_tokens, *, count, window
    ):
        captured_fit.update(
            {
                "contents": contents,
                "system_instruction": system_instruction,
                "max_output_tokens": max_output_tokens,
                "window": window,
            }
        )
        return original_fit_contents(
            contents,
            system_instruction,
            max_output_tokens,
            count=count,
            window=window,
        )

    monkeypatch.setattr(local_budget, "fit_contents", spy_fit_contents)
    captured_post = {}

    def fake_post(url, **kwargs):
        payload, decoded, content = _request_json_payload(url, kwargs["json"])
        captured_post.update(
            {"payload": payload, "decoded": decoded, "content": content}
        )
        return _ChatResponse("ok")

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    provider.run_generate(
        ["## Segment\npre\ud83dpost\n", "tail\ude00"],
        model=LOCAL_MODEL,
        system_instruction="sys\ud83d",
        max_output_tokens=7,
    )

    assert captured_fit["contents"] == ["## Segment\npre\ufffdpost\n", "tail\ufffd"]
    assert captured_fit["system_instruction"] == "sys\ufffd"
    assert captured_fit["window"] == 4096
    for item in captured_fit["contents"]:
        _assert_no_surrogate_codepoint(item)
    _assert_no_surrogate_codepoint(captured_fit["system_instruction"])

    messages = captured_post["payload"]["messages"]
    assert messages[0] == {"role": "system", "content": "sys\ufffd"}
    assert "pre\ufffdpost" in messages[1]["content"]
    assert "tail\ufffd" in messages[1]["content"]
    assert captured_post["decoded"].encode("utf-8") == captured_post["content"]


def test_run_generate_bundled_tokenize_and_chat_receive_normalized_text(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _bundled_endpoint)
    _patch_bundled_server(monkeypatch)
    tokenize_texts = []
    captured_chat = {}

    def fake_post(url, **kwargs):
        payload, decoded, content = _request_json_payload(url, kwargs["json"])
        if str(url).endswith("/tokenize"):
            text = payload["content"]
            tokenize_texts.append(text)
            return httpx.Response(
                200,
                request=httpx.Request("POST", url),
                json={"tokens": list(range(max(1, len(text) // 3)))},
            )
        if str(url).endswith("/v1/chat/completions"):
            captured_chat.update(
                {"payload": payload, "decoded": decoded, "content": content}
            )
            return _ChatResponse("ok")
        raise AssertionError(f"unexpected local provider URL: {url}")

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    provider.run_generate("pre\ud83dpost", model=LOCAL_MODEL, max_output_tokens=7)

    assert tokenize_texts
    assert "pre\ufffdpost" in tokenize_texts
    for text in tokenize_texts:
        _assert_no_surrogate_codepoint(text)
    message = captured_chat["payload"]["messages"][0]["content"]
    assert message == "pre\ufffdpost"
    _assert_no_surrogate_codepoint(message)
    assert captured_chat["decoded"].encode("utf-8") == captured_chat["content"]


def test_run_generate_confidential_endpoint_serializes_normalized_text_and_keeps_qwen_block(
    monkeypatch,
):
    provider = _provider()
    from solstone.think.providers.local_endpoint import LocalEndpoint

    endpoint = LocalEndpoint(
        base_url="https://spp.example.test",
        served_model_id="confidential-model",
        credential="confidential-token",
        is_bundled=False,
        is_confidential=True,
    )
    monkeypatch.setattr(provider, "resolve_local_endpoint", lambda: endpoint)
    monkeypatch.setattr(
        provider,
        "resolve_endpoint_served_window",
        lambda _endpoint: None,
    )
    monkeypatch.setattr(
        "solstone.think.services.spp_transport.confidential_egress_base_url",
        lambda _base_url: "http://127.0.0.1:4567",
    )
    captured = {}

    def fake_post(url, **kwargs):
        payload, decoded, content = _request_json_payload(url, kwargs["json"])
        captured.update(
            {"url": url, "payload": payload, "decoded": decoded, "content": content}
        )
        return _ChatResponse("ok")

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    provider.run_generate(
        "confidential\ud83dtext",
        model=LOCAL_MODEL,
        temperature=0.4,
        max_output_tokens=7,
    )

    assert captured["url"] == "http://127.0.0.1:4567/v1/chat/completions"
    assert captured["payload"]["messages"][0]["content"] == "confidential\ufffdtext"
    assert captured["payload"]["chat_template_kwargs"] == {"enable_thinking": False}
    assert captured["payload"]["top_p"] == 0.8
    assert captured["payload"]["top_k"] == 20
    assert captured["payload"]["min_p"] == 0.0
    assert captured["payload"]["presence_penalty"] == 1.5
    assert captured["decoded"].encode("utf-8") == captured["content"]


def test_run_generate_role_dict_multimodal_content_normalizes_text_and_preserves_image_identity(
    monkeypatch,
):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)
    monkeypatch.setattr(
        provider,
        "resolve_endpoint_served_window",
        lambda _endpoint: None,
        raising=False,
    )
    image = b"\x89PNG\r\n\x1a\npayload"
    contents = [{"role": "user", "content": ["caption\ud83d", image]}]
    encoded_parts = []

    def fake_encode_image_part(part):
        encoded_parts.append(part)
        assert part is image
        return "image/png", "encoded"

    monkeypatch.setattr(provider, "encode_image_part", fake_encode_image_part)
    captured = {}

    def fake_post(url, **kwargs):
        payload, decoded, content = _request_json_payload(url, kwargs["json"])
        captured.update({"payload": payload, "decoded": decoded, "content": content})
        return _ChatResponse("ok")

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    provider.run_generate(contents, model=LOCAL_MODEL, max_output_tokens=7)

    assert encoded_parts == [image]
    assert encoded_parts[0] is image
    message_content = captured["payload"]["messages"][0]["content"]
    assert message_content[0] == {"type": "text", "text": "caption\ufffd"}
    assert message_content[1] == {
        "type": "image_url",
        "image_url": {"url": "data:image/png;base64,encoded"},
    }
    _assert_no_surrogate_codepoint(message_content[0]["text"])
    assert captured["decoded"].encode("utf-8") == captured["content"]


def test_local_request_normalizer_preserves_clean_identity_and_non_mutation():
    provider = _provider()
    from PIL import Image

    image = Image.new("RGB", (1, 1))
    raw_bytes = b"\x89PNG\r\n\x1a\npayload"
    mutable_bytes = bytearray(raw_bytes)
    clean_text = "pre\U0001f600post"
    clean_list = [clean_text, image, raw_bytes, mutable_bytes]
    clean_tuple = (clean_text, image)
    clean_dict = {"role": "user", "content": clean_text, "image": image}

    assert provider._normalize_request_text(clean_text) is clean_text
    assert provider._normalize_request_text(clean_list) is clean_list
    assert provider._normalize_request_text(clean_tuple) is clean_tuple
    assert provider._normalize_request_text(clean_dict) is clean_dict
    assert provider._normalize_request_text(image) is image
    assert provider._normalize_request_text(raw_bytes) is raw_bytes
    assert provider._normalize_request_text(mutable_bytes) is mutable_bytes
    assert provider._normalize_request_text(None) is None

    dirty_text = "pre\ud83dpost"
    dirty_list = [dirty_text, image]
    dirty_tuple = (dirty_text, image)
    dirty_dict = {"content": dirty_text, "image": image}

    normalized_list = provider._normalize_request_text(dirty_list)
    normalized_tuple = provider._normalize_request_text(dirty_tuple)
    normalized_dict = provider._normalize_request_text(dirty_dict)

    assert normalized_list is not dirty_list
    assert normalized_list == ["pre\ufffdpost", image]
    assert normalized_list[1] is image
    assert dirty_list == [dirty_text, image]
    assert dirty_list[0] is dirty_text

    assert normalized_tuple is not dirty_tuple
    assert normalized_tuple == ("pre\ufffdpost", image)
    assert normalized_tuple[1] is image
    assert dirty_tuple == (dirty_text, image)

    assert normalized_dict is not dirty_dict
    assert normalized_dict == {"content": "pre\ufffdpost", "image": image}
    assert normalized_dict["image"] is image
    assert dirty_dict == {"content": dirty_text, "image": image}


def test_run_generate_does_not_mutate_contents_or_schema_when_normalizing(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)
    monkeypatch.setattr(
        provider,
        "resolve_endpoint_served_window",
        lambda _endpoint: None,
        raising=False,
    )
    image = b"\x89PNG\r\n\x1a\npayload"
    content_parts = ["pre\ud83dpost", image]
    message = {"role": "user", "content": content_parts}
    contents = [message]
    schema = {
        "type": "object",
        "properties": {
            "answer": {
                "type": "string",
                "pattern": r"^\w+$",
                "minLength": 1,
                "maxLength": 12,
            }
        },
    }
    contents_before = copy.deepcopy(contents)
    schema_before = copy.deepcopy(schema)
    contents_ref = contents
    message_ref = message
    content_parts_ref = content_parts
    prompt_ref = content_parts[0]
    schema_ref = schema
    captured = {}

    monkeypatch.setattr(
        provider,
        "encode_image_part",
        lambda part: ("image/png", "encoded") if part is image else None,
    )

    def fake_post(url, **kwargs):
        payload, decoded, content = _request_json_payload(url, kwargs["json"])
        captured.update({"payload": payload, "decoded": decoded, "content": content})
        return _ChatResponse("ok")

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    provider.run_generate(
        contents,
        model=LOCAL_MODEL,
        max_output_tokens=7,
        json_schema=schema,
    )

    assert contents is contents_ref
    assert contents == contents_before
    assert contents[0] is message_ref
    assert contents[0]["content"] is content_parts_ref
    assert contents[0]["content"][0] is prompt_ref
    assert contents[0]["content"][1] is image
    assert schema is schema_ref
    assert schema == schema_before

    message_content = captured["payload"]["messages"][0]["content"]
    assert message_content[0] == {"type": "text", "text": "pre\ufffdpost"}
    assert captured["payload"]["response_format"]["json_schema"]["schema"] == {
        "type": "object",
        "properties": {"answer": {"type": "string"}},
    }
    assert captured["decoded"].encode("utf-8") == captured["content"]


def test_run_generate_bundled_above_quarter_bounds_footprint(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _bundled_endpoint)
    window = 32768
    _patch_bundled_server(monkeypatch, window=window, slots=2, profile="capable")

    from solstone.think.providers import local_budget

    max_output_tokens = 16384
    input_budget = local_budget.compute_input_budget(max_output_tokens, window)
    monkeypatch.setattr(local_budget, "count_tokens", lambda text, _base_url: len(text))
    captured = {}

    def fake_post(url, json, timeout):
        del timeout
        if str(url).endswith("/v1/chat/completions"):
            captured["body"] = json
            return _ChatResponse("ok")
        raise AssertionError(f"unexpected local provider URL: {url}")

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    result = provider.run_generate(
        "x" * input_budget,
        model=LOCAL_MODEL,
        max_output_tokens=max_output_tokens,
    )
    request_budget = result["request_budget"]
    completion = request_budget["clamped_max_tokens"]

    assert captured["body"]["max_tokens"] == completion
    assert input_budget + completion + local_budget._SAFETY_MARGIN_TOKENS <= window


def test_run_generate_bundled_ac3_capable_49152_bounds_launched_pool(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _bundled_endpoint)

    from solstone.think import supervisor
    from solstone.think.providers import local_budget, local_server

    tier = local_server._CAPABLE_TIER
    _patch_bundled_server(
        monkeypatch,
        window=tier.context_tokens,
        slots=tier.parallel_slots,
        profile=tier.name,
    )
    plan = supervisor.LocalServerLaunchPlan(
        backend="vulkan",
        desired_fingerprint_json='{"provider":"local"}',
        desired_fingerprint_sha256="fp-local",
        binary_path=Path("/tmp/llama-server"),
        model_path=Path("/tmp/model.gguf"),
        context_tokens=tier.context_tokens,
        parallel_slots=tier.parallel_slots,
        prompt_cache_mib=tier.prompt_cache_mib,
    )
    launched_c = _launched_context_tokens(plan)
    max_output_tokens = 49152
    input_budget = local_budget.compute_input_budget(
        max_output_tokens,
        tier.context_tokens,
    )
    monkeypatch.setattr(local_budget, "count_tokens", lambda text, _base_url: len(text))
    captured = {}

    def fake_post(url, json, timeout):
        del timeout
        if str(url).endswith("/v1/chat/completions"):
            captured["body"] = json
            return _ChatResponse("ok")
        raise AssertionError(f"unexpected local provider URL: {url}")

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    result = provider.run_generate(
        "x" * input_budget,
        model=LOCAL_MODEL,
        max_output_tokens=max_output_tokens,
    )
    request_budget = result["request_budget"]
    completion = request_budget["clamped_max_tokens"]

    assert captured["body"]["max_tokens"] == completion
    assert (
        request_budget["estimated_prompt_tokens"]
        + request_budget["image_tokens"]
        + completion
    ) * request_budget["slots"] <= launched_c


@pytest.mark.parametrize(
    ("max_output_tokens", "expected_completion"),
    [(16384, 8192), (4096, 4096)],
)
def test_run_generate_bundled_bounds_above_and_below_quarter(
    monkeypatch,
    max_output_tokens,
    expected_completion,
):
    from solstone.think.providers import local_budget, local_server

    window = local_server._CAPABLE_TIER.context_tokens
    prompt_tokens = local_budget.compute_input_budget(max_output_tokens, window)

    result, captured, _provider_mod = _run_bundled_generate_capture(
        monkeypatch,
        contents="x" * prompt_tokens,
        window=window,
        slots=local_server._CAPABLE_TIER.parallel_slots,
        max_output_tokens=max_output_tokens,
        profile=local_server._CAPABLE_TIER.name,
    )

    request_budget = result["request_budget"]
    true_room = (
        window
        - request_budget["estimated_prompt_tokens"]
        - request_budget["image_tokens"]
        - local_budget._SAFETY_MARGIN_TOKENS
    )
    assert request_budget["clamped_max_tokens"] == expected_completion
    assert request_budget["clamped_max_tokens"] == true_room
    assert captured["body"]["max_tokens"] == expected_completion


def test_run_generate_bundled_clamp_is_prompt_derived_and_ample_room_unclamped(
    monkeypatch,
):
    from solstone.think.providers import local_budget, local_server

    tier = local_server._CAPABLE_TIER
    requested = 16384
    small, _small_captured, _provider_mod = _run_bundled_generate_capture(
        monkeypatch,
        contents="short",
        window=tier.context_tokens,
        slots=tier.parallel_slots,
        max_output_tokens=requested,
        profile=tier.name,
    )
    large_prompt = "x" * local_budget.compute_input_budget(
        requested, tier.context_tokens
    )
    large, _large_captured, _provider_mod = _run_bundled_generate_capture(
        monkeypatch,
        contents=large_prompt,
        window=tier.context_tokens,
        slots=tier.parallel_slots,
        max_output_tokens=requested,
        profile=tier.name,
    )

    assert small["request_budget"]["clamped_max_tokens"] == requested
    assert large["request_budget"]["clamped_max_tokens"] != requested
    assert (
        small["request_budget"]["clamped_max_tokens"]
        != large["request_budget"]["clamped_max_tokens"]
    )


def test_run_generate_bundled_role_dict_chat_payload_is_bounded_not_trimmed(
    monkeypatch,
):
    from solstone.think.providers import local_budget, local_server

    window = local_server._CAPABLE_TIER.context_tokens
    prompt = "x" * 30000
    result, captured, provider_mod = _run_bundled_generate_capture(
        monkeypatch,
        contents=[{"role": "user", "content": prompt}],
        window=window,
        slots=local_server._CAPABLE_TIER.parallel_slots,
        max_output_tokens=16384,
        profile=local_server._CAPABLE_TIER.name,
    )

    serialized = provider_mod._serialized_message_text(captured["body"]["messages"])
    assert serialized == prompt
    assert local_budget.TRUNCATION_MARKER not in serialized
    assert "input_budget" not in result
    assert (
        captured["body"]["max_tokens"]
        + len(serialized)
        + local_budget._SAFETY_MARGIN_TOKENS
        <= window
    )


def test_run_generate_bundled_role_dict_impossible_raises_context_budget(
    monkeypatch,
):
    from solstone.think.providers import local_server

    with pytest.raises(Exception) as exc:
        _run_bundled_generate_capture(
            monkeypatch,
            contents=[{"role": "user", "content": "x" * 32600}],
            window=local_server._CAPABLE_TIER.context_tokens,
            slots=local_server._CAPABLE_TIER.parallel_slots,
            max_output_tokens=4096,
            profile=local_server._CAPABLE_TIER.name,
        )

    assert getattr(exc.value, "reason_code", None) == "context_budget_exceeded"


@pytest.mark.parametrize(("image_count", "expected_image_tokens"), [(0, 0), (1, 2500)])
def test_run_generate_bundled_accounts_image_tokens(
    monkeypatch,
    image_count,
    expected_image_tokens,
):
    from solstone.think.providers import local_budget, local_server

    images = [f"budget-image-{idx}".encode("ascii") for idx in range(image_count)]
    contents = "hello" if image_count == 0 else ["hello", *images]

    result, captured, _provider_mod = _run_bundled_generate_capture(
        monkeypatch,
        contents=contents,
        window=local_server._CAPABLE_TIER.context_tokens,
        slots=local_server._CAPABLE_TIER.parallel_slots,
        max_output_tokens=49152,
        profile=local_server._CAPABLE_TIER.name,
    )

    request_budget = result["request_budget"]
    true_room = (
        request_budget["window"]
        - request_budget["estimated_prompt_tokens"]
        - expected_image_tokens
        - local_budget._SAFETY_MARGIN_TOKENS
    )
    assert request_budget["image_tokens"] == expected_image_tokens
    assert request_budget["clamped_max_tokens"] == min(49152, true_room)
    assert captured["body"]["max_tokens"] == request_budget["clamped_max_tokens"]


def test_run_generate_bundled_floor_two_images_trims_and_sends(monkeypatch):
    from solstone.think.providers import local_budget, local_server

    tier = local_server._FLOOR_TIER
    requested = 49152
    image_tokens = local_budget._ESTIMATED_IMAGE_TOKENS * 2
    effective_window = tier.context_tokens - image_tokens
    input_budget = local_budget.compute_input_budget(requested, effective_window)
    images = [b"budget-image-0", b"budget-image-1"]

    result, captured, _provider_mod = _run_bundled_generate_capture(
        monkeypatch,
        contents=[_block_that_trims_to_budget(input_budget), *images],
        window=tier.context_tokens,
        slots=tier.parallel_slots,
        max_output_tokens=requested,
        profile=tier.name,
    )

    request_budget = result["request_budget"]
    text_part = captured["body"]["messages"][0]["content"][0]["text"]
    assert local_budget.TRUNCATION_MARKER in text_part
    assert result["input_budget"]["budget_tokens"] == 8282
    assert request_budget["estimated_prompt_tokens"] == 8282
    assert request_budget["image_tokens"] == 5000
    assert request_budget["clamped_max_tokens"] == 2846
    assert captured["body"]["max_tokens"] == 2846


def test_run_generate_bundled_request_budget_and_clamp_log(caplog, monkeypatch):
    from solstone.think.providers import local_budget, local_server

    tier = local_server._CAPABLE_TIER
    requested = 49152
    input_budget = local_budget.compute_input_budget(requested, tier.context_tokens)

    with caplog.at_level(logging.INFO):
        result, _captured, _provider_mod = _run_bundled_generate_capture(
            monkeypatch,
            contents="x" * input_budget,
            window=tier.context_tokens,
            slots=tier.parallel_slots,
            max_output_tokens=requested,
            profile=tier.name,
        )

    request_budget = result["request_budget"]
    assert request_budget == {
        "window": tier.context_tokens,
        "slots": tier.parallel_slots,
        "estimated_prompt_tokens": input_budget,
        "image_tokens": 0,
        "clamped_max_tokens": 8192,
        "requested_max_output_tokens": requested,
    }
    assert any(
        "local bundled max_tokens clamped" in record.message
        for record in caplog.records
    )


def test_run_generate_bundled_image_content_impossible_raises_before_fit(
    monkeypatch,
):
    from solstone.think.providers import local_server

    images = [f"budget-image-{idx}".encode("ascii") for idx in range(7)]
    with pytest.raises(Exception) as exc:
        _run_bundled_generate_capture(
            monkeypatch,
            contents=["prompt", *images],
            window=local_server._FLOOR_TIER.context_tokens,
            slots=local_server._FLOOR_TIER.parallel_slots,
            max_output_tokens=49152,
            profile=local_server._FLOOR_TIER.name,
        )

    assert getattr(exc.value, "reason_code", None) == "context_budget_exceeded"
    assert "image content" in str(exc.value)
    assert "context window" in str(exc.value)


@pytest.mark.parametrize("image_count", [0, 1])
@pytest.mark.parametrize("tier_name", ["floor", "capable"])
def test_run_generate_bundled_ac3_fitted_prompt_bounds_launched_pool(
    monkeypatch,
    tier_name,
    image_count,
):
    from solstone.think import supervisor
    from solstone.think.providers import local_budget, local_server

    tier = (
        local_server._FLOOR_TIER if tier_name == "floor" else local_server._CAPABLE_TIER
    )
    requested = 49152
    images = [f"budget-image-{idx}".encode("ascii") for idx in range(image_count)]
    image_tokens = local_budget._ESTIMATED_IMAGE_TOKENS * image_count
    input_budget = local_budget.compute_input_budget(
        requested,
        tier.context_tokens - image_tokens,
    )
    block = _block_that_trims_to_budget(input_budget)
    contents = block if image_count == 0 else [block, *images]
    plan = supervisor.LocalServerLaunchPlan(
        backend="vulkan",
        desired_fingerprint_json='{"provider":"local"}',
        desired_fingerprint_sha256="fp-local",
        binary_path=Path("/tmp/llama-server"),
        model_path=Path("/tmp/model.gguf"),
        context_tokens=tier.context_tokens,
        parallel_slots=tier.parallel_slots,
        prompt_cache_mib=tier.prompt_cache_mib,
    )
    launched_c = _launched_context_tokens(plan)

    result, _captured, _provider_mod = _run_bundled_generate_capture(
        monkeypatch,
        contents=contents,
        window=tier.context_tokens,
        slots=tier.parallel_slots,
        max_output_tokens=requested,
        profile=tier.name,
    )

    resolved = local_budget.resolve_context_window()
    request_budget = result["request_budget"]
    assert resolved.window_tokens == tier.context_tokens
    assert request_budget["window"] == resolved.window_tokens
    assert request_budget["estimated_prompt_tokens"] == input_budget
    assert (
        request_budget["estimated_prompt_tokens"]
        + request_budget["image_tokens"]
        + request_budget["clamped_max_tokens"]
    ) * request_budget["slots"] <= launched_c


def test_run_generate_endpoint_clamps_large_default_budget_against_served_window(
    monkeypatch,
):
    provider = _provider()
    fake_endpoint = _FakeRejectEndpoint()
    fake_endpoint.install(monkeypatch)
    monkeypatch.setattr(provider, "resolve_local_endpoint", _qwen_byo_endpoint)

    result = provider.run_generate(
        "hello",
        model=LOCAL_MODEL,
        max_output_tokens=49152,
    )

    assert result["text"] == "ok"
    assert fake_endpoint.gets
    assert fake_endpoint.max_tokens
    assert all(token <= _FAKE_REJECT_WINDOW for token in fake_endpoint.max_tokens)


def test_run_agenerate_endpoint_clamps_window_sized_budget_against_served_window(
    monkeypatch,
):
    provider = _provider()
    fake_endpoint = _FakeRejectEndpoint()
    fake_endpoint.install(monkeypatch)
    monkeypatch.setattr(provider, "resolve_local_endpoint", _qwen_byo_endpoint)

    result = asyncio.run(
        provider.run_agenerate(
            "hello",
            model=LOCAL_MODEL,
            max_output_tokens=16384,
        )
    )

    assert result["text"] == "ok"
    assert fake_endpoint.gets
    assert fake_endpoint.max_tokens
    assert all(token <= _FAKE_REJECT_WINDOW for token in fake_endpoint.max_tokens)


def test_run_generate_endpoint_preflight_floor_skips_chat_post(monkeypatch):
    provider = _provider()
    fake_endpoint = _FakeRejectEndpoint()
    fake_endpoint.install(monkeypatch)
    monkeypatch.setattr(provider, "resolve_local_endpoint", _qwen_byo_endpoint)

    with pytest.raises(provider.ContextBudgetExceeded) as exc:
        provider.run_generate(
            [{"role": "user", "content": "x" * 50000}],
            model=LOCAL_MODEL,
            max_output_tokens=1024,
        )

    assert exc.value.reason_code == "context_budget_exceeded"
    assert fake_endpoint.gets
    assert fake_endpoint.posts == []


def test_validate_key_known_window_keeps_tiny_max_tokens(monkeypatch):
    provider = _provider()
    fake_endpoint = _FakeRejectEndpoint()
    fake_endpoint.install(monkeypatch)
    monkeypatch.setattr(provider, "resolve_local_endpoint", _qwen_byo_endpoint)

    assert provider.validate_key("local", "") == {"valid": True}
    assert fake_endpoint.max_tokens == [8]


def test_run_generate_endpoint_known_window_truncates_fittable_input(monkeypatch):
    provider = _provider()
    fake_endpoint = _FakeRejectEndpoint()
    fake_endpoint.install(monkeypatch)
    monkeypatch.setattr(provider, "resolve_local_endpoint", _qwen_byo_endpoint)
    chunks = [
        f"## 2026-06-23 09:{minute:02d}:00\n### Transcript\n"
        + (str(minute) * 3000)
        + "\n"
        for minute in range(20)
    ]

    result = provider.run_generate(
        "".join(chunks),
        model=LOCAL_MODEL,
        max_output_tokens=1024,
    )

    from solstone.think.providers import local_budget

    posted_content = fake_endpoint.posts[0]["json"]["messages"][0]["content"]
    assert local_budget.TRUNCATION_MARKER in posted_content
    assert result["input_budget"]["clipped"] is True
    assert result["input_budget"]["dropped_entries"] > 0
    assert result["request_budget"]["window"] == _FAKE_REJECT_WINDOW
    assert result["request_budget"]["clamped_max_tokens"] == 1024


def test_prepare_endpoint_request_explicit_window_is_not_divided_by_slots():
    provider = _provider()
    endpoint = _byo_endpoint(parallel_slots=2)
    served_window = 25000

    body, _input_budget, request_budget = provider._prepare_endpoint_request(
        endpoint=endpoint,
        served_window=served_window,
        contents="x" * 30000,
        system_instruction=None,
        temperature=0.3,
        max_output_tokens=12000,
        json_output=False,
        json_schema=None,
    )

    assert request_budget["window"] == served_window
    assert request_budget["slots"] == 2
    assert request_budget["estimated_prompt_tokens"] == 10000
    assert body["max_tokens"] == 12000


def test_endpoint_generate_branches_route_shared_prep(monkeypatch):
    provider = _provider()
    calls = []
    original = provider._prepare_endpoint_request
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)
    monkeypatch.setattr(
        provider,
        "resolve_endpoint_served_window",
        lambda _endpoint: None,
    )

    def spy_prepare(**kwargs):
        calls.append(kwargs)
        return original(**kwargs)

    def fake_post(url, **kwargs):
        return _ChatResponse("ok")

    class AsyncClient:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *_args):
            return None

        async def post(self, url, **kwargs):
            return _ChatResponse("ok")

    import httpx

    monkeypatch.setattr(provider, "_prepare_endpoint_request", spy_prepare)
    monkeypatch.setattr(httpx, "post", fake_post)
    monkeypatch.setattr(httpx, "AsyncClient", AsyncClient)

    provider.run_generate("hello", model=LOCAL_MODEL, max_output_tokens=7)
    asyncio.run(provider.run_agenerate("hello", model=LOCAL_MODEL, max_output_tokens=7))

    assert [call["max_output_tokens"] for call in calls] == [7, 7]
    assert [call["served_window"] for call in calls] == [None, None]


def _run_endpoint_generate(provider, mode: str, **kwargs):
    if mode == "sync":
        return provider.run_generate(**kwargs)
    return asyncio.run(provider.run_agenerate(**kwargs))


@pytest.mark.parametrize("mode", ["sync", "async"])
def test_endpoint_completion_overflow_reclamps_once(monkeypatch, mode):
    provider = _provider()
    fake_endpoint = _FakeRejectEndpoint()
    fake_endpoint.install(monkeypatch)
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)
    monkeypatch.setattr(
        provider,
        "resolve_endpoint_served_window",
        lambda _endpoint: None,
    )

    result = _run_endpoint_generate(
        provider,
        mode,
        contents="hello",
        model=LOCAL_MODEL,
        max_output_tokens=16384,
    )

    assert result["text"] == "ok"
    assert fake_endpoint.max_tokens == [16384, 16355]


@pytest.mark.parametrize("mode", ["sync", "async"])
def test_endpoint_completion_overflow_below_floor_is_budget_terminal(
    monkeypatch,
    mode,
):
    provider = _provider()
    low_room_body = _FAKE_F1_COMPLETION_OVERFLOW_BODY.replace(
        "13 tokens from the input messages and 16384 tokens for the completion",
        "16200 tokens from the input messages and 500 tokens for the completion",
    )
    fake_endpoint = _FakeRejectEndpoint(
        completion_overflow_body=low_room_body,
        force_completion_overflow=True,
    )
    fake_endpoint.install(monkeypatch)
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)
    monkeypatch.setattr(
        provider,
        "resolve_endpoint_served_window",
        lambda _endpoint: None,
    )

    with pytest.raises(provider.ContextBudgetExceeded) as exc:
        _run_endpoint_generate(
            provider,
            mode,
            contents="hello",
            model=LOCAL_MODEL,
            max_output_tokens=500,
        )

    assert exc.value.reason_code == "context_budget_exceeded"
    assert fake_endpoint.max_tokens == [500]


@pytest.mark.parametrize("mode", ["sync", "async"])
def test_endpoint_prompt_overflow_is_terminal_context_window(monkeypatch, mode):
    provider = _provider()
    fake_endpoint = _FakeRejectEndpoint(force_prompt_overflow=True)
    fake_endpoint.install(monkeypatch)
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)
    monkeypatch.setattr(
        provider,
        "resolve_endpoint_served_window",
        lambda _endpoint: None,
    )

    with pytest.raises(provider.LocalProviderError) as exc:
        _run_endpoint_generate(
            provider,
            mode,
            contents="hello",
            model=LOCAL_MODEL,
            max_output_tokens=1024,
        )

    assert exc.value.reason_code == "context_window_exceeded"
    assert fake_endpoint.max_tokens == [1024]


@pytest.mark.parametrize("mode", ["sync", "async"])
def test_endpoint_non_context_400_stays_contract_failed(monkeypatch, mode):
    provider = _provider()
    fake_endpoint = _FakeRejectEndpoint(force_non_context_400=True)
    fake_endpoint.install(monkeypatch)
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)
    monkeypatch.setattr(
        provider,
        "resolve_endpoint_served_window",
        lambda _endpoint: None,
    )

    with pytest.raises(provider.LocalProviderError) as exc:
        _run_endpoint_generate(
            provider,
            mode,
            contents="hello",
            model=LOCAL_MODEL,
            max_output_tokens=1024,
        )

    assert exc.value.reason_code == "local_endpoint_contract_failed"
    assert fake_endpoint.max_tokens == [1024]


@pytest.mark.parametrize("mode", ["sync", "async"])
def test_endpoint_second_context_400_after_reclamp_is_terminal(monkeypatch, mode):
    provider = _provider()
    fake_endpoint = _FakeRejectEndpoint(force_second_context_400=True)
    fake_endpoint.install(monkeypatch)
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)
    monkeypatch.setattr(
        provider,
        "resolve_endpoint_served_window",
        lambda _endpoint: None,
    )

    with pytest.raises(provider.LocalProviderError) as exc:
        _run_endpoint_generate(
            provider,
            mode,
            contents="hello",
            model=LOCAL_MODEL,
            max_output_tokens=16384,
        )

    assert exc.value.reason_code == "context_window_exceeded"
    assert fake_endpoint.max_tokens == [16384, 16355]


def test_run_generate_byo_unknown_window_request_body_matches_golden(
    monkeypatch,
):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)
    monkeypatch.setattr(
        provider,
        "resolve_endpoint_served_window",
        lambda _endpoint: None,
        raising=False,
    )
    captured = {}

    def fake_post(url, **kwargs):
        captured.update({"url": url, **kwargs})
        return _ChatResponse("ok")

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    provider.run_generate(
        "hello",
        model=LOCAL_MODEL,
        temperature=0.4,
        max_output_tokens=7,
    )

    assert captured["json"] == {
        "model": "served-model",
        "messages": [{"role": "user", "content": "hello"}],
        "temperature": 0.4,
        "max_tokens": 7,
        "stream": False,
    }


def test_run_generate_confidential_body_carries_qwen_block(monkeypatch):
    provider = _provider()
    from solstone.think.providers.local_endpoint import LocalEndpoint

    endpoint = LocalEndpoint(
        base_url="https://spp.example.test",
        served_model_id="confidential-model",
        credential="confidential-token",
        is_bundled=False,
        is_confidential=True,
    )
    captured = {}
    monkeypatch.setattr(provider, "resolve_local_endpoint", lambda: endpoint)
    monkeypatch.setattr(
        provider,
        "resolve_endpoint_served_window",
        lambda _endpoint: None,
    )
    monkeypatch.setattr(
        "solstone.think.services.spp_transport.confidential_egress_base_url",
        lambda _base_url: "http://127.0.0.1:4567",
    )

    def fake_post(url, **kwargs):
        captured.update({"url": url, **kwargs})
        return _ChatResponse("ok")

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    provider.run_generate(
        "hello",
        model=LOCAL_MODEL,
        temperature=0.4,
        max_output_tokens=7,
    )

    assert captured["url"] == "http://127.0.0.1:4567/v1/chat/completions"
    assert captured["json"]["chat_template_kwargs"] == {"enable_thinking": False}
    assert captured["json"]["top_p"] == 0.8
    assert captured["json"]["top_k"] == 20
    assert captured["json"]["min_p"] == 0.0
    assert captured["json"]["presence_penalty"] == 1.5


def test_run_generate_bundled_request_body_matches_golden(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _bundled_endpoint)
    _patch_bundled_server(monkeypatch)
    captured = {}

    def fake_post(url, **kwargs):
        if str(url).endswith("/tokenize"):
            return _FakeRejectEndpoint().post(url, **kwargs)
        if str(url).endswith("/v1/chat/completions"):
            captured.update({"url": url, **kwargs})
            return _ChatResponse("ok")
        raise AssertionError(f"unexpected local provider URL: {url}")

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    provider.run_generate(
        "hello",
        model=LOCAL_MODEL,
        temperature=0.4,
        max_output_tokens=7,
    )

    assert captured["json"] == {
        "model": LOCAL_MODEL,
        "messages": [{"role": "user", "content": "hello"}],
        "temperature": 0.4,
        "max_tokens": 7,
        "stream": False,
        "chat_template_kwargs": {"enable_thinking": False},
        "top_p": 0.8,
        "top_k": 20,
        "min_p": 0.0,
        "presence_penalty": 1.5,
    }


def test_run_generate_bundled_encodes_image_once(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _bundled_endpoint)
    _patch_bundled_server(monkeypatch)
    png = b"\x89PNG\r\n\x1a\npayload"
    calls = []
    captured = {}

    def count_encode(part):
        calls.append(part)
        return "image/png", base64.b64encode(b"encoded").decode("ascii")

    class TokenResponse:
        text = ""

        def raise_for_status(self):
            return None

        def json(self):
            return {"tokens": [1]}

    class ChatResponse:
        text = ""

        def raise_for_status(self):
            return None

        def json(self):
            return {
                "model": LOCAL_MODEL,
                "choices": [
                    {
                        "message": {"content": "ok"},
                        "finish_reason": "stop",
                    }
                ],
            }

    def fake_post(url, json, timeout):
        del timeout
        if url.endswith("/tokenize"):
            return TokenResponse()
        if url.endswith("/v1/chat/completions"):
            captured["body"] = json
            return ChatResponse()
        raise AssertionError(f"unexpected local provider URL: {url}")

    import httpx

    monkeypatch.setattr(provider, "encode_image_part", count_encode)
    monkeypatch.setattr(httpx, "post", fake_post)

    result = provider.run_generate(["look", png], model=LOCAL_MODEL)

    assert result["text"] == "ok"
    assert "body" in captured
    assert len(calls) == 1
    assert calls == [png]


def test_run_generate_byo_posts_to_normalized_endpoint_and_skips_connect(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)
    from solstone.think.providers import local_budget

    def fail_count(*_args, **_kwargs):
        raise AssertionError("count_tokens not expected")

    monkeypatch.setattr(local_budget, "count_tokens", fail_count)
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: (_ for _ in ()).throw(AssertionError("connect not expected")),
    )
    captured = {}

    class Response:
        def raise_for_status(self):
            return None

        def json(self):
            return {
                "choices": [
                    {
                        "message": {"content": "hello"},
                        "finish_reason": "stop",
                    }
                ],
            }

    def fake_post(url, **kwargs):
        captured.update({"url": url, **kwargs})
        return Response()

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    result = provider.run_generate("hello", model=LOCAL_MODEL)

    assert captured["url"] == "http://byo.example/openai/v1/chat/completions"
    assert captured["json"]["model"] == "served-model"
    for key in (
        "chat_template_kwargs",
        "top_p",
        "top_k",
        "min_p",
        "presence_penalty",
    ):
        assert key not in captured["json"]
    assert captured["headers"] == {"Authorization": "Bearer test-token-PLACEHOLDER"}
    assert local_budget.TRUNCATION_MARKER not in str(captured["json"])
    assert result["text"] == "hello"


def test_run_generate_byo_acquires_and_releases_permit(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(
        provider,
        "resolve_local_endpoint",
        lambda: _byo_endpoint(parallel_slots=1),
    )
    monkeypatch.setattr(
        provider,
        "resolve_endpoint_served_window",
        lambda _endpoint: None,
    )
    captured = {}

    def fake_post(url, **kwargs):
        captured.update({"url": url, **kwargs})
        return _ChatResponse()

    import httpx

    from solstone.think.providers import local_admission

    monkeypatch.setattr(httpx, "post", fake_post)

    result = provider.run_generate("hello", model=LOCAL_MODEL)

    assert result["text"] == "hello"
    assert captured["url"] == "http://byo.example/openai/v1/chat/completions"
    with local_admission.acquire_local_slot(1, 0.1) as permit:
        assert permit.slot_index == 0


def test_run_generate_byo_queue_timeout_preserves_exact_type_and_skips_post(
    monkeypatch,
):
    provider = _provider()
    monkeypatch.setattr(
        provider,
        "resolve_local_endpoint",
        lambda: _byo_endpoint(parallel_slots=1),
    )
    monkeypatch.setattr(
        provider,
        "resolve_endpoint_served_window",
        lambda _endpoint: None,
    )

    def fake_post(*_args, **_kwargs):
        raise AssertionError("httpx.post must not run after queue timeout")

    import httpx

    from solstone.think.providers import local_admission

    monkeypatch.setattr(httpx, "post", fake_post)

    holder = local_admission.acquire_local_slot(1, 0.1)
    try:
        with pytest.raises(local_admission.LocalAdmissionTimeout) as exc:
            provider.run_generate("hello", model=LOCAL_MODEL, timeout_s=0.03)
        assert exc.type is local_admission.LocalAdmissionTimeout
        assert exc.value.reason_code == "local_queue_timeout"
    finally:
        holder.release()


def test_run_generate_byo_http_timeout_uses_remaining_deadline(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(
        provider,
        "resolve_local_endpoint",
        lambda: _byo_endpoint(parallel_slots=1),
    )
    monkeypatch.setattr(
        provider,
        "resolve_endpoint_served_window",
        lambda _endpoint: None,
    )
    captured = {}
    times = iter([100.0, 100.0, 100.35])

    def fake_monotonic():
        try:
            return next(times)
        except StopIteration:
            return 100.35

    def fake_acquire(capacity, timeout_s, *, exclusive=False):
        captured["capacity"] = capacity
        captured["permit_timeout"] = timeout_s
        captured["exclusive"] = exclusive
        return contextlib.nullcontext()

    def fake_post(url, **kwargs):
        captured.update({"url": url, **kwargs})
        return _ChatResponse()

    import contextlib

    import httpx

    from solstone.think.providers import local_admission

    monkeypatch.setattr(provider.time, "monotonic", fake_monotonic)
    monkeypatch.setattr(local_admission, "acquire_local_slot", fake_acquire)
    monkeypatch.setattr(httpx, "post", fake_post)

    provider.run_generate("hello", model=LOCAL_MODEL, timeout_s=1.0)

    assert captured["capacity"] == 1
    assert captured["permit_timeout"] == pytest.approx(1.0)
    assert captured["exclusive"] is False
    assert captured["timeout"] == pytest.approx(0.65)


def test_run_generate_confidential_with_stray_slots_skips_admission(monkeypatch):
    provider = _provider()
    configured_endpoint = "https://spp.example.test"
    config = {
        "providers": {
            "local": {
                "endpoint_url": configured_endpoint,
                "served_model_id": "confidential-model",
                "credential": "confidential-token",
                "parallel_slots": 1,
            }
        },
        "services": {"confidential": {"account_id": "acct"}},
    }
    captured = {}
    current_time = {"value": 100.0}

    def fake_monotonic():
        return current_time["value"]

    def fail_acquire(*_args, **_kwargs):
        raise AssertionError("confidential generate must not acquire admission")

    def fake_post(url, **kwargs):
        captured.update({"url": url, **kwargs})
        return _ChatResponse()

    def fake_egress_base_url(base_url):
        current_time["value"] += 2.0
        return "http://127.0.0.1:4567" if base_url == configured_endpoint else base_url

    import httpx

    from solstone.think.providers import local_admission, local_endpoint

    monkeypatch.setattr(provider.time, "monotonic", fake_monotonic)
    monkeypatch.setattr(local_endpoint, "read_journal_config", lambda: config)
    monkeypatch.setattr(local_admission, "acquire_local_slot", fail_acquire)
    monkeypatch.setattr(
        "solstone.think.services.spp_transport.confidential_egress_base_url",
        fake_egress_base_url,
    )
    monkeypatch.setattr(httpx, "post", fake_post)

    result = provider.run_generate("hello", model=LOCAL_MODEL, timeout_s=1.0)

    assert result["text"] == "hello"
    assert captured["url"] == "http://127.0.0.1:4567/v1/chat/completions"
    assert captured["timeout"] == pytest.approx(1.0)


def test_run_generate_confidential_posts_to_forwarder_not_configured_endpoint(
    monkeypatch,
):
    provider = _provider()

    from solstone.think.providers import local_budget
    from solstone.think.providers.local_endpoint import LocalEndpoint

    configured_endpoint = "https://spp.example.test"
    forwarder = "http://127.0.0.1:4567"
    monkeypatch.setattr(
        provider,
        "resolve_local_endpoint",
        lambda: LocalEndpoint(
            base_url=configured_endpoint,
            served_model_id="confidential-model",
            credential="confidential-token",
            is_bundled=False,
        ),
    )
    monkeypatch.setattr(
        "solstone.think.services.spp_transport.confidential_egress_base_url",
        lambda base_url: forwarder if base_url == configured_endpoint else base_url,
    )
    monkeypatch.setattr(local_budget, "count_tokens", lambda *_args, **_kwargs: 1)
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: (_ for _ in ()).throw(AssertionError("connect not expected")),
    )
    captured = {}

    class Response:
        def raise_for_status(self):
            return None

        def json(self):
            return {
                "choices": [
                    {
                        "message": {"content": "hello"},
                        "finish_reason": "stop",
                    }
                ],
            }

    def fake_post(url, **kwargs):
        captured.update({"url": url, **kwargs})
        return Response()

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    result = provider.run_generate("hello", model=LOCAL_MODEL)

    assert result["text"] == "hello"
    assert captured["url"] == f"{forwarder}/v1/chat/completions"
    assert configured_endpoint not in captured["url"]


def test_run_agenerate_confidential_posts_to_forwarder_not_configured_endpoint(
    monkeypatch,
):
    provider = _provider()

    from solstone.think.providers.local_endpoint import LocalEndpoint

    configured_endpoint = "https://spp.example.test"
    forwarder = "http://127.0.0.1:4567"
    monkeypatch.setattr(
        provider,
        "resolve_local_endpoint",
        lambda: LocalEndpoint(
            base_url=configured_endpoint,
            served_model_id="confidential-model",
            credential="confidential-token",
            is_bundled=False,
        ),
    )
    monkeypatch.setattr(
        "solstone.think.services.spp_transport.confidential_egress_base_url",
        lambda base_url: forwarder if base_url == configured_endpoint else base_url,
    )
    captured = {}

    class Response:
        def raise_for_status(self):
            return None

        def json(self):
            return {
                "choices": [
                    {
                        "message": {"content": "hello"},
                        "finish_reason": "stop",
                    }
                ],
            }

    class AsyncClient:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *_args):
            return None

        async def post(self, url, **kwargs):
            captured.update({"url": url, **kwargs})
            return Response()

    import httpx

    monkeypatch.setattr(httpx, "AsyncClient", AsyncClient)

    result = asyncio.run(provider.run_agenerate("hello", model=LOCAL_MODEL))

    assert result["text"] == "hello"
    assert captured["url"] == f"{forwarder}/v1/chat/completions"
    assert configured_endpoint not in captured["url"]


def test_run_agenerate_byo_acquires_and_releases_permit(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(
        provider,
        "resolve_local_endpoint",
        lambda: _byo_endpoint(parallel_slots=1),
    )
    monkeypatch.setattr(
        provider,
        "resolve_endpoint_served_window",
        lambda _endpoint: None,
    )
    captured = {}

    class AsyncClient:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *_args):
            return None

        async def post(self, url, **kwargs):
            captured.update({"url": url, **kwargs})
            return _ChatResponse()

    import httpx

    from solstone.think.providers import local_admission

    monkeypatch.setattr(httpx, "AsyncClient", AsyncClient)

    result = asyncio.run(provider.run_agenerate("hello", model=LOCAL_MODEL))

    assert result["text"] == "hello"
    assert captured["url"] == "http://byo.example/openai/v1/chat/completions"
    with local_admission.acquire_local_slot(1, 0.1) as permit:
        assert permit.slot_index == 0


def test_run_agenerate_byo_queue_timeout_preserves_exact_type_and_skips_post(
    monkeypatch,
):
    provider = _provider()
    monkeypatch.setattr(
        provider,
        "resolve_local_endpoint",
        lambda: _byo_endpoint(parallel_slots=1),
    )
    entered = False

    class AsyncClient:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *_args):
            return None

        async def post(self, *_args, **_kwargs):
            nonlocal entered
            entered = True
            raise AssertionError("AsyncClient.post must not run after queue timeout")

    import httpx

    from solstone.think.providers import local_admission

    monkeypatch.setattr(httpx, "AsyncClient", AsyncClient)

    holder = local_admission.acquire_local_slot(1, 0.1)
    try:
        with pytest.raises(local_admission.LocalAdmissionTimeout) as exc:
            asyncio.run(
                provider.run_agenerate("hello", model=LOCAL_MODEL, timeout_s=0.03)
            )
        assert exc.type is local_admission.LocalAdmissionTimeout
        assert exc.value.reason_code == "local_queue_timeout"
        assert entered is False
    finally:
        holder.release()


def test_run_agenerate_byo_http_timeout_uses_remaining_deadline(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(
        provider,
        "resolve_local_endpoint",
        lambda: _byo_endpoint(parallel_slots=1),
    )
    captured = {}
    after_wait = False

    def fake_monotonic():
        return 200.25 if after_wait else 200.0

    async def fake_acquire(capacity, timeout_s, *, exclusive=False):
        nonlocal after_wait
        captured["capacity"] = capacity
        captured["permit_timeout"] = timeout_s
        captured["exclusive"] = exclusive
        after_wait = True
        return contextlib.nullcontext()

    class AsyncClient:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *_args):
            return None

        async def post(self, url, **kwargs):
            captured.update({"url": url, **kwargs})
            return _ChatResponse()

    import contextlib

    import httpx

    from solstone.think.providers import local_admission

    monkeypatch.setattr(provider.time, "monotonic", fake_monotonic)
    monkeypatch.setattr(local_admission, "acquire_local_slot_async", fake_acquire)
    monkeypatch.setattr(httpx, "AsyncClient", AsyncClient)

    asyncio.run(provider.run_agenerate("hello", model=LOCAL_MODEL, timeout_s=1.0))

    assert captured["capacity"] == 1
    assert captured["permit_timeout"] == pytest.approx(1.0)
    assert captured["exclusive"] is False
    assert captured["timeout"] == pytest.approx(0.75)


def test_run_agenerate_confidential_with_stray_slots_skips_admission(monkeypatch):
    provider = _provider()
    configured_endpoint = "https://spp.example.test"
    config = {
        "providers": {
            "local": {
                "endpoint_url": configured_endpoint,
                "served_model_id": "confidential-model",
                "credential": "confidential-token",
                "parallel_slots": 1,
            }
        },
        "services": {"confidential": {"account_id": "acct"}},
    }
    captured = {}
    current_time = {"value": 200.0}

    def fake_monotonic():
        return current_time["value"]

    async def fail_acquire(*_args, **_kwargs):
        raise AssertionError("confidential agenerate must not acquire admission")

    def fake_egress_base_url(base_url):
        current_time["value"] += 2.0
        return "http://127.0.0.1:4567" if base_url == configured_endpoint else base_url

    class AsyncClient:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *_args):
            return None

        async def post(self, url, **kwargs):
            captured.update({"url": url, **kwargs})
            return _ChatResponse()

    import httpx

    from solstone.think.providers import local_admission, local_endpoint

    monkeypatch.setattr(provider.time, "monotonic", fake_monotonic)
    monkeypatch.setattr(local_endpoint, "read_journal_config", lambda: config)
    monkeypatch.setattr(local_admission, "acquire_local_slot_async", fail_acquire)
    monkeypatch.setattr(
        "solstone.think.services.spp_transport.confidential_egress_base_url",
        fake_egress_base_url,
    )
    monkeypatch.setattr(httpx, "AsyncClient", AsyncClient)

    result = asyncio.run(
        provider.run_agenerate("hello", model=LOCAL_MODEL, timeout_s=1.0)
    )

    assert result["text"] == "hello"
    assert captured["url"] == "http://127.0.0.1:4567/v1/chat/completions"
    assert captured["timeout"] == pytest.approx(1.0)


def test_run_generate_byo_body_omits_bundled_request_fields(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)
    captured_posts = []

    class Response:
        def raise_for_status(self):
            return None

        def json(self):
            return {
                "choices": [
                    {
                        "message": {"content": "{}"},
                        "finish_reason": "stop",
                    }
                ],
            }

    def fake_post(url, **kwargs):
        captured_posts.append({"url": url, **kwargs})
        return Response()

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)
    schema = {
        "type": "object",
        "properties": {
            "items": {"type": "array", "items": {"type": "string"}},
        },
    }

    provider.run_generate(
        "hello",
        model=LOCAL_MODEL,
        temperature=0.4,
        max_output_tokens=7,
    )
    provider.run_generate(
        "hello",
        model=LOCAL_MODEL,
        temperature=0.5,
        max_output_tokens=11,
        json_schema=schema,
    )

    assert len(captured_posts) == 2
    assert captured_posts[0]["json"] == {
        "model": "served-model",
        "messages": [{"role": "user", "content": "hello"}],
        "temperature": 0.4,
        "max_tokens": 7,
        "stream": False,
    }
    assert captured_posts[1]["json"] == {
        "model": "served-model",
        "messages": [{"role": "user", "content": "hello"}],
        "temperature": 0.5,
        "max_tokens": 11,
        "stream": False,
        "response_format": {
            "type": "json_schema",
            "json_schema": {
                "name": "local_schema",
                "schema": provider._prepare_local_schema(schema),
                "strict": True,
            },
        },
    }
    for post in captured_posts:
        for key in (
            "chat_template_kwargs",
            "top_p",
            "top_k",
            "min_p",
            "presence_penalty",
        ):
            assert key not in post["json"]


def test_run_generate_byo_omits_auth_header_without_credential(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", lambda: _byo_endpoint(None))
    captured = {}

    class Response:
        def raise_for_status(self):
            return None

        def json(self):
            return {
                "choices": [{"message": {"content": "ok"}, "finish_reason": "stop"}]
            }

    def fake_post(url, **kwargs):
        captured.update({"url": url, **kwargs})
        return Response()

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    provider.run_generate("hello", model=LOCAL_MODEL)

    assert "headers" not in captured


def _load_schema(path: str) -> dict:
    return json.loads(Path(path).read_text(encoding="utf-8"))


def _sense_collection_bounds(schema: dict) -> dict[str, int]:
    properties = schema["properties"]
    return {
        "entities": properties["entities"]["maxItems"],
        "facets": properties["facets"]["maxItems"],
        "speakers": properties["speakers"]["maxItems"],
    }


def test_generate_schema_files_declare_only_safe_sense_collection_bounds():
    bounded_keys = {
        "minItems",
        "maxItems",
        "minLength",
        "maxLength",
        "minimum",
        "maximum",
    }
    sense = _load_schema("solstone/talent/sense.schema.json")

    assert _sense_collection_bounds(sense) == {
        "entities": 96,
        "facets": 16,
        "speakers": 16,
    }
    assert _schema_keyword_paths(sense, {"maxItems"}) == [
        "$/properties/entities/maxItems",
        "$/properties/facets/maxItems",
        "$/properties/speakers/maxItems",
    ]
    assert _schema_keyword_paths(sense, {"pattern", "minLength", "maxLength"}) == []

    for path in (
        "solstone/talent/participation.schema.json",
        "solstone/talent/participation_entry.schema.json",
    ):
        assert _schema_keyword_paths(_load_schema(path), bounded_keys) == []


def test_sense_collection_bounds_survive_runtime_and_local_schema_prep():
    from solstone.think.talent import hydrate_runtime_enums

    provider = _provider()
    sense = _load_schema("solstone/talent/sense.schema.json")

    hydrated = hydrate_runtime_enums(sense)
    prepared = provider._prepare_local_schema(hydrated)

    assert _sense_collection_bounds(hydrated) == {
        "entities": 96,
        "facets": 16,
        "speakers": 16,
    }
    assert _sense_collection_bounds(prepared) == {
        "entities": 96,
        "facets": 16,
        "speakers": 16,
    }


def test_local_input_budget_reserve_for_changed_caps():
    from solstone.think.providers.local_budget import compute_input_budget

    floor = 16384
    capable = 32768

    assert compute_input_budget(512, floor) - compute_input_budget(1024, floor) == 512
    assert compute_input_budget(2048, floor) - compute_input_budget(4096, floor) == 2048
    assert compute_input_budget(12288, floor) == compute_input_budget(6144, floor)
    assert compute_input_budget(12288, floor) == floor - 4096 - 256
    assert compute_input_budget(12288, capable) == capable - 8192 - 256
    assert compute_input_budget(6144, capable) == capable - 6144 - 256


def test_run_generate_byo_network_error_maps_to_unreachable(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)

    import httpx

    monkeypatch.setattr(
        httpx,
        "post",
        lambda *args, **kwargs: (_ for _ in ()).throw(
            httpx.ConnectError("connection refused")
        ),
    )

    with pytest.raises(provider.LocalProviderError) as exc:
        provider.run_generate("hello", model=LOCAL_MODEL)

    assert exc.value.reason_code == "local_endpoint_unreachable"
    assert str(exc.value) == provider.LOCAL_ENDPOINT_UNREACHABLE_COPY
    assert isinstance(exc.value.__cause__, httpx.ConnectError)


@pytest.mark.parametrize(
    "exc_name",
    [
        "ReadTimeout",
        "PoolTimeout",
        "WriteTimeout",
        "TimeoutException",
    ],
)
def test_classify_byo_generate_error_capacity_names_are_non_blocking(exc_name):
    provider = _provider()
    from solstone.convey.provider_readiness import is_blocking_reason

    inner = type(exc_name, (Exception,), {})(f"{exc_name} failed")
    exc = RuntimeError("outer")
    exc.__cause__ = inner

    classified = provider._classify_byo_generate_error(exc, _byo_endpoint())

    assert classified.reason_code == "local_capacity_exhausted"
    assert is_blocking_reason(classified.reason_code) is False


@pytest.mark.parametrize(
    "exc_name",
    [
        "ConnectError",
        "APIConnectionError",
        "ConnectTimeout",
        "NetworkError",
        "RequestError",
    ],
)
def test_classify_byo_generate_error_unreachable_names_are_blocking(exc_name):
    provider = _provider()
    from solstone.convey.provider_readiness import is_blocking_reason

    inner = type(exc_name, (Exception,), {})(f"{exc_name} failed")
    exc = RuntimeError("outer")
    exc.__cause__ = inner

    classified = provider._classify_byo_generate_error(exc, _byo_endpoint())

    assert classified.reason_code == "local_endpoint_unreachable"
    assert str(classified) == provider.LOCAL_ENDPOINT_UNREACHABLE_COPY
    assert is_blocking_reason(classified.reason_code) is True


def test_classify_byo_generate_error_capacity_wins_mixed_chain():
    provider = _provider()

    capacity = type("ReadTimeout", (Exception,), {})("read timeout")
    unreachable = type("ConnectError", (Exception,), {})("connect failed")
    outer = RuntimeError("outer")
    outer.__cause__ = unreachable
    unreachable.__cause__ = capacity

    classified = provider._classify_byo_generate_error(outer, _byo_endpoint())

    assert classified.reason_code == "local_capacity_exhausted"


def test_classify_byo_generate_error_500_stays_contract_failed():
    provider = _provider()

    class InternalServerError(Exception):
        status_code = 500

    classified = provider._classify_byo_generate_error(
        InternalServerError("server failed"),
        _byo_endpoint(),
    )

    assert classified.reason_code == "local_endpoint_contract_failed"
    assert str(classified) == provider.LOCAL_ENDPOINT_CONTRACT_COPY


@pytest.mark.parametrize(
    "body",
    [
        _FAKE_F1_COMPLETION_OVERFLOW_BODY,
        _FAKE_F2_PROMPT_OVERFLOW_BODY,
    ],
)
def test_classify_byo_generate_error_context_body_maps_to_context_window(body):
    provider = _provider()
    endpoint = _byo_endpoint()

    classified = provider._classify_byo_generate_error(
        _http_status_error(body),
        endpoint,
    )

    assert classified.reason_code == "context_window_exceeded"
    assert (
        str(classified)
        == "The configured endpoint rejected the request: prompt and completion "
        "exceed the served context window."
    )


@pytest.mark.parametrize(
    "body",
    [
        _FAKE_F1_COMPLETION_OVERFLOW_BODY,
        _FAKE_F2_PROMPT_OVERFLOW_BODY,
    ],
)
def test_classify_byo_cogitate_error_context_body_maps_to_context_window(body):
    from solstone.think.providers import local_endpoint

    class BadRequestError(RuntimeError):
        status_code = 400

        def __init__(self, message: str, payload: str) -> None:
            super().__init__(message)
            self.message = message
            self.body = payload

    exc = BadRequestError("litellm bad request wrapper", body)

    assert local_endpoint.classify_byo_cogitate_error(exc) == "context_window_exceeded"


def test_run_generate_byo_http_status_maps_to_contract_failed(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)

    import httpx

    request = httpx.Request("POST", "http://byo.example/openai/v1/chat/completions")
    response = httpx.Response(400, request=request)

    class Response:
        def raise_for_status(self):
            raise httpx.HTTPStatusError(
                "bad request", request=request, response=response
            )

    monkeypatch.setattr(httpx, "post", lambda *args, **kwargs: Response())

    with pytest.raises(provider.LocalProviderError) as exc:
        provider.run_generate("hello", model=LOCAL_MODEL)

    assert exc.value.reason_code == "local_endpoint_contract_failed"
    assert str(exc.value) == provider.LOCAL_ENDPOINT_CONTRACT_COPY


def test_run_generate_byo_invalid_shape_maps_to_response_invalid(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)

    class Response:
        def raise_for_status(self):
            return None

        def json(self):
            return {"choices": []}

    import httpx

    monkeypatch.setattr(httpx, "post", lambda *args, **kwargs: Response())

    with pytest.raises(provider.LocalProviderError) as exc:
        provider.run_generate("hello", model=LOCAL_MODEL)

    assert exc.value.reason_code == "provider_response_invalid"
    assert str(exc.value) == "No response from model."


def test_run_agenerate_byo_invalid_shape_maps_to_response_invalid(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)

    class Response:
        def raise_for_status(self):
            return None

        def json(self):
            return {"choices": []}

    class AsyncClient:
        async def __aenter__(self):
            return self

        async def __aexit__(self, exc_type, exc, tb):
            return None

        async def post(self, *_args, **_kwargs):
            return Response()

    import httpx

    monkeypatch.setattr(httpx, "AsyncClient", AsyncClient)

    with pytest.raises(provider.LocalProviderError) as exc:
        asyncio.run(provider.run_agenerate("hello", model=LOCAL_MODEL))

    assert exc.value.reason_code == "provider_response_invalid"
    assert str(exc.value) == "No response from model."


def test_run_generate_byo_json_decode_maps_to_contract_failed(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)

    class Response:
        def raise_for_status(self):
            return None

        def json(self):
            raise json.JSONDecodeError("bad json", "not-json", 0)

    import httpx

    monkeypatch.setattr(httpx, "post", lambda *args, **kwargs: Response())

    with pytest.raises(provider.LocalProviderError) as exc:
        provider.run_generate("hello", model=LOCAL_MODEL)

    assert exc.value.reason_code == "local_endpoint_contract_failed"
    assert str(exc.value) == provider.LOCAL_ENDPOINT_CONTRACT_COPY


def test_run_generate_malformed_response_matches_bundled_and_byo(monkeypatch):
    provider = _provider()
    malformed = {"choices": []}

    class Response:
        def raise_for_status(self):
            return None

        def json(self):
            return malformed

    def fake_post(*_args, **_kwargs):
        return Response()

    import httpx

    monkeypatch.setattr(httpx, "post", fake_post)

    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)
    with pytest.raises(provider.LocalProviderError) as byo_exc:
        provider.run_generate("hello", model=LOCAL_MODEL)

    monkeypatch.setattr(provider, "resolve_local_endpoint", _bundled_endpoint)
    _patch_bundled_server(monkeypatch)
    with pytest.raises(provider.LocalProviderError) as bundled_exc:
        provider.run_generate("hello", model=LOCAL_MODEL)

    assert byo_exc.value.reason_code == bundled_exc.value.reason_code
    assert byo_exc.value.reason_code == "provider_response_invalid"
    assert str(byo_exc.value) == str(bundled_exc.value)


def test_bundled_local_admission_timeout_reason_code_escapes(monkeypatch):
    provider = _provider()
    _patch_bundled_server(monkeypatch)

    from solstone.think.providers import local_admission

    def raise_timeout(*_args, **_kwargs):
        raise local_admission.LocalAdmissionTimeout("busy")

    async def raise_timeout_async(*_args, **_kwargs):
        raise local_admission.LocalAdmissionTimeout("busy")

    monkeypatch.setattr(local_admission, "acquire_local_slot", raise_timeout)
    monkeypatch.setattr(
        local_admission, "acquire_local_slot_async", raise_timeout_async
    )

    with pytest.raises(local_admission.LocalAdmissionTimeout) as sync_exc:
        provider.run_generate("hello", model=LOCAL_MODEL)
    assert sync_exc.value.reason_code == "local_queue_timeout"

    with pytest.raises(local_admission.LocalAdmissionTimeout) as async_exc:
        asyncio.run(provider.run_agenerate("hello", model=LOCAL_MODEL))
    assert async_exc.value.reason_code == "local_queue_timeout"

    with pytest.raises(local_admission.LocalAdmissionTimeout) as cogitate_exc:
        asyncio.run(provider.run_cogitate({"model": LOCAL_MODEL}))
    assert cogitate_exc.value.reason_code == "local_queue_timeout"


def test_run_cogitate_byo_acquires_permit_and_records_no_telemetry(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(
        provider,
        "resolve_local_endpoint",
        lambda: _byo_endpoint(parallel_slots=1),
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: (_ for _ in ()).throw(AssertionError("connect not expected")),
    )

    from solstone.think.providers import local_admission

    records = []
    monkeypatch.setattr(local_admission, "record_local_inference", records.append)

    async def fake_cogitate(*_args, slot_lease=None, **_kwargs):
        assert slot_lease is not None
        slot_lease.yield_slot()
        with local_admission.acquire_local_slot(1, 0.1) as nested:
            assert nested.slot_index == 0
        slot_lease.reacquire()
        with pytest.raises(local_admission.LocalAdmissionTimeout):
            local_admission.acquire_local_slot(1, 0.03)
        return "ok"

    monkeypatch.setattr(
        "solstone.think.providers.openhands.run_cogitate",
        fake_cogitate,
    )

    result = asyncio.run(
        provider.run_cogitate({"model": LOCAL_MODEL, "timeout_seconds": 1})
    )

    assert result == "ok"
    assert records == []
    with local_admission.acquire_local_slot(1, 0.1) as permit:
        assert permit.slot_index == 0


def test_run_cogitate_local_delegated_non_responsive_single_event(
    monkeypatch,
):
    provider = _provider()
    monkeypatch.setattr(
        provider,
        "resolve_local_endpoint",
        lambda: _byo_endpoint(parallel_slots=1),
    )
    terminal_event = {
        "event": "error",
        "error": "non-responsive output",
        "reason_code": NON_RESPONSIVE_REASON_CODE,
        "provider": "local",
        "terminal": True,
        "raw": [{"reason_code": NON_RESPONSIVE_REASON_CODE}],
    }

    async def fake_cogitate(*_args, on_event=None, slot_lease=None, **_kwargs):
        assert slot_lease is not None
        on_event(terminal_event)
        return None

    monkeypatch.setattr(
        "solstone.think.providers.openhands.run_cogitate",
        fake_cogitate,
    )
    events: list[dict] = []

    result = asyncio.run(
        provider.run_cogitate(
            {"model": LOCAL_MODEL, "timeout_seconds": 1},
            on_event=events.append,
        )
    )

    assert result is None
    assert events == [terminal_event]


def test_run_cogitate_byo_keeps_permit_for_non_sol_work(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(
        provider,
        "resolve_local_endpoint",
        lambda: _byo_endpoint(parallel_slots=1),
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: (_ for _ in ()).throw(AssertionError("connect not expected")),
    )

    from solstone.think.providers import local_admission

    async def fake_cogitate(*_args, slot_lease=None, **_kwargs):
        assert slot_lease is not None
        with pytest.raises(local_admission.LocalAdmissionTimeout):
            local_admission.acquire_local_slot(1, 0.03)
        return "ok"

    monkeypatch.setattr(
        "solstone.think.providers.openhands.run_cogitate",
        fake_cogitate,
    )

    assert (
        asyncio.run(provider.run_cogitate({"model": LOCAL_MODEL, "timeout_seconds": 1}))
        == "ok"
    )


@pytest.mark.parametrize("bundled", [False, True])
def test_run_cogitate_reacquire_timeout_preserves_exact_type(
    monkeypatch,
    bundled,
):
    provider = _provider()
    if bundled:
        _patch_bundled_server(monkeypatch)
    else:
        monkeypatch.setattr(
            provider,
            "resolve_local_endpoint",
            lambda: _byo_endpoint(parallel_slots=1),
        )
        monkeypatch.setattr(
            "solstone.think.providers.local_server.connect",
            lambda: (_ for _ in ()).throw(AssertionError("connect not expected")),
        )

    from solstone.think.providers import local_admission

    async def fake_cogitate(*_args, slot_lease=None, **_kwargs):
        assert slot_lease is not None
        slot_lease.yield_slot()
        holder = local_admission.acquire_local_slot(1, 0.1)
        try:
            slot_lease.reacquire()
        finally:
            holder.release()

    monkeypatch.setattr(
        "solstone.think.providers.openhands.run_cogitate",
        fake_cogitate,
    )

    with pytest.raises(local_admission.LocalAdmissionTimeout) as exc:
        asyncio.run(
            provider.run_cogitate({"model": LOCAL_MODEL, "timeout_seconds": 0.03})
        )

    assert exc.type is local_admission.LocalAdmissionTimeout
    assert exc.value.reason_code == "local_queue_timeout"
    assert not list(Path(local_admission._admission_dir()).glob("wait-*.ticket"))


def test_run_cogitate_byo_queue_timeout_preserves_exact_type(monkeypatch):
    provider = _provider()
    monkeypatch.setattr(
        provider,
        "resolve_local_endpoint",
        lambda: _byo_endpoint(parallel_slots=1),
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: (_ for _ in ()).throw(AssertionError("connect not expected")),
    )

    async def fail_if_called(*_args, **_kwargs):
        raise AssertionError("openhands must not run after queue timeout")

    from solstone.think.providers import local_admission

    monkeypatch.setattr(
        "solstone.think.providers.openhands.run_cogitate",
        fail_if_called,
    )

    holder = local_admission.acquire_local_slot(1, 0.1)
    try:
        with pytest.raises(local_admission.LocalAdmissionTimeout) as exc:
            asyncio.run(
                provider.run_cogitate({"model": LOCAL_MODEL, "timeout_seconds": 0.03})
            )
        assert exc.type is local_admission.LocalAdmissionTimeout
        assert exc.value.reason_code == "local_queue_timeout"
    finally:
        holder.release()


def test_run_cogitate_confidential_with_stray_slots_skips_admission(monkeypatch):
    provider = _provider()
    configured_endpoint = "https://spp.example.test"
    config = {
        "providers": {
            "local": {
                "endpoint_url": configured_endpoint,
                "served_model_id": "confidential-model",
                "credential": "confidential-token",
                "parallel_slots": 1,
            }
        },
        "services": {"confidential": {"account_id": "acct"}},
    }
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: (_ for _ in ()).throw(AssertionError("connect not expected")),
    )

    from solstone.think.providers import local_admission, local_endpoint

    async def fail_acquire(*_args, **_kwargs):
        raise AssertionError("confidential cogitate must not acquire admission")

    async def fake_cogitate(*_args, **_kwargs):
        return "ok"

    monkeypatch.setattr(local_endpoint, "read_journal_config", lambda: config)
    monkeypatch.setattr(local_admission, "acquire_local_slot_async", fail_acquire)
    monkeypatch.setattr(
        "solstone.think.providers.openhands.run_cogitate",
        fake_cogitate,
    )

    assert asyncio.run(provider.run_cogitate({"model": LOCAL_MODEL})) == "ok"


def test_run_generate_bundled_context_budget_exceeded_reason_code_escapes(
    monkeypatch,
):
    provider = _provider()
    _patch_bundled_server(monkeypatch)

    def raise_context_budget(**_kwargs):
        raise provider.ContextBudgetExceeded("too large")

    monkeypatch.setattr(provider, "_prepare_bundled_request", raise_context_budget)

    with pytest.raises(provider.ContextBudgetExceeded) as exc:
        provider.run_generate("hello", model=LOCAL_MODEL)

    assert exc.value.reason_code == "context_budget_exceeded"


def test_run_cogitate_byo_classified_error_uses_fixed_copy_and_redacts(
    monkeypatch,
):
    provider = _provider()
    token = "test-token-PLACEHOLDER"
    events: list[dict] = []

    class BadRequestError(RuntimeError):
        status_code = 400

    async def fail_cogitate(*_args, **_kwargs):
        raise BadRequestError(f"bad request with {token}")

    monkeypatch.setattr(
        provider, "resolve_local_endpoint", lambda: _byo_endpoint(token)
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: (_ for _ in ()).throw(AssertionError("connect not expected")),
    )
    monkeypatch.setattr(
        "solstone.think.providers.openhands.run_cogitate",
        fail_cogitate,
    )

    with pytest.raises(provider.LocalProviderError) as exc:
        asyncio.run(
            provider.run_cogitate({"model": LOCAL_MODEL}, on_event=events.append)
        )

    assert exc.value.reason_code == "local_endpoint_contract_failed"
    assert str(exc.value) == provider.LOCAL_ENDPOINT_CONTRACT_COPY
    assert token not in str(exc.value)
    assert getattr(exc.value, "_evented") is True
    assert events[0]["error"] == provider.LOCAL_ENDPOINT_CONTRACT_COPY
    assert events[0]["reason_code"] == "local_endpoint_contract_failed"
    assert token not in events[0]["trace"]


def test_run_cogitate_byo_context_error_event_caps_error_field(monkeypatch):
    from solstone.think.providers.shared import PROVIDER_ERROR_TEXT_CAP_CHARS

    provider = _provider()
    token = "SENTINEL-BYO-CONTEXT-CRED-219a"
    events: list[dict] = []

    class BadRequestError(RuntimeError):
        status_code = 400

        def __init__(self) -> None:
            body = _FAKE_F2_PROMPT_OVERFLOW_BODY + f" {token}" + ("x" * 6000)
            super().__init__(body)
            self.message = body
            self.body = body

    async def fail_cogitate(*_args, **_kwargs):
        raise BadRequestError()

    monkeypatch.setattr(
        provider, "resolve_local_endpoint", lambda: _byo_endpoint(token)
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: (_ for _ in ()).throw(AssertionError("connect not expected")),
    )
    monkeypatch.setattr(
        "solstone.think.providers.openhands.run_cogitate",
        fail_cogitate,
    )

    with pytest.raises(BadRequestError):
        asyncio.run(
            provider.run_cogitate({"model": LOCAL_MODEL}, on_event=events.append)
        )

    assert events[0]["reason_code"] == "context_window_exceeded"
    assert len(events[0]["error"]) <= PROVIDER_ERROR_TEXT_CAP_CHARS
    assert token not in events[0]["error"]
    assert token not in events[0]["trace"]


def test_run_cogitate_byo_connection_error_records_no_success_telemetry(
    monkeypatch,
):
    provider = _provider()
    sentinel = "SENTINEL-BYO-CRED-9f3a2b"
    events: list[dict] = []
    records: list[dict] = []

    monkeypatch.setattr(
        provider,
        "resolve_local_endpoint",
        lambda: _byo_endpoint(sentinel),
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: (_ for _ in ()).throw(AssertionError("connect not expected")),
    )

    from solstone.think.providers import local_admission

    monkeypatch.setattr(local_admission, "record_local_inference", records.append)

    async def fail_cogitate(*_args, **_kwargs):
        import httpx

        from solstone.think.providers.local_endpoint import redact_exception_credential

        exc = httpx.ConnectError(f"connection refused {sentinel}")
        raise redact_exception_credential(exc, sentinel)

    monkeypatch.setattr(
        "solstone.think.providers.openhands.run_cogitate",
        fail_cogitate,
    )

    with pytest.raises(provider.LocalProviderError) as exc:
        asyncio.run(
            provider.run_cogitate({"model": LOCAL_MODEL}, on_event=events.append)
        )

    assert exc.value.reason_code == "local_endpoint_unreachable"
    assert str(exc.value) == provider.LOCAL_ENDPOINT_UNREACHABLE_COPY
    assert sentinel not in str(exc.value)
    serialized = "".join(
        traceback.format_exception(
            type(exc.value),
            exc.value,
            exc.value.__traceback__,
        )
    )
    assert sentinel not in serialized
    assert sentinel not in json.dumps(events)
    assert records == []


def test_run_cogitate_talent_hook_error_bypasses_local_error_event(monkeypatch):
    provider = _provider()
    events: list[dict] = []
    hook_exc = TalentHookError(
        "post",
        "broken_hook",
        "chat",
        RuntimeError("hook exploded"),
    )

    async def fail_cogitate(*_args, **_kwargs):
        raise hook_exc

    monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint)
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: (_ for _ in ()).throw(AssertionError("connect not expected")),
    )
    monkeypatch.setattr(
        "solstone.think.providers.openhands.run_cogitate",
        fail_cogitate,
    )

    with pytest.raises(TalentHookError) as raised:
        asyncio.run(
            provider.run_cogitate({"model": LOCAL_MODEL}, on_event=events.append)
        )

    assert raised.value is hook_exc
    assert events == []
    assert not getattr(hook_exc, "_evented", False)


@pytest.mark.parametrize(
    ("credential", "expected_key"),
    [
        ("test-token-PLACEHOLDER", "test-token-PLACEHOLDER"),
        (None, "EMPTY"),
    ],
)
def test_openhands_local_byo_llm_kwargs(monkeypatch, credential, expected_key):
    from solstone.think.providers import openhands

    captured = {}

    class FakeLLM:
        def __init__(self, **kwargs):
            captured.update(kwargs)

    sdk_module = types.ModuleType("openhands.sdk")
    sdk_module.LLM = FakeLLM
    monkeypatch.setitem(sys.modules, "openhands.sdk", sdk_module)
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: (_ for _ in ()).throw(AssertionError("connect not expected")),
    )

    llm = openhands._build_llm(
        "local",
        LOCAL_MODEL,
        endpoint=_byo_endpoint(credential),
        served_window=None,
    )

    assert isinstance(llm, FakeLLM)
    assert captured == {
        "model": "openai/served-model",
        "base_url": "http://byo.example/openai/v1",
        "api_key": expected_key,
        "native_tool_calling": False,
        "timeout": openhands.LLM_TIMEOUT_S,
        "num_retries": openhands.LLM_NUM_RETRIES,
        "retry_min_wait": 1,
        "retry_max_wait": 2,
        "retry_multiplier": 1.0,
        "input_cost_per_token": 0,
        "output_cost_per_token": 0,
    }
    assert "max_input_tokens" not in captured
    assert "litellm_extra_body" not in captured
    waits = [
        min(
            captured["retry_max_wait"],
            max(
                captured["retry_min_wait"],
                captured["retry_multiplier"] * 2 ** (k - 1),
            ),
        )
        for k in range(1, captured["num_retries"])
    ]
    assert sum(waits) == 1.0
    assert sum(waits) < openhands.WALL_CLOCK_GRACE_S


def test_openhands_local_confidential_llm_uses_forwarder(monkeypatch):
    from solstone.think.providers import local_endpoint, openhands

    configured_endpoint = "https://spp.example.test"
    forwarder = "http://127.0.0.1:4567"
    captured = {}

    class FakeLLM:
        def __init__(self, **kwargs):
            captured.update(kwargs)

    sdk_module = types.ModuleType("openhands.sdk")
    sdk_module.LLM = FakeLLM
    monkeypatch.setitem(sys.modules, "openhands.sdk", sdk_module)
    endpoint = local_endpoint.LocalEndpoint(
        base_url=configured_endpoint,
        served_model_id="confidential-model",
        credential="confidential-token",
        is_bundled=False,
        is_confidential=True,
    )
    monkeypatch.setattr(
        "solstone.think.services.spp_transport.confidential_egress_base_url",
        lambda base_url: forwarder if base_url == configured_endpoint else base_url,
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_server.connect",
        lambda: (_ for _ in ()).throw(AssertionError("connect not expected")),
    )

    llm = openhands._build_llm(
        "local",
        LOCAL_MODEL,
        endpoint=endpoint,
        served_window=None,
    )

    assert isinstance(llm, FakeLLM)
    assert captured["base_url"] == f"{forwarder}/v1"
    assert configured_endpoint not in captured["base_url"]


def test_openhands_local_confidential_known_window_caps_and_extra_body(monkeypatch):
    from solstone.think.providers import local_endpoint, openhands

    captured = {}

    class FakeLLM:
        def __init__(self, **kwargs):
            captured.update(kwargs)

    sdk_module = types.ModuleType("openhands.sdk")
    sdk_module.LLM = FakeLLM
    monkeypatch.setitem(sys.modules, "openhands.sdk", sdk_module)
    endpoint = local_endpoint.LocalEndpoint(
        base_url="https://spp.example.test",
        served_model_id="confidential-model",
        credential="confidential-token",
        is_bundled=False,
        is_confidential=True,
    )
    monkeypatch.setattr(
        "solstone.think.services.spp_transport.confidential_egress_base_url",
        lambda _base_url: "http://127.0.0.1:4567",
    )

    openhands._build_llm(
        "local",
        LOCAL_MODEL,
        endpoint=endpoint,
        served_window=16384,
    )

    assert captured["max_input_tokens"] == 16384
    assert captured["max_output_tokens"] == 4096
    assert captured["litellm_extra_body"] == {
        "chat_template_kwargs": {"enable_thinking": False}
    }


def test_openhands_local_known_window_caps_without_extra_body(monkeypatch):
    from solstone.think.providers import openhands

    captured = {}

    class FakeLLM:
        def __init__(self, **kwargs):
            captured.update(kwargs)

    sdk_module = types.ModuleType("openhands.sdk")
    sdk_module.LLM = FakeLLM
    monkeypatch.setitem(sys.modules, "openhands.sdk", sdk_module)

    openhands._build_llm(
        "local",
        LOCAL_MODEL,
        endpoint=_qwen_byo_endpoint(),
        served_window=32768,
    )

    assert captured["max_input_tokens"] == 32768
    assert captured["max_output_tokens"] == 8192
    assert "litellm_extra_body" not in captured


def test_openhands_local_small_window_omits_caps_and_extra_body(monkeypatch):
    from solstone.think.providers import openhands

    captured = {}

    class FakeLLM:
        def __init__(self, **kwargs):
            captured.update(kwargs)

    sdk_module = types.ModuleType("openhands.sdk")
    sdk_module.LLM = FakeLLM
    monkeypatch.setitem(sys.modules, "openhands.sdk", sdk_module)

    openhands._build_llm(
        "local",
        LOCAL_MODEL,
        endpoint=_qwen_byo_endpoint(),
        served_window=8192,
    )

    assert "max_input_tokens" not in captured
    assert "max_output_tokens" not in captured
    assert "litellm_extra_body" not in captured


def test_local_context_window_split_floor_vs_tier():
    import inspect

    from solstone.think import supervisor
    from solstone.think.providers import local_server, openhands

    assert local_server.LOCAL_MIN_CONTEXT_TOKENS == 16384
    removed_name = "_".join(("LOCAL", "SERVER", "CONTEXT", "TOKENS"))
    assert not hasattr(local_server, removed_name)
    truth_src = inspect.getsource(supervisor._observe_linux_local_provider_truth)
    assert "select_server_tier" in truth_src
    assert "tier.context_tokens" in truth_src
    launcher_src = inspect.getsource(supervisor.start_local_server)
    assert "select_server_tier" not in launcher_src
    assert "plan.context_tokens" in inspect.getsource(
        supervisor._start_llama_local_server
    )
    assert '"16384"' not in launcher_src
    llm_src = inspect.getsource(openhands._build_llm)
    assert "LOCAL_MIN_CONTEXT_TOKENS" in llm_src


def test_select_server_tier_vram_thresholds():
    from solstone.think.providers import local_server

    cases = [
        (
            0,
            local_server.ServerTier(
                name="floor",
                context_tokens=16384,
                parallel_slots=1,
                prompt_cache_mib=0,
                resident_mib=4147,
            ),
        ),
        (
            15999,
            local_server.ServerTier(
                name="floor",
                context_tokens=16384,
                parallel_slots=1,
                prompt_cache_mib=0,
                resident_mib=4147,
            ),
        ),
        (
            16000,
            local_server.ServerTier(
                name="capable",
                context_tokens=32768,
                parallel_slots=2,
                prompt_cache_mib=2048,
                resident_mib=None,
            ),
        ),
        (
            24576,
            local_server.ServerTier(
                name="capable",
                context_tokens=32768,
                parallel_slots=2,
                prompt_cache_mib=2048,
                resident_mib=None,
            ),
        ),
    ]

    for vram_mib, expected in cases:
        tier = local_server.select_server_tier(vram_mib)
        assert tier == expected
        assert tier.context_tokens >= 16384
        assert tier.context_tokens > 0
    assert local_server._FLOOR_TIER.resident_mib == 4147
    assert local_server._CAPABLE_TIER.resident_mib is None


@pytest.mark.parametrize(
    ("props", "expected"),
    [
        ({"n_ctx": 32768}, 32768),
        ({"default_generation_settings": {"n_ctx": 16384}}, 16384),
        (
            {"n_ctx": 32768, "default_generation_settings": {"n_ctx": 16384}},
            32768,
        ),
        ({}, None),
        ({"default_generation_settings": {}}, None),
        ({"n_ctx": "abc"}, None),
        ({"n_ctx": None}, None),
        # Numeric strings are acceptable because _extract_n_ctx intentionally
        # uses int() coercion on reported llama-server values.
        ({"n_ctx": "32768"}, 32768),
    ],
)
def test_extract_n_ctx_props_shapes(props, expected):
    from solstone.think.providers import local_server

    assert local_server._extract_n_ctx(props) == expected


def test_read_server_context_props_fetch_props(monkeypatch):
    import httpx

    from solstone.think.providers import local_server

    class FakeResponse:
        status_code = 200

        def __init__(self, body=None, error: Exception | None = None):
            self.body = body
            self.error = error

        def json(self):
            if self.error is not None:
                raise self.error
            return self.body

    monkeypatch.setattr(
        httpx,
        "get",
        lambda url, timeout: FakeResponse({"n_ctx": 32768, "total_slots": 2}),
    )
    assert local_server.read_server_context_props(
        2468
    ) == local_server.ServerContextProps(
        n_ctx=32768,
        total_slots=2,
    )

    monkeypatch.setattr(
        httpx,
        "get",
        lambda url, timeout: FakeResponse(error=ValueError("bad json")),
    )
    assert local_server.read_server_context_props(2468) is None

    monkeypatch.setattr(httpx, "get", lambda url, timeout: FakeResponse(["n_ctx"]))
    assert local_server.read_server_context_props(2468) is None

    def raise_get(url, timeout):
        raise RuntimeError("network down")

    monkeypatch.setattr(httpx, "get", raise_get)
    assert local_server.read_server_context_props(2468) is None


def test_context_window_tokens_fallback(monkeypatch):
    from solstone.think import utils
    from solstone.think.providers import local_budget, local_server

    local_server.reset_parallel_slots_cache()
    monkeypatch.setattr(utils, "read_service_port", lambda service: 2468)
    monkeypatch.setattr(
        local_server,
        "read_server_context_props",
        lambda port: local_server.ServerContextProps(n_ctx=65536, total_slots=2),
    )
    monkeypatch.setattr(local_server, "read_local_context_window", lambda: None)
    assert local_budget.context_window_tokens() == 32768
    assert local_budget.resolve_context_window().slots == 2

    monkeypatch.setattr(local_server, "read_server_context_props", lambda port: None)
    monkeypatch.setattr(local_server, "read_local_context_window", lambda: 32768)
    assert local_budget.context_window_tokens() == 32768

    monkeypatch.setattr(utils, "read_service_port", lambda service: None)
    monkeypatch.setattr(local_server, "read_local_context_window", lambda: None)
    assert local_budget.context_window_tokens() == local_server.LOCAL_MIN_CONTEXT_TOKENS


@pytest.mark.parametrize(
    ("n_ctx", "total_slots", "expected_window"),
    [(65536, 2, 32768), (16384, 1, 16384)],
)
def test_context_window_tokens_divides_props_pool_by_total_slots(
    monkeypatch,
    n_ctx,
    total_slots,
    expected_window,
):
    from solstone.think import utils
    from solstone.think.providers import local_budget, local_server

    local_server.reset_parallel_slots_cache()
    monkeypatch.setattr(utils, "read_service_port", lambda service: 2468)
    monkeypatch.setattr(
        local_server,
        "read_server_context_props",
        lambda port: local_server.ServerContextProps(
            n_ctx=n_ctx,
            total_slots=total_slots,
        ),
    )
    monkeypatch.setattr(local_server, "read_local_context_window", lambda: None)

    resolution = local_budget.resolve_context_window()
    assert resolution.window_tokens == expected_window
    assert resolution.slots == total_slots


def test_context_window_tokens_props_without_total_slots_fails_closed_to_two(
    monkeypatch,
):
    from solstone.think import utils
    from solstone.think.providers import local_budget, local_server

    monkeypatch.setattr(utils, "read_service_port", lambda service: 2468)
    monkeypatch.setattr(
        local_server,
        "read_server_context_props",
        lambda port: local_server.ServerContextProps(n_ctx=65536, total_slots=None),
    )
    monkeypatch.setattr(local_server, "read_local_context_window", lambda: None)

    resolution = local_budget.resolve_context_window()
    assert resolution.window_tokens == 32768
    assert resolution.slots == local_server._CAPABLE_TIER.parallel_slots


def test_context_window_tokens_props_absent_sidecar_is_not_halved(monkeypatch):
    from solstone.think import utils
    from solstone.think.providers import local_budget, local_server

    monkeypatch.setattr(utils, "read_service_port", lambda service: 2468)
    monkeypatch.setattr(local_server, "read_server_context_props", lambda port: None)
    monkeypatch.setattr(local_server, "read_local_context_window", lambda: 32768)

    resolution = local_budget.resolve_context_window()
    assert resolution.window_tokens == 32768
    assert resolution.slots == local_server._CAPABLE_TIER.parallel_slots


def test_context_window_tokens_darwin_no_props_no_sidecar_uses_floor(monkeypatch):
    from solstone.think import utils
    from solstone.think.providers import local_budget, local_server

    monkeypatch.setattr(local_server.sys, "platform", "darwin")
    monkeypatch.setattr(utils, "read_service_port", lambda service: 2468)
    monkeypatch.setattr(local_server, "read_server_context_props", lambda port: None)
    monkeypatch.setattr(local_server, "read_local_context_window", lambda: None)

    assert local_budget.context_window_tokens() == local_server.LOCAL_MIN_CONTEXT_TOKENS


@pytest.mark.parametrize(
    ("props_total_slots", "capacity_slots", "expect_strict"),
    [(1, 1, False), (None, 2, False), (None, 1, True)],
)
def test_context_window_resolution_divisor_not_less_than_capacity(
    monkeypatch,
    props_total_slots,
    capacity_slots,
    expect_strict,
):
    from solstone.think import utils
    from solstone.think.providers import local_budget, local_server

    local_server.reset_parallel_slots_cache()
    monkeypatch.setattr(utils, "read_service_port", lambda service: 2468)
    monkeypatch.setattr(
        local_server,
        "read_server_context_props",
        lambda port: local_server.ServerContextProps(
            n_ctx=65536,
            total_slots=props_total_slots,
        ),
    )
    monkeypatch.setattr(
        local_server,
        "read_server_capacity",
        lambda: local_server.ServerCapacity(capacity_slots, "test", "capable"),
    )

    resolution = local_budget.resolve_context_window()
    reported_slots = local_server.read_server_capacity().parallel_slots
    assert resolution.slots >= reported_slots
    if expect_strict:
        assert resolution.slots > reported_slots


def _select_local_provider(monkeypatch) -> None:
    monkeypatch.setattr(
        "solstone.think.models.get_config",
        lambda: {
            "providers": {"active": {"provider": "local", "model": "local/qwen3.5-4b"}}
        },
    )


def _provider_local_readiness(
    *,
    binary_installed: bool = True,
    model_installed: bool = True,
    ram_sufficient: bool = True,
    gpu_available: bool = True,
    gpu_probe_ok: bool = True,
    binary_path: str = "/fake/llama-server",
) -> ReadinessOutcome:
    ready = binary_installed and model_installed
    return ReadinessOutcome(
        provider="local",
        status="ready" if ready else "missing-or-mismatched",
        reason_code="ready" if ready else "manifest_missing",
        target={"model_id": LOCAL_MODEL},
        install={
            "install_state": "idle",
            "install_error": None,
            "error_code": None,
            "attempt_id": None,
            "progress_bytes_received": None,
            "progress_bytes_total": None,
            "last_transition_at": None,
            "last_progress_at": None,
        },
        host={
            "ram_sufficient": ram_sufficient,
            "gpu_available": gpu_available,
            "gpu_probe_ok": gpu_probe_ok,
            "backend": "vulkan",
            "backend_reason": "test vulkan",
        },
        artifacts={
            "binary_installed": binary_installed,
            "model_installed": model_installed,
            "binary_path": binary_path,
            "model_path": "/tmp/model.gguf",
            "mmproj_path": None,
            "model_id": LOCAL_MODEL,
        },
        proof={
            "binary": {
                "status": "ready" if binary_installed else "missing-or-mismatched",
                "reason_code": "ready" if binary_installed else "manifest_missing",
                "cache_hit": False,
            },
            "model": {
                "status": "ready" if model_installed else "missing-or-mismatched",
                "reason_code": "ready" if model_installed else "manifest_missing",
                "cache_hit": False,
            },
        },
    )


def test_build_provider_status_local_not_selected_is_inert(monkeypatch):
    from solstone.think.providers import build_provider_status

    health_calls = []
    monkeypatch.setattr(
        "solstone.think.models.get_config",
        lambda: {
            "providers": {"active": {"provider": "google", "model": "gemini-3.5-flash"}}
        },
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_install.inspect_readiness",
        lambda: _provider_local_readiness(),
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_server.is_healthy",
        lambda: health_calls.append("health") or True,
    )

    status = build_provider_status(
        [{"name": "local", "label": "Local (on-device)", "env_key": ""}]
    )["local"]

    assert status["selected"] is False
    assert status["configured"] is True
    assert status["generate_ready"] is False
    assert status["cogitate_ready"] is False
    assert status["issues"] == []
    assert health_calls == []


def test_build_provider_status_local_readiness(monkeypatch):
    from solstone.think.providers import build_provider_status

    _select_local_provider(monkeypatch)
    monkeypatch.setattr(
        "solstone.think.providers.local_install.inspect_readiness",
        lambda: _provider_local_readiness(),
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_server.is_healthy", lambda: True
    )

    status = build_provider_status(
        [{"name": "local", "label": "Local (on-device)", "env_key": ""}]
    )["local"]

    assert status["configured"] is True
    assert status["generate_ready"] is True
    assert status["cogitate_ready"] is True
    assert status["issues"] == []


def test_build_provider_status_local_launch_failure_adds_probe_detail_and_hint(
    monkeypatch,
):
    from solstone.think.providers import build_provider_status

    _select_local_provider(monkeypatch)
    detail = "dyld: Library not loaded: @rpath/libllama.dylib"
    monkeypatch.setattr(
        "solstone.think.providers.local_install.inspect_readiness",
        lambda: _provider_local_readiness(),
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_server.is_healthy", lambda: False
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_install.probe_binary_runnable",
        lambda _path: (False, detail),
    )

    status = build_provider_status(
        [{"name": "local", "label": "Local (on-device)", "env_key": ""}]
    )["local"]

    assert status["issues"] == [
        f"failed to launch: {detail}",
        "run `journal install-provider local`",
    ]
    assert "server_unhealthy" not in status["issues"]


def test_build_provider_status_local_server_unhealthy_when_probe_runnable(
    monkeypatch,
):
    from solstone.think.providers import build_provider_status

    _select_local_provider(monkeypatch)
    monkeypatch.setattr(
        "solstone.think.providers.local_install.inspect_readiness",
        lambda: _provider_local_readiness(),
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_server.is_healthy", lambda: False
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_install.probe_binary_runnable",
        lambda _path: (True, None),
    )

    status = build_provider_status(
        [{"name": "local", "label": "Local (on-device)", "env_key": ""}]
    )["local"]

    assert status["issues"] == ["server_unhealthy"]


def test_build_provider_status_local_healthy_skips_probe(monkeypatch):
    from solstone.think.providers import build_provider_status

    _select_local_provider(monkeypatch)
    calls: list[str] = []

    def probe(_path):
        calls.append(_path)
        return False, "should not run"

    monkeypatch.setattr(
        "solstone.think.providers.local_install.inspect_readiness",
        lambda: _provider_local_readiness(),
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_server.is_healthy", lambda: True
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_install.probe_binary_runnable", probe
    )

    status = build_provider_status(
        [{"name": "local", "label": "Local (on-device)", "env_key": ""}]
    )["local"]

    assert status["issues"] == []
    assert calls == []


def test_local_provider_status_carries_install_hint_substring(monkeypatch):
    from solstone.think.providers import build_provider_status

    _select_local_provider(monkeypatch)
    monkeypatch.setattr(
        "solstone.think.providers.local_install.inspect_readiness",
        lambda: _provider_local_readiness(
            binary_installed=False,
            model_installed=False,
            ram_sufficient=False,
        ),
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_server.is_healthy", lambda: False
    )

    status = build_provider_status(
        [{"name": "local", "label": "Local (on-device)", "env_key": ""}]
    )["local"]

    assert status["configured"] is False
    assert status["generate_ready"] is False
    assert status["cogitate_ready"] is False
    assert status["issues"] == [
        "binary_missing",
        "model_missing",
        "run `journal install-provider local`",
    ]
    assert any("journal install-provider local" in issue for issue in status["issues"])


def test_local_provider_status_reports_gpu_unavailable_issue(monkeypatch):
    from solstone.think.providers import build_provider_status

    _select_local_provider(monkeypatch)
    monkeypatch.setattr(
        "solstone.think.providers.local_install.inspect_readiness",
        lambda: _provider_local_readiness(gpu_available=False),
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_server.is_healthy", lambda: True
    )

    status = build_provider_status(
        [{"name": "local", "label": "Local (on-device)", "env_key": ""}]
    )["local"]

    assert status["issues"] == ["gpu_unavailable"]


def test_build_provider_status_local_configured_ignores_ram_flag(monkeypatch):
    from solstone.think.providers import build_provider_status

    _select_local_provider(monkeypatch)
    monkeypatch.setattr(
        "solstone.think.providers.local_install.inspect_readiness",
        lambda: _provider_local_readiness(ram_sufficient=False),
    )
    monkeypatch.setattr(
        "solstone.think.providers.local_server.is_healthy", lambda: True
    )

    status = build_provider_status(
        [{"name": "local", "label": "Local (on-device)", "env_key": ""}]
    )["local"]

    assert status["configured"] is True
    assert status["generate_ready"] is True
    assert status["cogitate_ready"] is True
    assert status["issues"] == []


def test_local_server_connect_returns_healthy_service(monkeypatch):
    from solstone.think.providers import local_server

    monkeypatch.setattr(local_server, "read_service_port", lambda service: 2468)
    monkeypatch.setattr(
        local_server,
        "_fetch_health",
        lambda port: ("ready", None, {"loaded_model": "/path/to/snapshot"}),
    )

    info = local_server.connect()

    assert info.model_id == LOCAL_MODEL
    assert info.served_model_id == "/path/to/snapshot"
    assert info.base_url == "http://127.0.0.1:2468"
    assert info.state == local_server.STATE_READY


def test_resolve_served_model_id_returns_valid_loaded_model_verbatim():
    from solstone.think.providers import local_server

    assert (
        local_server._resolve_served_model_id({"loaded_model": "/snap/dir"})
        == "/snap/dir"
    )


def test_resolve_served_model_id_falls_back_when_loaded_model_absent():
    from solstone.think.providers import local_server

    assert local_server._resolve_served_model_id({}) == LOCAL_MODEL
    assert local_server._resolve_served_model_id(None) == LOCAL_MODEL


@pytest.mark.parametrize(
    "body",
    [
        {"loaded_model": None},
        {"loaded_model": ""},
        {"loaded_model": "   "},
        {"loaded_model": 123},
    ],
)
def test_resolve_served_model_id_rejects_invalid_loaded_model(body):
    from solstone.think.providers import local_server

    assert local_server._resolve_served_model_id(body) is None


def test_local_server_connect_missing_port_raises_named_copy(monkeypatch):
    from solstone.think.providers import local_server

    monkeypatch.setattr(local_server, "read_service_port", lambda service: None)

    with pytest.raises(local_server.LocalProviderError) as exc:
        local_server.connect()

    assert exc.value.reason_code == "local_model_not_ready"
    assert str(exc.value) == local_server.LOCAL_MODEL_NOT_READY_COPY


def test_local_server_connect_failed_health_raises_named_copy(monkeypatch):
    from solstone.think.providers import local_server

    monkeypatch.setattr(local_server, "read_service_port", lambda service: 2468)
    monkeypatch.setattr(
        local_server,
        "_fetch_health",
        lambda port: (local_server.STATE_FAILED, None, None),
    )

    with pytest.raises(local_server.LocalProviderError) as exc:
        local_server.connect()

    assert exc.value.reason_code == "local_model_not_ready"
    assert str(exc.value) == local_server.LOCAL_MODEL_NOT_READY_COPY


def test_local_server_connect_loading_health_raises_named_copy(monkeypatch):
    from solstone.think.providers import local_server

    monkeypatch.setattr(local_server, "read_service_port", lambda service: 2468)
    monkeypatch.setattr(
        local_server,
        "_fetch_health",
        lambda port: (local_server.STATE_LOADING, None, None),
    )

    with pytest.raises(local_server.LocalProviderError) as exc:
        local_server.connect()

    assert exc.value.reason_code == "local_model_loading"
    assert str(exc.value) == local_server.LOCAL_MODEL_NOT_READY_COPY


@pytest.mark.parametrize(
    "body",
    [
        {"loaded_model": None},
        {"loaded_model": ""},
    ],
)
def test_local_server_connect_invalid_loaded_model_raises_named_copy(monkeypatch, body):
    from solstone.think.providers import local_server

    monkeypatch.setattr(local_server, "read_service_port", lambda service: 2468)
    monkeypatch.setattr(
        local_server, "_fetch_health", lambda port: ("ready", None, body)
    )

    with pytest.raises(local_server.LocalProviderError) as exc:
        local_server.connect()

    assert exc.value.reason_code == "local_model_not_ready"
    assert str(exc.value) == local_server.LOCAL_MODEL_NOT_READY_COPY


def test_local_server_connect_linux_health_shape_uses_logical_model(monkeypatch):
    from solstone.think.providers import local_server

    monkeypatch.setattr(local_server, "read_service_port", lambda service: 2468)
    monkeypatch.setattr(
        local_server,
        "_fetch_health",
        lambda port: ("ready", None, {"status": "ok"}),
    )

    info = local_server.connect()

    assert info.model_id == LOCAL_MODEL
    assert info.served_model_id == LOCAL_MODEL


# --- read_server_parallel_slots ---------------------------------------------


def _local_journal(monkeypatch, tmp_path: Path) -> Path:
    from solstone.think.providers import local_server

    journal = tmp_path / "journal"
    (journal / "health").mkdir(parents=True)
    monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
    local_server.reset_parallel_slots_cache()
    return journal


def test_read_server_parallel_slots_prefers_live_props(monkeypatch, tmp_path):
    from solstone.think.providers import local_server

    journal = _local_journal(monkeypatch, tmp_path)
    (journal / "health" / "local.port").write_text("2468")
    # A launch-time context window that maps to the floor tier's single slot.
    (journal / "health" / "local.ctx").write_text(
        str(local_server._FLOOR_TIER.context_tokens)
    )
    monkeypatch.setattr(
        local_server,
        "fetch_props",
        lambda port, timeout_s=1.0: {"n_ctx": 32768, "total_slots": 2},
    )

    # /props is ground truth; it wins over the persisted tier.
    assert local_server.read_server_parallel_slots() == 2
    assert local_server.read_server_capacity() == local_server.ServerCapacity(
        parallel_slots=2,
        source="props",
        profile="capable",
    )


def test_read_server_parallel_slots_no_port_returns_floor(
    monkeypatch, tmp_path, caplog
):
    from solstone.think.providers import local_server

    _local_journal(monkeypatch, tmp_path)

    def _no_network(port, timeout_s=1.0):
        raise AssertionError("fetch_props must not run without a port")

    monkeypatch.setattr(local_server, "fetch_props", _no_network)

    caplog.set_level(logging.INFO)
    assert local_server.read_server_parallel_slots() == 1
    assert (
        "local_server_parallel_slots fallback slots=1 port=None "
        "context_tokens=None source=default" in caplog.text
    )


def test_server_capacity_uses_explicit_apple_profile(monkeypatch, tmp_path):
    from solstone.think.providers import local_server

    _local_journal(monkeypatch, tmp_path)
    monkeypatch.setattr(local_server.sys, "platform", "darwin")

    assert local_server.read_server_capacity() == local_server.ServerCapacity(
        parallel_slots=1,
        source="default",
        profile="apple",
    )


@pytest.mark.parametrize("slots", [1, 2])
def test_read_server_parallel_slots_falls_back_to_launched_tier(
    monkeypatch, tmp_path, slots
):
    from solstone.think.providers import local_server

    tier = local_server._FLOOR_TIER if slots == 1 else local_server._CAPABLE_TIER
    journal = _local_journal(monkeypatch, tmp_path)
    (journal / "health" / "local.port").write_text("2468")
    (journal / "health" / "local.ctx").write_text(str(tier.context_tokens))
    monkeypatch.setattr(local_server, "fetch_props", lambda port, timeout_s=1.0: None)

    assert local_server.read_server_parallel_slots() == tier.parallel_slots


def test_read_server_parallel_slots_unknown_context_window_returns_floor(
    monkeypatch, tmp_path
):
    from solstone.think.providers import local_server

    journal = _local_journal(monkeypatch, tmp_path)
    (journal / "health" / "local.port").write_text("2468")
    (journal / "health" / "local.ctx").write_text("99999")
    monkeypatch.setattr(local_server, "fetch_props", lambda port, timeout_s=1.0: None)

    assert local_server.read_server_parallel_slots() == 1


@pytest.mark.parametrize("props", [{}, {"total_slots": 0}, {"total_slots": "many"}])
def test_read_server_parallel_slots_rejects_unusable_total_slots(
    monkeypatch, tmp_path, props
):
    from solstone.think.providers import local_server

    journal = _local_journal(monkeypatch, tmp_path)
    (journal / "health" / "local.port").write_text("2468")
    monkeypatch.setattr(local_server, "fetch_props", lambda port, timeout_s=1.0: props)

    assert local_server.read_server_parallel_slots() == 1


def test_read_server_parallel_slots_is_memoized_and_resettable(monkeypatch, tmp_path):
    from solstone.think.providers import local_server

    journal = _local_journal(monkeypatch, tmp_path)
    (journal / "health" / "local.port").write_text("2468")
    calls = []

    def counting_props(port, timeout_s=1.0):
        calls.append(port)
        return {"total_slots": 2}

    monkeypatch.setattr(local_server, "fetch_props", counting_props)

    assert local_server.read_server_parallel_slots() == 2
    assert local_server.read_server_parallel_slots() == 2
    assert len(calls) == 1

    local_server.reset_parallel_slots_cache()
    assert local_server.read_server_parallel_slots() == 2
    assert len(calls) == 2


def test_local_response_telemetry_normalizes_llama_and_mlx_fields():
    provider = _provider()

    fields = provider._server_response_fields(
        {
            "timings": {
                "cache_n": 236,
                "prompt_n": 12,
                "prompt_ms": 30.5,
                "predicted_n": 35,
                "predicted_ms": 661.0,
                "slot_id": 1,
            },
            "usage": {
                "prompt_tokens": 248,
                "completion_tokens": 35,
            },
        }
    )

    assert fields == {
        "prompt_eval_ms": 30.5,
        "generation_ms": 661.0,
        "server_total_ms": 691.5,
        "prompt_tokens": 248,
        "generated_tokens": 35,
        "prompt_cached_tokens": 236,
        "selected_slot": 1,
        "prompt_cache_state": "warm",
    }
    assert provider._extract_usage(
        {
            "usage": {
                "prompt_tokens": 248,
                "completion_tokens": 35,
                "total_tokens": 283,
                "prompt_tokens_details": {"cached_tokens": 236},
            }
        }
    ) == {
        "input_tokens": 248,
        "output_tokens": 35,
        "total_tokens": 283,
        "cached_tokens": 236,
    }