From dc88060b79073a79562069932895a2fe3fef3206 Mon Sep 17 00:00:00 2001
From: Patrick Buckley <eous@users.noreply.github.com>
Date: Wed, 3 Jun 2026 03:53:43 -0700
Subject: [PATCH] refactor(core): session.messages is the canonical Turn
 trajectory
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

ChatSession.messages flips from list[dict] to list[Turn] — the in-memory
canonical trajectory.  Reads migrate to typed fields (turn.role, turn.text,
turn.tool_calls); appends and assignments go through turn_from_dict /
turns_from_dicts; the fork bulk-save and retry's multipart check read via
turn_to_dict.  _full_messages lowers Turns→dicts at the wire boundary — the
fold/repair and provider translators still consume dicts until the next slice.
The token-accounting helpers accept a dict or a Turn.

Non-session consumers migrate too: coordinator_idle_observer and eval to typed
fields (mypy-enumerated), and server's last-assistant extractor via turn_to_dict
(an Any-typed call site mypy could not flag).  An all-text multipart content
list (the unreadable-attachment placeholder path) now round-trips faithfully
through the adapter (single text block → str, multiple → list).

Tests that inspected session.messages as dicts read it through the
dicts_from_turns / turn_to_dict bridge; those that built it pass dicts through
turns_from_dicts / turn_from_dict.  Byte-identical wire harness; full non-live
suite green (7130).
---
 tests/test_cancel.py                          |  72 ++++---
 tests/test_coordinator_idle_observer.py       | 176 +++++++++++-------
 tests/test_idle_nudge_wake_integration.py     |  15 +-
 tests/test_memory_relevance.py                |  17 +-
 tests/test_notify_completion.py               |  65 ++++---
 tests/test_rewind_retry.py                    | 103 +++++-----
 tests/test_session.py                         | 162 +++++++++-------
 tests/test_session_attachments.py             |  24 +--
 tests/test_sessions.py                        |   3 +-
 tests/test_tool_truncation.py                 |   9 +-
 tests/test_trajectory.py                      |   9 +
 tests/test_watch_integration.py               |  13 +-
 .../console/coordinator_idle_observer.py      |  10 +-
 turnstone/core/session.py                     | 106 ++++++-----
 turnstone/core/trajectory.py                  |  11 +-
 turnstone/eval.py                             |  11 +-
 turnstone/server.py                           |   4 +-
 17 files changed, 471 insertions(+), 339 deletions(-)

diff --git a/tests/test_cancel.py b/tests/test_cancel.py
index 257e78d6..fa9f9261 100644
--- a/tests/test_cancel.py
+++ b/tests/test_cancel.py
@@ -9,6 +9,7 @@ from unittest.mock import MagicMock, patch
 import pytest
 
 from turnstone.core.session import ChatSession, GenerationCancelled, _CancelRef
+from turnstone.core.trajectory import dicts_from_turns, turn_from_dict
 
 
 class NullUI:
@@ -191,7 +192,7 @@ class TestCancelDuringStreaming:
         # raw "Hello world" without a marker would look like the
         # final assistant answer to a coord LLM reading the child's
         # transcript.
-        assistant_msgs = [m for m in session.messages if m["role"] == "assistant"]
+        assistant_msgs = [m for m in dicts_from_turns(session.messages) if m["role"] == "assistant"]
         assert len(assistant_msgs) == 1
         content = assistant_msgs[0]["content"]
         assert content.startswith("Hello world")
@@ -261,13 +262,14 @@ class TestCancelDuringToolExecution:
         # Session should be idle
         assert ui.states[-1] == "idle"
         # Cancelled tool calls should have synthesized results
-        tool_msgs = [m for m in session.messages if m["role"] == "tool"]
+        msgs = dicts_from_turns(session.messages)
+        tool_msgs = [m for m in msgs if m["role"] == "tool"]
         assert len(tool_msgs) == 1
         assert tool_msgs[0]["tool_call_id"] == "tc_1"
         assert "Cancelled by user" in tool_msgs[0]["content"]
         assert tool_msgs[0].get("is_error") is True
         # The assistant message with tool_calls should still be present
-        assistant_msgs = [m for m in session.messages if m.get("tool_calls")]
+        assistant_msgs = [m for m in msgs if m.get("tool_calls")]
         assert len(assistant_msgs) == 1
 
 
@@ -297,7 +299,7 @@ class TestCancelWhenIdle:
             session.send("hello")
 
         # Should complete normally
-        assistant_msgs = [m for m in session.messages if m["role"] == "assistant"]
+        assistant_msgs = [m for m in dicts_from_turns(session.messages) if m["role"] == "assistant"]
         assert len(assistant_msgs) == 1
         assert assistant_msgs[0]["content"] == "ok"
 
@@ -537,7 +539,7 @@ class TestStreamAbort:
         assert any("cancelled" in i.lower() for i in ui.infos)
         # Partial content preserved AND annotated with the
         # cancelled-before-completion marker.
-        assistant_msgs = [m for m in session.messages if m["role"] == "assistant"]
+        assistant_msgs = [m for m in dicts_from_turns(session.messages) if m["role"] == "assistant"]
         assert len(assistant_msgs) == 1
         content = assistant_msgs[0]["content"]
         assert content.startswith("Hello")
@@ -783,7 +785,7 @@ class TestForceCancelThreaded:
         assert old_done.wait(timeout=10), "orphaned thread did not exit"
 
         # The orphaned thread should NOT have appended its content
-        assistant_msgs = [m for m in session.messages if m["role"] == "assistant"]
+        assistant_msgs = [m for m in dicts_from_turns(session.messages) if m["role"] == "assistant"]
         # May have partial content from before cancel, but NOT the full
         # "Old content more" that would appear without the generation guard
         for msg in assistant_msgs:
@@ -834,7 +836,7 @@ class TestForceCancelThreaded:
 
         # The new generation should have completed successfully
         assert "idle" in ui.states
-        assistant_msgs = [m for m in session.messages if m["role"] == "assistant"]
+        assistant_msgs = [m for m in dicts_from_turns(session.messages) if m["role"] == "assistant"]
         assert any("Fresh response" in m.get("content", "") for m in assistant_msgs)
 
 
@@ -864,14 +866,16 @@ class TestSynthesizeCancelledResults:
         ui = self._ui_with_tool_result_tracking()
         session = _make_session(ui=ui)
         session.messages.append(
-            {
-                "role": "assistant",
-                "content": "calling tools",
-                "tool_calls": [
-                    {"id": "call_a", "function": {"name": "search", "arguments": "{}"}},
-                    {"id": "call_b", "function": {"name": "compute", "arguments": "{}"}},
-                ],
-            },
+            turn_from_dict(
+                {
+                    "role": "assistant",
+                    "content": "calling tools",
+                    "tool_calls": [
+                        {"id": "call_a", "function": {"name": "search", "arguments": "{}"}},
+                        {"id": "call_b", "function": {"name": "compute", "arguments": "{}"}},
+                    ],
+                },
+            )
         )
         session._msg_tokens.append(1)
 
@@ -888,25 +892,29 @@ class TestSynthesizeCancelledResults:
         assert all(tr[2] == "Cancelled by user." for tr in ui.tool_results)
         # And the message list has the synthesized tool entries
         # (preserves the prior contract).
-        tool_msgs = [m for m in session.messages if m.get("role") == "tool"]
+        tool_msgs = [m for m in dicts_from_turns(session.messages) if m.get("role") == "tool"]
         assert len(tool_msgs) == 2
 
     def test_skips_calls_already_answered(self, tmp_db):
         ui = self._ui_with_tool_result_tracking()
         session = _make_session(ui=ui)
         session.messages.append(
-            {
-                "role": "assistant",
-                "tool_calls": [
-                    {"id": "call_a", "function": {"name": "search", "arguments": "{}"}},
-                    {"id": "call_b", "function": {"name": "compute", "arguments": "{}"}},
-                ],
-            },
+            turn_from_dict(
+                {
+                    "role": "assistant",
+                    "tool_calls": [
+                        {"id": "call_a", "function": {"name": "search", "arguments": "{}"}},
+                        {"id": "call_b", "function": {"name": "compute", "arguments": "{}"}},
+                    ],
+                },
+            )
         )
         session._msg_tokens.append(1)
         # call_a already answered.
         session.messages.append(
-            {"role": "tool", "tool_call_id": "call_a", "content": "result"},
+            turn_from_dict(
+                {"role": "tool", "tool_call_id": "call_a", "content": "result"},
+            )
         )
         session._msg_tokens.append(1)
 
@@ -928,17 +936,19 @@ class TestSynthesizeCancelledResults:
         ui = _ExplodingUI()
         session = _make_session(ui=ui)
         session.messages.append(
-            {
-                "role": "assistant",
-                "tool_calls": [
-                    {"id": "call_a", "function": {"name": "search", "arguments": "{}"}},
-                ],
-            },
+            turn_from_dict(
+                {
+                    "role": "assistant",
+                    "tool_calls": [
+                        {"id": "call_a", "function": {"name": "search", "arguments": "{}"}},
+                    ],
+                },
+            )
         )
         session._msg_tokens.append(1)
 
         # Must not raise.
         session._synthesize_cancelled_results("Cancelled by user.")
 
-        tool_msgs = [m for m in session.messages if m.get("role") == "tool"]
+        tool_msgs = [m for m in dicts_from_turns(session.messages) if m.get("role") == "tool"]
         assert len(tool_msgs) == 1
diff --git a/tests/test_coordinator_idle_observer.py b/tests/test_coordinator_idle_observer.py
index a3db4bca..67cd5628 100644
--- a/tests/test_coordinator_idle_observer.py
+++ b/tests/test_coordinator_idle_observer.py
@@ -16,6 +16,7 @@ import pytest
 
 from turnstone.console.coordinator_idle_observer import CoordinatorIdleObserver
 from turnstone.core.nudge_queue import NudgeQueue
+from turnstone.core.trajectory import Turn, turns_from_dicts
 from turnstone.core.workstream import WorkstreamKind, WorkstreamState
 
 
@@ -73,7 +74,7 @@ class _FakeStorage:
 class _FakeSession:
     def __init__(self) -> None:
         self._nudge_queue = NudgeQueue()
-        self.messages: list[dict[str, Any]] = []
+        self.messages: list[Turn] = []
         self._wake_source_tag: str = ""
         self._metacog_state: dict[str, float] = {}
         self._mem_cfg = MagicMock(nudge_cooldown=300)
@@ -150,10 +151,12 @@ class TestEnqueueOnIdle:
         mgr, storage, ws = coord_setup
         _add_active_child(storage, ws_id="child-a", state="running")
         _add_active_child(storage, ws_id="child-b", state="thinking")
-        ws.session.messages = [
-            {"role": "user", "content": "go"},
-            {"role": "assistant", "content": "ok"},
-        ]
+        ws.session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "go"},
+                {"role": "assistant", "content": "ok"},
+            ]
+        )
 
         observer = CoordinatorIdleObserver(mgr, storage)
         observer.start()
@@ -181,10 +184,12 @@ class TestEnqueueOnIdle:
         _add_active_child(storage, state="closed")
         _add_active_child(storage, state="error")
         # ≥2 messages so should_nudge's message_count > 1 gate clears.
-        ws.session.messages = [
-            {"role": "user", "content": "go"},
-            {"role": "assistant", "content": "ok"},
-        ]
+        ws.session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "go"},
+                {"role": "assistant", "content": "ok"},
+            ]
+        )
         observer = CoordinatorIdleObserver(mgr, storage)
         observer.start()
         mgr.fire_state(ws.id, WorkstreamState.IDLE)
@@ -225,19 +230,21 @@ class TestWaitForWorkstreamSkip:
     def test_skips_when_last_assistant_used_wait(self, coord_setup):
         mgr, storage, ws = coord_setup
         _add_active_child(storage)
-        ws.session.messages = [
-            {"role": "user", "content": "kick off"},
-            {
-                "role": "assistant",
-                "content": None,
-                "tool_calls": [
-                    {
-                        "id": "call-1",
-                        "function": {"name": "wait_for_workstream", "arguments": "{}"},
-                    }
-                ],
-            },
-        ]
+        ws.session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "kick off"},
+                {
+                    "role": "assistant",
+                    "content": None,
+                    "tool_calls": [
+                        {
+                            "id": "call-1",
+                            "function": {"name": "wait_for_workstream", "arguments": "{}"},
+                        }
+                    ],
+                },
+            ]
+        )
         observer = CoordinatorIdleObserver(mgr, storage)
         observer.start()
         mgr.fire_state(ws.id, WorkstreamState.IDLE)
@@ -247,16 +254,21 @@ class TestWaitForWorkstreamSkip:
     def test_fires_when_last_assistant_used_different_tool(self, coord_setup):
         mgr, storage, ws = coord_setup
         _add_active_child(storage)
-        ws.session.messages = [
-            {"role": "user", "content": "go"},
-            {
-                "role": "assistant",
-                "content": None,
-                "tool_calls": [
-                    {"id": "call-1", "function": {"name": "spawn_workstream", "arguments": "{}"}}
-                ],
-            },
-        ]
+        ws.session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "go"},
+                {
+                    "role": "assistant",
+                    "content": None,
+                    "tool_calls": [
+                        {
+                            "id": "call-1",
+                            "function": {"name": "spawn_workstream", "arguments": "{}"},
+                        }
+                    ],
+                },
+            ]
+        )
         observer = CoordinatorIdleObserver(mgr, storage)
         observer.start()
         mgr.fire_state(ws.id, WorkstreamState.IDLE)
@@ -268,10 +280,12 @@ class TestHardCap:
         mgr, storage, ws = coord_setup
         _add_active_child(storage)
         # ≥2 messages so should_nudge's message_count > 1 gate clears.
-        ws.session.messages = [
-            {"role": "user", "content": "go"},
-            {"role": "assistant", "content": "ok"},
-        ]
+        ws.session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "go"},
+                {"role": "assistant", "content": "ok"},
+            ]
+        )
         observer = CoordinatorIdleObserver(mgr, storage)
         observer.start()
 
@@ -292,10 +306,12 @@ class TestHardCap:
         mgr, storage, ws = coord_setup
         _add_active_child(storage)
         # ≥2 messages so should_nudge's message_count > 1 gate clears.
-        ws.session.messages = [
-            {"role": "user", "content": "go"},
-            {"role": "assistant", "content": "ok"},
-        ]
+        ws.session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "go"},
+                {"role": "assistant", "content": "ok"},
+            ]
+        )
         observer = CoordinatorIdleObserver(mgr, storage)
         observer.start()
 
@@ -321,10 +337,12 @@ class TestHardCap:
         mgr, storage, ws = coord_setup
         _add_active_child(storage)
         # ≥2 messages so should_nudge's message_count > 1 gate clears.
-        ws.session.messages = [
-            {"role": "user", "content": "go"},
-            {"role": "assistant", "content": "ok"},
-        ]
+        ws.session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "go"},
+                {"role": "assistant", "content": "ok"},
+            ]
+        )
         observer = CoordinatorIdleObserver(mgr, storage)
         observer.start()
 
@@ -350,10 +368,12 @@ class TestCooldown:
         mgr, storage, ws = coord_setup
         _add_active_child(storage)
         # ≥2 messages so should_nudge's message_count > 1 gate clears.
-        ws.session.messages = [
-            {"role": "user", "content": "go"},
-            {"role": "assistant", "content": "ok"},
-        ]
+        ws.session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "go"},
+                {"role": "assistant", "content": "ok"},
+            ]
+        )
         observer = CoordinatorIdleObserver(mgr, storage)
         observer.start()
 
@@ -372,10 +392,12 @@ class TestStorageFailure:
     def test_storage_exception_is_swallowed(self, coord_setup):
         mgr, storage, ws = coord_setup
         # ≥2 messages so should_nudge's message_count > 1 gate clears.
-        ws.session.messages = [
-            {"role": "user", "content": "go"},
-            {"role": "assistant", "content": "ok"},
-        ]
+        ws.session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "go"},
+                {"role": "assistant", "content": "ok"},
+            ]
+        )
         storage.list_raises = True
         observer = CoordinatorIdleObserver(mgr, storage)
         observer.start()
@@ -389,10 +411,12 @@ class TestValidUntilPredicate:
         mgr, storage, ws = coord_setup
         _add_active_child(storage, ws_id="child-a", state="running")
         # ≥2 messages so should_nudge's message_count > 1 gate clears.
-        ws.session.messages = [
-            {"role": "user", "content": "go"},
-            {"role": "assistant", "content": "ok"},
-        ]
+        ws.session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "go"},
+                {"role": "assistant", "content": "ok"},
+            ]
+        )
         observer = CoordinatorIdleObserver(mgr, storage)
         observer.start()
         mgr.fire_state(ws.id, WorkstreamState.IDLE)
@@ -413,10 +437,12 @@ class TestValidUntilPredicate:
         mgr, storage, ws = coord_setup
         _add_active_child(storage, ws_id="child-a", state="running")
         # ≥2 messages so should_nudge's message_count > 1 gate clears.
-        ws.session.messages = [
-            {"role": "user", "content": "go"},
-            {"role": "assistant", "content": "ok"},
-        ]
+        ws.session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "go"},
+                {"role": "assistant", "content": "ok"},
+            ]
+        )
         observer = CoordinatorIdleObserver(mgr, storage)
         observer.start()
         mgr.fire_state(ws.id, WorkstreamState.IDLE)
@@ -432,10 +458,12 @@ class TestValidUntilPredicate:
         mgr, storage, ws = coord_setup
         _add_active_child(storage)
         # ≥2 messages so should_nudge's message_count > 1 gate clears.
-        ws.session.messages = [
-            {"role": "user", "content": "go"},
-            {"role": "assistant", "content": "ok"},
-        ]
+        ws.session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "go"},
+                {"role": "assistant", "content": "ok"},
+            ]
+        )
         observer = CoordinatorIdleObserver(mgr, storage)
         observer.start()
         mgr.fire_state(ws.id, WorkstreamState.IDLE)
@@ -456,10 +484,12 @@ class TestLifecycle:
         mgr, storage, ws = coord_setup
         _add_active_child(storage)
         # ≥2 messages so should_nudge's message_count > 1 gate clears.
-        ws.session.messages = [
-            {"role": "user", "content": "go"},
-            {"role": "assistant", "content": "ok"},
-        ]
+        ws.session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "go"},
+                {"role": "assistant", "content": "ok"},
+            ]
+        )
         observer = CoordinatorIdleObserver(mgr, storage)
         observer.start()
         observer.start()  # no-op
@@ -471,10 +501,12 @@ class TestLifecycle:
         mgr, storage, ws = coord_setup
         _add_active_child(storage)
         # ≥2 messages so should_nudge's message_count > 1 gate clears.
-        ws.session.messages = [
-            {"role": "user", "content": "go"},
-            {"role": "assistant", "content": "ok"},
-        ]
+        ws.session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "go"},
+                {"role": "assistant", "content": "ok"},
+            ]
+        )
         observer = CoordinatorIdleObserver(mgr, storage)
         observer.start()
         observer.shutdown()
diff --git a/tests/test_idle_nudge_wake_integration.py b/tests/test_idle_nudge_wake_integration.py
index 2adcf2f6..640a8aef 100644
--- a/tests/test_idle_nudge_wake_integration.py
+++ b/tests/test_idle_nudge_wake_integration.py
@@ -32,6 +32,7 @@ from tests.test_session_manager import FakeStorage
 from turnstone.core.idle_nudge_watcher import IdleNudgeWatcher
 from turnstone.core.session import ChatSession
 from turnstone.core.session_manager import SessionManager
+from turnstone.core.trajectory import dicts_from_turns, turn_from_dict
 from turnstone.core.workstream import Workstream, WorkstreamKind, WorkstreamState
 
 # ---------------------------------------------------------------------------
@@ -254,13 +255,14 @@ def test_idle_event_through_real_session_manager_drives_wake_send(real_mgr, tmp_
         # The synthesized empty user message landed in history with the
         # ``_source`` audit tag; the nudge follows it as a first-class
         # ``system`` turn (no _reminders side-channel).
-        user_msgs = [m for m in ws.session.messages if m.get("role") == "user"]
+        msgs = dicts_from_turns(ws.session.messages)
+        user_msgs = [m for m in msgs if m.get("role") == "user"]
         assert user_msgs, "expected a synthesized user message from the wake"
         wake_msg = user_msgs[-1]
         assert wake_msg["content"] == ""
         assert wake_msg.get("_source") == "system_nudge"
         assert "_reminders" not in wake_msg
-        sys_turns = [m for m in ws.session.messages if m.get("role") == "system"]
+        sys_turns = [m for m in msgs if m.get("role") == "system"]
         assert {
             "role": "system",
             "_source": "idle_children",
@@ -367,8 +369,8 @@ def test_coord_idle_with_active_children_emits_envelope_via_real_managers(coord_
 
         # Pretend the coord has already had a real conversation so
         # ``should_nudge``'s message_count > 1 gate passes.
-        coord.session.messages.append({"role": "user", "content": "spawn 2"})
-        coord.session.messages.append({"role": "assistant", "content": "ok"})
+        coord.session.messages.append(turn_from_dict({"role": "user", "content": "spawn 2"}))
+        coord.session.messages.append(turn_from_dict({"role": "assistant", "content": "ok"}))
 
         with (
             patch.object(coord.session, "_create_stream_with_retry", return_value=iter([])),
@@ -392,11 +394,12 @@ def test_coord_idle_with_active_children_emits_envelope_via_real_managers(coord_
         # The synthetic empty-user turn landed; the idle_children nudge
         # follows it as a first-class ``system`` turn containing both
         # children.
-        user_msgs = [m for m in coord.session.messages if m.get("role") == "user"]
+        msgs = dicts_from_turns(coord.session.messages)
+        user_msgs = [m for m in msgs if m.get("role") == "user"]
         wake_msg = user_msgs[-1]
         assert wake_msg["content"] == ""
         assert wake_msg.get("_source") == "system_nudge"
-        sys_turns = [m for m in coord.session.messages if m.get("role") == "system"]
+        sys_turns = [m for m in msgs if m.get("role") == "system"]
         idle_turns = [m for m in sys_turns if m["_source"] == "idle_children"]
         assert len(idle_turns) == 1
         text = idle_turns[0]["content"]
diff --git a/tests/test_memory_relevance.py b/tests/test_memory_relevance.py
index 4bfce98e..3d935a4a 100644
--- a/tests/test_memory_relevance.py
+++ b/tests/test_memory_relevance.py
@@ -8,6 +8,7 @@ from turnstone.core.memory_relevance import (
     extract_recent_context,
     score_memories,
 )
+from turnstone.core.trajectory import turns_from_dicts
 
 # ---------------------------------------------------------------------------
 # score_memories
@@ -303,7 +304,9 @@ class TestCompositionCandidateSelection:
     def test_recency_ceiling_regression(self, tmp_db):
         """Old relevant memory not in recency top-N still injected via search path."""
         session = _make_session(fetch_limit=5, relevance_k=3)
-        session.messages = [{"role": "user", "content": "postgres database configuration"}]
+        session.messages = turns_from_dicts(
+            [{"role": "user", "content": "postgres database configuration"}]
+        )
 
         old_mem = _make_mem(
             "ancient_db_config",
@@ -343,7 +346,7 @@ class TestCompositionCandidateSelection:
     def test_sparse_match_union_fills_candidate_pool(self, tmp_db):
         """Search returning < fetch_limit results unions with recency fillers."""
         session = _make_session(fetch_limit=5, relevance_k=4)
-        session.messages = [{"role": "user", "content": "unique_term xyzzy"}]
+        session.messages = turns_from_dicts([{"role": "user", "content": "unique_term xyzzy"}])
 
         hit_a = _make_mem("hit_alpha", content="unique_term xyzzy alpha", memory_id="m_ha")
         hit_b = _make_mem("hit_beta", content="unique_term xyzzy beta", memory_id="m_hb")
@@ -374,7 +377,7 @@ class TestCompositionCandidateSelection:
         evict the recency-only memory the bug had been surfacing.
         """
         session = _make_session(fetch_limit=10, relevance_k=3)
-        session.messages = [{"role": "user", "content": "configure host"}]
+        session.messages = turns_from_dicts([{"role": "user", "content": "configure host"}])
 
         # Search returns relevance_k=3 noise hits — enough to skip recency
         # under the OLD threshold, not enough to fill fetch_limit=10.
@@ -409,7 +412,7 @@ class TestCompositionCandidateSelection:
         sets out to improve.
         """
         session = _make_session(fetch_limit=10, relevance_k=3)
-        session.messages = [{"role": "user", "content": "alpha"}]
+        session.messages = turns_from_dicts([{"role": "user", "content": "alpha"}])
 
         # 5 search hits, none of which appear in recency.
         search_hits = [
@@ -451,7 +454,7 @@ class TestCompositionCandidateSelection:
         scopes = coord._visible_scopes()
         assert scopes == [("coordinator", "coord-1")]
         # And: search uses those same scopes (no global/user fan-in)
-        coord.messages = [{"role": "user", "content": "anything"}]
+        coord.messages = turns_from_dicts([{"role": "user", "content": "anything"}])
         with patch(
             "turnstone.core.session.search_visible_structured_memories",
             return_value=[],
@@ -492,13 +495,13 @@ class TestCompositionRerankFiltersWiring:
 
     def test_threshold_zero_uses_reorder_mode(self, tmp_db):
         session = _make_session()
-        session.messages = [{"role": "user", "content": "alpha"}]
+        session.messages = turns_from_dicts([{"role": "user", "content": "alpha"}])
         # threshold 0 (disabled floor) -> reorder mode -> no suppression.
         assert self._capture_rerank_filters(session, 0.0) is False
 
     def test_positive_threshold_uses_filter_mode(self, tmp_db):
         session = _make_session()
-        session.messages = [{"role": "user", "content": "alpha"}]
+        session.messages = turns_from_dicts([{"role": "user", "content": "alpha"}])
         # An active floor -> filter mode -> the reranker may empty the injection.
         assert self._capture_rerank_filters(session, 0.5) is True
 
diff --git a/tests/test_notify_completion.py b/tests/test_notify_completion.py
index 981134fb..1e4e8767 100644
--- a/tests/test_notify_completion.py
+++ b/tests/test_notify_completion.py
@@ -29,6 +29,7 @@ from turnstone.console.server import (
 )
 from turnstone.core.auth import AuthResult
 from turnstone.core.storage._sqlite import SQLiteBackend
+from turnstone.core.trajectory import turns_from_dicts
 from turnstone.server import (
     _deliver_notification,
     _extract_last_assistant_content,
@@ -210,23 +211,27 @@ class TestValidateNotifyTargets:
 class TestExtractLastAssistantContent:
     def test_string_content(self):
         session = MagicMock()
-        session.messages = [
-            {"role": "user", "content": "hello"},
-            {"role": "assistant", "content": "world"},
-        ]
+        session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "hello"},
+                {"role": "assistant", "content": "world"},
+            ]
+        )
         assert _extract_last_assistant_content(session) == "world"
 
     def test_structured_content(self):
         session = MagicMock()
-        session.messages = [
-            {
-                "role": "assistant",
-                "content": [
-                    {"type": "text", "text": "part one"},
-                    {"type": "text", "text": "part two"},
-                ],
-            },
-        ]
+        session.messages = turns_from_dicts(
+            [
+                {
+                    "role": "assistant",
+                    "content": [
+                        {"type": "text", "text": "part one"},
+                        {"type": "text", "text": "part two"},
+                    ],
+                },
+            ]
+        )
         assert _extract_last_assistant_content(session) == "part one\npart two"
 
     def test_empty_messages(self):
@@ -236,29 +241,33 @@ class TestExtractLastAssistantContent:
 
     def test_no_assistant_messages(self):
         session = MagicMock()
-        session.messages = [{"role": "user", "content": "hello"}]
+        session.messages = turns_from_dicts([{"role": "user", "content": "hello"}])
         assert _extract_last_assistant_content(session) == ""
 
     def test_picks_last_assistant(self):
         session = MagicMock()
-        session.messages = [
-            {"role": "assistant", "content": "first"},
-            {"role": "user", "content": "question"},
-            {"role": "assistant", "content": "second"},
-        ]
+        session.messages = turns_from_dicts(
+            [
+                {"role": "assistant", "content": "first"},
+                {"role": "user", "content": "question"},
+                {"role": "assistant", "content": "second"},
+            ]
+        )
         assert _extract_last_assistant_content(session) == "second"
 
     def test_skips_non_text_blocks(self):
         session = MagicMock()
-        session.messages = [
-            {
-                "role": "assistant",
-                "content": [
-                    {"type": "tool_use", "id": "123"},
-                    {"type": "text", "text": "result"},
-                ],
-            },
-        ]
+        session.messages = turns_from_dicts(
+            [
+                {
+                    "role": "assistant",
+                    "content": [
+                        {"type": "tool_use", "id": "123"},
+                        {"type": "text", "text": "result"},
+                    ],
+                },
+            ]
+        )
         assert _extract_last_assistant_content(session) == "result"
 
 
diff --git a/tests/test_rewind_retry.py b/tests/test_rewind_retry.py
index 0f13d7c5..542b1049 100644
--- a/tests/test_rewind_retry.py
+++ b/tests/test_rewind_retry.py
@@ -5,6 +5,7 @@ from __future__ import annotations
 from unittest.mock import MagicMock
 
 from turnstone.core.session import ChatSession
+from turnstone.core.trajectory import turn_to_dict, turns_from_dicts
 
 # ---------------------------------------------------------------------------
 # Helpers
@@ -90,31 +91,35 @@ def _make_session(tmp_db) -> ChatSession:
 
 def _populate_simple(session: ChatSession) -> None:
     """Populate with 2 simple turns (no tool calls)."""
-    session.messages = [
-        {"role": "user", "content": "Hello"},
-        {"role": "assistant", "content": "Hi there!"},
-        {"role": "user", "content": "How are you?"},
-        {"role": "assistant", "content": "I'm fine."},
-    ]
+    session.messages = turns_from_dicts(
+        [
+            {"role": "user", "content": "Hello"},
+            {"role": "assistant", "content": "Hi there!"},
+            {"role": "user", "content": "How are you?"},
+            {"role": "assistant", "content": "I'm fine."},
+        ]
+    )
     session._msg_tokens = [10, 20, 10, 20]
 
 
 def _populate_with_tools(session: ChatSession) -> None:
     """Populate with 2 turns, first has tool calls."""
-    session.messages = [
-        {"role": "user", "content": "Write a test"},
-        {
-            "role": "assistant",
-            "content": "",
-            "tool_calls": [
-                {"id": "tc1", "function": {"name": "bash", "arguments": '{"cmd":"echo hi"}'}}
-            ],
-        },
-        {"role": "tool", "tool_call_id": "tc1", "content": "hi"},
-        {"role": "assistant", "content": "Done."},
-        {"role": "user", "content": "Fix the import"},
-        {"role": "assistant", "content": "Fixed."},
-    ]
+    session.messages = turns_from_dicts(
+        [
+            {"role": "user", "content": "Write a test"},
+            {
+                "role": "assistant",
+                "content": "",
+                "tool_calls": [
+                    {"id": "tc1", "function": {"name": "bash", "arguments": '{"cmd":"echo hi"}'}}
+                ],
+            },
+            {"role": "tool", "tool_call_id": "tc1", "content": "hi"},
+            {"role": "assistant", "content": "Done."},
+            {"role": "user", "content": "Fix the import"},
+            {"role": "assistant", "content": "Fixed."},
+        ]
+    )
     session._msg_tokens = [10, 20, 10, 20, 10, 20]
 
 
@@ -130,10 +135,12 @@ class TestFindTurnBoundaries:
 
     def test_single_turn(self, tmp_db):
         session = _make_session(tmp_db)
-        session.messages = [
-            {"role": "user", "content": "Hello"},
-            {"role": "assistant", "content": "Hi!"},
-        ]
+        session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "Hello"},
+                {"role": "assistant", "content": "Hi!"},
+            ]
+        )
         assert session._find_turn_boundaries() == [0]
 
     def test_multi_turn(self, tmp_db):
@@ -165,8 +172,8 @@ class TestRewind:
         removed = session.rewind(1)
         assert removed == 2  # user + assistant
         assert len(session.messages) == 2
-        assert session.messages[0]["content"] == "Hello"
-        assert session.messages[1]["content"] == "Hi there!"
+        assert turn_to_dict(session.messages[0])["content"] == "Hello"
+        assert turn_to_dict(session.messages[1])["content"] == "Hi there!"
         assert len(session._msg_tokens) == 2
 
     def test_rewind_all_turns(self, tmp_db):
@@ -196,7 +203,7 @@ class TestRewind:
         removed = session.rewind(1)
         assert removed == 2  # user "Fix the import" + assistant "Fixed."
         assert len(session.messages) == 4
-        assert session.messages[-1]["content"] == "Done."
+        assert turn_to_dict(session.messages[-1])["content"] == "Done."
 
     def test_rewind_tokens_sync(self, tmp_db):
         """_msg_tokens stays in sync with messages."""
@@ -219,7 +226,7 @@ class TestRetry:
         assert msg == "How are you?"
         # Only Turn 1 remains, without the second user message
         assert len(session.messages) == 2
-        assert session.messages[-1]["content"] == "Hi there!"
+        assert turn_to_dict(session.messages[-1])["content"] == "Hi there!"
 
     def test_retry_empty(self, tmp_db):
         session = _make_session(tmp_db)
@@ -249,10 +256,18 @@ class TestRetry:
     def test_retry_multipart_content_returns_none(self, tmp_db):
         """retry() should refuse multipart (vision/image) messages."""
         session = _make_session(tmp_db)
-        session.messages = [
-            {"role": "user", "content": [{"type": "text", "text": "describe this"}]},
-            {"role": "assistant", "content": "It's an image."},
-        ]
+        session.messages = turns_from_dicts(
+            [
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "text", "text": "describe this"},
+                        {"type": "image_url", "image_url": {"url": "data:image/png;base64,AA=="}},
+                    ],
+                },
+                {"role": "assistant", "content": "It's an image."},
+            ]
+        )
         session._msg_tokens = [10, 20]
         assert session.retry() is None
         # Messages should be unchanged
@@ -261,10 +276,12 @@ class TestRetry:
     def test_retry_none_content_returns_none(self, tmp_db):
         """retry() should handle content=None gracefully."""
         session = _make_session(tmp_db)
-        session.messages = [
-            {"role": "user", "content": None},
-            {"role": "assistant", "content": "Ok."},
-        ]
+        session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": None},
+                {"role": "assistant", "content": "Ok."},
+            ]
+        )
         session._msg_tokens = [10, 20]
         assert session.retry() is None
 
@@ -386,12 +403,14 @@ class TestRewindDBSync:
         save_message(ws_id, "user", "Bye")
         save_message(ws_id, "assistant", "Goodbye!")
 
-        session.messages = [
-            {"role": "user", "content": "Hello"},
-            {"role": "assistant", "content": "Hi!"},
-            {"role": "user", "content": "Bye"},
-            {"role": "assistant", "content": "Goodbye!"},
-        ]
+        session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "Hello"},
+                {"role": "assistant", "content": "Hi!"},
+                {"role": "user", "content": "Bye"},
+                {"role": "assistant", "content": "Goodbye!"},
+            ]
+        )
         session._msg_tokens = [5, 5, 5, 5]
 
         session.rewind(1)
diff --git a/tests/test_session.py b/tests/test_session.py
index 1a0828ad..b86e26a6 100644
--- a/tests/test_session.py
+++ b/tests/test_session.py
@@ -11,6 +11,12 @@ from unittest.mock import MagicMock, patch
 import pytest
 
 from turnstone.core.session import _IMAGE_EXTENSIONS, _IMAGE_SIZE_CAP, ChatSession
+from turnstone.core.trajectory import (
+    dicts_from_turns,
+    turn_from_dict,
+    turn_to_dict,
+    turns_from_dicts,
+)
 
 
 class NullUI:
@@ -198,7 +204,7 @@ class TestChatSessionConstruction:
         assert len(full) == len(session.system_messages)
 
         # Add a user message
-        session.messages.append({"role": "user", "content": "hello"})
+        session.messages.append(turn_from_dict({"role": "user", "content": "hello"}))
         full = session._full_messages()
         assert len(full) == len(session.system_messages) + 1
         assert full[-1]["role"] == "user"
@@ -834,10 +840,12 @@ class TestTitleRetry:
 
         session = _make_session()
         session._title_generated = True
-        session.messages = [
-            {"role": "user", "content": "Hello"},
-            {"role": "assistant", "content": "Hi there"},
-        ]
+        session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "Hello"},
+                {"role": "assistant", "content": "Hi there"},
+            ]
+        )
         # Mock provider to raise
         session._provider = MagicMock()
         session._provider.get_capabilities.return_value = ModelCapabilities()
@@ -852,10 +860,12 @@ class TestTitleRetry:
 
         session = _make_session()
         session._title_generated = True
-        session.messages = [
-            {"role": "user", "content": "Hello"},
-            {"role": "assistant", "content": "Hi there"},
-        ]
+        session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "Hello"},
+                {"role": "assistant", "content": "Hi there"},
+            ]
+        )
         result = MagicMock()
         result.content = "Test Title"
         session._provider = MagicMock()
@@ -874,10 +884,12 @@ class TestTitleRetry:
 
         session = _make_session()
         session._title_generated = True
-        session.messages = [
-            {"role": "user", "content": "Hello"},
-            {"role": "assistant", "content": "Hi there"},
-        ]
+        session.messages = turns_from_dicts(
+            [
+                {"role": "user", "content": "Hello"},
+                {"role": "assistant", "content": "Hi there"},
+            ]
+        )
         original_ws_id = session._ws_id
         result = MagicMock()
         result.content = "Test Title"
@@ -2750,11 +2762,11 @@ class TestMemoryCompositionDeferral:
         # __init__ composed against an empty history -> no query yet.
         assert session._system_composed_with_context is False
         # A whitespace-only "turn" (e.g. a wake send("")) is not a real query.
-        session.messages.append({"role": "user", "content": "   "})
+        session.messages.append(turn_from_dict({"role": "user", "content": "   "}))
         session._init_system_messages()
         assert session._system_composed_with_context is False
         # A real user message flips it (one-shot).
-        session.messages.append({"role": "user", "content": "what is the weather"})
+        session.messages.append(turn_from_dict({"role": "user", "content": "what is the weather"}))
         session._init_system_messages()
         assert session._system_composed_with_context is True
 
@@ -2768,7 +2780,7 @@ class TestMemoryCompositionDeferral:
         real_init = session._init_system_messages
 
         def spy_init():
-            seen_queries.append(extract_recent_context(session.messages))
+            seen_queries.append(extract_recent_context(dicts_from_turns(session.messages)))
             real_init()
 
         responses = [{"role": "assistant", "content": "ok"}]
@@ -2827,14 +2839,14 @@ class TestMetacognitiveBuffers:
         legacy ``_reminders`` side-channel splice.  The user turn content
         stays clean — the nudge is its own role=system trajectory turn."""
         session = _make_session()
-        session.messages.append({"role": "user", "content": "hello there"})
+        session.messages.append(turn_from_dict({"role": "user", "content": "hello there"}))
         session._msg_tokens.append(1)
         session._queue_user_advisory("correction", "ALERT_TEXT")
         with patch("turnstone.core.session.save_message"):
             session._emit_pending_user_nudges()
         # User turn untouched; a system turn now follows it.
-        assert session.messages[-2] == {"role": "user", "content": "hello there"}
-        assert session.messages[-1] == {
+        assert turn_to_dict(session.messages[-2]) == {"role": "user", "content": "hello there"}
+        assert turn_to_dict(session.messages[-1]) == {
             "role": "system",
             "_source": "correction",
             "content": "ALERT_TEXT",
@@ -2845,24 +2857,24 @@ class TestMetacognitiveBuffers:
 
     def test_emit_user_nudges_noop_when_buffer_empty(self, tmp_db):
         session = _make_session()
-        session.messages.append({"role": "user", "content": "untouched"})
+        session.messages.append(turn_from_dict({"role": "user", "content": "untouched"}))
         session._msg_tokens.append(1)
         pre_len = len(session.messages)
         with patch("turnstone.core.session.save_message"):
             session._emit_pending_user_nudges()
         # No nudges → no system turn appended.
         assert len(session.messages) == pre_len
-        assert session.messages[-1]["role"] == "user"
+        assert turn_to_dict(session.messages[-1])["role"] == "user"
 
     def test_emit_user_nudges_appends_one_system_turn_per_nudge(self, tmp_db):
         session = _make_session()
-        session.messages.append({"role": "user", "content": "user text"})
+        session.messages.append(turn_from_dict({"role": "user", "content": "user text"}))
         session._msg_tokens.append(1)
         session._queue_user_advisory("denial", "FIRST")
         session._queue_user_advisory("correction", "SECOND")
         with patch("turnstone.core.session.save_message"):
             session._emit_pending_user_nudges()
-        sys_turns = [m for m in session.messages if m.get("role") == "system"]
+        sys_turns = [m for m in dicts_from_turns(session.messages) if m.get("role") == "system"]
         assert sys_turns == [
             {"role": "system", "_source": "denial", "content": "FIRST"},
             {"role": "system", "_source": "correction", "content": "SECOND"},
@@ -3031,12 +3043,13 @@ class TestMetacognitiveBuffers:
             session.send("first")
 
         # Role sequence: the nudge follows the clean tool message.
-        roles = [m.get("role") for m in session.messages]
+        msgs = dicts_from_turns(session.messages)
+        roles = [m.get("role") for m in msgs]
         assert roles == ["user", "assistant", "tool", "system", "assistant"], (
             f"expected the tool_error nudge as a system turn after the tool, got {roles!r}"
         )
-        assert session.messages[2]["content"] == "boom"  # clean tool output
-        sys_turn = session.messages[3]
+        assert msgs[2]["content"] == "boom"  # clean tool output
+        sys_turn = msgs[3]
         assert sys_turn["_source"] == "tool_error"
         assert sys_turn["content"] == "you hit an error; check memory"
 
@@ -3085,16 +3098,17 @@ class TestMetacognitiveBuffers:
         # Role sequence: user -> assistant(tool_calls) -> tool ->
         # system(user_interjection) -> assistant(ack).  No trailing user
         # row — the interjection is operator-context, not user input.
-        roles = [m.get("role") for m in session.messages]
+        msgs = dicts_from_turns(session.messages)
+        roles = [m.get("role") for m in msgs]
         assert roles == ["user", "assistant", "tool", "system", "assistant"], (
             f"expected user->assistant->tool->system->assistant, got {roles!r}"
         )
         # Tool message content is the bare output — no envelope.
-        tool_msg = session.messages[2]
+        tool_msg = msgs[2]
         assert tool_msg["content"] == "ok"
         assert "<tool_output>" not in tool_msg["content"]
         # The system turn carries the queued interjection.
-        sys_turn = session.messages[3]
+        sys_turn = msgs[3]
         assert sys_turn["_source"] == "user_interjection"
         assert sys_turn["content"].endswith("User message: typed during tool")
         # The tool row was saved clean; a system row carries the interjection.
@@ -3139,13 +3153,14 @@ class TestMetacognitiveBuffers:
             session._title_generated = True
             session.send("first")
 
-        roles = [m.get("role") for m in session.messages]
+        msgs = dicts_from_turns(session.messages)
+        roles = [m.get("role") for m in msgs]
         # Single trailing user row carrying the feedback before the
         # final assistant ack.
         assert roles == ["user", "assistant", "tool", "user", "assistant"], (
             f"expected feedback-as-user-row sequence, got {roles!r}"
         )
-        assert session.messages[3]["content"] == "y, use full path"
+        assert msgs[3]["content"] == "y, use full path"
         # Persisted via _append_user_turn -> save_message("user", ...).
         user_saves = [
             c for c in save_msg.call_args_list if len(c.args) >= 3 and c.args[1] == "user"
@@ -3213,7 +3228,8 @@ class TestMetacognitiveBuffers:
             session._title_generated = True
             session.send("first")
 
-        roles = [m.get("role") for m in session.messages]
+        msgs = dicts_from_turns(session.messages)
+        roles = [m.get("role") for m in msgs]
         # NO back-to-back user rows.  Pre-fix the user_feedback
         # appended a separate user row and the queue drained another:
         # roles == [..., "user", "user", ...] which broke strict
@@ -3226,7 +3242,7 @@ class TestMetacognitiveBuffers:
         assert roles == ["user", "assistant", "tool", "user", "assistant"], (
             f"expected single-trailing-user shape, got {roles!r}"
         )
-        flushed_content = session.messages[3]["content"]
+        flushed_content = msgs[3]["content"]
         # The two pieces are joined by the canonical separator.
         assert flushed_content == "y, use full path\n\nlate arrival"
         # Queue cleared.
@@ -3241,7 +3257,7 @@ class TestMetacognitiveBuffers:
         appended = session._flush_queued_messages(prefix="hello")
         assert appended is True
         assert len(session.messages) == pre_count + 1
-        last = session.messages[-1]
+        last = turn_to_dict(session.messages[-1])
         assert last["role"] == "user"
         assert last["content"] == "hello"
 
@@ -3254,7 +3270,7 @@ class TestMetacognitiveBuffers:
         session.queue_message("b", queue_msg_id="q-b")
         appended = session._flush_queued_messages(prefix="approve")
         assert appended is True
-        last = session.messages[-1]
+        last = turn_to_dict(session.messages[-1])
         assert last["role"] == "user"
         assert last["content"] == "approve\n\na\n\nb"
         assert session._queued_messages == {}
@@ -3422,14 +3438,15 @@ class TestMetacognitiveBuffers:
             session.send("first")
 
         # In-memory: tool row keeps the list content; a system turn follows.
-        tool_msg = next(m for m in session.messages if m.get("role") == "tool")
+        msgs = dicts_from_turns(session.messages)
+        tool_msg = next(m for m in msgs if m.get("role") == "tool")
         text_parts = [
             p["text"]
             for p in tool_msg["content"]
             if isinstance(p, dict) and p.get("type") == "text"
         ]
         assert text_parts == ["the chart shows X"]
-        sys_turn = next(m for m in session.messages if m.get("role") == "system")
+        sys_turn = next(m for m in msgs if m.get("role") == "system")
         assert sys_turn["_source"] == "user_interjection"
         assert sys_turn["content"].endswith("User message: inspect the histogram")
 
@@ -3459,10 +3476,11 @@ class TestMetacognitiveBuffers:
 
         # User turn landed clean; the start nudge follows it as a system turn.
         assert session.messages, "user message should have been appended"
-        user_turns = [m for m in session.messages if m.get("role") == "user"]
+        msgs = dicts_from_turns(session.messages)
+        user_turns = [m for m in msgs if m.get("role") == "user"]
         assert user_turns[-1]["content"] == "first user message"
         assert "_reminders" not in user_turns[-1]
-        sys_turns = [m for m in session.messages if m.get("role") == "system"]
+        sys_turns = [m for m in msgs if m.get("role") == "system"]
         assert any(m["_source"] == "start" for m in sys_turns), (
             f"expected a start system turn, got {sys_turns!r}"
         )
@@ -3478,7 +3496,7 @@ class TestMetacognitiveBuffers:
         line is gone.  No ``on_info`` should fire from the drain."""
         session = _make_session()
         session.ui = MagicMock()
-        session.messages.append({"role": "user", "content": "noted"})
+        session.messages.append(turn_from_dict({"role": "user", "content": "noted"}))
         session._msg_tokens.append(1)
         session._queue_user_advisory("correction", "watch out")
         with patch("turnstone.core.session.save_message"):
@@ -3494,7 +3512,7 @@ class TestMetacognitiveBuffers:
         operator bubble in lockstep with the originating tab."""
         session = _make_session()
         session.ui = MagicMock()
-        session.messages.append({"role": "user", "content": "noted"})
+        session.messages.append(turn_from_dict({"role": "user", "content": "noted"}))
         session._msg_tokens.append(1)
         session._queue_user_advisory("correction", "watch out")
         with patch("turnstone.core.session.save_message"):
@@ -3511,13 +3529,13 @@ class TestMetacognitiveBuffers:
         session = _make_session()
         session.ui = MagicMock()
         session.ui.on_system_turn.side_effect = RuntimeError("queue full")
-        session.messages.append({"role": "user", "content": "noted"})
+        session.messages.append(turn_from_dict({"role": "user", "content": "noted"}))
         session._msg_tokens.append(1)
         session._queue_user_advisory("correction", "watch out")
         with patch("turnstone.core.session.save_message"):
             session._emit_pending_user_nudges()
         # The system turn was appended despite the hook raising.
-        assert session.messages[-1] == {
+        assert turn_to_dict(session.messages[-1]) == {
             "role": "system",
             "_source": "correction",
             "content": "watch out",
@@ -3566,8 +3584,8 @@ class TestApplyPostExecuteAdvisories:
         so seed two messages to mirror that.
         """
         session._mem_cfg.nudges = True
-        session.messages.append({"role": "user", "content": "hi"})
-        session.messages.append({"role": "assistant", "content": "ok"})
+        session.messages.append(turn_from_dict({"role": "user", "content": "hi"}))
+        session.messages.append(turn_from_dict({"role": "assistant", "content": "ok"}))
 
     def test_three_identical_calls_fire_warning_and_advisory(self, tmp_db):
         session = _make_session()
@@ -3774,7 +3792,7 @@ class TestUpdateTokenTableMsgsParam:
     def test_uses_provided_msgs_skips_re_application(self, tmp_db):
         session = _make_session()
         session._last_usage = {"prompt_tokens": 100, "completion_tokens": 50}
-        session.messages.append({"role": "user", "content": "hi"})
+        session.messages.append(turn_from_dict({"role": "user", "content": "hi"}))
         # Patch _prepare_wire_messages to detect a redundant re-fold.
         with patch.object(
             session,
@@ -3792,7 +3810,7 @@ class TestUpdateTokenTableMsgsParam:
         can't) pre-build the wire copy still get a sane calibration."""
         session = _make_session()
         session._last_usage = {"prompt_tokens": 100, "completion_tokens": 50}
-        session.messages.append({"role": "user", "content": "hi"})
+        session.messages.append(turn_from_dict({"role": "user", "content": "hi"}))
         with patch.object(
             session,
             "_prepare_wire_messages",
@@ -3911,7 +3929,7 @@ class TestUserAdvisoryCancelClear:
         )
         # The queued message landed in history before the second turn.
         user_texts: list[str] = []
-        for m in session.messages:
+        for m in dicts_from_turns(session.messages):
             if m.get("role") != "user":
                 continue
             content = m.get("content")
@@ -3986,12 +4004,13 @@ class TestDeliverWakeNudge:
         assert _user_pending(session) == []
         # Empty-content user message was appended; the nudge follows it as
         # a first-class system turn (no _reminders side-channel).
-        user_msgs = [m for m in session.messages if m.get("role") == "user"]
+        msgs = dicts_from_turns(session.messages)
+        user_msgs = [m for m in msgs if m.get("role") == "user"]
         assert user_msgs, "wake should append a synthetic user message"
         wake_msg = user_msgs[-1]
         assert wake_msg["content"] == ""
         assert "_reminders" not in wake_msg
-        sys_turns = [m for m in session.messages if m.get("role") == "system"]
+        sys_turns = [m for m in msgs if m.get("role") == "system"]
         assert {"role": "system", "_source": "idle_children", "content": "your kids"} in sys_turns
 
     def test_marks_source_tag_on_synthesized_user_msg(self, tmp_db):
@@ -4013,7 +4032,7 @@ class TestDeliverWakeNudge:
             patch("turnstone.core.session.save_message"),
         ):
             session.deliver_wake_nudge_from_queue()
-        user_msgs = [m for m in session.messages if m.get("role") == "user"]
+        user_msgs = [m for m in dicts_from_turns(session.messages) if m.get("role") == "user"]
         wake_msg = user_msgs[-1]
         assert wake_msg.get("_source") == "system_nudge"
 
@@ -4056,7 +4075,7 @@ class TestDeliverWakeNudge:
         session._queue_user_advisory("denial", "don't do that next time")
         # Force enough memory + message context that should_nudge would
         # otherwise fire a fresh correction nudge.
-        session.messages.append({"role": "user", "content": "earlier"})
+        session.messages.append(turn_from_dict({"role": "user", "content": "earlier"}))
         with (
             patch.object(session, "_create_stream_with_retry", return_value=iter([])),
             patch.object(
@@ -4116,7 +4135,7 @@ class TestDeliverWakeNudge:
         ):
             session.deliver_wake_nudge_from_queue()
 
-        user_msgs = [m for m in session.messages if m.get("role") == "user"]
+        user_msgs = [m for m in dicts_from_turns(session.messages) if m.get("role") == "user"]
         # Two user messages: the wake's synthetic empty turn (with
         # _source) AND the flushed real user input (without _source).
         wake_msg = next(m for m in user_msgs if m.get("content") == "")
@@ -4145,10 +4164,11 @@ class TestDeliverWakeNudge:
         ):
             session.deliver_wake_nudge_from_queue()
         # The nudge system turn landed and stays (persistent history).
-        sys_turns = [m for m in session.messages if m.get("role") == "system"]
+        msgs = dicts_from_turns(session.messages)
+        sys_turns = [m for m in msgs if m.get("role") == "system"]
         assert any(m["_source"] == "denial" and m["content"] == "leftover" for m in sys_turns)
         # No legacy delivered flag anywhere.
-        assert all("_reminders_delivered" not in m for m in session.messages)
+        assert all("_reminders_delivered" not in m for m in msgs)
         # Wake tag cleared even on exception (finally block).
         assert session._wake_source_tag == ""
 
@@ -4230,14 +4250,16 @@ class TestReminderSidechannelIsolation:
         the summary text and outlive the turn it advised."""
         session = _make_session()
         session.messages.append(
-            {
-                "role": "user",
-                "content": "user said this",
-                "_reminders": [{"type": "correction", "text": "SECRET_NUDGE_TEXT"}],
-            }
+            turn_from_dict(
+                {
+                    "role": "user",
+                    "content": "user said this",
+                    "_reminders": [{"type": "correction", "text": "SECRET_NUDGE_TEXT"}],
+                }
+            )
         )
-        session.messages.append({"role": "assistant", "content": "ok"})
-        summary = session._format_messages_for_summary(session.messages)
+        session.messages.append(turn_from_dict({"role": "assistant", "content": "ok"}))
+        summary = session._format_messages_for_summary(dicts_from_turns(session.messages))
         assert "SECRET_NUDGE_TEXT" not in summary
         assert "<system-reminder>" not in summary
         assert "user said this" in summary
@@ -4249,16 +4271,18 @@ class TestReminderSidechannelIsolation:
         clean even when ``_reminders`` is populated."""
         session = _make_session()
         session.messages.append(
-            {
-                "role": "user",
-                "content": "first message body",
-                "_reminders": [{"type": "start", "text": "SECRET_NUDGE_TEXT"}],
-            }
+            turn_from_dict(
+                {
+                    "role": "user",
+                    "content": "first message body",
+                    "_reminders": [{"type": "start", "text": "SECRET_NUDGE_TEXT"}],
+                }
+            )
         )
         # Mirror the loop at session.py:_generate_title that pulls the
         # first user message into the title prompt.
         extracted_user = ""
-        for m in session.messages:
+        for m in dicts_from_turns(session.messages):
             content = m.get("content") or ""
             if isinstance(content, list):
                 content = " ".join(p.get("text", "") for p in content if isinstance(p, dict))
@@ -4290,7 +4314,7 @@ class TestReminderSidechannelIsolation:
 
         wake_msgs = [
             m
-            for m in resumed_fork.messages
+            for m in dicts_from_turns(resumed_fork.messages)
             if m.get("role") == "user" and m.get("_source") == "system_nudge"
         ]
         assert len(wake_msgs) == 1
diff --git a/tests/test_session_attachments.py b/tests/test_session_attachments.py
index e4f3c6e8..9e23b53e 100644
--- a/tests/test_session_attachments.py
+++ b/tests/test_session_attachments.py
@@ -12,6 +12,7 @@ from turnstone.core.memory import (
     register_workstream,
 )
 from turnstone.core.session import ChatSession
+from turnstone.core.trajectory import dicts_from_turns, turn_to_dict
 
 PNG_1x1 = (
     b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01"
@@ -56,12 +57,12 @@ class TestPlainTextUnchanged:
     def test_no_attachments_stores_string_content(self, tmp_db, mock_openai_client):
         s = _make_session(mock_openai_client)
         _run_send(s, "hello")
-        assert s.messages[-1] == {"role": "user", "content": "hello"}
+        assert turn_to_dict(s.messages[-1]) == {"role": "user", "content": "hello"}
 
     def test_empty_attachments_list_stores_string_content(self, tmp_db, mock_openai_client):
         s = _make_session(mock_openai_client)
         _run_send(s, "hello", attachments=[])
-        assert s.messages[-1] == {"role": "user", "content": "hello"}
+        assert turn_to_dict(s.messages[-1]) == {"role": "user", "content": "hello"}
 
 
 class TestMultipartBuild:
@@ -75,7 +76,7 @@ class TestMultipartBuild:
             content=PNG_1x1,
         )
         _run_send(s, "what is this?", attachments=[att])
-        msg = s.messages[-1]
+        msg = turn_to_dict(s.messages[-1])
         assert msg["role"] == "user"
         assert isinstance(msg["content"], list)
         assert msg["content"][0] == {"type": "text", "text": "what is this?"}
@@ -93,7 +94,7 @@ class TestMultipartBuild:
             content=b"# hi\n",
         )
         _run_send(s, "summarize", attachments=[att])
-        msg = s.messages[-1]
+        msg = turn_to_dict(s.messages[-1])
         doc = msg["content"][1]
         assert doc == {
             "type": "document",
@@ -112,9 +113,10 @@ class TestMultipartBuild:
             Attachment("a3", "second.md", "text/markdown", "text", b"B"),
         ]
         _run_send(s, "look", attachments=atts)
-        types = [p["type"] for p in s.messages[-1]["content"]]
+        msg = turn_to_dict(s.messages[-1])
+        types = [p["type"] for p in msg["content"]]
         assert types == ["text", "image_url", "document", "document"]
-        docs = [p for p in s.messages[-1]["content"] if p["type"] == "document"]
+        docs = [p for p in msg["content"] if p["type"] == "document"]
         assert docs[0]["document"]["data"] == "A"
         assert docs[1]["document"]["data"] == "B"
 
@@ -122,7 +124,7 @@ class TestMultipartBuild:
         s = _make_session(mock_openai_client)
         att = Attachment("a1", "bad.bin", "text/plain", "text", b"\xff\xfe")
         _run_send(s, "read this", attachments=[att])
-        parts = s.messages[-1]["content"]
+        parts = turn_to_dict(s.messages[-1])["content"]
         assert any(
             p.get("type") == "text" and p.get("text") == "[unreadable attachment: bad.bin]"
             for p in parts
@@ -225,7 +227,7 @@ class TestProviderIntegration:
         ]
         _run_send(s, "look at both", attachments=atts)
 
-        _, converted = AnthropicProvider()._convert_messages([s.messages[-1]])
+        _, converted = AnthropicProvider()._convert_messages(dicts_from_turns([s.messages[-1]]))
         assert len(converted) == 1
         content = converted[0]["content"]
         types = [p["type"] for p in content]
@@ -250,7 +252,7 @@ class TestProviderIntegration:
             Attachment("a2", "notes.md", "text/markdown", "text", b"hi"),
         ]
         _run_send(s, "desc", attachments=atts)
-        meta = s.messages[-1].get("_attachments_meta")
+        meta = turn_to_dict(s.messages[-1]).get("_attachments_meta")
         assert meta == [
             {"kind": "image", "filename": "dog.png", "mime_type": "image/png"},
             {"kind": "text", "filename": "notes.md", "mime_type": "text/markdown"},
@@ -264,7 +266,7 @@ class TestProviderIntegration:
         s = _make_session(mock_openai_client)
         atts = [Attachment("a1", "x.md", "text/markdown", "text", b"x")]
         _run_send(s, "hi", attachments=atts)
-        out = sanitize_messages([s.messages[-1]])
+        out = sanitize_messages(dicts_from_turns([s.messages[-1]]))
         for k in out[0]:
             assert not k.startswith("_"), f"{k!r} leaked to wire"
 
@@ -277,7 +279,7 @@ class TestProviderIntegration:
         ]
         _run_send(s, "review", attachments=atts)
 
-        out = sanitize_messages([s.messages[-1]])
+        out = sanitize_messages(dicts_from_turns([s.messages[-1]]))
         parts = out[0]["content"]
         types = [p["type"] for p in parts]
         assert types == ["text", "text"]
diff --git a/tests/test_sessions.py b/tests/test_sessions.py
index 7548c276..c97ad1d2 100644
--- a/tests/test_sessions.py
+++ b/tests/test_sessions.py
@@ -19,6 +19,7 @@ from turnstone.core.memory import (
 )
 from turnstone.core.session import ChatSession
 from turnstone.core.storage import get_storage
+from turnstone.core.trajectory import turn_to_dict
 
 # ── Workstream registration ───────────────────────────────────────────
 
@@ -298,7 +299,7 @@ class TestResumeWorkstream:
         assert result is True
         assert session._ws_id == "old_ws_123"
         assert len(session.messages) == 2
-        assert session.messages[0]["content"] == "hello world"
+        assert turn_to_dict(session.messages[0])["content"] == "hello world"
         assert session._title_generated is True
 
     def test_resume_nonexistent_returns_false(self, tmp_db, mock_openai_client):
diff --git a/tests/test_tool_truncation.py b/tests/test_tool_truncation.py
index f9ade55c..ac295827 100644
--- a/tests/test_tool_truncation.py
+++ b/tests/test_tool_truncation.py
@@ -7,6 +7,7 @@ from unittest.mock import MagicMock, patch
 import pytest
 
 from turnstone.core.session import ChatSession
+from turnstone.core.trajectory import turns_from_dicts
 
 # ---------------------------------------------------------------------------
 # Helpers
@@ -144,7 +145,7 @@ class TestContextOverflowRecovery:
     """Test that context-length errors trigger compact-and-retry."""
 
     def test_openai_context_length_error_triggers_compact(self, session):
-        session.messages = [{"role": "user", "content": "hi"}]
+        session.messages = turns_from_dicts([{"role": "user", "content": "hi"}])
         session._msg_tokens = [1]
 
         call_count = 0
@@ -175,7 +176,7 @@ class TestContextOverflowRecovery:
         assert call_count == 2
 
     def test_anthropic_prompt_too_long_triggers_compact(self, session):
-        session.messages = [{"role": "user", "content": "hi"}]
+        session.messages = turns_from_dicts([{"role": "user", "content": "hi"}])
         session._msg_tokens = [1]
 
         call_count = 0
@@ -205,7 +206,7 @@ class TestContextOverflowRecovery:
         compact_mock.assert_called_once_with(auto=True)
 
     def test_non_context_error_propagates(self, session):
-        session.messages = [{"role": "user", "content": "hi"}]
+        session.messages = turns_from_dicts([{"role": "user", "content": "hi"}])
         session._msg_tokens = [1]
 
         with (
@@ -222,7 +223,7 @@ class TestContextOverflowRecovery:
             session.send("hello")
 
     def test_compact_failure_raises_original_error(self, session):
-        session.messages = [{"role": "user", "content": "hi"}]
+        session.messages = turns_from_dicts([{"role": "user", "content": "hi"}])
         session._msg_tokens = [1]
 
         with (
diff --git a/tests/test_trajectory.py b/tests/test_trajectory.py
index bc12ece5..efb9bed8 100644
--- a/tests/test_trajectory.py
+++ b/tests/test_trajectory.py
@@ -38,6 +38,15 @@ _ROUNDTRIP: list[dict[str, Any]] = [
         ],
         "_attachments_meta": [{"kind": "image", "filename": "x.png", "mime_type": "image/png"}],
     },
+    {
+        # All-text multipart list (the unreadable-attachment placeholder path)
+        # stays a list — must not collapse to a joined string.
+        "role": "user",
+        "content": [
+            {"type": "text", "text": "read this"},
+            {"type": "text", "text": "[unreadable attachment: bad.bin]"},
+        ],
+    },
     {"role": "assistant", "content": "hi there"},
     {"role": "assistant", "content": ""},  # empty assistant (no text, no tools)
     {
diff --git a/tests/test_watch_integration.py b/tests/test_watch_integration.py
index b58c3412..ac16e54c 100644
--- a/tests/test_watch_integration.py
+++ b/tests/test_watch_integration.py
@@ -31,6 +31,7 @@ import pytest
 from tests._helpers import patch_session_storage
 from turnstone.core.session import ChatSession
 from turnstone.core.storage import get_storage
+from turnstone.core.trajectory import dicts_from_turns
 from turnstone.core.watch import WatchRunner
 
 
@@ -120,10 +121,11 @@ def test_watch_fires_then_user_send_drains_envelope(tmp_db, monkeypatch):
     # The watch text lands as a first-class operator-context ``system``
     # turn following the user turn (the production drain seam now emits
     # ``make_system_turn`` rather than the legacy ``_reminders`` splice).
-    user_msgs = [m for m in session.messages if m.get("role") == "user"]
+    msgs = dicts_from_turns(session.messages)
+    user_msgs = [m for m in msgs if m.get("role") == "user"]
     assert user_msgs, "expected a user message in history"
     assert "_reminders" not in user_msgs[-1]
-    sys_turns = [m for m in session.messages if m.get("role") == "system"]
+    sys_turns = [m for m in msgs if m.get("role") == "system"]
     assert any(
         m["_source"] == "watch_triggered" and "watch payload body" in m["content"]
         for m in sys_turns
@@ -176,10 +178,9 @@ def test_three_back_to_back_watch_fires_drain_into_one_turn(tmp_db, monkeypatch)
 
     # All three drained into three operator-context system turns after
     # the single user turn.
+    msgs = dicts_from_turns(session.messages)
     watch_turns = [
-        m
-        for m in session.messages
-        if m.get("role") == "system" and m.get("_source") == "watch_triggered"
+        m for m in msgs if m.get("role") == "system" and m.get("_source") == "watch_triggered"
     ]
     assert len(watch_turns) == 3
     bodies = [m["content"] for m in watch_turns]
@@ -187,7 +188,7 @@ def test_three_back_to_back_watch_fires_drain_into_one_turn(tmp_db, monkeypatch)
     assert any("fire two" in b for b in bodies)
     assert any("fire three" in b for b in bodies)
     # And there's exactly ONE assistant turn (not three).
-    assistant_turns = [m for m in session.messages if m.get("role") == "assistant"]
+    assistant_turns = [m for m in msgs if m.get("role") == "assistant"]
     assert len(assistant_turns) == 1
 
 
diff --git a/turnstone/console/coordinator_idle_observer.py b/turnstone/console/coordinator_idle_observer.py
index 2e0f4549..606e0949 100644
--- a/turnstone/console/coordinator_idle_observer.py
+++ b/turnstone/console/coordinator_idle_observer.py
@@ -41,6 +41,7 @@ from turnstone.core.metacognition import (
     format_idle_children_nudge,
     should_nudge,
 )
+from turnstone.core.trajectory import Role
 from turnstone.core.workstream import WorkstreamKind, WorkstreamState
 
 if TYPE_CHECKING:
@@ -250,13 +251,10 @@ class CoordinatorIdleObserver:
         ``wait_for_workstream`` tool call, return ``True``.
         """
         for msg in reversed(session.messages):
-            if msg.get("role") != "assistant":
+            if msg.role is not Role.ASSISTANT:
                 continue
-            for tc in msg.get("tool_calls") or []:
-                fn = tc.get("function", {}) or {}
-                if fn.get("name") == "wait_for_workstream":
-                    return True
-            return False  # found the most recent assistant turn — done
+            # The most recent assistant turn — done after this one.
+            return any(tc.name == "wait_for_workstream" for tc in msg.tool_calls)
         return False
 
     def _active_children(self, ws: Workstream) -> list[dict[str, str]]:
diff --git a/turnstone/core/session.py b/turnstone/core/session.py
index bd6366ff..cb6fec56 100644
--- a/turnstone/core/session.py
+++ b/turnstone/core/session.py
@@ -123,6 +123,14 @@ from turnstone.core.tools import (
     TASK_AUTO_TOOLS,
     merge_mcp_tools,
 )
+from turnstone.core.trajectory import (
+    Role,
+    Turn,
+    dicts_from_turns,
+    turn_from_dict,
+    turn_to_dict,
+    turns_from_dicts,
+)
 from turnstone.core.watch import WATCH_REMINDER_OPTIONAL_KEYS
 from turnstone.core.web import check_ssrf, strip_html
 from turnstone.core.workstream import WorkstreamKind
@@ -1018,7 +1026,10 @@ class ChatSession:
         self._ws_id = ws_id or uuid.uuid4().hex
         self._title_generated = False
         self._read_files: set[str] = set()
-        self.messages: list[dict[str, Any]] = []
+        # The canonical in-memory trajectory.  Wire prep (fold/repair) + the
+        # provider translators still consume dicts, so ``_full_messages`` lowers
+        # Turns→dicts at that boundary until those layers migrate.
+        self.messages: list[Turn] = []
         self._last_usage: dict[str, int] | None = None
         self._msg_tokens: list[int] = []  # parallel to self.messages
         self._system_tokens = 0  # tokens for system_messages
@@ -2375,13 +2386,10 @@ class ChatSession:
             user_msg = ""
             asst_msg = ""
             for m in self.messages:
-                content = m.get("content") or ""
-                # Handle multi-part content (vision messages)
-                if isinstance(content, list):
-                    content = " ".join(p.get("text", "") for p in content if isinstance(p, dict))
-                if m["role"] == "user" and not user_msg:
+                content = m.text  # joins text blocks; multipart attachments contribute none
+                if m.role is Role.USER and not user_msg:
                     user_msg = content[:300]
-                elif m["role"] == "assistant" and not asst_msg:
+                elif m.role is Role.ASSISTANT and not asst_msg:
                     asst_msg = content[:200]
                 if user_msg and asst_msg:
                     break
@@ -2474,12 +2482,12 @@ class ChatSession:
         so the resumed/forked workstream behaves identically to the
         original.  Returns True on success.
         """
-        messages = load_messages(ws_id)
-        if not messages:
+        turns = turns_from_dicts(load_messages(ws_id))
+        if not turns:
             return False
         if not fork:
             self._ws_id = ws_id
-        self.messages = messages
+        self.messages = turns
         self._read_files.clear()
         self._repeat_detector.clear()
         self._last_usage = None
@@ -2491,7 +2499,7 @@ class ChatSession:
         log.info(
             "Resuming ws=%s: %d messages, provider=%s, model=%s",
             ws_id,
-            len(messages),
+            len(self.messages),
             type(self._provider).__name__,
             self.model,
         )
@@ -2574,7 +2582,8 @@ class ChatSession:
         if fork:
             # Bulk-insert all messages in a single transaction for performance.
             bulk_rows: list[dict[str, Any]] = []
-            for msg in self.messages:
+            for turn in self.messages:
+                msg = turn_to_dict(turn)
                 tc = msg.get("tool_calls")
                 tc_json = json.dumps(tc) if tc else None
                 # Provider-fidelity blocks ride the in-memory
@@ -2828,7 +2837,7 @@ class ChatSession:
         if self.instructions:
             dev_parts.append("")
             dev_parts.append(self.instructions)
-        context = extract_recent_context(self.messages)
+        context = extract_recent_context(dicts_from_turns(self.messages))
         if context.strip():
             # Composed against a real user-message query at least once; send()
             # uses this to know the deferred first-turn recompose is done.
@@ -2871,8 +2880,12 @@ class ChatSession:
         self._agent_system_messages = list(new_system_messages)
 
     def _full_messages(self) -> list[dict[str, Any]]:
-        """System messages + conversation history."""
-        return self.system_messages + self.messages
+        """System messages + conversation history, lowered to wire dicts.
+
+        ``self.messages`` is the canonical ``Turn`` trajectory; the wire-prep and
+        provider layers still consume dicts, so the Turns are lowered here (this
+        boundary moves inward when those layers migrate)."""
+        return self.system_messages + dicts_from_turns(self.messages)
 
     def _prepare_wire_messages(
         self,
@@ -3563,7 +3576,7 @@ class ChatSession:
                 }
                 for a in attachments
             ]
-        self.messages.append(user_msg)
+        self.messages.append(turn_from_dict(user_msg))
         self._msg_tokens.append(max(1, int(self._msg_char_count(user_msg) / self._chars_per_token)))
         # DB row stores the raw text only; attachment bytes are written
         # content-addressed into workstream_attachments and the ordered id-list
@@ -3696,7 +3709,7 @@ class ChatSession:
         before the wire by ``sanitize_messages``).
         """
         turn = make_system_turn(source, content, **meta)
-        self.messages.append(turn)
+        self.messages.append(turn_from_dict(turn))
         self._msg_tokens.append(max(1, int(self._msg_char_count(turn) / self._chars_per_token)))
         save_message(
             self._ws_id,
@@ -3784,7 +3797,10 @@ class ChatSession:
         # flips True inside the recompose, so this fires once and the prefix
         # stays cache-stable after. Per-turn refresh is the larger redesign on
         # another branch.
-        if not self._system_composed_with_context and extract_recent_context(self.messages).strip():
+        if (
+            not self._system_composed_with_context
+            and extract_recent_context(dicts_from_turns(self.messages)).strip()
+        ):
             self._init_system_messages()
 
         try:
@@ -3863,7 +3879,7 @@ class ChatSession:
                 # actually counted.
                 self._update_token_table(assistant_msg, msgs=msgs)
                 self._print_status_line()  # Report usage for EVERY API call
-                self.messages.append(assistant_msg)
+                self.messages.append(turn_from_dict(assistant_msg))
                 # Clear per-turn inflight buffers — the assistant
                 # message is now in the history list a refresh would
                 # replay, so the in_progress_snapshot shouldn't re-
@@ -4063,7 +4079,7 @@ class ChatSession:
                     tool_is_error = self._tool_error_flags.pop(tc_id, False)
                     if tool_is_error:
                         tool_msg["is_error"] = True
-                    self.messages.append(tool_msg)
+                    self.messages.append(turn_from_dict(tool_msg))
 
                     # Token estimation — image content uses a fixed heuristic
                     if isinstance(output, list):
@@ -4188,7 +4204,7 @@ class ChatSession:
                 else:
                     msg["content"] = "[generation cancelled before completion]"
                 save_message(self._ws_id, "assistant", msg["content"], event_id=self._ui_event_id())
-                self.messages.append(msg)
+                self.messages.append(turn_from_dict(msg))
                 tok_est = max(
                     1,
                     int(self._msg_char_count(msg) / self._chars_per_token),
@@ -4248,7 +4264,7 @@ class ChatSession:
         assistant_idx = None
         for i in range(len(self.messages) - 1, -1, -1):
             msg = self.messages[i]
-            if msg.get("role") == "assistant" and msg.get("tool_calls"):
+            if msg.role is Role.ASSISTANT and msg.tool_calls:
                 assistant_idx = i
                 break
         if assistant_idx is None:
@@ -4257,22 +4273,15 @@ class ChatSession:
         # Collect tool_call IDs that already have results
         answered_ids: set[str] = set()
         for msg in self.messages[assistant_idx + 1 :]:
-            if msg.get("role") == "tool":
-                answered_ids.add(msg.get("tool_call_id", ""))
+            if msg.role is Role.TOOL:
+                answered_ids.add(msg.tool_call_id or "")
 
         # Synthesize results for unanswered tool_calls
-        for tc in self.messages[assistant_idx].get("tool_calls", []):
-            tc_id = tc.get("id", "")
-            func_name = tc.get("function", {}).get("name", "")
+        for tc in self.messages[assistant_idx].tool_calls:
+            tc_id = tc.id
+            func_name = tc.name
             if tc_id and tc_id not in answered_ids:
-                self.messages.append(
-                    {
-                        "role": "tool",
-                        "tool_call_id": tc_id,
-                        "content": reason,
-                        "is_error": True,
-                    }
-                )
+                self.messages.append(Turn.tool(tc_id, reason, is_error=True))
                 self._msg_tokens.append(1)
                 save_message(
                     self._ws_id,
@@ -4302,7 +4311,7 @@ class ChatSession:
 
     def _find_turn_boundaries(self) -> list[int]:
         """Return indices of user messages in self.messages (turn start positions)."""
-        return [i for i, m in enumerate(self.messages) if m["role"] == "user"]
+        return [i for i, m in enumerate(self.messages) if m.role is Role.USER]
 
     def rewind(self, n: int) -> int:
         """Drop the last *n* complete turns from the conversation.
@@ -4334,7 +4343,7 @@ class ChatSession:
         if not boundaries:
             return None
         last_user_idx = boundaries[-1]
-        content = self.messages[last_user_idx].get("content")
+        content = turn_to_dict(self.messages[last_user_idx]).get("content")
         # Multipart messages (vision/images) have list-type content;
         # retry only supports plain text.
         if not isinstance(content, str) or not content:
@@ -4751,7 +4760,7 @@ class ChatSession:
     _IMAGE_TOKENS = 1000
 
     @staticmethod
-    def _msg_text_chars(msg: dict[str, Any]) -> tuple[int, int, int]:
+    def _msg_text_chars(msg: dict[str, Any] | Turn) -> tuple[int, int, int]:
         """Return ``(text_chars, image_count, doc_chars)`` for a message.
 
         Counts textual content + structural overhead (role, tool_call
@@ -4763,7 +4772,12 @@ class ChatSession:
         calibration — provider-native document blocks (Anthropic) and
         inlined text (OpenAI/Google) tokenize differently, so it's
         safer to exclude them from the text calibration.
+
+        Accepts a wire dict or a canonical ``Turn`` (the latter is lowered to
+        its dict form so the char accounting matches what the provider sees).
         """
+        if isinstance(msg, Turn):
+            msg = turn_to_dict(msg)
         content = msg.get("content")
         n = 0
         images = 0
@@ -4791,12 +4805,12 @@ class ChatSession:
         n += len(msg.get("tool_call_id", ""))
         return n, images, doc_chars
 
-    def _msg_char_count(self, msg: dict[str, Any]) -> int:
+    def _msg_char_count(self, msg: dict[str, Any] | Turn) -> int:
         """Count characters in a message, including structural overhead.
 
         Includes role markers, tool_call IDs, image placeholders, and
         document-part characters so that the budget estimate reflects
-        the full payload the provider sees.
+        the full payload the provider sees.  Accepts a wire dict or a ``Turn``.
         """
         text_chars, images, doc_chars = self._msg_text_chars(msg)
         return text_chars + doc_chars + int(images * self._IMAGE_TOKENS * self._chars_per_token)
@@ -4949,8 +4963,8 @@ class ChatSession:
         last_user_content = None
         if auto:
             for m in reversed(self.messages):
-                if m["role"] == "user":
-                    last_user_content = m.get("content") or ""
+                if m.role is Role.USER:
+                    last_user_content = m.text or ""
                     break
 
         to_summarize = self.messages
@@ -4980,7 +4994,7 @@ class ChatSession:
             return
 
         # Build summary prompt and call model
-        formatted = self._format_messages_for_summary(selected)
+        formatted = self._format_messages_for_summary(dicts_from_turns(selected))
         summary_msgs = [
             {
                 "role": "system",
@@ -5070,7 +5084,7 @@ class ChatSession:
         before_tokens = self._system_tokens + sum(self._msg_tokens)
         summary_user = {"role": "user", "content": "[Conversation summary]"}
         summary_asst = {"role": "assistant", "content": summary}
-        self.messages = [summary_user, summary_asst]
+        self.messages = turns_from_dicts([summary_user, summary_asst])
         # File contents are gone after compaction — force re-read before edit_file
         self._read_files.clear()
         self._repeat_detector.clear()
@@ -5363,7 +5377,7 @@ class ChatSession:
 
         heuristic_verdicts = judge.evaluate(
             pending,
-            list(self.messages),  # snapshot — daemon thread must not see mutations
+            dicts_from_turns(self.messages),  # snapshot — daemon thread must not see mutations
             callback=_on_verdict,
             cancel_event=cancel_event,
         )
@@ -12437,7 +12451,7 @@ class ChatSession:
             # Clear history when toggling ON if it contains tool messages,
             # because the API rejects tool-call history without tool definitions
             if self.creative_mode and any(
-                m.get("tool_calls") or m.get("role") == "tool" for m in self.messages
+                m.tool_calls or m.role is Role.TOOL for m in self.messages
             ):
                 self.messages.clear()
                 self._read_files.clear()
diff --git a/turnstone/core/trajectory.py b/turnstone/core/trajectory.py
index 86678257..f6ac1f74 100644
--- a/turnstone/core/trajectory.py
+++ b/turnstone/core/trajectory.py
@@ -199,13 +199,16 @@ def _content_from_raw(raw: Any) -> tuple[ContentBlock, ...]:
 def _content_to_raw(content: tuple[ContentBlock, ...]) -> str | list[dict[str, Any]]:
     """Inverse of :func:`_content_from_raw`.
 
-    All-text content collapses to a plain string (the 95% case, and what an empty
-    turn round-trips to); any non-text block forces the multipart list form.
+    A single text block collapses to a plain string (the 95% case, and what an
+    empty turn round-trips to); multiple blocks — or any non-text block — take
+    the multipart list form.  The multiple-block rule preserves an all-*text*
+    multipart list (the unreadable-attachment placeholder path emits one) instead
+    of collapsing it to a joined string.
     """
     if not content:
         return ""
-    if all(isinstance(b, TextBlock) for b in content):
-        return "".join(b.text for b in content if isinstance(b, TextBlock))
+    if len(content) == 1 and isinstance(content[0], TextBlock):
+        return content[0].text
     parts: list[dict[str, Any]] = []
     for b in content:
         if isinstance(b, TextBlock):
diff --git a/turnstone/eval.py b/turnstone/eval.py
index 55e8c30f..ca340b8a 100644
--- a/turnstone/eval.py
+++ b/turnstone/eval.py
@@ -36,6 +36,7 @@ from turnstone.core.providers import LLMProvider, create_client, create_provider
 from turnstone.core.session import ChatSession
 from turnstone.core.storage import init_storage, reset_storage
 from turnstone.core.tools import INTERACTIVE_TOOLS, PRIMARY_KEY_MAP
+from turnstone.core.trajectory import Role, turn_from_dict
 
 # Eval evaluates interactive-session agent behaviour — coordinator tools
 # require a console-hosted session and aren't exercised by the harness.
@@ -280,7 +281,7 @@ class HeadlessSession(ChatSession):
             tool: str, args: dict, result: str (truncated), turn: int
         """
         self.tool_call_log = []
-        self.messages.append({"role": "user", "content": user_input})
+        self.messages.append(turn_from_dict({"role": "user", "content": user_input}))
         self._msg_tokens.append(max(1, int(len(user_input) / self._chars_per_token)))
 
         for turn in range(max_turns):
@@ -321,7 +322,7 @@ class HeadlessSession(ChatSession):
                 # Cap parallel tool calls to prevent degenerate repetition
                 assistant_msg["tool_calls"] = result.tool_calls[:10]
 
-            self.messages.append(assistant_msg)
+            self.messages.append(turn_from_dict(assistant_msg))
             msg_len = len(assistant_msg.get("content") or "")
             self._msg_tokens.append(max(1, int(msg_len / self._chars_per_token)))
 
@@ -401,7 +402,7 @@ class HeadlessSession(ChatSession):
                     "tool_call_id": tc_id,
                     "content": raw_output,
                 }
-                self.messages.append(tool_msg)
+                self.messages.append(turn_from_dict(tool_msg))
                 self._msg_tokens.append(max(1, int(len(output) / self._chars_per_token)))
 
         return self.tool_call_log
@@ -507,8 +508,8 @@ def _run_single_test(
 
                 # Extract results before releasing session
                 for msg in reversed(session.messages):
-                    if msg["role"] == "assistant" and msg.get("content"):
-                        final_content = msg["content"]
+                    if msg.role is Role.ASSISTANT and msg.text:
+                        final_content = msg.text
                         break
                 message_count = len(session.messages)
                 total_usage = session.total_usage
diff --git a/turnstone/server.py b/turnstone/server.py
index 0bc2a68c..fb31e8d1 100644
--- a/turnstone/server.py
+++ b/turnstone/server.py
@@ -88,6 +88,7 @@ from turnstone.core.session_ui_base import (
     fire_judge_verdict_metric,
 )
 from turnstone.core.tools import TOOLS  # noqa: F401 — available for introspection
+from turnstone.core.trajectory import turn_to_dict
 from turnstone.core.web_helpers import version_html as _version_html
 from turnstone.core.workstream import (
     Workstream,
@@ -1627,7 +1628,8 @@ def _validate_notify_targets(raw: Any) -> tuple[str, str]:
 
 def _extract_last_assistant_content(session: Any) -> str:
     """Return the text content of the last assistant message."""
-    for msg in reversed(session.messages):
+    for turn in reversed(session.messages):
+        msg = turn_to_dict(turn)
         if msg.get("role") == "assistant":
             content = msg.get("content", "")
             if isinstance(content, str):