feat(core): phase 2 — every single-shot lane speaks Turn IR (#827)

create_completion now has exactly one caller: model_turn. The π-side
lanes migrate off hand-built OpenAI dicts:

- _utility_completion (title gen, compaction, web-fetch extraction)
  takes list[Turn] and runs the session's primary lane through
  model_turn; its three call sites build Turn.system/Turn.user.
- perception.describe builds a by-reference trajectory (AttachmentRef +
  the prebuilt parts via resolve_attachments, reintroduced on
  model_turn with its first caller and pinned by tests) — Turn IR never
  carries inline media bytes, matching the main loop's wire path. Its
  temperature=0.2 pin is gone (house rule).
- eval HeadlessSession's loop lowers system prompts through the
  turns_from_dicts bridge and appends result.turn; the parallel-call
  cap now also drops the native lane on a capped turn (a capped mirror
  with a full native lane would replay orphan tool blocks).
- optimizer: all five sites (diversifier, observer, analyst loop,
  tool optimizer, prompt optimizer) build Turn IR through per-function
  lanes; every temperature pin (0.8/0.3/0.3/0.3/0.6) removed per house
  rule — sampling behavior belongs in the model's configuration.

Test mocks move to the shared full-shape helper where the model_turn
re-ingest now runs; perception/attachment tests assert the
by-reference placeholder + resolver contract instead of inline parts.

tests/test_compaction_crossing.py

@@ -115,7 +115,7 @@ class TestSummaryTurnProvenance:
             session._generate_title()
 
         uc.assert_called_once()
-        prompt = uc.call_args[0][0][-1]["content"]
+        prompt = uc.call_args[0][0][-1].text
         assert COMPACTION_SUMMARY_LABEL in prompt  # titled FROM the real message
 
 

tests/test_cooperative_compaction.py

@@ -685,7 +685,7 @@ class TestChunkedCompaction:
         recorded: list[int] = []
 
         def fake_uc(messages, **_kwargs):
-            body = messages[1]["content"]
+            body = messages[1].text
             prefix = session._COMPACT_USER_PREFIX
             if body.startswith(prefix):
                 body = body[len(prefix) :]

tests/test_perception.py

@@ -14,22 +14,50 @@ if TYPE_CHECKING:
 
 
 class _StubProvider:
-    """Minimal LLMProvider stand-in: counts calls, can fail the first N."""
+    """Minimal LLMProvider stand-in: counts calls, can fail the first N.
+
+    ``describe`` routes through ``model_turn``, so the stub carries the lane
+    surface (``provider_name``, ``get_capabilities``) and returns a full
+    ``CompletionResult`` shape, and it records the ``resolve_attachments``
+    callback the translator would use to materialize the by-reference parts.
+    """
+
+    provider_name = "openai-compatible"
 
     def __init__(self, *, content: str = "a description", fail_times: int = 0) -> None:
         self.calls = 0
         self._content = content
         self._fail_times = fail_times
         self.last_messages: list[dict[str, Any]] | None = None
+        self.last_resolve: Any = None
+
+    def get_capabilities(self, model: str) -> Any:
+        from turnstone.core.providers._protocol import ModelCapabilities
+
+        return ModelCapabilities()
 
     def create_completion(
-        self, *, client: Any, model: str, messages: list[dict[str, Any]], **_: Any
+        self,
+        *,
+        client: Any,
+        model: str,
+        messages: list[dict[str, Any]],
+        resolve_attachments: Any = None,
+        **_: Any,
     ) -> SimpleNamespace:
         self.calls += 1
         self.last_messages = messages
+        self.last_resolve = resolve_attachments
         if self.calls <= self._fail_times:
             raise RuntimeError("backend down")
-        return SimpleNamespace(content=self._content)
+        return SimpleNamespace(
+            content=self._content,
+            tool_calls=None,
+            finish_reason="stop",
+            usage=None,
+            provider_blocks=[],
+            reasoning="",
+        )
 
 
 @pytest.fixture(autouse=True)
@@ -43,14 +71,18 @@ def _parts() -> list[dict[str, Any]]:
     return [{"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}]
 
 
-def test_describe_builds_prompt_then_parts() -> None:
+def test_describe_lowers_prompt_then_by_reference_parts() -> None:
     prov = _StubProvider(content="desc")
     out = perception.describe(provider=prov, client=object(), model="m", parts=_parts())  # type: ignore[arg-type]
     assert out == "desc"
     assert prov.last_messages is not None
     content = prov.last_messages[0]["content"]
     assert content[0]["type"] == "text"  # prompt leads
-    assert content[1]["type"] == "image_url"  # attachment parts follow
+    # The attachment rides by reference; the translator materializes it via
+    # the threaded resolver, which must return the prebuilt parts verbatim.
+    assert content[1]["attachment_id"] == "perception-input"
+    assert prov.last_resolve is not None
+    assert prov.last_resolve(["perception-input"]) == {"perception-input": _parts()}
 
 
 def test_describe_empty_parts_skips_backend() -> None:

tests/test_session.py

@@ -11,6 +11,7 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 
+from tests._session_helpers import mock_completion_result
 from turnstone.core.session import _IMAGE_EXTENSIONS, _IMAGE_SIZE_CAP, ChatSession
 from turnstone.core.trajectory import (
     Turn,
@@ -1730,7 +1731,7 @@ class TestTitleRetry:
                 {"role": "assistant", "content": "Hi there"},
             ]
         )
-        result = MagicMock()
+        result = mock_completion_result()
         result.content = "Test Title"
         session._provider = MagicMock()
         session._provider.get_capabilities.return_value = ModelCapabilities()
@@ -1755,7 +1756,7 @@ class TestTitleRetry:
         session = _make_session()
         session._title_generated = True
         session.messages = turns_from_dicts([{"role": "user", "content": "hi"}])
-        result = MagicMock()
+        result = mock_completion_result()
         result.content = (
             "<think>The user greets me; a fitting title would be...</think>\n\n"
             '**"Cluster Routing Deep-Dive"**'
@@ -1789,7 +1790,7 @@ class TestTitleRetry:
         session = _make_session()
         session._title_generated = True
         session.messages = turns_from_dicts([{"role": "user", "content": "hi"}])
-        result = MagicMock()
+        result = mock_completion_result()
         result.content = "<think>still reasoning, never closed before the cap"
         session._provider = MagicMock()
         session._provider.get_capabilities.return_value = ModelCapabilities()
@@ -1819,7 +1820,7 @@ class TestTitleRetry:
             session = _make_session()
             session._title_generated = True
             session.messages = turns_from_dicts([{"role": "user", "content": "hi"}])
-            result = MagicMock()
+            result = mock_completion_result()
             result.content = content
             session._provider = MagicMock()
             session._provider.get_capabilities.return_value = ModelCapabilities()
@@ -1842,7 +1843,7 @@ class TestTitleRetry:
         session = _make_session()
         session._title_generated = True
         session.messages = turns_from_dicts([{"role": "user", "content": "hi"}])
-        result = MagicMock()
+        result = mock_completion_result()
         result.content = "Story " * 40  # 240 chars on one line
         session._provider = MagicMock()
         session._provider.get_capabilities.return_value = ModelCapabilities()
@@ -1869,7 +1870,7 @@ class TestTitleRetry:
             ]
         )
         original_ws_id = session._ws_id
-        result = MagicMock()
+        result = mock_completion_result()
         result.content = "Test Title"
         session._provider = MagicMock()
         session._provider.get_capabilities.return_value = ModelCapabilities()
@@ -7613,7 +7614,7 @@ def test_utility_completion_records_aux_usage():
         ),
     )
 
-    session._utility_completion([{"role": "user", "content": "hi"}])
+    session._utility_completion([Turn.user("hi")])
 
     assert len(ui.aux_calls) == 1
     rec = ui.aux_calls[0]
@@ -7638,11 +7639,11 @@ def test_utility_completion_defers_temperature_to_session():
     session._provider.get_capabilities.return_value = ModelCapabilities()
     session._provider.create_completion.return_value = CompletionResult(content="x")
 
-    session._utility_completion([{"role": "user", "content": "hi"}])
+    session._utility_completion([Turn.user("hi")])
     _, kw = session._provider.create_completion.call_args
     assert kw["temperature"] == 0.42  # deferred to the session/registry value
 
-    session._utility_completion([{"role": "user", "content": "hi"}], temperature=0.9)
+    session._utility_completion([Turn.user("hi")], temperature=0.9)
     _, kw2 = session._provider.create_completion.call_args
     assert kw2["temperature"] == 0.9  # explicit override still honored
 

tests/test_session_attachments.py

@@ -2,11 +2,11 @@
 
 from __future__ import annotations
 
-from types import SimpleNamespace
 from unittest.mock import MagicMock
 
 import pytest
 
+from tests._session_helpers import mock_completion_result
 from turnstone.core import perception
 from turnstone.core.attachments import Attachment
 from turnstone.core.memory import (
@@ -561,7 +561,7 @@ class TestPerceptionFallback:
         """Wire a stub perception backend onto the session; return the provider mock."""
         perception._clear_perception_cache_for_test()
         prov = MagicMock()
-        prov.create_completion.return_value = SimpleNamespace(content=content)
+        prov.create_completion.return_value = mock_completion_result(content)
         s._config_store = MagicMock()
         s._config_store.get = lambda k, *a: "omni" if k == "perception.model_alias" else ""
         s._registry = MagicMock()
@@ -603,9 +603,15 @@ class TestPerceptionFallback:
         )
         assert part["type"] == "text"
         assert "DESCRIPTION" in part["text"]
-        # the perception model was handed the rasterized pages, not the raw PDF
+        # the perception model was handed the rasterized pages, not the raw
+        # PDF: the wire carries the prompt + a by-reference placeholder, and
+        # the threaded resolver materializes the page parts at the translator.
         sent = prov.create_completion.call_args.kwargs["messages"][0]["content"]
-        assert [p["type"] for p in sent] == ["text", "image_url", "image_url"]
+        assert sent[0]["type"] == "text"
+        assert sent[1]["attachment_id"] == "perception-input"
+        resolver = prov.create_completion.call_args.kwargs["resolve_attachments"]
+        pages = resolver(["perception-input"])["perception-input"]
+        assert [p["type"] for p in pages] == ["image_url", "image_url"]
 
     def test_audio_perception_when_omni_and_no_stt(self, tmp_db, mock_openai_client):
         s = _make_session(mock_openai_client)

tests/test_session_chat_reasoning_replay.py

@@ -30,6 +30,7 @@ from tests._session_helpers import make_session as _make_session
 from turnstone.core.providers._anthropic import AnthropicProvider
 from turnstone.core.providers._openai_chat import OpenAIChatCompletionsProvider
 from turnstone.core.providers._openai_responses import OpenAIResponsesProvider
+from turnstone.core.trajectory import turns_from_dicts
 
 
 def _vllm_registry(*, replay: bool = True, alias: str = "qwen3") -> Any:
@@ -403,7 +404,12 @@ class TestCallSitesInvokeMaybeAttach:
         def capture_completion(**kwargs: Any) -> Any:
             captured.update(kwargs)
             return SimpleNamespace(
-                content="", tool_calls=[], usage=None, raw_blocks=None, provider_blocks=None
+                content="",
+                tool_calls=None,
+                finish_reason="stop",
+                usage=None,
+                provider_blocks=[],
+                reasoning="",
             )
 
         provider = OpenAIChatCompletionsProvider()
@@ -417,7 +423,7 @@ class TestCallSitesInvokeMaybeAttach:
             ),
         ):
             session._utility_completion(
-                messages=[_assistant_msg_with_thinking("from utility")],
+                turns_from_dicts([_assistant_msg_with_thinking("from utility")]),
             )
 
         msgs_sent = captured["messages"]

tests/test_session_replay_reasoning.py

@@ -27,6 +27,7 @@ from typing import Any
 from unittest.mock import MagicMock, patch
 
 from tests._session_helpers import make_session as _make_session
+from turnstone.core.trajectory import Turn
 
 
 def _registry_with_flag(persist: bool = True, replay: bool = False) -> Any:
@@ -626,7 +627,14 @@ class TestUtilityCompletionPassesFlag:
 
         def capture_completion(**kwargs: Any) -> Any:
             captured.update(kwargs)
-            return SimpleNamespace(content="title", finish_reason="stop", usage=None)
+            return SimpleNamespace(
+                content="title",
+                tool_calls=None,
+                finish_reason="stop",
+                usage=None,
+                provider_blocks=[],
+                reasoning="",
+            )
 
         mock_provider = MagicMock()
         mock_provider.create_completion = capture_completion
@@ -637,7 +645,7 @@ class TestUtilityCompletionPassesFlag:
             patch.object(session, "_provider_extra_params", return_value=None),
         ):
             session._utility_completion(
-                messages=[{"role": "user", "content": "summarize"}],
+                [Turn.user("summarize")],
                 max_tokens=512,
                 temperature=0.3,
             )

turnstone/core/model_turn.py

@@ -506,6 +506,7 @@ def model_turn(
     reasoning_effort: str = "medium",
     mint: Callable[[str], str] | None = None,
     wire_id_map: dict[str, str] | None = None,
+    resolve_attachments: Callable[[list[str]], dict[str, Any]] | None = None,
 ) -> ModelTurnResult:
     """Advance a trajectory by one model turn: lower, sample, re-ingest.
 
@@ -524,6 +525,13 @@ def model_turn(
     default when unset.  House rule: never pin a temperature in code; pass
     an explicit float only to relay an operator-resolved knob.
 
+    *resolve_attachments* materializes by-reference ``AttachmentRef``
+    content at the provider translator (``{type: kind, attachment_id}``
+    placeholders → inline parts; one id may expand to several parts, e.g.
+    a rasterized PDF).  Turn IR never carries inline media bytes — a lane
+    with non-text content passes refs plus this resolver, exactly like the
+    main loop's wire path.
+
     *mint* rewrites each returned tool call's id (provider-original →
     caller-scoped) before the Turn is built; the native blocks keep the
     provider ids verbatim (they are never rewritten — they may sit under a
@@ -563,6 +571,8 @@ def model_turn(
             lane.registry, lane.alias, caps=lane.capabilities
         ),
     }
+    if resolve_attachments is not None:
+        call_kwargs["resolve_attachments"] = resolve_attachments
     effective_temperature = temperature if temperature is not None else lane.temperature
     if effective_temperature is not None:
         call_kwargs["temperature"] = effective_temperature

turnstone/core/perception.py

@@ -20,10 +20,11 @@ configured STT model still wins for audio — perception only fills the remainin
 gap.  Point it at an omni model (text+vision+audio) to cover every modality from
 one alias; a vision-only model covers image/PDF and is simply skipped for audio.
 
-The call goes through the provider abstraction's ``create_completion`` (the same
-path the intent judge uses for its secondary model), so any provider works; the
-parts are OpenAI-shaped (``image_url`` / ``input_audio``) and the provider
-translates them to its own wire form.
+The call goes through :func:`turnstone.core.model_turn.model_turn` (the shared
+plant-call seam, #827), so any provider works: the trajectory carries the
+attachment by reference and the pre-built OpenAI-shaped parts (``image_url`` /
+``input_audio``) materialize at the provider translator via the
+``resolve_attachments`` callback, exactly like the main loop's wire path.
 """
 
 from __future__ import annotations
@@ -32,12 +33,25 @@ import threading
 from typing import TYPE_CHECKING, Any
 
 from turnstone.core.log import get_logger
+from turnstone.core.model_turn import model_turn, resolve_lane
+from turnstone.core.trajectory import AttachmentRef, Role, TextBlock, Turn
 
 if TYPE_CHECKING:
     from turnstone.core.providers._protocol import LLMProvider
 
 log = get_logger(__name__)
 
+# The single by-reference id inside a perception trajectory.  Perception
+# trajectories are ephemeral one-shot requests (never persisted, never
+# displayed), so the id only needs intra-request consistency between the
+# placeholder and the resolver mapping.
+_PERCEPTION_REF_ID = "perception-input"
+
+# Placeholder ``kind`` per part shape — cosmetic for an ephemeral trajectory
+# (the resolver replaces the placeholder wholesale) but kept honest for
+# debuggability.
+_PART_KINDS = {"image_url": "image", "input_audio": "audio"}
+
 # Config key naming the model used for perception fallbacks.
 PERCEPTION_SETTING = "perception.model_alias"
 
@@ -64,21 +78,34 @@ def describe(
 ) -> str:
     """Perceive ``parts`` via the perception model, returning the text.
 
-    ``parts`` are OpenAI-shaped content parts — ``image_url`` for image/PDF-page
-    perception, ``input_audio`` for audio (the provider translates them to its
-    own wire shape).  Raises :class:`PerceptionBackendError` if the backend call
-    fails.  Never caches — see :func:`describe_cached`.
+    ``parts`` are pre-built OpenAI-shaped content parts — ``image_url`` for
+    image/PDF-page perception, ``input_audio`` for audio.  The trajectory
+    carries them by reference; ``model_turn`` hands the resolver to the
+    provider translator, which materializes the placeholder into these exact
+    parts (one ref may expand to many, e.g. a rasterized PDF).  Temperature
+    is not pinned (house rule) — the perception model's own configuration
+    governs sampling.  Raises :class:`PerceptionBackendError` if the backend
+    call fails.  Never caches — see :func:`describe_cached`.
     """
     if not parts:
         return ""
-    messages = [{"role": "user", "content": [{"type": "text", "text": prompt}, *parts]}]
+    kind = _PART_KINDS.get(str(parts[0].get("type", "")), "document")
+    turns = [
+        Turn(
+            role=Role.USER,
+            content=(
+                TextBlock(prompt),
+                AttachmentRef(attachment_id=_PERCEPTION_REF_ID, kind=kind),
+            ),
+        )
+    ]
+    lane = resolve_lane(provider, client, model)
     try:
-        result = provider.create_completion(
-            client=client,
-            model=model,
-            messages=messages,
+        result = model_turn(
+            lane,
+            turns,
             max_tokens=4096,
-            temperature=0.2,
+            resolve_attachments=lambda _ids: {_PERCEPTION_REF_ID: parts},
         )
     except Exception as exc:
         raise PerceptionBackendError(f"perception backend failed: {exc}") from exc

turnstone/core/session.py

@@ -225,7 +225,6 @@ if TYPE_CHECKING:
     from turnstone.core.output_guard import OutputAssessment
     from turnstone.core.output_guard_judge import OutputGuardJudge, OutputJudgeVerdict
     from turnstone.core.providers import (
-        CompletionResult,
         LLMProvider,
         ModelCapabilities,
         StreamChunk,
@@ -3311,18 +3310,15 @@ class ChatSession:
 
             result = self._utility_completion(
                 [
-                    {
-                        "role": "system",
-                        "content": (
-                            "# Instructions\n\n"
-                            "You are a conversation title generator. "
-                            "The user will show you the opening of a conversation. "
-                            "Respond with ONLY a short title (3-8 words). "
-                            "Do NOT answer the conversation. Do NOT explain. "
-                            "Output ONLY the title text, nothing else."
-                        ),
-                    },
-                    {"role": "user", "content": snippet},
+                    Turn.system(
+                        "# Instructions\n\n"
+                        "You are a conversation title generator. "
+                        "The user will show you the opening of a conversation. "
+                        "Respond with ONLY a short title (3-8 words). "
+                        "Do NOT answer the conversation. Do NOT explain. "
+                        "Output ONLY the title text, nothing else."
+                    ),
+                    Turn.user(snippet),
                 ],
                 max_tokens=_TITLE_MAX_TOKENS,
             )
@@ -4745,18 +4741,17 @@ class ChatSession:
 
     def _utility_completion(
         self,
-        messages: list[dict[str, Any]],
+        turns: list[Turn],
         *,
         max_tokens: int = 4096,
         temperature: float | None = None,
         reasoning_effort: str = "low",
-    ) -> CompletionResult:
-        """Run a lightweight internal completion (title gen, compaction, extraction).
+    ) -> ModelTurnResult:
+        """Run a lightweight internal completion (title gen, compaction,
+        extraction) through ``model_turn`` on the session's primary lane.
 
-        Threads ``reasoning_effort`` through both the direct keyword (for
-        commercial providers) and ``extra_params`` (for local model servers)
-        so callers don't need to duplicate it.  ``max_tokens`` is clamped to
-        the model's advertised output limit so small models don't error.
+        ``max_tokens`` is clamped to the model's advertised output limit so
+        small models don't error.
 
         ``temperature`` defaults to the session temperature (``self.temperature``)
         — the same operator/registry-resolved value the main turn uses — rather
@@ -4767,17 +4762,21 @@ class ChatSession:
         """
         caps = self._get_capabilities()
         clamped = min(max_tokens, caps.max_output_tokens) if caps.max_output_tokens else max_tokens
-        messages = self._maybe_attach_vllm_chat_reasoning(messages, self._provider)
-        result = self._provider.create_completion(
+        lane = ModelLane(
+            provider=self._provider,
             client=self.client,
             model=self.model,
-            messages=messages,
+            alias=self._model_alias or "",
+            capabilities=caps,
+            extra_params=self._provider_extra_params(),
+            registry=self._registry,
+        )
+        result = model_turn(
+            lane,
+            turns,
             max_tokens=clamped,
             temperature=self.temperature if temperature is None else temperature,
             reasoning_effort=reasoning_effort,
-            extra_params=self._provider_extra_params(),
-            capabilities=caps,
-            replay_reasoning_to_model=self._resolve_replay_reasoning_to_model(caps=caps),
         )
         # Utility completions (title gen, compaction, web-fetch extraction)
         # bypass the streaming on_status path — record their usage so the
@@ -7365,10 +7364,10 @@ class ChatSession:
         so the caller can abort the whole compaction before any message swap.
         """
         summary_msgs = [
-            {"role": "system", "content": system_prompt},
-            {"role": "user", "content": self._COMPACT_USER_PREFIX + body},
+            Turn.system(system_prompt),
+            Turn.user(self._COMPACT_USER_PREFIX + body),
         ]
-        result: CompletionResult | None = None
+        result: ModelTurnResult | None = None
         for attempt in range(self._MAX_RETRIES + 1):
             try:
                 result = self._utility_completion(
@@ -15359,7 +15358,7 @@ class ChatSession:
                 # it.  No consumer evaluates ``.text`` on a sub-agent tool turn
                 # today.  The proper by-reference representation needs the
                 # attachment resolver wired into this sub-agent's
-                # ``create_completion`` (it currently isn't) plus content-
+                # ``model_turn`` call (it currently isn't) plus content-
                 # addressed byte storage — deferred to the recall/persist work
                 # where that attachment path is already in scope.
                 agent_turns.append(Turn.tool(tc_dict["id"], output, is_error=is_tool_error))
@@ -16589,24 +16588,18 @@ class ChatSession:
         try:
             result = self._utility_completion(
                 [
-                    {
-                        "role": "system",
-                        "content": (
-                            "You are a web content extraction assistant. "
-                            "Answer the user's question using ONLY the "
-                            "provided page content. Be concise and factual. "
-                            "If the content doesn't contain the answer, say so."
-                        ),
-                    },
-                    {
-                        "role": "user",
-                        "content": (
-                            f"Page URL: {url}\n"
-                            f"Page content ({original_len} chars):\n\n"
-                            f"{text}\n\n---\n"
-                            f"Question: {question}"
-                        ),
-                    },
+                    Turn.system(
+                        "You are a web content extraction assistant. "
+                        "Answer the user's question using ONLY the "
+                        "provided page content. Be concise and factual. "
+                        "If the content doesn't contain the answer, say so."
+                    ),
+                    Turn.user(
+                        f"Page URL: {url}\n"
+                        f"Page content ({original_len} chars):\n\n"
+                        f"{text}\n\n---\n"
+                        f"Question: {question}"
+                    ),
                 ],
                 max_tokens=min(self.max_tokens, self.context_window // 4),
                 reasoning_effort=self.reasoning_effort,

turnstone/core/session_ui_base.py

@@ -3145,7 +3145,7 @@ class SessionUIBase:
 
         Title generation, conversation compaction, web-fetch
         summarisation, and task sub-agents all run through the
-        provider's non-streaming ``create_completion`` and never reach
+        non-streaming ``model_turn`` seam and never reach
         :meth:`on_status` — so without this their tokens are invisible to
         the governance usage dashboard, undercounting real consumption
         (potentially by a large factor for agent-heavy workstreams).

turnstone/eval/core.py

@@ -30,11 +30,12 @@ from typing import Any
 
 from openai import OpenAI
 
+from turnstone.core.model_turn import ModelLane, model_turn
 from turnstone.core.providers import LLMProvider, create_client, create_provider
 from turnstone.core.session import ChatSession
 from turnstone.core.storage import get_storage, init_storage, reset_storage
 from turnstone.core.tools import INTERACTIVE_TOOLS, PRIMARY_KEY_MAP
-from turnstone.core.trajectory import Role, turn_from_dict
+from turnstone.core.trajectory import Role, Turn, turn_from_dict, turns_from_dicts
 
 # Eval evaluates interactive-session agent behaviour — coordinator tools
 # require a console-hosted session and aren't exercised by the harness.
@@ -299,9 +300,22 @@ class HeadlessSession(ChatSession):
             tool: str, args: dict, result: str (truncated), turn: int
         """
         self.tool_call_log = []
-        self.messages.append(turn_from_dict({"role": "user", "content": user_input}))
+        self.messages.append(Turn.user(user_input))
         self._msg_tokens.append(max(1, int(len(user_input) / self._chars_per_token)))
 
+        # The eval lane — resolved once per run, like the sub-agent seam.
+        # ``temperature`` relays the harness's operator-resolved knob per
+        # call below (house rule: relay, never pin).
+        lane = ModelLane(
+            provider=self._provider,
+            client=self.client,
+            model=self.model,
+            alias=self._model_alias or "",
+            capabilities=self._get_capabilities(),
+            extra_params=self._provider_extra_params(),
+            registry=self._registry,
+        )
+
         for turn in range(max_turns):
             if self._cancelled.is_set():
                 break
@@ -310,20 +324,20 @@ class HeadlessSession(ChatSession):
                 _log(f"{log_prefix}  turn {turn}: calling API...", dim=True)
 
             t0 = time.monotonic()
-            msgs = self._full_messages()
+            # System prompts live as wire dicts on the session; bridge them to
+            # Turn IR so the whole trajectory lowers through the shared seam.
+            turns = turns_from_dicts(self.system_messages) + self.messages
 
             if self._cancelled.is_set():
                 break
 
-            result = self._provider.create_completion(
-                client=self.client,
-                model=self.model,
-                messages=msgs,
+            result = model_turn(
+                lane,
+                turns,
                 tools=self._eval_tools,
                 max_tokens=self.max_tokens,
                 temperature=self.temperature,
                 reasoning_effort=self.reasoning_effort,
-                extra_params=self._provider_extra_params(),
             )
             elapsed = time.monotonic() - t0
 
@@ -331,17 +345,21 @@ class HeadlessSession(ChatSession):
                 self._total_usage["prompt"] += result.usage.prompt_tokens
                 self._total_usage["completion"] += result.usage.completion_tokens
 
-            assistant_msg: dict[str, Any] = {
-                "role": "assistant",
-                "content": result.content or None,
-            }
+            # Cap parallel tool calls to prevent degenerate repetition.  The
+            # cap applies to the mirror AND the appended turn; a capped turn
+            # drops its native lane rather than replaying orphan native
+            # tool_use blocks the mirror no longer carries.
+            capped_calls = (result.tool_calls or [])[:10]
+            assistant_turn = result.turn
+            if len(result.tool_calls) > len(capped_calls):
+                assistant_turn = Turn(
+                    role=Role.ASSISTANT,
+                    content=assistant_turn.content,
+                    tool_calls=assistant_turn.tool_calls[: len(capped_calls)],
+                )
 
-            if result.tool_calls:
-                # Cap parallel tool calls to prevent degenerate repetition
-                assistant_msg["tool_calls"] = result.tool_calls[:10]
-
-            self.messages.append(turn_from_dict(assistant_msg))
-            msg_len = len(assistant_msg.get("content") or "")
+            self.messages.append(assistant_turn)
+            msg_len = len(result.content or "")
             self._msg_tokens.append(max(1, int(msg_len / self._chars_per_token)))
 
             # Log usage and content
@@ -355,27 +373,27 @@ class HeadlessSession(ChatSession):
                     f"{log_prefix}  turn {turn}: response in {elapsed:.1f}s{toks}",
                     dim=True,
                 )
-                if assistant_msg["content"]:
-                    text = assistant_msg["content"][:200]
-                    if len(assistant_msg["content"]) > 200:
+                if result.content:
+                    text = result.content[:200]
+                    if len(result.content) > 200:
                         text += "..."
                     _log(f"{log_prefix}    content: {text}", dim=True)
 
-            if not result.tool_calls:
+            if not capped_calls:
                 if verbose:
                     _log(f"{log_prefix}  turn {turn}: no tool calls, done", dim=True)
                 break
 
             # Log tool calls
             if verbose:
-                names = [tc["function"]["name"] for tc in result.tool_calls]
+                names = [tc["function"]["name"] for tc in capped_calls]
                 _log(f"{log_prefix}  turn {turn}: tools -> {names}")
 
             # Execute tools (NullUI discards session output; tools return
             # results as strings, not via stdout)
-            results, _ = self._execute_tools(assistant_msg["tool_calls"])
+            results, _ = self._execute_tools(capped_calls)
 
-            for tc, (tc_id, raw_output) in zip(assistant_msg["tool_calls"], results, strict=False):
+            for tc, (tc_id, raw_output) in zip(capped_calls, results, strict=False):
                 # Flatten list content (image tool results) to text for logging
                 if isinstance(raw_output, list):
                     output = " ".join(
@@ -415,6 +433,9 @@ class HeadlessSession(ChatSession):
                     _log(f"{log_prefix}    {func_name}({arg_summary})", dim=False)
                     _log(f"{log_prefix}      -> {result_preview}", dim=True)
 
+                # ``raw_output`` may be multipart (image tool results carry
+                # by-reference parts) — the dict↔Turn strangler bridge is the
+                # canonical adapter for that shape.
                 tool_msg = {
                     "role": "tool",
                     "tool_call_id": tc_id,

turnstone/optimizer.py

@@ -27,8 +27,10 @@ from typing import Any
 
 from openai import OpenAI
 
+from turnstone.core.model_turn import model_turn, resolve_lane
 from turnstone.core.providers import LLMProvider, create_provider
 from turnstone.core.session import ChatSession
+from turnstone.core.trajectory import Role, Turn
 from turnstone.eval.core import (
     _MCP_ONLY_TOOLS,
     BOLD,
@@ -237,6 +239,9 @@ def _diversify_prompts(
     user_prompt is always included as the first variant.
     """
     prov = provider or create_provider("openai")
+    # Temperature is not pinned (house rule) — sampling diversity for the
+    # paraphraser belongs in the diversifier model's own configuration.
+    lane = resolve_lane(prov, client, model)
     result: dict[str, list[str]] = {}
 
     for ci, case in enumerate(cases):
@@ -291,15 +296,10 @@ def _diversify_prompts(
         )
 
         try:
-            cr = prov.create_completion(
-                client=client,
-                model=model,
-                messages=[
-                    {"role": "system", "content": DIVERSIFIER_SYSTEM},
-                    {"role": "user", "content": user_content},
-                ],
+            cr = model_turn(
+                lane,
+                [Turn.system(DIVERSIFIER_SYSTEM), Turn.user(user_content)],
                 max_tokens=8192,
-                temperature=0.8,
                 reasoning_effort="low",
             )
             raw = (cr.content or "").strip()
@@ -447,15 +447,10 @@ def _observe_and_update_optimizer(
     )
 
     prov = provider or create_provider("openai")
-    cr = prov.create_completion(
-        client=client,
-        model=model,
-        messages=[
-            {"role": "system", "content": OBSERVER_SYSTEM},
-            {"role": "user", "content": user_content},
-        ],
+    cr = model_turn(
+        resolve_lane(prov, client, model),
+        [Turn.system(OBSERVER_SYSTEM), Turn.user(user_content)],
         max_tokens=8192,
-        temperature=0.3,
         reasoning_effort="low",
     )
 
@@ -762,52 +757,50 @@ def _run_analyst(
         )
 
     prov = provider or create_provider("openai")
-    messages: list[dict[str, Any]] = [
-        {"role": "system", "content": analyst_system},
-        {"role": "user", "content": user_content},
+    turns: list[Turn] = [
+        Turn.system(analyst_system),
+        Turn.user(user_content),
     ]
 
-    # Multi-turn loop: let the analyst call tools up to 5 rounds
+    # Multi-turn loop: let the analyst call tools up to 5 rounds.
+    # Temperature is not pinned (house rule) — the analyst model's own
+    # configuration governs sampling.
+    lane = resolve_lane(prov, client, model)
     max_turns = 5
     for _turn in range(max_turns):
-        cr = prov.create_completion(
-            client=client,
-            model=model,
-            messages=messages,
+        mtr = model_turn(
+            lane,
+            turns,
             tools=_ANALYST_TOOLS,
             max_tokens=8192,
-            temperature=0.3,
             reasoning_effort="medium",
         )
 
-        assistant_msg: dict[str, Any] = {
-            "role": "assistant",
-            "content": cr.content or None,
-        }
-        if cr.tool_calls:
-            assistant_msg["tool_calls"] = cr.tool_calls[:5]
-        messages.append(assistant_msg)
+        # Same degenerate-repetition cap as before; a capped turn drops its
+        # native lane rather than replay orphan native tool blocks the
+        # mirror no longer carries.
+        capped = mtr.tool_calls[:5]
+        assistant_turn = mtr.turn
+        if len(mtr.tool_calls) > len(capped):
+            assistant_turn = Turn.assistant(
+                mtr.content, tool_calls=mtr.turn.tool_calls[: len(capped)]
+            )
+        turns.append(assistant_turn)
 
-        if not cr.tool_calls:
+        if not capped:
             break
 
         # Execute tool calls
-        for tc in assistant_msg["tool_calls"]:
+        for tc in capped:
             func_name = tc["function"]["name"]
             output = _exec_analyst_tool(func_name, tc["function"]["arguments"])
-            messages.append(
-                {
-                    "role": "tool",
-                    "tool_call_id": tc["id"],
-                    "content": output,
-                }
-            )
+            turns.append(Turn.tool(tc["id"], output))
 
     # Extract final text response
     result = ""
-    for msg in reversed(messages):
-        if msg["role"] == "assistant" and msg.get("content"):
-            result = msg["content"]
+    for t in reversed(turns):
+        if t.role is Role.ASSISTANT and t.text:
+            result = t.text
             break
 
     # Strip reasoning tags if present
@@ -920,15 +913,10 @@ def _propose_tool_overrides(
     )
 
     prov = provider or create_provider("openai")
-    cr = prov.create_completion(
-        client=client,
-        model=model,
-        messages=[
-            {"role": "system", "content": TOOL_OPTIMIZER_SYSTEM},
-            {"role": "user", "content": user_content},
-        ],
+    cr = model_turn(
+        resolve_lane(prov, client, model),
+        [Turn.system(TOOL_OPTIMIZER_SYSTEM), Turn.user(user_content)],
         max_tokens=8192,
-        temperature=0.3,
         reasoning_effort="medium",
     )
 
@@ -1067,15 +1055,10 @@ def _propose_prompt_modification(
     )
 
     prov = provider or create_provider("openai")
-    cr = prov.create_completion(
-        client=client,
-        model=model,
-        messages=[
-            {"role": "system", "content": optimizer_system},
-            {"role": "user", "content": user_content},
-        ],
+    cr = model_turn(
+        resolve_lane(prov, client, model),
+        [Turn.system(optimizer_system), Turn.user(user_content)],
         max_tokens=16384,
-        temperature=0.6,
         reasoning_effort="medium",
     )