turnstone/tests/test_model_turn.py

"""Unit tests for the ``model_turn`` plant-call primitive (#827).

The agent-path tests in ``test_session.py`` exercise ``model_turn`` through
``_run_agent`` (native-lane replay, blank-id gate, minted-id nesting); these
pin the module's own contract directly so the judges (phase 1b) and the
single-shot lanes (phase 2) can build on it without re-deriving semantics.
"""

from __future__ import annotations

import ast
import inspect
import logging
import textwrap
import threading
import time
from types import SimpleNamespace
from typing import Any
from unittest.mock import MagicMock

import pytest

import turnstone.core.model_turn as model_turn_mod
from tests._session_helpers import as_stream
from turnstone.core.model_turn import (
    ModelLane,
    finalize_provider_blocks,
    maybe_attach_vllm_chat_reasoning,
    model_turn,
    resolve_lane,
    resolve_model_binding,
    synth_reasoning_block,
)
from turnstone.core.providers._protocol import (
    CompletionResult,
    IncompleteStreamError,
    ModelCapabilities,
    ProviderRequestMetrics,
    StreamChunk,
    UsageInfo,
    serialized_tool_chars,
)
from turnstone.core.session import ChatSession
from turnstone.core.trajectory import AttachmentRef, Role, ToolCall, Turn


class _FakeProvider:
    """Records every ``create_streaming`` call; replays scripted results
    as single-chunk streams via the shared ``as_stream`` adapter
    (multi-chunk accumulation is pinned by the dedicated ``drain_stream``
    unit tests)."""

    provider_name = "openai-compatible"

    def __init__(self, results: list[CompletionResult]) -> None:
        self.results = list(results)
        self.calls: list[dict[str, Any]] = []

    def get_capabilities(self, model: str) -> ModelCapabilities:
        return ModelCapabilities()

    def create_streaming(self, **kwargs: Any) -> list[StreamChunk]:
        self.calls.append(kwargs)
        return as_stream(self.results.pop(0))


def _fake_registry(
    *,
    capabilities: dict[str, Any] | None = None,
    server_compat: dict[str, Any] | None = None,
    replay: bool = False,
    temperature: float | None = None,
) -> MagicMock:
    cfg = SimpleNamespace(
        capabilities=capabilities or {},
        server_compat=server_compat or {},
        replay_reasoning_to_model=replay,
        temperature=temperature,
    )
    reg = MagicMock()
    reg.get_config.return_value = cfg
    return reg


def _lane(provider: _FakeProvider, **kw: Any) -> ModelLane:
    return ModelLane(provider=provider, client=object(), model="m", **kw)


def test_chat_session_has_no_raw_provider_facing_holders() -> None:
    """Keep #979's architectural closure stronger than a text grep."""
    tree = ast.parse(textwrap.dedent(inspect.getsource(ChatSession)))
    violations: list[str] = []
    for node in ast.walk(tree):
        if not isinstance(node, ast.Attribute):
            continue
        if (
            isinstance(node.value, ast.Name)
            and node.value.id == "self"
            and node.attr in {"_provider", "client"}
        ):
            violations.append(f"self.{node.attr}")
        if node.attr == "retryable_error_names":
            violations.append("direct retryable_error_names read")
        if node.attr in {"provider", "client"}:
            if isinstance(node.value, ast.Name) and node.value.id.endswith("lane"):
                violations.append(f"{node.value.id}.{node.attr}")
            if isinstance(node.value, ast.Attribute) and node.value.attr == "lane":
                violations.append(f"binding.lane.{node.attr}")

    assert violations == []


def test_prepare_wire_observes_canonical_argument_legalization() -> None:
    """The caller hook runs after Turn IR projection has legalized arguments."""
    provider = _FakeProvider([CompletionResult(content="ok")])
    seen: list[list[dict[str, Any]]] = []

    def prepare(messages: list[dict[str, Any]], _lane: ModelLane) -> list[dict[str, Any]]:
        seen.append(messages)
        return messages

    result = model_turn(
        _lane(provider),
        [
            Turn.assistant(
                tool_calls=(ToolCall(id="call-bad", name="lookup", arguments="not-json"),)
            ),
            Turn.tool("call-bad", "handled"),
        ],
        prepare_wire=prepare,
    )

    assert result.content == "ok"
    assistant = next(message for message in seen[0] if message["role"] == "assistant")
    assert assistant["tool_calls"][0]["function"]["arguments"] == "{}"


def test_backend_auth_token_binds_sdk_credential_once() -> None:
    """Dynamic credentials use SDK with_options, not an override header."""
    provider = _FakeProvider([CompletionResult(content="ok")])
    base_client = MagicMock()
    bound_client = object()
    base_client.with_options.return_value = bound_client
    lane = ModelLane(provider=provider, client=base_client, model="m", alias="gateway")

    result = model_turn(
        lane,
        [Turn.user("hello")],
        backend_auth_token="minted-token",
    )

    assert result.content == "ok"
    base_client.with_options.assert_called_once_with(api_key="minted-token")
    assert provider.calls[0]["client"] is bound_client
    assert "extra_headers" not in provider.calls[0]


def test_result_carries_exact_serving_tool_definition_size() -> None:
    """Token calibration consumes the tool list sent to this lane."""
    provider = _FakeProvider([CompletionResult(content="ok")])
    tools = [
        {
            "type": "function",
            "function": {"name": "lookup", "description": "Find a value"},
        }
    ]

    result = model_turn(_lane(provider), [Turn.user("hello")], tools=tools)

    assert result.tool_def_chars == serialized_tool_chars(tools)
    assert result.serving_model == "m"


def test_result_prefers_final_provider_native_tool_definition_size() -> None:
    """Adapter metrics win over the pre-provider OpenAI-shaped schemas."""

    class _NativeMetricsProvider(_FakeProvider):
        def create_streaming(self, **kwargs: Any) -> list[StreamChunk]:
            metrics = kwargs["request_metrics_ref"]
            metrics.append(ProviderRequestMetrics(serialized_tool_chars=1_234))
            return super().create_streaming(**kwargs)

    provider = _NativeMetricsProvider([CompletionResult(content="ok")])
    result = model_turn(
        _lane(provider),
        [Turn.user("hello")],
        tools=[{"type": "function", "function": {"name": "lookup"}}],
    )

    assert result.tool_def_chars == 1_234


def test_entra_app_lane_resolver_never_issues_placeholder_client() -> None:
    """A resolver-carrying lane binds its app token before the provider call."""
    provider = _FakeProvider([CompletionResult(content="ok")])
    placeholder_client = MagicMock(name="backend-auth-placeholder-unused")
    bound_client = object()
    placeholder_client.with_options.return_value = bound_client
    resolver = MagicMock(return_value="app-token")
    auth_config = MagicMock(name="pinned-auth-config")
    lane = ModelLane(
        provider=provider,
        client=placeholder_client,
        model="m",
        alias="app-gateway",
        backend_auth_resolver=resolver,
        backend_auth_config=auth_config,
    )

    model_turn(lane, [Turn.user("hello")])

    resolver.assert_called_once_with("app-gateway", auth_config)
    placeholder_client.with_options.assert_called_once_with(api_key="app-token")
    assert provider.calls[0]["client"] is bound_client


def test_resolve_model_binding_canonicalizes_empty_alias_to_default() -> None:
    """The empty spelling must not erase live flags or dynamic auth identity."""
    provider = _FakeProvider([])
    client = object()
    cfg = SimpleNamespace(capabilities={}, server_compat={})
    registry = MagicMock()
    registry.default = "default-gateway"
    registry.resolve_binding.return_value = (client, "model", cfg, provider, 7)

    binding = resolve_model_binding(registry, "")

    registry.resolve_binding.assert_called_once_with("default-gateway")
    assert binding.lane.alias == "default-gateway"
    assert binding.lane.client is client
    assert binding.config is cfg
    assert binding.registry_generation == 7


def test_model_turn_materializes_before_admission_and_mints_inside_hold() -> None:
    order: list[str] = []

    class _Gate:
        held = False

        def acquire(self, *, cancel_ref: Any = None) -> Any:
            del cancel_ref
            order.append("acquire")
            gate = self

            class _Lease:
                def __enter__(self) -> None:
                    gate.held = True
                    order.append("enter")

                def __exit__(self, *_exc: object) -> None:
                    gate.held = False
                    order.append("release")

            return _Lease()

    gate = _Gate()
    bound_client = object()
    base_client = MagicMock()
    base_client.with_options.return_value = bound_client

    def _resolve(ids: list[str]) -> dict[str, Any]:
        assert ids == ["image-1"]
        assert not gate.held
        order.append("materialize")
        return {
            "image-1": {
                "type": "image_url",
                "image_url": {"url": "data:image/png;base64,abc"},
            }
        }

    def _auth(_alias: str, _cfg: Any) -> str:
        assert gate.held
        order.append("auth")
        return "minted-token"

    class _Provider(_FakeProvider):
        def create_streaming(self, **kwargs: Any) -> Any:
            assert gate.held
            assert kwargs["client"] is bound_client
            assert kwargs["resolve_attachments"] is None
            order.append("dispatch")
            self.calls.append(kwargs)

            def _stream() -> Any:
                assert gate.held
                order.append("drain")
                yield from as_stream(CompletionResult(content="ok"))

            return _stream()

    provider = _Provider([])
    lane = ModelLane(
        provider=provider,
        client=base_client,
        model="m",
        alias="primary",
        backend_auth_resolver=_auth,
        admission=gate,  # type: ignore[arg-type]
    )
    turn = Turn(Role.USER, (AttachmentRef(attachment_id="image-1", kind="image"),))

    result = model_turn(lane, [turn], resolve_attachments=_resolve)

    assert result.content == "ok"
    assert order == ["materialize", "acquire", "enter", "auth", "dispatch", "drain", "release"]
    assert result.wire_msgs == [
        {
            "role": "user",
            "content": [
                {
                    "type": "image_url",
                    "image_url": {"url": "data:image/png;base64,abc"},
                }
            ],
        }
    ]


def test_model_turn_releases_admission_before_retry_backoff(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from turnstone.core.deadline import StreamAbortRef

    order: list[str] = []

    class _Gate:
        held = False
        acquire_calls = 0

        def acquire(self, *, cancel_ref: Any = None) -> Any:
            del cancel_ref
            self.acquire_calls += 1
            gate = self

            class _Lease:
                def __enter__(self) -> None:
                    assert not gate.held
                    gate.held = True
                    order.append("enter")

                def __exit__(self, *_exc: object) -> None:
                    gate.held = False
                    order.append("release")

            return _Lease()

    gate = _Gate()
    provider = _FlakyProvider([IncompleteStreamError("retry"), CompletionResult(content="ok")])
    dispatch = provider.create_streaming

    def _dispatch(**kwargs: Any) -> Any:
        assert gate.held
        order.append("dispatch")
        return dispatch(**kwargs)

    provider.create_streaming = _dispatch  # type: ignore[method-assign]

    def _sleep(_delay: float) -> None:
        assert not gate.held
        order.append("backoff")

    monkeypatch.setattr(model_turn_mod, "time", SimpleNamespace(sleep=_sleep))
    lane = ModelLane(
        provider=provider,
        client=object(),
        model="m",
        admission=gate,  # type: ignore[arg-type]
    )
    ref = StreamAbortRef()

    result = model_turn(lane, [Turn.user("x")], cancel_ref=ref)

    assert result.content == "ok"
    assert gate.acquire_calls == 2
    assert ref.dispatch_count == 2
    assert order == ["enter", "dispatch", "release", "backoff", "enter", "dispatch", "release"]


def test_same_alias_attachment_work_completes_before_outer_admission() -> None:
    from turnstone.core.admission import ModelAdmission

    gate = ModelAdmission("primary", 1)
    nested_completed = False

    def _resolve(ids: list[str]) -> dict[str, Any]:
        nonlocal nested_completed
        assert ids == ["image-1"]
        # Models a nested perception call using the same alias.  This would
        # block forever if the outer model_turn had already taken the slot.
        with gate.acquire():
            nested_completed = True
        return {"image-1": {"type": "image_url", "image_url": {"url": "data:x"}}}

    provider = _FakeProvider([CompletionResult(content="ok")])
    lane = ModelLane(provider=provider, client=object(), model="m", admission=gate)
    turn = Turn(Role.USER, (AttachmentRef(attachment_id="image-1", kind="image"),))

    assert model_turn(lane, [turn], resolve_attachments=_resolve).content == "ok"
    assert nested_completed
    assert gate.snapshot().in_flight == 0


def test_model_turn_releases_admission_when_eager_create_fails() -> None:
    from turnstone.core.admission import ModelAdmission

    class _CreateFailureProvider(_FakeProvider):
        def create_streaming(self, **kwargs: Any) -> Any:
            self.calls.append(kwargs)
            raise RuntimeError("connect failed")

    gate = ModelAdmission("primary", 1)
    provider = _CreateFailureProvider([])
    lane = ModelLane(provider=provider, client=object(), model="m", admission=gate)

    with pytest.raises(RuntimeError, match="connect failed"):
        model_turn(lane, [Turn.user("x")])

    assert len(provider.calls) == 1
    assert gate.snapshot().in_flight == 0


def test_model_turn_cancelled_while_queued_never_dispatches() -> None:
    from turnstone.core.admission import ModelAdmission
    from turnstone.core.deadline import DeadlineCancelledError, StreamAbortRef

    gate = ModelAdmission("primary", 1)
    holder = gate.acquire()
    provider = _FakeProvider([CompletionResult(content="never")])
    lane = ModelLane(provider=provider, client=object(), model="m", admission=gate)
    cancel_ref = StreamAbortRef()
    errors: list[BaseException] = []

    def _call() -> None:
        try:
            model_turn(lane, [Turn.user("x")], cancel_ref=cancel_ref)
        except BaseException as exc:  # test records the worker's exact exit
            errors.append(exc)

    thread = threading.Thread(target=_call, daemon=True)
    thread.start()
    deadline = time.monotonic() + 1.0
    while gate.snapshot().queued != 1:
        if time.monotonic() >= deadline:
            holder.release()
            raise AssertionError("model turn did not queue")
        time.sleep(0.005)
    cancel_ref.abort()
    thread.join(1.0)
    holder.release()

    assert not thread.is_alive()
    assert len(errors) == 1
    assert isinstance(errors[0], DeadlineCancelledError)
    assert provider.calls == []


class _FlakyProvider:
    """Scripted drain-time deaths: each script entry is either a
    ``CompletionResult`` (streamed normally) or an exception instance
    (raised mid-iteration — AFTER ``create_streaming`` returned, exactly
    where a real mid-body wire death surfaces)."""

    provider_name = "openai-compatible"
    retryable_error_names: frozenset[str] = frozenset({"IncompleteStreamError"})

    def __init__(self, script: list[Any]) -> None:
        self.script = list(script)
        self.calls: list[dict[str, Any]] = []

    def get_capabilities(self, model: str) -> ModelCapabilities:
        return ModelCapabilities()

    def create_streaming(self, **kwargs: Any) -> Any:
        self.calls.append(kwargs)
        item = self.script.pop(0)

        def _iter() -> Any:
            if isinstance(item, BaseException):
                raise item
            yield from as_stream(item)

        return _iter()


def test_model_turn_retries_transient_mid_stream_death(monkeypatch: pytest.MonkeyPatch) -> None:
    # The retired non-streaming transport read the whole body inside the
    # SDK's retried request, so single-shot lanes never saw a mid-body wire
    # blip — the drain-scoped loop is that retry's new home.
    monkeypatch.setattr("turnstone.core.model_turn._DRAIN_RETRY_BASE_DELAY", 0.0)
    provider = _FlakyProvider(
        [
            IncompleteStreamError("stream died mid-response"),
            CompletionResult(content="second try"),
        ]
    )
    lane = ModelLane(provider=provider, client=object(), model="m")

    result = model_turn(lane, [Turn.user("x")])

    assert result.content == "second try"
    assert len(provider.calls) == 2


def test_model_turn_gives_up_after_retry_budget(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr("turnstone.core.model_turn._DRAIN_RETRY_BASE_DELAY", 0.0)
    provider = _FlakyProvider([IncompleteStreamError(f"death {i}") for i in range(5)])
    lane = ModelLane(provider=provider, client=object(), model="m")

    with pytest.raises(IncompleteStreamError):
        model_turn(lane, [Turn.user("x")])

    # One initial issue + _DRAIN_RETRIES re-issues, then it propagates.
    assert len(provider.calls) == 3


def test_model_turn_retry_backs_off_between_attempts(monkeypatch: pytest.MonkeyPatch) -> None:
    # Instant re-issues are guaranteed to re-hit a still-active rate
    # limit/overload — the loop paces like the SDK request retry it
    # replaces: 0.5s base, doubling, ±50% jitter.
    sleeps: list[float] = []
    monkeypatch.setattr(model_turn_mod, "time", SimpleNamespace(sleep=sleeps.append))
    provider = _FlakyProvider(
        [
            IncompleteStreamError("death 1"),
            IncompleteStreamError("death 2"),
            CompletionResult(content="ok"),
        ]
    )
    lane = ModelLane(provider=provider, client=object(), model="m")

    result = model_turn(lane, [Turn.user("x")])

    assert result.content == "ok"
    assert len(sleeps) == 2
    assert 0.25 <= sleeps[0] <= 0.75  # 0.5 * jitter[0.5, 1.5)
    assert 0.5 <= sleeps[1] <= 1.5  # 1.0 * jitter[0.5, 1.5)


def test_model_turn_does_not_retry_unrecognized_errors() -> None:
    provider = _FlakyProvider([RuntimeError("schema violation")])
    lane = ModelLane(provider=provider, client=object(), model="m")

    with pytest.raises(RuntimeError, match="schema violation"):
        model_turn(lane, [Turn.user("x")])

    assert len(provider.calls) == 1


def test_model_turn_abort_during_backoff_suppresses_reissue(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    # The deadline can abandon the worker while it sleeps between
    # attempts — the wake-up must die with the original failure, not
    # issue one more full request from an abandoned thread.
    from turnstone.core.deadline import StreamAbortRef

    ref = StreamAbortRef()
    monkeypatch.setattr(model_turn_mod, "time", SimpleNamespace(sleep=lambda _delay: ref.abort()))
    provider = _FlakyProvider(
        [IncompleteStreamError("transient death"), CompletionResult(content="never")]
    )
    lane = ModelLane(provider=provider, client=object(), model="m")

    with pytest.raises(IncompleteStreamError, match="transient death"):
        model_turn(lane, [Turn.user("x")], cancel_ref=ref)

    assert len(provider.calls) == 1


def test_model_turn_does_not_retry_after_abort(
    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
    # A call the deadline abandoned must not have its request resurrected
    # behind its back: in the field the abort closes the stream and the
    # drain dies with an error that LOOKS retryable.  The fixture models
    # only the shape of that — abort landing after dispatch, drain raising
    # IncompleteStreamError — because the aborted ref is what gates the
    # re-issue regardless of which of the two produced the error.  This is
    # the RE-ISSUE gate; of the tests below, two cover the pre-dispatch reads
    # and the third pins the raised message.
    from turnstone.core.deadline import StreamAbortRef

    provider = _FlakyProvider(
        [IncompleteStreamError("closed by abort"), CompletionResult(content="never")]
    )
    lane = ModelLane(provider=provider, client=object(), model="m")
    ref = StreamAbortRef()
    dispatch = provider.create_streaming

    def _abort_after_dispatch(**kwargs: Any) -> Any:
        stream = dispatch(**kwargs)
        ref.abort()  # the deadline daemon fires; the request is already out
        return stream

    monkeypatch.setattr(provider, "create_streaming", _abort_after_dispatch)

    with (
        caplog.at_level(logging.WARNING, logger="turnstone.core.model_turn"),
        pytest.raises(IncompleteStreamError),
    ):
        model_turn(lane, [Turn.user("x")], cancel_ref=ref)

    assert len(provider.calls) == 1
    # Isolates THIS gate from the post-backoff one its sibling covers.  The
    # abort is read where the failure surfaces, so the loop never announces a
    # re-issue it will not make; delete that gate and the backoff arm still
    # ends at one dispatch, but it logs on the way — which is what makes this
    # assertion, and not the call count, the discriminating one.
    assert "model_turn.drain_retry" not in caplog.text


def test_abort_landing_during_the_backend_auth_mint_still_never_dispatches() -> None:
    # The window an entry-only check cannot see: on a dynamic-auth alias
    # the resolve can block for seconds on a cache miss, so an abort can
    # land after the entry read and before the request.  The read
    # immediately before create_streaming is what covers it.
    from turnstone.core.deadline import DeadlineCancelledError, StreamAbortRef

    provider = _FakeProvider([CompletionResult(content="never")])
    ref = StreamAbortRef()
    client = MagicMock()

    def _abort_during_mint(alias: str, config: Any | None) -> str:
        assert alias == "obo-gateway"
        assert config is None
        ref.abort()  # the user hits Stop while the mint is blocked
        return "minted-token"

    lane = ModelLane(
        provider=provider,
        client=client,
        model="m",
        alias="obo-gateway",
        backend_auth_resolver=_abort_during_mint,
    )

    with pytest.raises(DeadlineCancelledError):
        model_turn(lane, [Turn.user("x")], cancel_ref=ref)

    assert provider.calls == []


def test_abort_during_failed_backend_auth_mint_masks_auth_error() -> None:
    from turnstone.core.deadline import DeadlineCancelledError, StreamAbortRef
    from turnstone.core.model_backend_auth import BackendAuthUnavailableError

    provider = _FakeProvider([CompletionResult(content="never")])
    ref = StreamAbortRef()

    def _abort_then_fail(alias: str, config: Any | None) -> str:
        assert alias == "obo-gateway"
        assert config is None
        ref.abort()
        raise BackendAuthUnavailableError("mint failed")

    lane = ModelLane(
        provider=provider,
        client=MagicMock(),
        model="m",
        alias="obo-gateway",
        backend_auth_resolver=_abort_then_fail,
    )

    with pytest.raises(DeadlineCancelledError):
        model_turn_mod.lane_call_client(lane, cancel_ref=ref)

    assert provider.calls == []


def test_pre_dispatch_abort_precedes_the_backend_auth_mint() -> None:
    # Placement of the FIRST read: an already-abandoned call skips the
    # resolve entirely.  On a cache miss that resolve is a network mint
    # under a cluster-wide lock, so this is work worth not doing — but the
    # invariant itself rides the read before create_streaming, not this one.
    from turnstone.core.deadline import DeadlineCancelledError, StreamAbortRef

    provider = _FakeProvider([CompletionResult(content="never")])
    resolver = MagicMock(return_value="minted-token")
    client = MagicMock()
    lane = ModelLane(
        provider=provider,
        client=client,
        model="m",
        alias="obo-gateway",
        backend_auth_resolver=resolver,
    )
    ref = StreamAbortRef()
    ref.abort()

    with pytest.raises(DeadlineCancelledError):
        model_turn(lane, [Turn.user("x")], cancel_ref=ref)

    resolver.assert_not_called()
    client.with_options.assert_not_called()
    assert provider.calls == []


def test_abort_during_wire_preparation_precedes_backend_auth_mint() -> None:
    """A Stop observed after lowering must not redeem a backend credential."""
    from turnstone.core.deadline import DeadlineCancelledError, StreamAbortRef

    provider = _FakeProvider([CompletionResult(content="never")])
    resolver = MagicMock(return_value="minted-token")
    client = MagicMock()
    ref = StreamAbortRef()
    lane = ModelLane(
        provider=provider,
        client=client,
        model="m",
        alias="obo-gateway",
        backend_auth_resolver=resolver,
    )

    def prepare(
        messages: list[dict[str, Any]],
        _lane: ModelLane,
    ) -> list[dict[str, Any]]:
        ref.abort()
        return messages

    with pytest.raises(DeadlineCancelledError):
        model_turn(
            lane,
            [Turn.user("x")],
            cancel_ref=ref,
            prepare_wire=prepare,
        )

    resolver.assert_not_called()
    client.with_options.assert_not_called()
    assert provider.calls == []


def test_pre_dispatch_abort_does_not_read_as_a_context_overflow() -> None:
    # A latent coupling, pinned deliberately rather than a live path: today
    # compaction's ``except`` arm re-checks the session first and raises
    # GenerationCancelled, and ``_stop_retrying`` short-circuits on the class
    # gate, so this message never reaches ``_is_ctx_overflow``.  It would the
    # moment either shortcut moves — and ``_is_ctx_overflow`` classifies an
    # unrecognized class by TEXT, so an overflow reading would send the
    # compaction lane subdividing.  The message is a wire contract; pin it.
    from turnstone.core.deadline import DeadlineCancelledError, StreamAbortRef
    from turnstone.core.session import _is_ctx_overflow

    provider = _FakeProvider([CompletionResult(content="never")])
    lane = ModelLane(provider=provider, client=object(), model="m")
    ref = StreamAbortRef()
    ref.abort()

    with pytest.raises(DeadlineCancelledError) as excinfo:
        model_turn(lane, [Turn.user("x")], cancel_ref=ref)

    assert not _is_ctx_overflow(excinfo.value)


def _real_semantics_store(**stored: Any) -> SimpleNamespace:
    """A ConfigStore fake with the REAL ``get()`` semantics.

    A stored key returns its value; a never-stored key returns the
    SETTINGS registry default — which for the sampling keys IS the unset
    sentinel (``None`` / ``""``).  The old fakes returned ``None`` on any
    miss, which masked the default-on-miss collision the round-2 review
    caught: never fake a store rung more forgiving than the real one.
    """
    from turnstone.core.settings_registry import SETTINGS

    def _get(key: str, default: Any = ...) -> Any:
        if key in stored:
            return stored[key]
        if default is not ...:
            return default
        defn = SETTINGS.get(key)
        return defn.default if defn else None

    return SimpleNamespace(get=_get)


def test_model_turn_lowers_turns_and_threads_lane_config() -> None:
    caps = ModelCapabilities(max_output_tokens=1234)
    extra = {"chat_template_kwargs": {"enable_thinking": True}}
    provider = _FakeProvider([CompletionResult(content="hi")])
    lane = _lane(provider, capabilities=caps, extra_params=extra)

    result = model_turn(
        lane,
        [Turn.user("x")],
        tools=[{"type": "function", "function": {"name": "f", "parameters": {}}}],
        max_tokens=99,
        temperature=0.1,
        reasoning_effort="low",
    )

    (call,) = provider.calls
    assert call["messages"][0]["role"] == "user"
    assert call["messages"][0]["content"] == "x"
    assert call["capabilities"] is caps
    assert call["extra_params"] is extra
    assert call["max_tokens"] == 99
    assert call["temperature"] == 0.1
    assert call["reasoning_effort"] == "low"
    # No registry on the lane → the operator replay flag resolves False.
    assert call["replay_reasoning_to_model"] is False
    assert result.turn.role is Role.ASSISTANT
    assert result.content == "hi"
    assert result.finish_reason == "stop"


def test_model_turn_returns_usage_verbatim() -> None:
    usage = UsageInfo(prompt_tokens=9, completion_tokens=1, total_tokens=10)
    provider = _FakeProvider([CompletionResult(content="", usage=usage)])
    result = model_turn(_lane(provider), [Turn.user("x")])
    # Value equality, not identity: ``drain_stream`` max-merges usage across
    # chunks into its own instance so it never mutates the provider's object.
    assert result.usage == usage


def test_mint_rewrites_mirror_records_map_and_native_keeps_original() -> None:
    provider = _FakeProvider(
        [
            CompletionResult(
                content="",
                tool_calls=[
                    {
                        "id": "call_0",
                        "type": "function",
                        "function": {"name": "f", "arguments": "{}"},
                    }
                ],
                provider_blocks=[{"type": "tool_use", "id": "call_0", "name": "f"}],
            )
        ]
    )
    wire_id_map: dict[str, str] = {}
    result = model_turn(
        _lane(provider),
        [Turn.user("x")],
        mint=lambda original: f"parent::r1s1::{original}",
        wire_id_map=wire_id_map,
    )

    # The mirror (execution view) and the Turn both carry the minted id …
    assert result.tool_calls[0]["id"] == "parent::r1s1::call_0"
    assert result.turn.tool_calls[0].id == "parent::r1s1::call_0"
    # … the map records the recovery path …
    assert wire_id_map == {"parent::r1s1::call_0": "call_0"}
    # … and the native block keeps the provider-original id verbatim (it may
    # sit under a reasoning signature and is never rewritten).
    assert result.turn.native is not None
    assert result.turn.native.blocks[0]["id"] == "call_0"
    assert result.turn.native.producer == "openai-compatible"


def test_restore_maps_minted_ids_back_on_the_wire() -> None:
    minted = "parent::r1s1::call_0"
    provider = _FakeProvider([CompletionResult(content="done")])
    turns = [
        Turn.user("go"),
        Turn.assistant("", tool_calls=(ToolCall(id=minted, name="f", arguments="{}"),)),
        Turn.tool(minted, "result"),
    ]

    model_turn(_lane(provider), turns, wire_id_map={minted: "call_0"})

    (call,) = provider.calls
    assistant = next(m for m in call["messages"] if m["role"] == "assistant")
    tool = next(m for m in call["messages"] if m["role"] == "tool")
    assert assistant["tool_calls"][0]["id"] == "call_0"
    assert tool["tool_call_id"] == "call_0"


def test_blank_ids_repair_native_lane_pairwise() -> None:
    # Google-compat shape: blank id on BOTH the mirror and the raw fidelity
    # block.  The manufactured uuid lands in both (positional pairing), so
    # the thought_signature-bearing block SURVIVES instead of the turn
    # degrading to loose reasoning text — the unblock for the Gemini judge
    # evidence loop on blank-id compat responses.
    provider = _FakeProvider(
        [
            CompletionResult(
                content="",
                tool_calls=[
                    {"id": "", "type": "function", "function": {"name": "f", "arguments": "{}"}}
                ],
                provider_blocks=[
                    {
                        "id": "",
                        "type": "function",
                        "function": {"name": "f", "arguments": "{}"},
                        "thought_signature": "sig123",
                    }
                ],
                reasoning="thought",
            )
        ]
    )
    result = model_turn(_lane(provider), [Turn.user("x")])

    manufactured = result.tool_calls[0]["id"]
    assert manufactured.startswith("call_")
    assert result.turn.native is not None
    blocks = list(result.turn.native.blocks)
    # The fidelity block survives, id-agreeing with the mirror, signature
    # untouched; the loose reasoning still synthesizes alongside it.
    assert blocks[0]["id"] == manufactured
    assert blocks[0]["thought_signature"] == "sig123"
    assert blocks[-1]["type"] == "reasoning_text"


def test_blank_id_repair_never_rewrites_nonblank_ids() -> None:
    provider = _FakeProvider(
        [
            CompletionResult(
                content="",
                tool_calls=[
                    {"id": "call_7", "type": "function", "function": {"name": "a"}},
                    {"id": "", "type": "function", "function": {"name": "b"}},
                ],
                provider_blocks=[
                    {"id": "call_7", "type": "function", "function": {"name": "a"}},
                    {"id": "", "type": "function", "function": {"name": "b"}},
                ],
            )
        ]
    )
    result = model_turn(_lane(provider), [Turn.user("x")])
    assert result.turn.native is not None
    blocks = list(result.turn.native.blocks)
    # Provider-assigned id untouched (it may sit under a signature) …
    assert blocks[0]["id"] == "call_7"
    # … only the blank one was manufactured, agreeing with its mirror twin.
    assert blocks[1]["id"] == result.tool_calls[1]["id"]
    assert blocks[1]["id"].startswith("call_")


def test_blank_id_pairing_mismatch_falls_back_to_reasoning_text_drop() -> None:
    # Two mirror calls but only one client block: no trustworthy pairing —
    # the total drop rule (the #825-converged fallback) keeps only the
    # loose-text reasoning synth.
    provider = _FakeProvider(
        [
            CompletionResult(
                content="",
                tool_calls=[
                    {"id": "", "type": "function", "function": {"name": "a"}},
                    {"id": "", "type": "function", "function": {"name": "b"}},
                ],
                provider_blocks=[{"type": "function", "id": "", "function": {"name": "a"}}],
                reasoning="thought",
            )
        ]
    )
    result = model_turn(_lane(provider), [Turn.user("x")])
    assert result.tool_calls[0]["id"].startswith("call_")
    assert result.turn.native is not None
    assert [b["type"] for b in result.turn.native.blocks] == ["reasoning_text"]
    assert result.turn.native.blocks[0]["text"] == "thought"


def test_orphan_client_tool_blocks_stripped_when_no_tool_calls() -> None:
    provider = _FakeProvider(
        [
            CompletionResult(
                content="truncated",
                tool_calls=None,
                provider_blocks=[{"type": "tool_use", "id": "x", "name": "f"}],
            )
        ]
    )
    result = model_turn(_lane(provider), [Turn.user("x")])
    # A tool_use with no mirrored call would replay with no matching
    # tool_result — the finalize gate strips it, leaving no lane at all.
    assert result.turn.native is None


def test_live_operator_flags_reresolve_per_call() -> None:
    registry = _fake_registry(replay=False)
    provider = _FakeProvider([CompletionResult(content="a"), CompletionResult(content="b")])
    lane = _lane(provider, alias="ali", registry=registry)

    model_turn(lane, [Turn.user("x")])
    # Operator flips the toggle mid-session (admin write → registry reload).
    registry.get_config.return_value.replay_reasoning_to_model = True
    model_turn(lane, [Turn.user("x")])

    first, second = provider.calls
    assert first["replay_reasoning_to_model"] is False
    assert second["replay_reasoning_to_model"] is True


def test_resolve_lane_respects_preresolved_values() -> None:
    provider = _FakeProvider([])
    caps = ModelCapabilities(max_output_tokens=7)
    lane = resolve_lane(provider, object(), "m", capabilities=caps, extra_params={"k": "v"})
    assert lane.capabilities is caps
    assert lane.extra_params == {"k": "v"}
    # Explicit None is a valid resolved value, distinct from "resolve for me".
    lane_none = resolve_lane(provider, object(), "m", capabilities=caps, extra_params=None)
    assert lane_none.extra_params is None


def test_resolve_lane_merges_registry_capability_overrides() -> None:
    provider = _FakeProvider([])
    registry = _fake_registry(capabilities={"max_output_tokens": 42, "not_a_field": 1})
    lane = resolve_lane(provider, object(), "m", alias="ali", registry=registry)
    assert lane.capabilities is not None
    assert lane.capabilities.max_output_tokens == 42


def test_vllm_attach_gates() -> None:
    from turnstone.core.providers._openai_chat import OpenAIChatCompletionsProvider

    msgs = [
        {"role": "user", "content": "q"},
        {
            "role": "assistant",
            "content": "a",
            "_provider_content": [{"type": "reasoning_text", "text": "cot"}],
        },
    ]
    # Non-Chat-Completions provider: untouched (identity).
    assert maybe_attach_vllm_chat_reasoning(msgs, _FakeProvider([]), None, "ali") is msgs  # type: ignore[arg-type]

    chat = OpenAIChatCompletionsProvider()
    # All three gates open → reasoning field attached.
    on = _fake_registry(server_compat={"server_type": "vllm"}, replay=True)
    out = maybe_attach_vllm_chat_reasoning(msgs, chat, on, "ali")
    assert out[1]["reasoning"] == "cot"
    # Operator flag off → untouched.
    off = _fake_registry(server_compat={"server_type": "vllm"}, replay=False)
    assert maybe_attach_vllm_chat_reasoning(msgs, chat, off, "ali") is msgs
    # Wrong server type → untouched.
    sglang = _fake_registry(server_compat={"server_type": "sglang"}, replay=True)
    assert maybe_attach_vllm_chat_reasoning(msgs, chat, sglang, "ali") is msgs


def test_synth_reasoning_block_appends_with_source_and_skips_native() -> None:
    registry = _fake_registry(server_compat={"server_type": "vllm"})
    fidelity = [{"type": "tool_calls", "raw": True}]
    out = synth_reasoning_block(fidelity, ["thought"], registry=registry, alias="ali")
    # Appends (Google fidelity blocks survive) and tags the source server.
    assert out[0] is fidelity[0]
    assert out[1] == {"type": "reasoning_text", "text": "thought", "source": "vllm"}
    # A native reasoning-bearing block suppresses synthesis (identity return).
    native = [{"type": "thinking", "thinking": "t", "signature": "s"}]
    assert synth_reasoning_block(native, ["thought"]) is native


def test_finalize_keeps_full_lane_with_tool_calls_and_clean_ids() -> None:
    blocks = [
        {"type": "thinking", "thinking": "t", "signature": "s"},
        {"type": "tool_use", "id": "toolu_1", "name": "f"},
    ]
    out = finalize_provider_blocks(blocks, [""], has_tool_calls=True)
    assert out == blocks


def test_mint_without_wire_id_map_raises() -> None:
    provider = _FakeProvider([])
    with pytest.raises(ValueError, match="wire_id_map"):
        model_turn(_lane(provider), [Turn.user("x")], mint=lambda o: f"p::{o}")
    # Nothing reached the provider — the guard fires before lowering.
    assert provider.calls == []


def test_temperature_inherits_lane_value_when_caller_omits() -> None:
    provider = _FakeProvider([CompletionResult(content="")])
    lane = _lane(provider, temperature=1.3)
    model_turn(lane, [Turn.user("x")])
    assert provider.calls[0]["temperature"] == 1.3


def test_temperature_caller_value_wins_over_lane() -> None:
    provider = _FakeProvider([CompletionResult(content="")])
    lane = _lane(provider, temperature=1.3)
    model_turn(lane, [Turn.user("x")], temperature=0.9)
    assert provider.calls[0]["temperature"] == 0.9


def test_temperature_unresolved_passes_none_and_wire_omits_it() -> None:
    # No caller value, no lane value → model_turn passes temperature=None,
    # and the PROVIDER layer omits the field from the wire so the server
    # default applies (house rule: code never pins one).  Both halves are
    # pinned: a Python-signature default of 0.5 anywhere on this path is a
    # hidden universal pin — the exact bug the second xhigh review caught.
    provider = _FakeProvider([CompletionResult(content="")])
    model_turn(_lane(provider), [Turn.user("x")])
    assert provider.calls[0]["temperature"] is None

    from turnstone.core.providers._openai_common import apply_temperature

    kwargs: dict[str, Any] = {}
    apply_temperature(kwargs, ModelCapabilities(), None, "medium")
    assert "temperature" not in kwargs  # None never reaches the wire
    apply_temperature(kwargs, ModelCapabilities(), 1.0, "medium")
    assert kwargs["temperature"] == 1.0  # a real value still does


def test_resolve_lane_global_config_store_rung() -> None:
    # The global rung fires only when the operator actually STORED a
    # value; the registry default is the unset sentinel (None), so an
    # untouched install resolves None → the wire omits the field.
    provider = _FakeProvider([])
    registry = _fake_registry(temperature=None)
    store = _real_semantics_store(**{"model.temperature": 1.0})
    lane = resolve_lane(provider, object(), "m", alias="ali", registry=registry, config_store=store)
    assert lane.temperature == 1.0
    # The per-model value wins over the global rung.
    registry2 = _fake_registry(temperature=0.3)
    lane2 = resolve_lane(
        provider, object(), "m", alias="ali", registry=registry2, config_store=store
    )
    assert lane2.temperature == 0.3
    # Never-stored global → None (the round-2 headline: ConfigStore.get
    # must NOT manufacture a wire value on a miss).
    lane3 = resolve_lane(
        provider,
        object(),
        "m",
        alias="ali",
        registry=_fake_registry(temperature=None),
        config_store=_real_semantics_store(),
    )
    assert lane3.temperature is None


def test_resolve_lane_reasoning_effort_operator_rungs() -> None:
    # The lane carries the OPERATOR rungs only: per-model config → stored
    # global setting → None.  The in-code model definition (caps default)
    # applies at the model_turn call, so an operator-silent lane stays
    # None — the assignment scheme's "if not set, we don't send it".
    provider = _FakeProvider([])
    # Per-model config wins.
    reg = _fake_registry()
    reg.get_config.return_value.reasoning_effort = "high"
    lane = resolve_lane(provider, object(), "m", alias="ali", registry=reg)
    assert lane.reasoning_effort == "high"
    # Stored global setting rung.
    reg2 = _fake_registry()
    reg2.get_config.return_value.reasoning_effort = None
    store = _real_semantics_store(**{"model.reasoning_effort": "low"})
    lane2 = resolve_lane(provider, object(), "m", alias="ali", registry=reg2, config_store=store)
    assert lane2.reasoning_effort == "low"
    # Empty string at any rung is the unset sentinel (a valid settings
    # choice meaning fall through) — with a never-stored global (registry
    # default IS "") the lane stays operator-silent.
    reg3 = _fake_registry()
    reg3.get_config.return_value.reasoning_effort = ""
    lane3 = resolve_lane(
        provider, object(), "m", alias="ali", registry=reg3, config_store=_real_semantics_store()
    )
    assert lane3.reasoning_effort is None
    # Bare lane (no registry, no store): None.
    assert resolve_lane(provider, object(), "m").reasoning_effort is None


def test_model_turn_effort_lower_rungs() -> None:
    # Below the lane's operator rungs, model_turn applies exactly one more
    # rung — the in-code model definition (caps) — then None (wire
    # omission).  There is deliberately NO caller-default rung: a
    # code-chosen effort is an unvetted token on local vocabularies
    # (effort_passthrough forwards verbatim) and flips template thinking
    # toggles the operator never engaged.
    # Bare hand-built lane: nothing anywhere → the provider receives None.
    provider = _FakeProvider([CompletionResult(content="")])
    model_turn(_lane(provider), [Turn.user("x")])
    assert provider.calls[0]["reasoning_effort"] is None

    # In-code model definition rung: a declared caps default applies…
    caps = ModelCapabilities(default_reasoning_effort="high")
    provider2 = _FakeProvider([CompletionResult(content="")])
    model_turn(_lane(provider2, capabilities=caps), [Turn.user("x")])
    assert provider2.calls[0]["reasoning_effort"] == "high"

    # …loses to an operator value on the lane…
    provider3 = _FakeProvider([CompletionResult(content="")])
    model_turn(
        _lane(provider3, capabilities=caps, reasoning_effort="xhigh"),
        [Turn.user("x")],
    )
    assert provider3.calls[0]["reasoning_effort"] == "xhigh"

    # …and to an explicit relay (the "none" knob stays distinct from unset).
    provider4 = _FakeProvider([CompletionResult(content="")])
    model_turn(
        _lane(provider4, capabilities=caps),
        [Turn.user("x")],
        reasoning_effort="none",
    )
    assert provider4.calls[0]["reasoning_effort"] == "none"


def test_model_turn_fetches_config_once_per_call() -> None:
    # ONE get_config per plant call feeds both live flags (replay + vLLM
    # attach) — a hot-reload between them cannot mix config generations
    # within a single request.
    registry = _fake_registry(replay=True)
    provider = _FakeProvider([CompletionResult(content="")])
    lane = _lane(provider, alias="ali", registry=registry)
    model_turn(lane, [Turn.user("x")])
    assert registry.get_config.call_count == 1


def test_resolve_lane_inherits_config_temperature() -> None:
    provider = _FakeProvider([])
    registry = _fake_registry(temperature=0.7)
    lane = resolve_lane(provider, object(), "m", alias="ali", registry=registry)
    assert lane.temperature == 0.7
    # Exactly ONE config fetch feeds caps + extra_params + temperature —
    # no cross-generation mixing on a registry hot-reload.
    assert registry.get_config.call_count == 1


def test_resolve_lane_survives_get_config_raise() -> None:
    provider = _FakeProvider([])
    registry = MagicMock()
    registry.get_config.side_effect = ValueError("Unknown model alias")
    lane = resolve_lane(provider, object(), "m", alias="gone", registry=registry)
    # Every facet degrades to its miss behavior instead of raising into a
    # caller's constructor (the judge alias-resolution abort case).
    assert lane.capabilities is not None
    assert lane.extra_params is None
    assert lane.temperature is None


def test_resolve_capabilities_survives_get_config_raise() -> None:
    from turnstone.core.model_turn import resolve_capabilities

    provider = _FakeProvider([])
    registry = MagicMock()
    registry.get_config.side_effect = KeyError("gone")
    caps = resolve_capabilities(provider, "m", "gone", registry)
    assert caps == ModelCapabilities()


def test_inline_tags_segregate_to_native_reasoning_text_and_clean_content() -> None:
    # A passthrough server's tagged content, drained through the real seam:
    # the turn's text is IR-clean and the extracted reasoning lands in the
    # native lane as the path-3 synth block (so it survives reload and the
    # operator-gated replay), never in any consumer-visible content.
    provider = _FakeProvider([CompletionResult(content="<think>plan</think>answer")])
    result = model_turn(_lane(provider), [Turn.user("q")])
    assert result.content == "answer"
    assert result.turn.text == "answer"
    assert result.turn.native is not None
    synth = [b for b in result.turn.native.blocks if b.get("type") == "reasoning_text"]
    assert len(synth) == 1
    assert synth[0]["text"] == "plan"


def test_synth_bail_is_silent_and_leaks_nothing(
    caplog: pytest.LogCaptureFixture,
) -> None:
    # Bailing on an existing native reasoning block is the ROUTINE no-op on
    # Anthropic/Responses lanes (reasoning_delta mirrors the block): no log
    # event here, and reasoning text never reaches a log payload.  The
    # genuinely anomalous shape (inline-EXTRACTED text beside a native
    # block) is logged at the drain, where it is distinguishable.
    import logging

    from turnstone.core.model_turn import synth_reasoning_block

    secret_reasoning = "the plan nobody logs"
    with caplog.at_level(logging.DEBUG):
        blocks = synth_reasoning_block(
            [{"type": "thinking", "thinking": "native"}], [secret_reasoning]
        )
    assert blocks == [{"type": "thinking", "thinking": "native"}]
    assert secret_reasoning not in caplog.text


def test_capability_bool_overrides_coerced() -> None:
    """The capabilities dict is hand-edited JSON: a string "false" is
    truthy, and left raw it would flip every downstream truthiness read
    (a ``server_parses_reasoning: "false"`` typo silently turning the
    inline tag scan off is #940 reopened by punctuation).  Recognized
    spellings coerce, ints pass through ``bool()``, and an unrecognized
    value drops the key so the field keeps its default."""
    from turnstone.core.model_turn import apply_capability_overrides

    base = ModelCapabilities()
    off = apply_capability_overrides(base, {"server_parses_reasoning": "false"})
    assert off.server_parses_reasoning is False
    on = apply_capability_overrides(base, {"server_parses_reasoning": "true"})
    assert on.server_parses_reasoning is True
    coerced = apply_capability_overrides(base, {"supports_vision": 1, "supports_tools": 0})
    assert coerced.supports_vision is True
    assert coerced.supports_tools is False
    # Unrecognized string: key dropped, default kept; non-bool fields untouched.
    kept = apply_capability_overrides(
        base, {"server_parses_reasoning": "maybe", "thinking_mode": "manual"}
    )
    assert kept.server_parses_reasoning is False
    assert kept.thinking_mode == "manual"