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 logging
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,
    synth_reasoning_block,
)
from turnstone.core.providers._protocol import (
    CompletionResult,
    IncompleteStreamError,
    ModelCapabilities,
    StreamChunk,
    UsageInfo,
)
from turnstone.core.trajectory import 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_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_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")
    lane = ModelLane(
        provider=provider,
        client=placeholder_client,
        model="m",
        alias="app-gateway",
        backend_auth_resolver=resolver,
    )

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

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


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) -> str:
        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_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_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"